""" Vismatix Enterprise Developer SDK for Python (API v3) Unified Social Media Matrix & Content Scheduling Engine Zero external dependencies required (uses Python standard library). """ import hmac import hashlib import json import time import uuid import urllib.parse import urllib.request import urllib.error from typing import Optional, Dict, Any, List class VismatixError(Exception): """Base exception for Vismatix API errors.""" def __init__(self, code: str, message: str, status_code: int = 400, request_id: Optional[str] = None): super().__init__(f"[{code}] {message} (HTTP {status_code})") self.code = code self.message = message self.status_code = status_code self.request_id = request_id VismatixError = VismatixError class VismatixClient: """Official high-performance Python client for Vismatix API v3.""" def __init__( self, api_key: str, base_url: str = "https://vismatix.online", api_version: str = "v3", sign_requests: bool = True, timeout: int = 15 ): self.api_key = api_key self.base_url = base_url.rstrip("/") self.api_version = api_version self.sign_requests = sign_requests self.timeout = timeout self.key_hash = hashlib.sha256(api_key.encode()).hexdigest() def _request( self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, idempotency_key: Optional[str] = None ) -> Dict[str, Any]: """Execute signed HTTP request against API v3 / v3 with automatic error handling.""" clean_endpoint = endpoint.lstrip("/") path = clean_endpoint if clean_endpoint.startswith("api/") else f"api/{self.api_version}/{clean_endpoint}" url = f"{self.base_url}/{path}" if params: query_string = urllib.parse.urlencode({key: value for key, value in params.items() if value is not None}) if query_string: url = f"{url}?{query_string}" body_bytes = json.dumps(data).encode("utf-8") if data is not None else b"" headers = { "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", "User-Agent": "Vismatix-Python-SDK/2.0", } if data is not None: headers["Content-Type"] = "application/json" if idempotency_key: headers["Idempotency-Key"] = idempotency_key # Cryptographic HMAC SHA-256 signature if self.sign_requests: timestamp = int(time.time()) content_hash = hashlib.sha256(body_bytes).hexdigest() # Sign the same path sent in the request, without its query string. payload = f"{timestamp}.{method.upper()}.{path}.{content_hash}" sig = hmac.new(self.key_hash.encode(), payload.encode(), hashlib.sha256).hexdigest() headers["X-Vismatix-Signature"] = f"t={timestamp},v3={sig}" req = urllib.request.Request(url, data=body_bytes if method.upper() in ["POST", "PUT", "PATCH"] else None, headers=headers, method=method.upper()) try: with urllib.request.urlopen(req, timeout=self.timeout) as response: content = response.read().decode("utf-8") return json.loads(content) if content else {} except urllib.error.HTTPError as e: raw_err = e.read().decode("utf-8") try: err_data = json.loads(raw_err) err_obj = err_data.get("error", {}) code = err_obj.get("code", "HTTP_ERROR") msg = err_obj.get("message", e.reason) req_id = err_data.get("request_id") except Exception: code = f"HTTP_{e.code}" msg = raw_err or e.reason req_id = None raise VismatixError(code, msg, e.code, req_id) except Exception as e: raise VismatixError("NETWORK_ERROR", str(e), 500) # --- System & Health --- def get_health(self) -> Dict[str, Any]: """Check system diagnostics, database latency, and cache operational status.""" return self._request("GET", "health") # --- Account & Profile (Requires Account Master Key: vmtx_u_...) --- def get_account(self) -> Dict[str, Any]: """Retrieve authenticated account profile, usage stats, and active keys.""" return self._request("GET", "account") def update_profile( self, name: Optional[str] = None, email: Optional[str] = None, avatar_url: Optional[str] = None ) -> Dict[str, Any]: """Update user profile information.""" data = {} if name is not None: data["name"] = name if email is not None: data["email"] = email if avatar_url is not None: data["avatar_url"] = avatar_url return self._request("PUT", "account/profile", data=data) def update_settings(self, preferences: Dict[str, Any]) -> Dict[str, Any]: """Update account preferences, notification settings, and dashboard defaults.""" return self._request("PUT", "account/settings", data={"preferences": preferences}) # --- Projects Management --- def list_projects(self) -> Dict[str, Any]: """List all accessible projects.""" return self._request("GET", "projects") def create_project( self, name: str, description: Optional[str] = None, color: Optional[str] = None, brand_voice: Optional[str] = None, target_audience: Optional[str] = None ) -> Dict[str, Any]: """Create a new project workspace (Requires Account Master Key: vmtx_u_...).""" data = { "name": name, "description": description, "color": color, "brand_voice": brand_voice, "target_audience": target_audience, } return self._request("POST", "projects", data=data) def get_project(self, project_id: int) -> Dict[str, Any]: """Retrieve project details and metrics.""" return self._request("GET", f"projects/{project_id}") def update_project(self, project_id: int, **kwargs) -> Dict[str, Any]: """Update project settings, brand voice, and goals.""" return self._request("PUT", f"projects/{project_id}", data=kwargs) def delete_project(self, project_id: int) -> Dict[str, Any]: """Permanently delete a project (Requires Account Master Key: vmtx_u_...).""" return self._request("DELETE", f"projects/{project_id}") # --- Posts Management --- def list_posts( self, status: Optional[str] = None, platform_id: Optional[int] = None, project_id: Optional[int] = None, search: Optional[str] = None, limit: int = 25 ) -> Dict[str, Any]: """List posts with virality scores and filtering.""" params = { "status": status, "platform_id": platform_id, "project_id": project_id, "search": search, "limit": limit } return self._request("GET", "posts", params=params) def get_post(self, post_id: int) -> Dict[str, Any]: """Retrieve details of a specific post.""" return self._request("GET", f"posts/{post_id}") def create_post( self, platform_id: int, title: str, description: str = "", status: str = "scheduled", time_block_id: Optional[int] = None, campaign: Optional[str] = None, tags: Optional[str] = None, is_evergreen: bool = False, idempotency_key: Optional[str] = None ) -> Dict[str, Any]: """Create a new post with automatic idempotency protection.""" data = { "platform_id": platform_id, "title": title, "description": description, "status": status, "time_block_id": time_block_id, "campaign": campaign, "tags": tags, "is_evergreen": is_evergreen, } key = idempotency_key or str(uuid.uuid4()) return self._request("POST", "posts", data=data, idempotency_key=key) def publish_post(self, post_id: int) -> Dict[str, Any]: """Instantly publish post and dispatch external webhooks.""" return self._request("POST", f"posts/{post_id}/publish-now") def smart_reschedule(self, post_id: int) -> Dict[str, Any]: """Automatically reschedule post to next peak audience engagement slot.""" return self._request("POST", f"posts/{post_id}/smart-reschedule") def delete_post(self, post_id: int) -> Dict[str, Any]: """Delete a post.""" return self._request("DELETE", f"posts/{post_id}") def duplicate_post( self, post_id: int, title: Optional[str] = None, platform_id: Optional[int] = None, time_block_id: Optional[int] = None, status: Optional[str] = None ) -> Dict[str, Any]: """Clone an existing post with optional platform or timing overrides.""" data = {k: v for k, v in { "title": title, "platform_id": platform_id, "time_block_id": time_block_id, "status": status, }.items() if v is not None} return self._request("POST", f"posts/{post_id}/duplicate", data=data) def virality_audit( self, post_id: Optional[int] = None, title: Optional[str] = None, description: Optional[str] = None ) -> Dict[str, Any]: """Perform algorithmic virality, readability, and hook audit without external AI.""" if post_id: return self._request("POST", f"posts/{post_id}/virality-audit") data = {"title": title or "", "description": description or ""} return self._request("POST", "posts/virality-audit", data=data) # --- Matrix Grid (7-Day x 12-Slot) --- def get_matrix(self, project_id: Optional[int] = None) -> Dict[str, Any]: """Retrieve full 84-cell calendar matrix representation.""" params = {"project_id": project_id} if project_id else None return self._request("GET", "matrix", params=params) def export_matrix(self, project_id: Optional[int] = None) -> Dict[str, Any]: """Export complete 84-cell weekly agenda and occupancy metrics as JSON.""" params = {"project_id": project_id} if project_id else None return self._request("GET", "matrix/export", params=params) def bulk_shift_matrix(self, days: int, project_id: Optional[int] = None) -> Dict[str, Any]: """Bulk shift all scheduled posts across the 7-day weekly canvas by +/- N days.""" data = {"days": days, "project_id": project_id} return self._request("POST", "matrix/bulk-shift", data=data) def toggle_peak_slot(self, slot_id: int) -> Dict[str, Any]: """Toggle golden peak engagement hour on a matrix time block.""" return self._request("PATCH", f"matrix/slots/{slot_id}/peak") def move_post_slot(self, post_id: int, target_day: int, target_slot: int) -> Dict[str, Any]: """Relocate a post across weekly matrix cells (Day 1-7, Slot 1-12).""" data = { "post_id": post_id, "target_day": target_day, "target_slot": target_slot } return self._request("POST", "matrix/move-post", data=data) # --- Tasks & Workflow Management --- def list_tasks( self, project_id: Optional[int] = None, status: Optional[str] = None, priority: Optional[str] = None, category: Optional[str] = None ) -> Dict[str, Any]: """List tasks and workflow checklists.""" params = { "project_id": project_id, "status": status, "priority": priority, "category": category } return self._request("GET", "tasks", params=params) def create_task( self, title: str, project_id: Optional[int] = None, priority: str = "medium", category: Optional[str] = None, due_date: Optional[str] = None, description: Optional[str] = None ) -> Dict[str, Any]: """Create a new action item or workflow task.""" data = { "title": title, "project_id": project_id, "priority": priority, "category": category, "due_date": due_date, "description": description, } return self._request("POST", "tasks", data=data) def get_task(self, task_id: int) -> Dict[str, Any]: """Get task details.""" return self._request("GET", f"tasks/{task_id}") def update_task(self, task_id: int, **fields) -> Dict[str, Any]: """Update an existing task.""" return self._request("PUT", f"tasks/{task_id}", data=fields) def toggle_task(self, task_id: int) -> Dict[str, Any]: """Toggle task completion state.""" return self._request("PATCH", f"tasks/{task_id}/toggle") def delete_task(self, task_id: int) -> Dict[str, Any]: """Delete a task.""" return self._request("DELETE", f"tasks/{task_id}") # --- Ideas Scratchpad --- def list_ideas(self, project_id: Optional[int] = None, category: Optional[str] = None) -> Dict[str, Any]: """List brainstormed ideas and scratchpad notes.""" params = {"project_id": project_id, "category": category} return self._request("GET", "ideas", params=params) def get_idea(self, idea_id: int) -> Dict[str, Any]: """Get idea details.""" return self._request("GET", f"ideas/{idea_id}") def update_idea(self, idea_id: int, **fields) -> Dict[str, Any]: """Update an existing idea.""" return self._request("PUT", f"ideas/{idea_id}", data=fields) def create_idea( self, title: str, project_id: Optional[int] = None, notes: Optional[str] = None, category: Optional[str] = None, color: Optional[str] = None ) -> Dict[str, Any]: """Save a new creative idea to the scratchpad.""" data = { "title": title, "project_id": project_id, "notes": notes, "category": category, "color": color, } return self._request("POST", "ideas", data=data) def convert_idea_to_draft( self, idea_id: int, platform_id: int, scheduled_at: Optional[str] = None ) -> Dict[str, Any]: """Convert a scratchpad idea directly into a scheduled draft post.""" data = { "platform_id": platform_id, "scheduled_at": scheduled_at } return self._request("PATCH", f"ideas/{idea_id}/convert-to-draft", data=data) def delete_idea(self, idea_id: int) -> Dict[str, Any]: """Delete an idea.""" return self._request("DELETE", f"ideas/{idea_id}") # --- Webhooks & Event Streaming --- def list_webhooks(self) -> Dict[str, Any]: """List registered webhooks subscriptions.""" return self._request("GET", "webhooks") def create_webhook( self, name: str, url: str, secret: Optional[str] = None, events: Optional[List[str]] = None, is_active: bool = True ) -> Dict[str, Any]: """Register a new webhook subscription with SSRF validation and HMAC signing.""" data = { "name": name, "url": url, "secret": secret, "events": events or ["post.created", "post.scheduled"], "is_active": is_active, } return self._request("POST", "webhooks", data=data) def get_webhook(self, webhook_id: int) -> Dict[str, Any]: """Get webhook details.""" return self._request("GET", f"webhooks/{webhook_id}") def update_webhook( self, webhook_id: int, name: Optional[str] = None, url: Optional[str] = None, secret: Optional[str] = None, events: Optional[List[str]] = None, is_active: Optional[bool] = None ) -> Dict[str, Any]: """Update webhook endpoint URL, secret, or subscribed events.""" data = {k: v for k, v in { "name": name, "url": url, "secret": secret, "events": events, "is_active": is_active, }.items() if v is not None} return self._request("PUT", f"webhooks/{webhook_id}", data=data) def test_webhook(self, webhook_id: int) -> Dict[str, Any]: """Dispatch real-time diagnostic test ping to verify endpoint connectivity.""" return self._request("POST", f"webhooks/{webhook_id}/test") def delete_webhook(self, webhook_id: int) -> Dict[str, Any]: """Delete a webhook subscription.""" return self._request("DELETE", f"webhooks/{webhook_id}") # --- Supported Platforms --- def get_platforms(self) -> Dict[str, Any]: """List supported social platforms with character limits and capabilities.""" return self._request("GET", "platforms") # --- Hashtag & Category Presets --- def get_presets(self, category: Optional[str] = None, search: Optional[str] = None) -> Dict[str, Any]: """List hashtag presets with optional filtering.""" params = {} if category: params["category"] = category if search: params["search"] = search return self._request("GET", "presets", params=params or None) def create_preset( self, name: str, tags: str, color: Optional[str] = None, category: Optional[str] = None ) -> Dict[str, Any]: """Create a new hashtag preset.""" data = { "name": name, "tags": tags, "color": color or "#06b6d4", "category": category or "General", } return self._request("POST", "presets", data=data) def delete_preset(self, preset_id: int) -> Dict[str, Any]: """Delete a hashtag preset.""" return self._request("DELETE", f"presets/{preset_id}") # --- Project Duplication --- def duplicate_project( self, project_id: int, name: Optional[str] = None, include_tasks: bool = False ) -> Dict[str, Any]: """Duplicate an existing workspace and its agenda layout (vmtx_u_ only).""" data = {"name": name, "include_tasks": include_tasks} return self._request("POST", f"projects/{project_id}/duplicate", data=data) # --- Matrix Slots Querying --- def get_matrix_slots( self, project_id: Optional[int] = None, day_of_week: Optional[int] = None, time_slot: Optional[int] = None, is_peak: Optional[bool] = None, is_occupied: Optional[bool] = None ) -> Dict[str, Any]: """Query individual matrix slots with filtering.""" params = {} if project_id is not None: params["project_id"] = project_id if day_of_week is not None: params["day_of_week"] = day_of_week if time_slot is not None: params["time_slot"] = time_slot if is_peak is not None: params["is_peak"] = str(is_peak).lower() if is_occupied is not None: params["is_occupied"] = str(is_occupied).lower() return self._request("GET", "matrix/slots", params=params or None) # --- Bulk Posts Operations --- def bulk_update_posts_status(self, post_ids: List[int], status: str) -> Dict[str, Any]: """Batch update statuses of multiple posts.""" return self._request("POST", "posts/bulk-status", data={"ids": post_ids, "status": status}) def bulk_delete_posts(self, post_ids: List[int]) -> Dict[str, Any]: """Batch delete multiple posts.""" return self._request("POST", "posts/bulk-delete", data={"ids": post_ids}) def export_posts_csv(self, status: Optional[str] = None, project_id: Optional[int] = None) -> bytes: """Export posts spreadsheet as CSV.""" params = {} if status: params["status"] = status if project_id: params["project_id"] = project_id url = f"{self.base_url}/posts/export-csv" headers = self._get_headers("GET", "posts/export-csv") req = urllib.request.Request(url, headers=headers, method="GET") with urllib.request.urlopen(req) as resp: return resp.read() # --- Audit Trail & Observability --- def get_audit_logs(self, limit: int = 20) -> Dict[str, Any]: """Retrieve unified audit trail of recent API and workspace activities (vmtx_u_ only).""" return self._request("GET", "audit-logs", params={"limit": limit}) # --- Telemetry & Analytics --- def get_analytics(self) -> Dict[str, Any]: """Retrieve cadence telemetry, platform distribution, and peak hours coverage.""" return self._request("GET", "analytics/overview") VismatixClient = VismatixClient if __name__ == "__main__": import sys print("Vismatix Python SDK v3 loaded successfully.")