lauadminandClaude Sonnet 5
d13e7ce2b3
Adds NpTrackingClient (Protocol + real/stub impls) and an ARQ worker that polls Nova Poshta every minute for orders without a receipt, writing status, net COD amount (Контроль оплати), and payment status onto the order. Surfaced in the orders table and detail modal. Marks plan stages 3-4 done in README. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""Реальный клиент Nova Poshta (`TrackingDocument.getStatusDocuments`)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from app.core.config import Settings
|
|
from app.schemas.tracking import TrackingStatusOut
|
|
from app.services.nova_poshta.client import NovaPoshtaError
|
|
|
|
_API_URL = "https://api.novaposhta.ua/v2.0/json/"
|
|
|
|
# NP отклоняет запросы с более чем 100 накладными за раз.
|
|
_MAX_DOCUMENTS_PER_REQUEST = 100
|
|
|
|
|
|
class NpTrackingClient:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._api_key = settings.nova_poshta_api_key
|
|
|
|
async def get_statuses(self, *, waybill_numbers: list[str]) -> list[TrackingStatusOut]:
|
|
if not waybill_numbers:
|
|
return []
|
|
if len(waybill_numbers) > _MAX_DOCUMENTS_PER_REQUEST:
|
|
raise NovaPoshtaError(
|
|
f"Слишком много ТТН за один запрос: {len(waybill_numbers)} "
|
|
f"(максимум {_MAX_DOCUMENTS_PER_REQUEST})"
|
|
)
|
|
|
|
body = {
|
|
"apiKey": self._api_key,
|
|
"modelName": "TrackingDocument",
|
|
"calledMethod": "getStatusDocuments",
|
|
"methodProperties": {
|
|
"Documents": [{"DocumentNumber": number, "Phone": ""} for number in waybill_numbers]
|
|
},
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(_API_URL, json=body)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if not data.get("success"):
|
|
errors = data.get("errors") or []
|
|
message = "; ".join(str(error) for error in errors)
|
|
raise NovaPoshtaError(f"NP вернул ошибку: {message or 'неизвестная ошибка'}")
|
|
|
|
return [TrackingStatusOut.model_validate(item) for item in data.get("data", [])]
|