"""Sklik (Seznam) "Drak" JSON API client. Protocol (verified against the official seznam/api-examples JSON example): * Endpoint: ``{SKLIK_BASE_URL}/{method}`` e.g. .../drak/json/v5/campaigns.list * HTTP POST, body = a JSON ARRAY of positional arguments. * ``client.loginByToken`` takes the API token as its single argument and returns ``{"status":200,"session":"...",...}``. * Every authenticated method takes the user struct ``{"session": ...}`` (optionally ``"userId"``) as its FIRST argument, followed by the method's own arguments. * Every response is an object containing ``status`` (HTTP-style int), ``statusMessage``, a refreshed ``session``, plus method-specific data. The proxy is stateless: it logs in with ``X-Sklik-Token`` per request to obtain a session, then performs the requested call. The token and session are never logged. """ from __future__ import annotations from typing import Any import httpx from .. import config from ..errors import UpstreamError from ..logging_config import get_logger logger = get_logger(__name__) # Sklik report data is paginated; readReport is called with an offset/limit # window until all rows are fetched. Keep the page size conservative. _REPORT_PAGE_LIMIT = 100 # Hard stop so a misbehaving upstream can't loop forever. _REPORT_MAX_PAGES = 1000 class SklikClient: """Performs JSON-RPC calls against the Sklik Drak API.""" def __init__(self, token: str, user_id: int | None = None) -> None: self._token = token self._user_id = user_id self._session: str | None = None async def __aenter__(self) -> "SklikClient": self._http = httpx.AsyncClient(timeout=config.HTTP_TIMEOUT_SECONDS) return self async def __aexit__(self, *exc: Any) -> None: await self._http.aclose() async def _call(self, method: str, args: list[Any]) -> dict: """Low-level: POST a JSON array of args to ``/{method}``.""" url = f"{config.SKLIK_BASE_URL}/{method}" try: resp = await self._http.post(url, json=args) except httpx.TimeoutException as exc: raise UpstreamError( f"Sklik request timed out ({method}).", status=504 ) from exc except httpx.HTTPError as exc: raise UpstreamError( f"Sklik is unreachable ({method}): {exc}", status=502 ) from exc try: payload = resp.json() except ValueError as exc: raise UpstreamError( f"Sklik returned a non-JSON response ({method}).", status=502, upstream_status=resp.status_code, body={"raw": resp.text}, ) from exc if not isinstance(payload, dict): raise UpstreamError( f"Unexpected Sklik response shape ({method}).", status=502, body=payload, ) status = payload.get("status") # Sklik conveys business errors in the body with an HTTP-style status. # 200 OK, 206 partially OK, 301 "user is serviced" are all acceptable. if status not in (200, 206, 301): raise UpstreamError( payload.get("statusMessage", f"Sklik error on {method}."), status=400 if isinstance(status, int) and 400 <= status < 500 else 502, upstream_status=status if isinstance(status, int) else None, body=payload, ) # Refresh our session from every response (Sklik rotates it). new_session = payload.get("session") if isinstance(new_session, str) and new_session: self._session = new_session return payload async def login(self) -> dict: """Exchange the API token for a session. Idempotent per client.""" payload = await self._call("client.loginByToken", [self._token]) if not self._session: raise UpstreamError( "Sklik login succeeded but returned no session.", status=502, body=payload, ) return payload def _user_struct(self) -> dict: user: dict[str, Any] = {"session": self._session} if self._user_id is not None: user["userId"] = self._user_id return user async def call(self, method: str, args: list[Any] | None = None) -> dict: """Authenticated call: prepends the user/session struct to ``args``. Logs in first if no session is held yet. ``method`` is e.g. ``campaigns.list``; ``args`` are the method arguments AFTER the user struct. """ if method == "client.loginByToken": # Login is handled by login(); never forward the bare token here. raise UpstreamError( "client.loginByToken cannot be called directly; the proxy " "manages the session.", status=400, ) if not self._session: await self.login() full_args = [self._user_struct()] + list(args or []) return await self._call(method, full_args) async def fetch_report( self, entity: str, report_args: list[Any] ) -> dict: """Create a stats report for ``entity`` then read all of its rows. ``entity`` is e.g. ``campaigns``/``groups``/``ads``/``keywords``. Calls ``{entity}.createReport`` with ``report_args`` (the restriction + display-options structs), then pages through ``{entity}.readReport`` until every row is collected. """ created = await self.call(f"{entity}.createReport", report_args) report_id = created.get("reportId") if not report_id: raise UpstreamError( f"{entity}.createReport returned no reportId.", status=502, body=created, ) total = created.get("totalCount", 0) rows: list[Any] = [] offset = 0 pages = 0 while True: page = await self.call( f"{entity}.readReport", [ report_id, { "offset": offset, "limit": _REPORT_PAGE_LIMIT, "allowEmptyStatistics": False, }, ], ) batch = page.get("report") or [] rows.extend(batch) pages += 1 offset += _REPORT_PAGE_LIMIT if len(batch) < _REPORT_PAGE_LIMIT: break if pages >= _REPORT_MAX_PAGES: logger.warning( "Sklik %s.readReport hit the %d-page safety cap (collected " "%d rows); result may be truncated.", entity, _REPORT_MAX_PAGES, len(rows), ) break return { "reportId": report_id, "totalCount": total, "returnedCount": len(rows), "truncated": pages >= _REPORT_MAX_PAGES, "report": rows, }