lauadminandClaude Opus 5.5
528a869657
Checkbox relays Nova Poshta's "To many requests" (20000401501) as a 4xx third_party.generic error, not a 429, so bursts of receipt requests ended up as failed receipts. Such responses are now CheckboxRateLimitedError: the receipt stays pending and the worker job is retried with arq.Retry after the "Try again after N seconds" delay plus backoff. NP timeouts relayed the same way are treated as unavailable (unknown outcome). The HTTP client also sends Checkbox requests one at a time with a CHECKBOX_MIN_REQUEST_INTERVAL_MS pause and signs the cashier in once for concurrent jobs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
234 lines
9.9 KiB
Python
234 lines
9.9 KiB
Python
"""Реальный клиент Checkbox (ЕТТН-чеки, `/api/v1/ettn`).
|
||
|
||
Авторизация — токен кассира по PIN-коду (`/api/v1/cashier/signinPinCode`).
|
||
Токен кэшируется в памяти процесса по кассе; на 401 — один повторный вход.
|
||
|
||
Запросы из одного процесса идут строго по одному с паузой
|
||
`CHECKBOX_MIN_REQUEST_INTERVAL_MS`: при создании ЕТТН Checkbox синхронно ходит
|
||
в API Новой Почты, а та на частые вызовы отвечает «To many requests».
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
import time
|
||
import uuid
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from app.core.config import Settings
|
||
from app.schemas.checkbox import EttnOut, EttnStatus
|
||
from app.services.checkbox.client import (
|
||
CheckboxCredentials,
|
||
CheckboxError,
|
||
CheckboxRateLimitedError,
|
||
CheckboxUnavailableError,
|
||
)
|
||
|
||
_PROVIDER = "novapost"
|
||
_TIMEOUT = httpx.Timeout(30, connect=10)
|
||
# Сколько страниц списка ЕТТН просматривать при сверке после таймаута.
|
||
# Список отдаётся от новых к старым; больше 50 за страницу Checkbox не принимает.
|
||
_FIND_PAGES = 3
|
||
_FIND_PAGE_SIZE = 50
|
||
|
||
# Ошибки Новой Почты, которые Checkbox отдаёт 4xx с `code=third_party.*`:
|
||
# {"message": "To many requests", "info": ["Try again after 1 seconds"],
|
||
# "errorCodes": ["20000401501"]} — лимит частоты НП;
|
||
# "cURL error 28: SSL connection timeout" — Checkbox не дождался НП.
|
||
_NP_RATE_LIMIT_RE = re.compile(r"20000401501|too? many requests", re.IGNORECASE)
|
||
_NP_RETRY_AFTER_RE = re.compile(r"try again after (\d+) second", re.IGNORECASE)
|
||
_NP_TIMEOUT_RE = re.compile(r"curl error|timed? ?out", re.IGNORECASE)
|
||
_MIN_RETRY_AFTER = 1.0
|
||
|
||
|
||
def _error_message(response: httpx.Response) -> str:
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
return f"HTTP {response.status_code}: {response.text[:500]}"
|
||
if isinstance(data, dict):
|
||
message = data.get("message")
|
||
detail = data.get("detail")
|
||
# Ошибки со стороны Новой Почты Checkbox отдаёт как
|
||
# {"code": "third_party.*", "message": "Internal Server Error"} —
|
||
# без кода кассир видит только бесполезное «Internal Server Error».
|
||
if code := data.get("code"):
|
||
message = f"{message or 'Ошибка'} ({code})"
|
||
if isinstance(detail, list) and detail:
|
||
parts = [
|
||
f"{'.'.join(str(p) for p in item.get('loc', []))}: {item.get('msg')}"
|
||
for item in detail
|
||
if isinstance(item, dict)
|
||
]
|
||
return "; ".join(filter(None, [message, *parts]))
|
||
if message:
|
||
return str(message)
|
||
if detail:
|
||
return str(detail)
|
||
return f"HTTP {response.status_code}"
|
||
|
||
|
||
def _retry_after(response: httpx.Response) -> float:
|
||
seconds: float = 0
|
||
if match := _NP_RETRY_AFTER_RE.search(response.text):
|
||
seconds = float(match.group(1))
|
||
elif (header := response.headers.get("Retry-After", "")).isdigit():
|
||
seconds = float(header)
|
||
return max(seconds, _MIN_RETRY_AFTER)
|
||
|
||
|
||
def _transient_error(response: httpx.Response) -> CheckboxUnavailableError | None:
|
||
"""Ответ, после которого запрос можно повторить, или None, если ошибка окончательная."""
|
||
if response.status_code == 429:
|
||
return CheckboxRateLimitedError(_error_message(response), _retry_after(response))
|
||
if response.status_code >= 500:
|
||
return CheckboxUnavailableError(_error_message(response))
|
||
if response.status_code >= 400 and '"third_party.' in response.text:
|
||
if _NP_RATE_LIMIT_RE.search(response.text):
|
||
return CheckboxRateLimitedError(_error_message(response), _retry_after(response))
|
||
if _NP_TIMEOUT_RE.search(response.text):
|
||
return CheckboxUnavailableError(_error_message(response))
|
||
return None
|
||
|
||
|
||
class HttpCheckboxClient:
|
||
def __init__(self, settings: Settings) -> None:
|
||
self._base_url = settings.checkbox_base_url.rstrip("/")
|
||
self._client_headers = {
|
||
"X-Client-Name": settings.checkbox_client_name,
|
||
"X-Client-Version": settings.checkbox_client_version,
|
||
}
|
||
self._tokens: dict[uuid.UUID, str] = {}
|
||
self._min_interval = settings.checkbox_min_request_interval_ms / 1000
|
||
self._send_lock = asyncio.Lock()
|
||
self._sign_in_lock = asyncio.Lock()
|
||
self._last_sent_at = float("-inf")
|
||
|
||
async def _send(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
headers: dict[str, str],
|
||
json: Any = None,
|
||
params: dict[str, Any] | None = None,
|
||
) -> httpx.Response:
|
||
# Строго по одному запросу с паузой — параллельные задачи worker'а
|
||
# иначе пачкой упираются в лимит Новой Почты.
|
||
async with self._send_lock:
|
||
delay = self._last_sent_at + self._min_interval - time.monotonic()
|
||
if delay > 0:
|
||
await asyncio.sleep(delay)
|
||
self._last_sent_at = time.monotonic()
|
||
try:
|
||
async with httpx.AsyncClient(base_url=self._base_url, timeout=_TIMEOUT) as client:
|
||
response = await client.request(
|
||
method, path, headers=headers, json=json, params=params
|
||
)
|
||
except httpx.HTTPError as exc:
|
||
raise CheckboxUnavailableError(f"Checkbox недоступен: {exc!r}") from exc
|
||
if error := _transient_error(response):
|
||
raise error
|
||
return response
|
||
|
||
async def _token(self, creds: CheckboxCredentials, *, stale: str | None = None) -> str:
|
||
"""Токен кассира; `stale` — отвергнутый Checkbox'ом (401), его не переиспользуем.
|
||
|
||
Под блокировкой: параллельные задачи без токена входят один раз,
|
||
остальные берут токен, полученный первой.
|
||
"""
|
||
async with self._sign_in_lock:
|
||
token = self._tokens.get(creds.cash_register_id)
|
||
if token is not None and token != stale:
|
||
return token
|
||
return await self._sign_in(creds)
|
||
|
||
async def _sign_in(self, creds: CheckboxCredentials) -> str:
|
||
response = await self._send(
|
||
"POST",
|
||
"/api/v1/cashier/signinPinCode",
|
||
headers={**self._client_headers, "X-License-Key": creds.license_key},
|
||
json={"pin_code": creds.pin_code},
|
||
)
|
||
if response.status_code != 200:
|
||
raise CheckboxError(f"Вход кассира не удался: {_error_message(response)}")
|
||
token = response.json()["access_token"]
|
||
self._tokens[creds.cash_register_id] = token
|
||
return token
|
||
|
||
async def _request(
|
||
self,
|
||
creds: CheckboxCredentials,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
json: Any = None,
|
||
params: dict[str, Any] | None = None,
|
||
) -> httpx.Response:
|
||
token = await self._token(creds)
|
||
for attempt in range(2):
|
||
response = await self._send(
|
||
method,
|
||
path,
|
||
headers={
|
||
**self._client_headers,
|
||
"X-License-Key": creds.license_key,
|
||
"Authorization": f"Bearer {token}",
|
||
},
|
||
json=json,
|
||
params=params,
|
||
)
|
||
if response.status_code == 401 and attempt == 0:
|
||
token = await self._token(creds, stale=token)
|
||
continue
|
||
break
|
||
if response.status_code >= 400:
|
||
raise CheckboxError(_error_message(response))
|
||
return response
|
||
|
||
async def sign_in(self, creds: CheckboxCredentials) -> None:
|
||
await self._sign_in(creds)
|
||
|
||
async def create_ettn(self, creds: CheckboxCredentials, body: dict[str, Any]) -> EttnOut:
|
||
response = await self._request(creds, "POST", "/api/v1/ettn", json=body)
|
||
return EttnOut.model_validate(response.json())
|
||
|
||
async def get_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> EttnOut:
|
||
response = await self._request(creds, "GET", f"/api/v1/ettn/{ettn_id}")
|
||
return EttnOut.model_validate(response.json())
|
||
|
||
async def find_ettn(self, creds: CheckboxCredentials, waybill_number: str) -> EttnOut | None:
|
||
"""Ищет неотменённый ЕТТН-чек по номеру ТТН за последнюю неделю.
|
||
|
||
Нужен после таймаута создания: POST мог дойти до Checkbox, и слепой
|
||
повтор создал бы второй чек на ту же ТТН.
|
||
"""
|
||
date_from = (datetime.now(UTC) - timedelta(days=7)).isoformat()
|
||
for page in range(_FIND_PAGES):
|
||
response = await self._request(
|
||
creds,
|
||
"GET",
|
||
"/api/v1/ettn",
|
||
params={
|
||
"provider": _PROVIDER,
|
||
"date_from": date_from,
|
||
"limit": _FIND_PAGE_SIZE,
|
||
"offset": page * _FIND_PAGE_SIZE,
|
||
},
|
||
)
|
||
items = response.json()
|
||
for item in items:
|
||
ettn = EttnOut.model_validate(item)
|
||
if ettn.ettn_number == waybill_number and ettn.status != EttnStatus.CANCELLED:
|
||
return ettn
|
||
if len(items) < _FIND_PAGE_SIZE:
|
||
break
|
||
return None
|
||
|
||
async def delete_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> None:
|
||
await self._request(creds, "DELETE", f"/api/v1/ettn/{ettn_id}")
|