""" Primenza Python SDK v1 Usage: pip install requests """ import requests class PrimenzaClient: def __init__(self, api_key: str, base_url: str = "https://primenza.com/api/v1"): self.api_key = api_key self.base_url = base_url.rstrip("/") self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "X-API-Key": api_key, "Content-Type": "application/json", "Accept": "application/json", }) def _request(self, method: str, path: str, json=None): r = self.session.request(method, f"{self.base_url}{path}", json=json, timeout=30) r.raise_for_status() body = r.json() if body.get("errors"): raise RuntimeError(body["errors"][0].get("message", "API request failed")) return body.get("data", body) def dashboard(self, domain_id=None): q = f"?domain_id={domain_id}" if domain_id else "" return self._request("GET", f"/dashboard{q}") def domains(self): return self._request("GET", "/domains") def domain(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}") def create_domain(self, url: str, name: str = None): return self._request("POST", "/domains", {"url": url, "name": name}) def delete_domain(self, domain_id: int): return self._request("DELETE", f"/domains/{domain_id}") def verify_domain(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/verify") def detect_tech(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/detect-tech") def pages(self, domain_id: int, page: int = 1): return self._request("GET", f"/domains/{domain_id}/pages?page={page}") def crawl(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/crawl") def audit(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/audit") def audits(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/audits") def seo(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/seo") def run_seo(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/seo/run") def geo(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/geo") def run_geo(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/geo/run") def ai_visibility(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/ai-visibility") def run_ai_visibility(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/ai-visibility/run") def performance(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/performance") def run_performance(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/performance/run") def links(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/links") def run_links(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/links/run") def content_gaps(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/content-gaps") def discover_content_gaps(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/content-gaps/discover") def schema(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/schema") def robots(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/robots") def deploy_robots(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/robots/deploy") def deploy_llms(self, domain_id: int, type_: str = "llms"): return self._request("POST", f"/domains/{domain_id}/llms-txt/deploy", {"type": type_}) def competitors(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/competitors") def add_competitor(self, domain_id: int, url: str): return self._request("POST", f"/domains/{domain_id}/competitors", {"url": url}) def refresh_competitor(self, domain_id: int, competitor_id: int): return self._request("POST", f"/domains/{domain_id}/competitors/{competitor_id}/refresh") def remove_competitor(self, domain_id: int, competitor_id: int): return self._request("DELETE", f"/domains/{domain_id}/competitors/{competitor_id}") def update_domain(self, domain_id: int, **fields): return self._request("PATCH", f"/domains/{domain_id}", fields) def tenant(self): return self._request("GET", "/tenant") def schedules(self): return self._request("GET", "/schedules") def knowledge_graph(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/knowledge-graph") def build_knowledge_graph(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/knowledge-graph/build") def citation(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/citation") def deploy_preview(self, domain_id: int, artifact: str = "robots"): return self._request("GET", f"/domains/{domain_id}/deploy/preview?artifact={artifact}") def monitoring(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/monitoring") def reports(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/reports") def agents(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/agents") def register_agent(self, domain_id: int, name: str, type_: str): return self._request("POST", f"/domains/{domain_id}/agents", {"name": name, "type": type_}) def test_webhook(self): return self._request("POST", "/webhooks/test") def graphql(self, query: str, variables: dict = None): return self._request("POST", "/graphql", {"query": query, "variables": variables or {}}) def create_schedule(self, data: dict): return self._request("POST", "/schedules", data) def update_schedule(self, schedule_id: int, data: dict): return self._request("PATCH", f"/schedules/{schedule_id}", data) def delete_schedule(self, schedule_id: int): return self._request("DELETE", f"/schedules/{schedule_id}") def workspaces(self): return self._request("GET", "/workspaces") def create_workspace(self, data: dict): return self._request("POST", "/workspaces", data) def update_workspace(self, workspace_id: int, data: dict): return self._request("PATCH", f"/workspaces/{workspace_id}", data) def delete_workspace(self, workspace_id: int): return self._request("DELETE", f"/workspaces/{workspace_id}") def team(self): return self._request("GET", "/team") def invite_team_member(self, data: dict): return self._request("POST", "/team/invitations", data) def update_team_role(self, user_id: int, role: str): return self._request("PATCH", f"/team/{user_id}", {"role": role}) def remove_team_member(self, user_id: int): return self._request("DELETE", f"/team/{user_id}") def cancel_invitation(self, invitation_id: int): return self._request("DELETE", f"/team/invitations/{invitation_id}") def settings(self): return self._request("GET", "/settings") def update_settings(self, data: dict): return self._request("PATCH", "/settings", data) def create_api_key(self, data: dict): return self._request("POST", "/api-keys", data) def delete_api_key(self, key_id: int): return self._request("DELETE", f"/api-keys/{key_id}") def audit_logs(self, query: str = ""): return self._request("GET", f"/audit-logs{f'?{query}' if query else ''}") def notifications(self): return self._request("GET", "/notifications") def mark_notification_read(self, alert_id: int): return self._request("POST", f"/notifications/{alert_id}/read") def mark_all_notifications_read(self): return self._request("POST", "/notifications/read-all") def regenerate_verification(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/verification/regenerate") def update_verification_method(self, domain_id: int, method: str): return self._request("PATCH", f"/domains/{domain_id}/verification-method", {"method": method}) def activate_domain(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/activate") def generate_content_gap_outline(self, domain_id: int, gap_id: int): return self._request("POST", f"/domains/{domain_id}/content-gaps/{gap_id}/outline") def generate_content_gap_article(self, domain_id: int, gap_id: int): return self._request("POST", f"/domains/{domain_id}/content-gaps/{gap_id}/article") def dismiss_content_gap(self, domain_id: int, gap_id: int): return self._request("POST", f"/domains/{domain_id}/content-gaps/{gap_id}/dismiss") def content_optimizer(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/content-optimizer") def refresh_content_optimizer(self, domain_id: int): return self._request("POST", f"/domains/{domain_id}/content-optimizer/refresh") def simulate_ai_visibility(self, domain_id: int, data: dict = None): return self._request("POST", f"/domains/{domain_id}/ai-visibility/simulate", data or {}) def install_agent(self, domain_id: int, data: dict): return self._request("POST", f"/domains/{domain_id}/agents/install", data) def generate_schema(self, domain_id: int, data: dict = None): return self._request("POST", f"/domains/{domain_id}/schema/generate", data or {}) def deploy_schema(self, domain_id: int, data: dict = None): return self._request("POST", f"/domains/{domain_id}/schema/deploy", data or {}) def update_robots(self, domain_id: int, content: str): return self._request("PUT", f"/domains/{domain_id}/robots", {"content": content}) def generate_llms(self, domain_id: int, type_: str = "llms"): return self._request("POST", f"/domains/{domain_id}/llms-txt/generate", {"type": type_}) def store_subdomain(self, domain_id: int, data: dict): return self._request("POST", f"/domains/{domain_id}/subdomains", data) def destroy_subdomain(self, domain_id: int, subdomain_id: int): return self._request("DELETE", f"/domains/{domain_id}/subdomains/{subdomain_id}") def download_report(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/reports/download") def download_report_pdf(self, domain_id: int): return self._request("GET", f"/domains/{domain_id}/reports/download/pdf") def suggest_citation(self, domain_id: int, page_id: int, data: dict = None): return self._request("POST", f"/domains/{domain_id}/citation/{page_id}/suggest", data or {}) def approve_agent_command(self, domain_id: int, command_id: int): return self._request("POST", f"/domains/{domain_id}/agents/commands/{command_id}/approve") class PrimenzaAgentClient: def __init__(self, agent_token: str, base_url: str = "https://primenza.com/api/v1"): self.base_url = base_url.rstrip("/") self.session = requests.Session() self.session.headers.update({ "X-Agent-Token": agent_token, "Content-Type": "application/json", "Accept": "application/json", }) def poll(self): r = self.session.post(f"{self.base_url}/agent/poll", timeout=30) r.raise_for_status() body = r.json() return body.get("data", body) def report(self, command_id: int, result: dict): r = self.session.post(f"{self.base_url}/agent/report", json={"command_id": command_id, "result": result}, timeout=30) r.raise_for_status() body = r.json() return body.get("data", body)