anycrawler.py

from __future__ import annotations

import json
from typing import Any, Dict, Optional
from urllib import error as urllib_error
from urllib import request as urllib_request


def _normalize_base_url(base_url: str) -> str:
    return base_url.rstrip("/")


def _parse_optional_number(value: Optional[str]) -> Optional[int]:
    if value in (None, ""):
        return None

    try:
        return int(value)
    except ValueError:
        try:
            return int(float(value))
        except ValueError:
            return None


def _build_meta(status: int, headers: Any) -> Dict[str, Any]:
    return {
        "status": status,
        "requestId": headers.get("x-request-id"),
        "creditsReserved": _parse_optional_number(headers.get("x-credits-reserved")),
        "creditsUsed": _parse_optional_number(headers.get("x-credits-used")),
        "browserMsUsed": _parse_optional_number(headers.get("x-browser-ms-used")),
    }


def _parse_json(text: str) -> Any:
    if not text:
        return None

    return json.loads(text)


class AnyCrawlerError(Exception):
    def __init__(
        self,
        message: str,
        *,
        status: int,
        error_code: Optional[str],
        data: Any,
        meta: Dict[str, Any],
    ) -> None:
        super().__init__(message)
        self.status = status
        self.error_code = error_code
        self.error_message = message
        self.data = data
        self.meta = meta


class AnyCrawlerClient:
    def __init__(
        self,
        api_key: str,
        *,
        base_url: str = "https://api.anycrawler.com",
        timeout: float = 60.0,
        headers: Optional[Dict[str, str]] = None,
    ) -> None:
        if not api_key:
            raise ValueError("AnyCrawlerClient requires an api_key.")

        self.api_key = api_key
        self.base_url = _normalize_base_url(base_url)
        self.timeout = timeout
        self.default_headers = headers or {}

    def get_account_usage(self) -> Dict[str, Any]:
        return self._request("GET", "/v1/account/usage")

    def crawl_page(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self._request("POST", "/v1/crawl/page", payload)

    def crawl_screenshot(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self._request("POST", "/v1/crawl/screenshot", payload)

    def search(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self._request("POST", "/v1/search", payload)

    def search_page(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self._request("POST", "/v1/search/page", payload)

    def search_news(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self._request("POST", "/v1/search/news", payload)

    def _request(
        self,
        method: str,
        pathname: str,
        payload: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            **self.default_headers,
        }

        data: Optional[bytes] = None

        if payload is not None:
            headers["Content-Type"] = "application/json"
            data = json.dumps(payload).encode("utf-8")

        request = urllib_request.Request(
            f"{self.base_url}{pathname}",
            data=data,
            headers=headers,
            method=method,
        )

        try:
            with urllib_request.urlopen(request, timeout=self.timeout) as response:
                status = response.getcode()
                raw_text = response.read().decode("utf-8")
                parsed = _parse_json(raw_text)
                meta = _build_meta(status, response.headers)
                return {
                    "data": parsed,
                    "meta": meta,
                }
        except urllib_error.HTTPError as exc:
            raw_text = exc.read().decode("utf-8", errors="replace")
            parsed = _parse_json(raw_text)
            meta = _build_meta(exc.code, exc.headers)
            body = parsed if isinstance(parsed, dict) else {}
            error_code = body.get("error_code") if isinstance(body.get("error_code"), str) else None
            error_message = (
                body.get("error_message")
                if isinstance(body.get("error_message"), str)
                else body.get("message")
                if isinstance(body.get("message"), str)
                else f"AnyCrawler request failed with status {exc.code}."
            )
            raise AnyCrawlerError(
                error_message,
                status=exc.code,
                error_code=error_code,
                data=parsed,
                meta=meta,
            ) from exc
        except urllib_error.URLError as exc:
            meta = {
                "status": 0,
                "requestId": None,
                "creditsReserved": None,
                "creditsUsed": None,
                "browserMsUsed": None,
            }
            raise AnyCrawlerError(
                f"AnyCrawler request failed before receiving an HTTP response: {exc.reason}",
                status=0,
                error_code=None,
                data=None,
                meta=meta,
            ) from exc