JiriUhlir
246 lines
8.8 KiB
Python
246 lines
8.8 KiB
Python
"""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_list(
|
|
self,
|
|
entity: str,
|
|
restriction_filter: dict[str, Any],
|
|
display_columns: list[str] | None = None,
|
|
) -> dict:
|
|
"""Page through ``{entity}.list`` and collect every row.
|
|
|
|
``{entity}.list`` takes ``(restrictionFilter, displayOptions)`` and
|
|
returns the rows under a key named after the entity (``campaigns``,
|
|
``groups``, ``ads``). Paging is offset/limit, same as readReport.
|
|
"""
|
|
rows: list[Any] = []
|
|
offset = 0
|
|
pages = 0
|
|
total: int | None = None
|
|
|
|
while True:
|
|
display: dict[str, Any] = {
|
|
"offset": offset,
|
|
"limit": config.SKLIK_LIST_PAGE_LIMIT,
|
|
}
|
|
if display_columns:
|
|
display["displayColumns"] = display_columns
|
|
|
|
page = await self.call(f"{entity}.list", [restriction_filter, display])
|
|
batch = page.get(entity) or []
|
|
rows.extend(batch)
|
|
if total is None and isinstance(page.get("totalCount"), int):
|
|
total = page["totalCount"]
|
|
|
|
pages += 1
|
|
offset += config.SKLIK_LIST_PAGE_LIMIT
|
|
if len(batch) < config.SKLIK_LIST_PAGE_LIMIT:
|
|
break
|
|
if pages >= config.SKLIK_LIST_MAX_PAGES:
|
|
logger.warning(
|
|
"Sklik %s.list hit the %d-page safety cap (collected %d "
|
|
"rows); result is truncated.",
|
|
entity,
|
|
config.SKLIK_LIST_MAX_PAGES,
|
|
len(rows),
|
|
)
|
|
break
|
|
|
|
return {
|
|
"totalCount": total if total is not None else len(rows),
|
|
"returnedCount": len(rows),
|
|
"truncated": pages >= config.SKLIK_LIST_MAX_PAGES,
|
|
entity: rows,
|
|
}
|
|
|
|
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,
|
|
}
|