Add similarweb-analytics Docker sandbox skill

similarweb-analytics/scripts/runtime/data_api.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""Minimal ApiClient runtime for SimilarWeb skill.
+
+Implements the subset of Manus-style interface used by the skill:
+    ApiClient().call_api(api_name, path_params={"domain": ...}, query={...})
+
+Primary mode: Similarweb official API (requires SIMILARWEB_API_KEY)
+Fallback mode: RapidAPI similarweb13 domain snapshot (requires RAPIDAPI_KEY)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any, Dict, Mapping, Optional
+
+
+class ApiError(RuntimeError):
+    pass
+
+
+@dataclass
+class EndpointSpec:
+    path: str
+
+
+OFFICIAL_ENDPOINTS: Dict[str, EndpointSpec] = {
+    "SimilarWeb/get_global_rank": EndpointSpec("/v1/website/{domain}/global-rank/global-rank"),
+    "SimilarWeb/get_visits_total": EndpointSpec("/v1/website/{domain}/total-traffic-and-engagement/visits"),
+    "SimilarWeb/get_unique_visit": EndpointSpec("/v1/website/{domain}/deduplicated-audience/deduplicated-audience"),
+    "SimilarWeb/get_bounce_rate": EndpointSpec("/v1/website/{domain}/total-traffic-and-engagement/bounce-rate"),
+    "SimilarWeb/get_traffic_sources_desktop": EndpointSpec("/v1/website/{domain}/traffic-sources/desktop"),
+    "SimilarWeb/get_traffic_sources_mobile": EndpointSpec("/v1/website/{domain}/traffic-sources/mobile-web"),
+    "SimilarWeb/get_total_traffic_by_country": EndpointSpec("/v1/website/{domain}/geography/total-traffic-and-engagement"),
+}
+
+
+class ApiClient:
+    def __init__(
+        self,
+        *,
+        similarweb_api_key: Optional[str] = None,
+        similarweb_base_url: Optional[str] = None,
+        rapidapi_key: Optional[str] = None,
+        rapidapi_host: Optional[str] = None,
+        timeout: int = 30,
+    ) -> None:
+        self.similarweb_api_key = similarweb_api_key or os.getenv("SIMILARWEB_API_KEY")
+        self.similarweb_base_url = (
+            similarweb_base_url
+            or os.getenv("SIMILARWEB_BASE_URL")
+            or "https://api.similarweb.com"
+        ).rstrip("/")
+        self.rapidapi_key = rapidapi_key or os.getenv("RAPIDAPI_KEY")
+        self.rapidapi_host = rapidapi_host or os.getenv("RAPIDAPI_SIMILARWEB_HOST") or "similarweb13.p.rapidapi.com"
+        self.timeout = timeout
+
+    def call_api(
+        self,
+        api_name: str,
+        *,
+        path_params: Optional[Mapping[str, Any]] = None,
+        query: Optional[Mapping[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        path_params = dict(path_params or {})
+        query = dict(query or {})
+
+        domain = str(path_params.get("domain", "")).strip()
+        if not domain:
+            raise ApiError("path_params.domain is required")
+
+        if self.similarweb_api_key:
+            return self._call_official(api_name, domain=domain, query=query)
+
+        if self.rapidapi_key:
+            return self._call_rapidapi_snapshot(api_name, domain=domain, query=query)
+
+        raise ApiError(
+            "No credentials configured. Set SIMILARWEB_API_KEY (preferred) or RAPIDAPI_KEY."
+        )
+
+    def _call_official(self, api_name: str, *, domain: str, query: Dict[str, Any]) -> Dict[str, Any]:
+        spec = OFFICIAL_ENDPOINTS.get(api_name)
+        if not spec:
+            raise ApiError(f"Unsupported api_name for official mode: {api_name}")
+
+        path = spec.path.format(domain=domain)
+        q = self._clean_query(query)
+        q["api_key"] = self.similarweb_api_key
+        url = f"{self.similarweb_base_url}{path}?{urllib.parse.urlencode(q)}"
+
+        req = urllib.request.Request(url=url, method="GET")
+        return self._do_request(req, mode="official", api_name=api_name, url=url)
+
+    def _call_rapidapi_snapshot(self, api_name: str, *, domain: str, query: Dict[str, Any]) -> Dict[str, Any]:
+        encoded_domain = urllib.parse.quote(domain)
+        url = f"https://{self.rapidapi_host}/v2/getdomain?domain={encoded_domain}"
+        headers = {
+            "x-rapidapi-key": self.rapidapi_key or "",
+            "x-rapidapi-host": self.rapidapi_host,
+        }
+        req = urllib.request.Request(url=url, method="GET", headers=headers)
+
+        resp = self._do_request(req, mode="rapidapi", api_name=api_name, url=url)
+        return {
+            "_adapter": {
+                "mode": "rapidapi",
+                "note": "Using /v2/getdomain snapshot fallback; not 1:1 with official endpoint schema.",
+                "requested_api": api_name,
+                "requested_query": query,
+            },
+            "data": resp,
+        }
+
+    @staticmethod
+    def _clean_query(query: Mapping[str, Any]) -> Dict[str, Any]:
+        out: Dict[str, Any] = {}
+        for k, v in query.items():
+            if v is None:
+                continue
+            if isinstance(v, bool):
+                out[k] = "true" if v else "false"
+            else:
+                out[k] = str(v)
+        return out
+
+    def _do_request(self, req: urllib.request.Request, *, mode: str, api_name: str, url: str) -> Dict[str, Any]:
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                body = resp.read().decode("utf-8", errors="replace")
+                data = json.loads(body) if body else {}
+                return {
+                    "_meta": {
+                        "mode": mode,
+                        "api_name": api_name,
+                        "http_status": resp.status,
+                        "url": url,
+                    },
+                    "response": data,
+                }
+        except urllib.error.HTTPError as exc:
+            body = exc.read().decode("utf-8", errors="replace")
+            try:
+                parsed = json.loads(body)
+            except Exception:
+                parsed = {"raw": body}
+            raise ApiError(
+                json.dumps(
+                    {
+                        "http_status": exc.code,
+                        "mode": mode,
+                        "api_name": api_name,
+                        "url": url,
+                        "error": parsed,
+                    },
+                    ensure_ascii=False,
+                )
+            )
+        except urllib.error.URLError as exc:
+            raise ApiError(f"Network error for {url}: {exc}")
+
+
+__all__ = ["ApiClient", "ApiError"]