"""Клиент Rentu External API v2. Один файл без зависимостей кроме requests. Копируйте в проект целиком. from rentu_api import Rentu api = Rentu("YOUR_CLIENT_ID", "YOUR_CLIENT_SECRET") for sc in api.shopping_centers(): print(sc["id"], sc["title"], sc["time_zone"]) Что берёт на себя: * обновление токена до истечения, без лишних запросов к /auth/token; * повтор при 429 по Retry-After и при 503 с нарастающей паузой; * постраничный обход; * разбор конверта ответа и понятное исключение с машинным кодом ошибки. Документация: https://rentu.ru/new-api-docs/ """ from __future__ import annotations import datetime as dt import time from typing import Any, Iterator import requests PRODUCTION = "https://api.rentu.ru/api/external/v2" STAGE = "https://stage-api.rentu.ru/api/external/v2" # Повторять имеет смысл только эти. Остальное либо ошибка в параметрах, либо # состояние, которое само не изменится: прав нет, тариф не тот, ТЦ заблокирован. RETRYABLE = frozenset({"rights_unavailable", "traffic_unavailable", "internal_error"}) class RentuError(RuntimeError): """Ошибка API. Разбирайте по code, а не по тексту: тексты меняются.""" def __init__(self, code: str, message: str, status: int, details: Any = None): super().__init__(f"{code}: {message}") self.code = code self.message = message self.status = status self.details = details @property def retryable(self) -> bool: return self.code in RETRYABLE class Rentu: def __init__( self, client_id: str, client_secret: str, base_url: str = PRODUCTION, timeout: int = 60, max_retries: int = 4, ): self._auth = (client_id, client_secret) self.base_url = base_url.rstrip("/") self.timeout = timeout self.max_retries = max_retries self._session = requests.Session() self._token: str | None = None self._expires_at = 0.0 # ---------- транспорт ---------- def _headers(self) -> dict[str, str]: # Обновляемся за минуту до истечения: так не ловим 401 на границе. if self._token is None or time.time() > self._expires_at - 60: client_id, client_secret = self._auth resp = self._session.post( f"{self.base_url}/auth/token", json={"client_id": client_id, "client_secret": client_secret}, timeout=self.timeout, ) body = resp.json() if not body.get("success"): error = body.get("error", {}) raise RentuError( error.get("code", "unauthorized"), error.get("message", "не удалось получить токен"), resp.status_code, ) self._token = body["access_token"] self._expires_at = time.time() + body["expires_in"] return {"Authorization": f"Bearer {self._token}", "Accept": "application/json"} def request(self, method: str, path: str, **params: Any) -> dict[str, Any]: json_body = params.pop("json", None) attempt = 0 while True: resp = self._session.request( method, f"{self.base_url}{path}", headers=self._headers(), params=params or None, json=json_body, timeout=self.timeout, ) body = resp.json() if body.get("success"): return body error = body.get("error", {}) code = error.get("code", "unknown") if code == "token_expired": self._token = None continue if code == "rate_limited" and attempt < self.max_retries: time.sleep(int(resp.headers.get("Retry-After", 60))) attempt += 1 continue if code in RETRYABLE and attempt < self.max_retries: time.sleep(2 ** attempt) attempt += 1 continue raise RentuError(code, error.get("message", ""), resp.status_code, error.get("details")) def get(self, path: str, **params: Any) -> dict[str, Any]: return self.request("GET", path, **params) def pages(self, path: str, per_page: int = 500, **params: Any) -> Iterator[dict[str, Any]]: """Обход постраничных методов. Отдаёт строки, а не страницы.""" page = 1 while True: body = self.get(path, page=page, per_page=per_page, **params) yield from body["data"] meta = body.get("meta", {}).get("pagination") if not meta or page * meta["per_page"] >= meta["total"]: return page += 1 # ---------- справочники ---------- def shopping_centers(self) -> list[dict[str, Any]]: return self.get("/shopping_centers")["data"] def shops(self, sc_id: str, archived: bool = False) -> list[dict[str, Any]]: return list(self.pages(f"/sc/{sc_id}/shops", archived=str(archived).lower())) def kkts(self, sc_id: str, shop_id: str, archived: bool = False) -> list[dict[str, Any]]: return self.get(f"/sc/{sc_id}/shops/{shop_id}/kkts", archived=str(archived).lower())["data"] def traffic_areas(self, sc_id: str) -> list[dict[str, Any]]: return self.get(f"/sc/{sc_id}/traffic_areas")["data"] def turnover_config(self, sc_id: str, shop_id: str) -> dict[str, Any]: return self.get(f"/sc/{sc_id}/shops/{shop_id}/turnover_config")["data"] # ---------- данные ---------- def by_days(self, sc_id: str, shop_id: str, start: dt.date, end: dt.date) -> list[dict[str, Any]]: return list(self.pages( f"/sc/{sc_id}/reports/shops/{shop_id}/by_days", per_page=366, start_date=start.isoformat(), end_date=end.isoformat(), )) def by_months(self, sc_id: str, shop_id: str, start: dt.date, end: dt.date) -> list[dict[str, Any]]: return self.get( f"/sc/{sc_id}/reports/shops/{shop_id}/by_months", start_date=start.isoformat(), end_date=end.isoformat(), )["data"] def receipts(self, sc_id: str, shop_id: str, start: dt.date, end: dt.date, by: str = "created", document_types: str | None = None) -> Iterator[dict[str, Any]]: """Чеки за период. by='created' по времени пробития, 'received' по поступлению. Окно ограничено 31 днём на любом тарифе, поэтому длинные периоды режем сами: снаружи это выглядит одним обходом. """ params = {"document_types": document_types} if document_types else {} for window_start, window_end in _windows(start, end, days=31): yield from self.pages( f"/sc/{sc_id}/shops/{shop_id}/receipts", per_page=1000, **{f"start_{by}_date": window_start.isoformat(), f"end_{by}_date": window_end.isoformat()}, **params, ) def attendance(self, sc_id: str, start: dt.date, end: dt.date, granularity: str = "day", scope: str = "sc", **extra: Any) -> list[dict[str, Any]]: return self.get( f"/sc/{sc_id}/attendance", start_date=start.isoformat(), end_date=end.isoformat(), granularity=granularity, scope=scope, **extra, )["data"] def anomalies(self, sc_id: str, start: dt.date, end: dt.date, **filters: Any) -> Iterator[dict[str, Any]]: return self.pages( f"/sc/{sc_id}/anomalies", start_date=start.isoformat(), end_date=end.isoformat(), **filters, ) def create_event(self, sc_id: str, **event: Any) -> dict[str, Any] | None: """Создаёт событие. Возвращает None, если такое уже есть: это не ошибка.""" try: return self.request("POST", f"/sc/{sc_id}/events", json=event)["data"] except RentuError as error: if error.code == "already_exists_in_sc": return None raise # ---------- вспомогательное ---------- @staticmethod def today(sc: dict[str, Any]) -> dt.date: """Сегодня в таймзоне торгового центра, а не вашего сервера. Границы суток API считает по зоне ТЦ, и «вчера» у дальневосточного центра наступает раньше, чем у вас. """ from zoneinfo import ZoneInfo return dt.datetime.now(ZoneInfo(sc["time_zone"])).date() @staticmethod def rubles(kopecks: int) -> float: """Копейки в рубли. Все суммы API отдаёт целыми копейками.""" return kopecks / 100 def _windows(start: dt.date, end: dt.date, days: int) -> Iterator[tuple[dt.date, dt.date]]: while start <= end: stop = min(start + dt.timedelta(days=days - 1), end) yield start, stop start = stop + dt.timedelta(days=1)