474 lines
19 KiB
Python
474 lines
19 KiB
Python
import logging
|
||
import re
|
||
import time
|
||
from collections.abc import Iterator
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
from django.conf import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MoySkladError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MoySkladRemoteImage:
|
||
external_id: str
|
||
filename: str
|
||
title: str
|
||
size: int
|
||
updated: str
|
||
download_href: str
|
||
|
||
|
||
class MoySkladClient:
|
||
DEFAULT_BASE_URL = "https://api.moysklad.ru/api/remap/1.2"
|
||
MAX_IMAGE_BYTES = 25 * 1024 * 1024
|
||
ASSORTMENT_ENTITY_TYPES = ("product", "variant", "bundle", "service")
|
||
GET_RETRY_DELAYS = (2.0, 5.0, 10.0)
|
||
CONNECT_TIMEOUT_SECONDS = 30.0
|
||
READ_TIMEOUT_SECONDS = 90.0
|
||
|
||
def __init__(self, *, auth_type: str, credentials: dict[str, str]):
|
||
self._project_reference_cache = {}
|
||
self.base_url = (
|
||
getattr(settings, "MOYSKLAD_API_BASE_URL", self.DEFAULT_BASE_URL).rstrip(
|
||
"/"
|
||
)
|
||
+ "/"
|
||
)
|
||
headers = {
|
||
"Accept": "application/json;charset=utf-8",
|
||
"Accept-Encoding": "gzip",
|
||
"User-Agent": "EUSRR-Commerce/1.0",
|
||
}
|
||
auth = None
|
||
if auth_type == "token":
|
||
token = re.sub(
|
||
r"^Bearer\s+",
|
||
"",
|
||
credentials.get("token", "").strip(),
|
||
flags=re.IGNORECASE,
|
||
).strip()
|
||
if not token:
|
||
raise MoySkladError("Не указан токен МойСклад.")
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
elif auth_type == "basic":
|
||
login = credentials.get("login", "").strip()
|
||
password = credentials.get("password", "")
|
||
if not login or not password:
|
||
raise MoySkladError("Не указаны логин или пароль МойСклад.")
|
||
auth = httpx.BasicAuth(login, password)
|
||
else:
|
||
raise MoySkladError("Неподдерживаемый способ авторизации.")
|
||
self.client = httpx.Client(
|
||
base_url=self.base_url,
|
||
headers=headers,
|
||
auth=auth,
|
||
timeout=httpx.Timeout(
|
||
connect=self.CONNECT_TIMEOUT_SECONDS,
|
||
read=self.READ_TIMEOUT_SECONDS,
|
||
write=self.CONNECT_TIMEOUT_SECONDS,
|
||
pool=self.CONNECT_TIMEOUT_SECONDS,
|
||
),
|
||
follow_redirects=False,
|
||
)
|
||
|
||
def close(self):
|
||
self.client.close()
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_value, traceback):
|
||
self.close()
|
||
|
||
def _get(self, path: str, *, params: dict[str, Any] | None = None) -> dict:
|
||
endpoint = path.lstrip("/")
|
||
retryable_errors = (
|
||
httpx.ReadTimeout,
|
||
httpx.ConnectTimeout,
|
||
httpx.ReadError,
|
||
)
|
||
total_attempts = len(self.GET_RETRY_DELAYS) + 1
|
||
payload: Any = None
|
||
|
||
for attempt in range(1, total_attempts + 1):
|
||
try:
|
||
response = self.client.get(endpoint, params=params)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
break
|
||
except retryable_errors as exc:
|
||
log_context = {
|
||
"event": "commerce.moysklad.get_retry",
|
||
"endpoint": endpoint,
|
||
"request_offset": (params or {}).get("offset"),
|
||
"request_limit": (params or {}).get("limit"),
|
||
"attempt": attempt,
|
||
"max_attempts": total_attempts,
|
||
"error_type": type(exc).__name__,
|
||
"failure_reason": str(exc)[:500],
|
||
}
|
||
if attempt >= total_attempts:
|
||
logger.error(
|
||
"MoySklad GET failed after retries: %s",
|
||
type(exc).__name__,
|
||
extra={
|
||
**log_context,
|
||
"event": "commerce.moysklad.get_failed",
|
||
"retries_exhausted": True,
|
||
},
|
||
exc_info=True,
|
||
)
|
||
raise MoySkladError(
|
||
"МойСклад не ответил после повторных попыток."
|
||
) from exc
|
||
|
||
retry_delay = self.GET_RETRY_DELAYS[attempt - 1]
|
||
logger.warning(
|
||
"MoySklad GET transient failure %s; retrying in %.1f seconds",
|
||
type(exc).__name__,
|
||
retry_delay,
|
||
extra={
|
||
**log_context,
|
||
"event": "commerce.moysklad.get_retry",
|
||
"retry_in_seconds": retry_delay,
|
||
},
|
||
)
|
||
time.sleep(retry_delay)
|
||
except httpx.HTTPStatusError as exc:
|
||
if (
|
||
exc.response.status_code in {429, 500, 502, 503, 504}
|
||
and attempt < total_attempts
|
||
):
|
||
retry_delay = self.GET_RETRY_DELAYS[attempt - 1]
|
||
logger.warning(
|
||
"MoySklad HTTP %s; retrying GET",
|
||
exc.response.status_code,
|
||
extra={
|
||
"event": "commerce.moysklad.get_retry",
|
||
"endpoint": endpoint,
|
||
"attempt": attempt,
|
||
"request_offset": (params or {}).get("offset"),
|
||
},
|
||
)
|
||
time.sleep(retry_delay)
|
||
continue
|
||
raise MoySkladError(self._http_status_message(exc.response)) from exc
|
||
except (httpx.HTTPError, ValueError) as exc:
|
||
logger.error(
|
||
"MoySklad GET failed without retry: %s",
|
||
type(exc).__name__,
|
||
extra={
|
||
"event": "commerce.moysklad.get_failed",
|
||
"endpoint": endpoint,
|
||
"request_offset": (params or {}).get("offset"),
|
||
"request_limit": (params or {}).get("limit"),
|
||
"error_type": type(exc).__name__,
|
||
"failure_reason": str(exc)[:500],
|
||
},
|
||
exc_info=True,
|
||
)
|
||
raise MoySkladError(
|
||
"Не удалось получить корректный ответ от МойСклад."
|
||
) from exc
|
||
if not isinstance(payload, dict):
|
||
raise MoySkladError("МойСклад вернул ответ неожиданного формата.")
|
||
return payload
|
||
|
||
@classmethod
|
||
def _http_status_message(cls, response: httpx.Response) -> str:
|
||
status = response.status_code
|
||
if status in {401, 403}:
|
||
message = "МойСклад отклонил реквизиты или у пользователя нет прав API."
|
||
elif status == 400:
|
||
message = "МойСклад не принял запрос. Проверьте access token и права пользователя."
|
||
elif status == 429:
|
||
message = "МойСклад временно ограничил частоту запросов."
|
||
else:
|
||
message = f"МойСклад вернул ошибку HTTP {status}."
|
||
detail = cls._error_detail(response)
|
||
return f"{message} Ответ сервиса: {detail}" if detail else message
|
||
|
||
def product_images(
|
||
self,
|
||
entity_type: str,
|
||
external_id: str,
|
||
) -> list[MoySkladRemoteImage]:
|
||
if entity_type not in {"product", "variant", "bundle"}:
|
||
return []
|
||
payload = self._get(f"/entity/{entity_type}/{external_id}/images")
|
||
rows = payload.get("rows") or []
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("МойСклад вернул некорректный список изображений.")
|
||
|
||
return self.parse_product_images(rows)
|
||
|
||
@staticmethod
|
||
def parse_product_images(rows: list[dict]) -> list[MoySkladRemoteImage]:
|
||
images = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
meta = row.get("meta") if isinstance(row.get("meta"), dict) else {}
|
||
href = str(meta.get("href") or "").rstrip("/")
|
||
download_href = str(meta.get("downloadHref") or "").strip()
|
||
image_id = href.rsplit("/", 1)[-1]
|
||
if not image_id or not download_href:
|
||
continue
|
||
images.append(
|
||
MoySkladRemoteImage(
|
||
external_id=image_id[:255],
|
||
filename=str(row.get("filename") or "image")[:255],
|
||
title=str(row.get("title") or "")[:255],
|
||
size=max(0, int(row.get("size") or 0)),
|
||
updated=str(row.get("updated") or ""),
|
||
download_href=download_href,
|
||
)
|
||
)
|
||
return images
|
||
|
||
def download_image(self, download_href: str) -> tuple[bytes, str]:
|
||
source = urlparse(download_href)
|
||
api = urlparse(self.base_url)
|
||
if source.scheme != "https" or source.netloc != api.netloc:
|
||
raise MoySkladError("МойСклад вернул недопустимую ссылку на изображение.")
|
||
|
||
try:
|
||
with self.client.stream(
|
||
"GET",
|
||
download_href,
|
||
follow_redirects=True,
|
||
) as response:
|
||
response.raise_for_status()
|
||
declared_size = int(response.headers.get("Content-Length") or 0)
|
||
if declared_size > self.MAX_IMAGE_BYTES:
|
||
raise MoySkladError("Изображение МойСклад превышает лимит 25 МБ.")
|
||
chunks = []
|
||
received = 0
|
||
for chunk in response.iter_bytes():
|
||
received += len(chunk)
|
||
if received > self.MAX_IMAGE_BYTES:
|
||
raise MoySkladError(
|
||
"Изображение МойСклад превышает лимит 25 МБ."
|
||
)
|
||
chunks.append(chunk)
|
||
content_type = response.headers.get("Content-Type", "").split(";", 1)[0]
|
||
except MoySkladError:
|
||
raise
|
||
except httpx.HTTPError as exc:
|
||
raise MoySkladError("Не удалось скачать изображение из МойСклад.") from exc
|
||
|
||
if not chunks:
|
||
raise MoySkladError("МойСклад вернул пустое изображение.")
|
||
if content_type and not (
|
||
content_type.startswith("image/")
|
||
or content_type == "application/octet-stream"
|
||
):
|
||
raise MoySkladError("МойСклад вернул файл неподдерживаемого типа.")
|
||
return b"".join(chunks), content_type
|
||
|
||
@staticmethod
|
||
def _error_detail(response: httpx.Response) -> str:
|
||
try:
|
||
payload = response.json()
|
||
except ValueError:
|
||
return ""
|
||
candidates: list[Any] = []
|
||
if isinstance(payload, dict):
|
||
errors = payload.get("errors")
|
||
if isinstance(errors, list):
|
||
candidates.extend(errors)
|
||
candidates.append(payload)
|
||
for candidate in candidates:
|
||
if not isinstance(candidate, dict):
|
||
continue
|
||
for key in ("error", "message", "detail"):
|
||
value = candidate.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
return " ".join(value.split())[:500]
|
||
return ""
|
||
|
||
def employee_context(self) -> dict:
|
||
return self._get("/context/employee")
|
||
|
||
def sales_channel_names(self) -> dict[str, str]:
|
||
payload = self._get("/entity/saleschannel", params={"limit": 100})
|
||
rows = payload.get("rows") or []
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("МойСклад вернул некорректный список каналов продаж.")
|
||
return {
|
||
str(row.get("id")): str(row.get("name") or "")
|
||
for row in rows
|
||
if isinstance(row, dict) and row.get("id")
|
||
}
|
||
|
||
def iter_archived_assortment(self, *, limit: int = 100) -> Iterator[dict]:
|
||
page_size = min(limit, 100)
|
||
for entity_type in self.ASSORTMENT_ENTITY_TYPES:
|
||
offset = 0
|
||
while True:
|
||
payload = self._get(
|
||
f"/entity/{entity_type}",
|
||
params={
|
||
"limit": page_size,
|
||
"offset": offset,
|
||
"filter": "archived=true",
|
||
"expand": "images",
|
||
},
|
||
)
|
||
rows = payload.get("rows", [])
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError(
|
||
"В ответе архивного ассортимента отсутствует список rows."
|
||
)
|
||
for row in rows:
|
||
if isinstance(row, dict):
|
||
yield row
|
||
if len(rows) < page_size:
|
||
break
|
||
offset += page_size
|
||
|
||
def iter_assortment(
|
||
self, *, limit: int = 100, include_archived: bool = True
|
||
) -> Iterator[dict]:
|
||
offset = 0
|
||
seen_ids: set[str] = set()
|
||
page_size = min(limit, 100)
|
||
while True:
|
||
payload = self._get(
|
||
"/entity/assortment",
|
||
params={
|
||
"limit": page_size,
|
||
"offset": offset,
|
||
"stockMode": "all",
|
||
"expand": "images",
|
||
},
|
||
)
|
||
rows = payload.get("rows", [])
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("В ответе ассортимента отсутствует список rows.")
|
||
for row in rows:
|
||
if isinstance(row, dict):
|
||
external_id = str(row.get("id") or "")
|
||
if external_id:
|
||
seen_ids.add(external_id)
|
||
yield row
|
||
if len(rows) < page_size:
|
||
break
|
||
offset += page_size
|
||
|
||
if include_archived:
|
||
for row in self.iter_archived_assortment(limit=page_size):
|
||
external_id = str(row.get("id") or "")
|
||
if external_id and external_id in seen_ids:
|
||
continue
|
||
yield row
|
||
|
||
def iter_documents(self, entity_type: str, *, limit: int = 100) -> Iterator[dict]:
|
||
if entity_type not in {"demand", "salesreturn"}:
|
||
raise MoySkladError("Неподдерживаемый тип документа МойСклад.")
|
||
offset = 0
|
||
while True:
|
||
payload = self._get(
|
||
f"/entity/{entity_type}",
|
||
params={
|
||
"limit": limit,
|
||
"offset": offset,
|
||
"expand": "positions.assortment,state,salesChannel,rate.currency,agent,project",
|
||
},
|
||
)
|
||
rows = payload.get("rows", [])
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("В ответе документов отсутствует список rows.")
|
||
for row in rows:
|
||
if isinstance(row, dict):
|
||
yield row
|
||
if len(rows) < limit:
|
||
return
|
||
offset += limit
|
||
|
||
def _payment_reference(self, reference):
|
||
if not reference or reference.get("name"):
|
||
return reference or {}
|
||
meta = reference.get("meta") or {}
|
||
kind = meta.get("type")
|
||
external_id = (
|
||
reference.get("id")
|
||
or urlparse(meta.get("href", "")).path.rstrip("/").split("/")[-1]
|
||
)
|
||
if kind not in {
|
||
"counterparty",
|
||
"organization",
|
||
"employee",
|
||
"expenseitem",
|
||
"currency",
|
||
} or not re.fullmatch(r"[0-9a-fA-F-]{36}", external_id):
|
||
raise MoySkladError("Некорректная ссылка в платеже МойСклад.")
|
||
key = (kind, external_id)
|
||
if key not in self._project_reference_cache:
|
||
data = self._get(f"/entity/{kind}/{external_id}")
|
||
self._project_reference_cache[key] = {
|
||
k: data[k]
|
||
for k in ["id", "meta", "name", "isoCode", "code"]
|
||
if k in data
|
||
}
|
||
return self._project_reference_cache[key]
|
||
|
||
def _payment_references(self, row):
|
||
# Resolve shared references once per run instead of repeatedly asking
|
||
# the API to expand hundreds of the same objects on every large page.
|
||
row = dict(row)
|
||
for key in ["expenseItem", "agent"]:
|
||
row[key] = self._payment_reference(row.get(key))
|
||
rate = dict(row.get("rate") or {})
|
||
rate["currency"] = self._payment_reference(rate.get("currency"))
|
||
row["rate"] = rate
|
||
return row
|
||
|
||
def iter_project_data(
|
||
self, entity_type: str, *, archived=False, limit=1000
|
||
) -> Iterator[dict]:
|
||
if entity_type not in {"project", "paymentout", "cashout"}:
|
||
raise MoySkladError("Неподдерживаемый тип данных проекта.")
|
||
offset = 0
|
||
while True:
|
||
params = {"limit": limit, "offset": offset}
|
||
if entity_type == "project":
|
||
params["filter"] = f"archived={str(archived).lower()}"
|
||
payload = self._get(f"/entity/{entity_type}", params=params)
|
||
rows = payload.get("rows")
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("В ответе проектов отсутствует список rows.")
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
raise MoySkladError("Некорректная запись проекта.")
|
||
yield row if entity_type == "project" else self._payment_references(row)
|
||
if len(rows) < limit:
|
||
return
|
||
offset += limit
|
||
|
||
def iter_stock_by_store(self, *, limit: int = 1000) -> Iterator[dict]:
|
||
offset = 0
|
||
while True:
|
||
page_size = min(limit, 1000)
|
||
payload = self._get(
|
||
"/report/stock/bystore",
|
||
params={"limit": page_size, "offset": offset},
|
||
)
|
||
rows = payload.get("rows", [])
|
||
if not isinstance(rows, list):
|
||
raise MoySkladError("В отчёте остатков отсутствует список rows.")
|
||
for row in rows:
|
||
if isinstance(row, dict):
|
||
yield row
|
||
if len(rows) < page_size:
|
||
return
|
||
offset += page_size
|