serpmax_client.py

"""
serpmax_client.py — Python SDK for the Serpmax.ru API.

Docs: https://serpmax.ru/api-documentation
Auth: Bearer token via Authorization header
Rate limit: 60 requests / 60 seconds
"""

import requests


BASE_URL = "https://serpmax.ru"
TIMEOUT = 15


class SerpmaxAPIError(Exception):
    """Raised when the API returns an error response."""

    def __init__(self, message: str, status_code: int | None = None):
        super().__init__(message)
        self.status_code = status_code


class SerpmaxAPI:
    """Lightweight client for the Serpmax.ru REST API.

    Usage::

        client = SerpmaxAPI(api_key="your_api_key_here")
        user = client.get_user()
    """

    def __init__(self, api_key: str, base_url: str = BASE_URL):
        if not api_key or not api_key.strip():
            raise ValueError("api_key must not be empty.")
        try:
            api_key.encode("latin-1")
        except UnicodeEncodeError as exc:
            raise ValueError(
                "api_key contains invalid characters. "
                "Expected an ASCII token from your account settings."
            ) from exc
        self._session = requests.Session()
        self._session.headers.update(
            {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            }
        )
        self.base_url = base_url.rstrip("/")

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{self.base_url}{path}"
        try:
            response = self._session.request(
                method, url, timeout=TIMEOUT, **kwargs
            )
        except requests.exceptions.ConnectionError as exc:
            raise SerpmaxAPIError(f"Cannot connect to {self.base_url}: {exc}") from exc
        except requests.exceptions.Timeout as exc:
            raise SerpmaxAPIError(f"Request timed out after {TIMEOUT}s.") from exc

        if response.status_code == 401:
            raise SerpmaxAPIError("Unauthorized — check your API key.", 401)
        if response.status_code == 403:
            raise SerpmaxAPIError("Forbidden — your plan may not include API access.", 403)
        if response.status_code == 429:
            raise SerpmaxAPIError("Rate limit exceeded. Retry after 60 seconds.", 429)
        if not response.ok:
            try:
                detail = response.json()
            except ValueError:
                detail = response.text
            raise SerpmaxAPIError(
                f"API error {response.status_code}: {detail}", response.status_code
            )

        return response.json()

    def _get(self, path: str, params: dict | None = None) -> dict:
        return self._request("GET", path, params=params)

    def _post(self, path: str, data: dict | None = None) -> dict:
        return self._request("POST", path, data=data)

    def _delete(self, path: str) -> None:
        self._request("DELETE", path)

    # ------------------------------------------------------------------
    # User
    # ------------------------------------------------------------------

    def get_user(self) -> dict:
        """Return the authenticated user's profile and plan details."""
        return self._get("/api/user")

    # ------------------------------------------------------------------
    # Websites
    # ------------------------------------------------------------------

    def list_websites(self, page: int = 1, **filters) -> dict:
        """List all websites tracked by the authenticated user.

        Args:
            page: Page number (default 1).
            **filters: Optional filter params (e.g. host="example.com").
        """
        return self._get("/api/websites", params={"page": page, **filters})

    def get_website(self, website_id: int) -> dict:
        """Fetch a single website by its ID."""
        return self._get(f"/api/websites/{website_id}")

    def update_website(self, website_id: int, **fields) -> dict:
        """Update settings for a website (POST with ID acts as PATCH).

        Common fields: domain_id, audit_check_interval, is_public,
                       password, notifications_mode, sitemap_url.
        """
        return self._post(f"/api/websites/{website_id}", data=fields)

    def delete_website(self, website_id: int) -> None:
        """Permanently delete a website and all its audits."""
        self._delete(f"/api/websites/{website_id}")

    # ------------------------------------------------------------------
    # Audits
    # ------------------------------------------------------------------

    def list_audits(self, page: int = 1, **filters) -> dict:
        """List all audits for the authenticated user.

        Args:
            page: Page number (default 1).
            **filters: Optional filter params (e.g. website_id=42, host="example.com").
        """
        return self._get("/api/audits", params={"page": page, **filters})

    def get_audit(self, audit_id: int) -> dict:
        """Fetch a single audit by its ID.

        The audit object contains the full technical report including:
        score, ttfb, response_time, is_https, is_ssl_valid, http_protocol,
        page_size, http_requests, issues (major/moderate/minor), and more.
        """
        return self._get(f"/api/audits/{audit_id}")

    def create_audit(self, url: str, audit_type: str = "single", **options) -> dict:
        """Create a new audit (or re-audit an existing URL).

        Args:
            url: The URL to audit (required for types 'single' and 'html').
            audit_type: One of 'single', 'bulk', 'sitemap', 'html'.
            **options: Optional fields — audit_check_interval, is_public,
                       password, notifications_mode, domain_id, html (for type='html'),
                       urls (for type='bulk'/'sitemap').

        Returns:
            A single audit dict (or list of dicts for bulk/sitemap).
        """
        return self._post("/api/audits", data={"url": url, "type": audit_type, **options})

    def update_audit(self, audit_id: int, **fields) -> dict:
        """Update settings for an existing audit.

        Common fields: domain_id, audit_check_interval, is_public,
                       password, notifications_mode.
        """
        return self._post(f"/api/audits/{audit_id}", data=fields)

    def delete_audit(self, audit_id: int) -> None:
        """Permanently delete an audit."""
        self._delete(f"/api/audits/{audit_id}")

    # ------------------------------------------------------------------
    # Domains (custom white-label domains)
    # ------------------------------------------------------------------

    def list_domains(self, page: int = 1) -> dict:
        """List all custom domains registered by the user."""
        return self._get("/api/domains", params={"page": page})

    def get_domain(self, domain_id: int) -> dict:
        """Fetch a single custom domain by its ID."""
        return self._get(f"/api/domains/{domain_id}")

    def create_domain(self, host: str, scheme: str = "https://", **options) -> dict:
        """Register a new custom white-label domain.

        Args:
            host: Domain hostname, e.g. "audit.mycompany.com".
            scheme: "https://" (default) or "http://".
            **options: custom_index_url, custom_not_found_url.
        """
        return self._post("/api/domains", data={"host": host, "scheme": scheme, **options})

    def delete_domain(self, domain_id: int) -> None:
        """Delete a custom domain."""
        self._delete(f"/api/domains/{domain_id}")