"""Небольшой синхронный клиент WOK без сторонних зависимостей.""" from __future__ import annotations import json from typing import Any, Callable, Iterable from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode, urlparse from urllib.request import Request, urlopen class WokApiError(RuntimeError): """Ошибка API с сохранёнными кодом ответа и безопасным JSON-телом.""" def __init__(self, status: int, message: str, *, payload: Any = None, retry_after: str | None = None): super().__init__(message) self.status = status self.payload = payload self.retry_after = retry_after Transport = Callable[[Request, float], Any] class WokApi: """Клиент публичных маршрутов WOK Steam API.""" def __init__(self, api_key: str, *, base_url: str = "https://woksteamapi.com", timeout: float = 30.0, transport: Transport | None = None): parsed = urlparse(base_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("base_url должен быть абсолютным HTTP(S) URL") if not isinstance(api_key, str) or not api_key.strip(): raise ValueError("api_key не должен быть пустым") if timeout <= 0: raise ValueError("timeout должен быть больше нуля") self.base_url = base_url.rstrip("/") self.timeout = float(timeout) self._api_key = api_key.strip() self._transport = transport or self._open @staticmethod def _open(request: Request, timeout: float): return urlopen(request, timeout=timeout) @staticmethod def _decode(raw: bytes) -> Any: if not raw: return None try: return json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return {"error": "invalid_json_response"} @staticmethod def _error_message(payload: Any) -> str: """Извлечь сообщение и из простого, и из структурного error.""" if not isinstance(payload, dict): return "API request failed" message = payload.get("message") if isinstance(message, str) and message: return message error = payload.get("error") if isinstance(error, str) and error: return error if isinstance(error, dict): for field in ("message", "code"): value = error.get(field) if isinstance(value, str) and value: return value return "API request failed" def _request(self, method: str, path: str, *, query: dict[str, Any] | None = None, body: dict[str, Any] | None = None, authenticated: bool = True) -> Any: if not path.startswith("/"): raise ValueError("path должен начинаться с /") clean_query = { key: value for key, value in (query or {}).items() if value is not None and value != "" } suffix = f"?{urlencode(clean_query, doseq=True)}" if clean_query else "" raw_body = (json.dumps(body, separators=(",", ":")).encode("utf-8") if body is not None else None) headers = { "Accept": "application/json", "User-Agent": "wok-api-python/1.1.0", } if authenticated: headers["Authorization"] = f"Bearer {self._api_key}" if raw_body is not None: headers["Content-Type"] = "application/json" request = Request( f"{self.base_url}{path}{suffix}", data=raw_body, headers=headers, method=method, ) try: with self._transport(request, self.timeout) as response: return self._decode(response.read()) except HTTPError as exc: payload = self._decode(exc.read()) raise WokApiError( int(exc.code), self._error_message(payload), payload=payload, retry_after=exc.headers.get("Retry-After"), ) from exc except URLError as exc: raise WokApiError(0, "Network request failed") from exc @staticmethod def _values(values: Iterable[str], name: str) -> list[str]: if isinstance(values, (str, bytes)): raise ValueError(f"{name} должен быть списком") result = [str(value) for value in values] if not result or any(not value for value in result): raise ValueError(f"{name} не должен быть пустым") return result def inventory(self, steam_id: str, *, game: str = "cs2", no_cache: bool = False, top: int = 10, include_inspect: bool = False) -> dict[str, Any]: return self._request("GET", "/v1/inventory", query={ "steam_id": steam_id, "game": game, "no_cache": int(no_cache), "top": top, "include_inspect": int(include_inspect), }) def inventories(self, steamids: Iterable[str], *, games: Iterable[str] = ("cs2",), no_cache: bool = False, top: int = 10, include_inspect: bool = False) -> dict[str, Any]: return self._request("POST", "/v1/inventories", body={ "steamids": self._values(steamids, "steamids"), "games": self._values(games, "games"), "no_cache": bool(no_cache), "top": int(top), "include_inspect": bool(include_inspect), }) def compatibility_inventory(self, steam_id: str, *, game: str = "cs2", no_cache: bool = False, grouped: bool = True) -> list[dict[str, Any]]: return self._request("GET", "/steam/api/inventory", query={ "steam_id": steam_id, "game": game, "currency": "USD", "no_cache": int(no_cache), "group": int(grouped), }) def compatibility_profile(self, steam_id: str) -> dict[str, Any]: return self._request("GET", "/steam/api/profile", query={ "steam_id": steam_id, }) def profile(self, steam_id: str) -> dict[str, Any]: return self._request("GET", "/v1/profile", query={"steam_id": steam_id}) def profiles(self, steamids: Iterable[str]) -> dict[str, dict[str, Any]]: return self._request("POST", "/v1/profiles", body={ "steamids": self._values(steamids, "steamids"), }) def player_security(self, steam_id: str) -> dict[str, Any]: return self._request("GET", "/v1/player-security", query={ "steam_id": steam_id, }) def friends(self, steam_id: str, *, limit: int = 100, cursor: str | None = None) -> dict[str, Any]: return self._request("GET", "/v1/friends", query={ "steam_id": steam_id, "limit": limit, "cursor": cursor, }) def price(self, game: str, name: str) -> dict[str, Any]: return self._request("GET", "/v1/price", query={ "game": game, "name": name, }) def prices(self, game: str, names: Iterable[str]) -> dict[str, Any]: return self._request("POST", "/v1/prices", body={ "game": game, "names": self._values(names, "names"), }) def items(self, *, game: str = "cs2", query: str | None = None, source: str | None = None, limit: int = 50, cursor: str | None = None) -> dict[str, Any]: return self._request("GET", "/v1/items", query={ "game": game, "q": query, "source": source, "limit": limit, "cursor": cursor, }) def price_history(self, game: str, name: str, *, days: int = 30, source: str | None = None) -> dict[str, Any]: return self._request("GET", "/v1/price-history", query={ "game": game, "name": name, "days": days, "source": source, }) def inspect_cs2(self, inspect_link: str) -> dict[str, Any]: """float_value читать по float, paint_seed и paint_index по их флагам.""" return self._request("POST", "/v1/cs2/inspect", body={ "inspect_link": inspect_link, }) def create_crawl(self, steamids: Iterable[str], *, games: Iterable[str] = ("cs2",)) -> dict[str, Any]: return self._request("POST", "/v1/crawl", body={ "steamids": self._values(steamids, "steamids"), "games": self._values(games, "games"), }) def crawl_status(self, batch_id: str) -> dict[str, Any]: return self._request("GET", f"/v1/crawl/{quote(batch_id, safe='')}") def faceit_player(self, steam_id: str, *, game: str = "cs2") -> dict[str, Any]: return self._request("GET", "/faceit/data/v4/players", query={ "game": game, "game_player_id": steam_id, }) def faceit_stats(self, player_id: str, *, game: str = "cs2") -> dict[str, Any]: player = quote(player_id, safe="") game_name = quote(game, safe="") return self._request( "GET", f"/faceit/data/v4/players/{player}/stats/{game_name}") def faceit_history(self, player_id: str, *, game: str = "cs2", limit: int = 5) -> dict[str, Any]: player = quote(player_id, safe="") return self._request( "GET", f"/faceit/data/v4/players/{player}/history", query={"game": game, "limit": limit}, ) def health(self) -> dict[str, Any]: return self._request("GET", "/healthz", authenticated=False) __all__ = ["WokApi", "WokApiError"]