923 lines
32 KiB
Python
923 lines
32 KiB
Python
import logging
|
||
import re
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import datetime
|
||
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
|
||
from time import monotonic
|
||
from typing import Any, Callable
|
||
from urllib.parse import urlparse
|
||
|
||
from celery.exceptions import SoftTimeLimitExceeded
|
||
from dateutil import parser as date_parser
|
||
from django.db import transaction
|
||
from django.utils import timezone
|
||
|
||
from commerce.integrations.moysklad import MoySkladClient
|
||
from commerce.media import sync_moysklad_product_images
|
||
from commerce.models import (
|
||
IntegrationConnection,
|
||
IntegrationSyncStage,
|
||
Product,
|
||
ProductBinding,
|
||
ProductKind,
|
||
Sale,
|
||
SaleEventType,
|
||
SaleLine,
|
||
StockSnapshot,
|
||
)
|
||
from commerce.services.credentials import decrypt_credentials
|
||
from commerce.services.ingestion import ingest_source_record
|
||
from commerce.services.matching import match_product
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class MoySkladSyncStats:
|
||
projects: int = 0
|
||
project_payments: int = 0
|
||
processed: int = 0
|
||
created_products: int = 0
|
||
matched_products: int = 0
|
||
stock_snapshots: int = 0
|
||
images: int = 0
|
||
image_errors: int = 0
|
||
sales: int = 0
|
||
sale_lines: int = 0
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return asdict(self)
|
||
|
||
|
||
def _external_id(row: dict[str, Any]) -> str:
|
||
value = str(row.get("id") or "").strip()
|
||
if value:
|
||
return value
|
||
href = str(row.get("meta", {}).get("href") or "").rstrip("/")
|
||
return urlparse(href).path.rstrip("/").rsplit("/", 1)[-1]
|
||
|
||
|
||
def _barcode(row: dict[str, Any]) -> str:
|
||
for item in row.get("barcodes") or []:
|
||
if isinstance(item, str) and item.strip():
|
||
return item.strip()
|
||
if isinstance(item, dict):
|
||
for value in item.values():
|
||
if isinstance(value, str) and value.strip():
|
||
return value.strip()
|
||
return ""
|
||
|
||
|
||
def _decimal(value: Any) -> Decimal:
|
||
try:
|
||
return Decimal(str(value or 0))
|
||
except (InvalidOperation, TypeError, ValueError):
|
||
return Decimal("0")
|
||
|
||
|
||
def _money(value: Any) -> Decimal:
|
||
return (_decimal(value) / Decimal("100")).quantize(
|
||
Decimal("0.01"),
|
||
rounding=ROUND_HALF_UP,
|
||
)
|
||
|
||
|
||
def _position_revenue(
|
||
document: dict[str, Any], position: dict[str, Any]
|
||
) -> tuple[Decimal, Decimal, Decimal]:
|
||
"""Return source-compatible gross, discount and added VAT amounts.
|
||
|
||
MoySklad may return a fractional price in kopecks. Rounding that unit price
|
||
before multiplying by quantity changes the document total. It may also keep
|
||
VAT outside the position price while including it in the document ``sum``.
|
||
All arithmetic therefore stays at source precision until the line total.
|
||
"""
|
||
|
||
quantity = _decimal(position.get("quantity"))
|
||
base_gross = quantity * _decimal(position.get("price")) / Decimal("100")
|
||
discount_percent = _decimal(position.get("discount"))
|
||
base_net = base_gross * (Decimal("100") - discount_percent) / Decimal("100")
|
||
multiplier = Decimal("1")
|
||
if document.get("vatIncluded") is False and bool(position.get("vatEnabled")):
|
||
multiplier += _decimal(position.get("vat")) / Decimal("100")
|
||
|
||
gross = (base_gross * multiplier).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
net = (base_net * multiplier).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
base_net_rounded = base_net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
return gross, max(Decimal("0"), gross - net), net - base_net_rounded
|
||
|
||
|
||
def _source_updated_at(row: dict[str, Any]) -> datetime | None:
|
||
value = row.get("updated")
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = date_parser.parse(str(value))
|
||
except (TypeError, ValueError, OverflowError):
|
||
return None
|
||
if timezone.is_naive(parsed):
|
||
parsed = timezone.make_aware(parsed)
|
||
return parsed
|
||
|
||
|
||
def _product_kind(row: dict[str, Any]) -> str:
|
||
source_type = str(row.get("meta", {}).get("type") or "")
|
||
if source_type == "service":
|
||
return ProductKind.SERVICE
|
||
if source_type == "bundle":
|
||
return ProductKind.BUNDLE
|
||
return ProductKind.PRODUCT
|
||
|
||
|
||
def _reference_external_id(value: Any) -> str:
|
||
if not isinstance(value, dict):
|
||
return ""
|
||
direct = str(value.get("id") or "").strip()
|
||
if direct:
|
||
return direct
|
||
meta = value.get("meta") if isinstance(value.get("meta"), dict) else {}
|
||
href = str(meta.get("href") or "").rstrip("/")
|
||
return urlparse(href).path.rstrip("/").rsplit("/", 1)[-1]
|
||
|
||
|
||
def _reference_name(value: Any) -> str:
|
||
return str(value.get("name") or "") if isinstance(value, dict) else ""
|
||
|
||
|
||
def _price_value(value: Any) -> str | None:
|
||
if not isinstance(value, dict) or "value" not in value:
|
||
return None
|
||
return str(_money(value.get("value")))
|
||
|
||
|
||
def _binding_metadata(row: dict[str, Any]) -> dict[str, Any]:
|
||
sale_prices = []
|
||
for price in row.get("salePrices") or []:
|
||
if not isinstance(price, dict):
|
||
continue
|
||
price_type = (
|
||
price.get("priceType") if isinstance(price.get("priceType"), dict) else {}
|
||
)
|
||
sale_prices.append(
|
||
{
|
||
"name": str(price_type.get("name") or ""),
|
||
"external_id": _reference_external_id(price_type),
|
||
"value": str(_money(price.get("value"))),
|
||
}
|
||
)
|
||
attributes = []
|
||
for attribute in row.get("attributes") or []:
|
||
if isinstance(attribute, dict):
|
||
attributes.append(
|
||
{
|
||
"external_id": str(attribute.get("id") or ""),
|
||
"name": str(attribute.get("name") or ""),
|
||
"type": str(attribute.get("type") or ""),
|
||
"value": attribute.get("value"),
|
||
}
|
||
)
|
||
return {
|
||
"provider": "moysklad",
|
||
"entity_type": str((row.get("meta") or {}).get("type") or ""),
|
||
"external_code": str(row.get("externalCode") or ""),
|
||
"parent_external_id": _reference_external_id(row.get("product")),
|
||
"purchase_price": _price_value(row.get("buyPrice")),
|
||
"minimum_price": _price_value(row.get("minPrice")),
|
||
"sale_prices": sale_prices,
|
||
"uom": {
|
||
"external_id": _reference_external_id(row.get("uom")),
|
||
"name": _reference_name(row.get("uom")),
|
||
},
|
||
"folder": {
|
||
"external_id": _reference_external_id(row.get("productFolder")),
|
||
"name": str(row.get("pathName") or ""),
|
||
},
|
||
"weight": row.get("weight"),
|
||
"volume": row.get("volume"),
|
||
"vat": row.get("effectiveVat", row.get("vat")),
|
||
"attributes": attributes,
|
||
"characteristics": row.get("characteristics") or [],
|
||
"source_updated_at": str(row.get("updated") or ""),
|
||
}
|
||
|
||
|
||
def _update_product_from_source(product: Product, row: dict[str, Any]) -> None:
|
||
external_id = _external_id(row)
|
||
external_sku = str(row.get("article") or row.get("code") or "").strip()
|
||
product.name = str(row.get("name") or external_sku or external_id)[:255]
|
||
product.barcode = _barcode(row)[:64]
|
||
product.description = str(row.get("description") or "")
|
||
product.kind = _product_kind(row)
|
||
product.is_active = not bool(row.get("archived"))
|
||
product.save(
|
||
update_fields=[
|
||
"name",
|
||
"barcode",
|
||
"description",
|
||
"kind",
|
||
"is_active",
|
||
"updated_at",
|
||
]
|
||
)
|
||
|
||
|
||
def _candidate_sku(row: dict[str, Any], external_id: str) -> str:
|
||
raw = str(row.get("article") or row.get("code") or "").strip()
|
||
if not raw:
|
||
raw = f"moysklad-{external_id}"
|
||
normalized = re.sub(r"\s+", "-", raw)[:128]
|
||
if not Product.objects.filter(sku__iexact=normalized).exists():
|
||
return normalized
|
||
suffix = external_id.replace("-", "")[:10]
|
||
base = normalized[: max(1, 117 - len(suffix))]
|
||
candidate = f"{base}-ms-{suffix}"[:128]
|
||
sequence = 2
|
||
while Product.objects.filter(sku__iexact=candidate).exists():
|
||
tail = f"-{sequence}"
|
||
candidate = f"{base[: 128 - len(tail)]}{tail}"
|
||
sequence += 1
|
||
return candidate
|
||
|
||
|
||
@transaction.atomic
|
||
def normalize_moysklad_assortment_row(
|
||
*, connection: IntegrationConnection, row: dict[str, Any], captured_at: datetime
|
||
) -> tuple[bool, bool, bool]:
|
||
source = connection.source
|
||
external_id = _external_id(row)
|
||
if not external_id:
|
||
return False, False, False
|
||
external_sku = str(row.get("article") or row.get("code") or "").strip()
|
||
barcode = _barcode(row)
|
||
ingest_source_record(
|
||
source=source,
|
||
entity_type="product",
|
||
external_id=external_id,
|
||
payload=row,
|
||
source_updated_at=_source_updated_at(row),
|
||
)
|
||
|
||
existing_binding = (
|
||
ProductBinding.objects.select_related("product")
|
||
.filter(source=source, external_id=external_id)
|
||
.first()
|
||
)
|
||
product = existing_binding.product if existing_binding else None
|
||
if product is None and str((row.get("meta") or {}).get("type") or "") != "variant":
|
||
candidate = match_product(
|
||
source=source,
|
||
external_id=external_id,
|
||
external_sku=external_sku,
|
||
barcode=barcode,
|
||
)
|
||
if (
|
||
candidate.product
|
||
and not ProductBinding.objects.filter(
|
||
product=candidate.product,
|
||
source=source,
|
||
).exists()
|
||
):
|
||
product = candidate.product
|
||
created = False
|
||
matched = product is not None
|
||
if product is None:
|
||
product = Product.objects.create(
|
||
sku=_candidate_sku(row, external_id),
|
||
name=str(row.get("name") or external_sku or external_id)[:255],
|
||
barcode=barcode[:64],
|
||
description=str(row.get("description") or ""),
|
||
kind=_product_kind(row),
|
||
is_active=not bool(row.get("archived")),
|
||
)
|
||
created = True
|
||
|
||
if created or not product.bindings.exclude(source=source).exists():
|
||
_update_product_from_source(product, row)
|
||
|
||
binding, _ = ProductBinding.objects.update_or_create(
|
||
source=source,
|
||
external_id=external_id,
|
||
defaults={
|
||
"product": product,
|
||
"external_sku": external_sku[:128],
|
||
"metadata": _binding_metadata(row),
|
||
"is_primary": True,
|
||
},
|
||
)
|
||
parent_external_id = str(binding.metadata.get("parent_external_id") or "")
|
||
if parent_external_id:
|
||
parent_binding = ProductBinding.objects.filter(
|
||
source=source,
|
||
external_id=parent_external_id,
|
||
).first()
|
||
if parent_binding and parent_binding.product_id != product.pk:
|
||
product.parent_id = parent_binding.product_id
|
||
product.save(update_fields=["parent", "updated_at"])
|
||
has_stock = any(key in row for key in ("stock", "reserve", "inTransit"))
|
||
return created, matched, has_stock and product.kind != ProductKind.SERVICE
|
||
|
||
|
||
def reconcile_moysklad_variant_parents(
|
||
source,
|
||
*,
|
||
progress_callback: Callable[[int, int], None] | None = None,
|
||
batch_size: int = 500,
|
||
) -> int:
|
||
updated = 0
|
||
bindings = {
|
||
binding.external_id: binding
|
||
for binding in ProductBinding.objects.filter(source=source).select_related(
|
||
"product"
|
||
)
|
||
}
|
||
total = len(bindings)
|
||
changed_products = []
|
||
for processed, binding in enumerate(bindings.values(), start=1):
|
||
parent_external_id = str(binding.metadata.get("parent_external_id") or "")
|
||
parent_binding = bindings.get(parent_external_id)
|
||
if (
|
||
parent_binding
|
||
and parent_binding.product_id != binding.product_id
|
||
and binding.product.parent_id != parent_binding.product_id
|
||
):
|
||
binding.product.parent_id = parent_binding.product_id
|
||
binding.product.updated_at = timezone.now()
|
||
changed_products.append(binding.product)
|
||
updated += 1
|
||
if progress_callback and (processed % batch_size == 0 or processed == total):
|
||
progress_callback(processed, total)
|
||
if changed_products:
|
||
Product.objects.bulk_update(
|
||
changed_products,
|
||
["parent", "updated_at"],
|
||
batch_size=batch_size,
|
||
)
|
||
return updated
|
||
|
||
|
||
def normalize_moysklad_stock_row(
|
||
*, connection: IntegrationConnection, row: dict[str, Any], captured_at: datetime
|
||
) -> int:
|
||
return sync_moysklad_stock_rows(
|
||
connection=connection,
|
||
rows=[row],
|
||
captured_at=captured_at,
|
||
)
|
||
|
||
|
||
def _bulk_upsert_stock_snapshots(snapshots, *, batch_size: int) -> int:
|
||
if not snapshots:
|
||
return 0
|
||
with transaction.atomic():
|
||
StockSnapshot.objects.bulk_create(
|
||
snapshots,
|
||
batch_size=batch_size,
|
||
update_conflicts=True,
|
||
update_fields=[
|
||
"warehouse_name",
|
||
"quantity",
|
||
"reserved",
|
||
"in_transit",
|
||
"metadata",
|
||
],
|
||
unique_fields=[
|
||
"product",
|
||
"source",
|
||
"warehouse_external_id",
|
||
"captured_at",
|
||
],
|
||
)
|
||
return len(snapshots)
|
||
|
||
|
||
def _stock_snapshot(
|
||
*,
|
||
product_id,
|
||
source,
|
||
external_product_id,
|
||
store_stock,
|
||
captured_at,
|
||
):
|
||
warehouse_external_id = _reference_external_id(store_stock)
|
||
if not warehouse_external_id:
|
||
return None
|
||
return StockSnapshot(
|
||
product_id=product_id,
|
||
source=source,
|
||
warehouse_external_id=warehouse_external_id,
|
||
captured_at=captured_at,
|
||
warehouse_name=str(store_stock.get("name") or "")[:255],
|
||
quantity=_decimal(store_stock.get("stock")),
|
||
reserved=_decimal(store_stock.get("reserve")),
|
||
in_transit=_decimal(store_stock.get("inTransit")),
|
||
metadata={
|
||
"provider": "moysklad",
|
||
"external_product_id": external_product_id,
|
||
},
|
||
)
|
||
|
||
|
||
def sync_moysklad_stock_rows(
|
||
*,
|
||
connection: IntegrationConnection,
|
||
rows,
|
||
captured_at: datetime,
|
||
batch_size: int = 500,
|
||
progress_callback: Callable[[int, int], None] | None = None,
|
||
) -> int:
|
||
source = connection.source
|
||
product_ids = dict(
|
||
ProductBinding.objects.filter(source=source).values_list(
|
||
"external_id",
|
||
"product_id",
|
||
)
|
||
)
|
||
synchronized = 0
|
||
processed_rows = 0
|
||
pending = {}
|
||
|
||
for row in rows:
|
||
processed_rows += 1
|
||
external_product_id = _external_id(row)
|
||
product_id = product_ids.get(external_product_id)
|
||
if product_id:
|
||
for store_stock in row.get("stockByStore") or []:
|
||
if not isinstance(store_stock, dict):
|
||
continue
|
||
snapshot = _stock_snapshot(
|
||
product_id=product_id,
|
||
source=source,
|
||
external_product_id=external_product_id,
|
||
store_stock=store_stock,
|
||
captured_at=captured_at,
|
||
)
|
||
if snapshot is None:
|
||
continue
|
||
pending[(product_id, snapshot.warehouse_external_id)] = snapshot
|
||
if processed_rows % batch_size == 0:
|
||
synchronized += _bulk_upsert_stock_snapshots(
|
||
list(pending.values()), batch_size=batch_size
|
||
)
|
||
pending.clear()
|
||
if progress_callback:
|
||
progress_callback(processed_rows, synchronized)
|
||
synchronized += _bulk_upsert_stock_snapshots(
|
||
list(pending.values()), batch_size=batch_size
|
||
)
|
||
if progress_callback and processed_rows % batch_size:
|
||
progress_callback(processed_rows, synchronized)
|
||
return synchronized
|
||
|
||
|
||
def _document_positions(row: dict[str, Any]) -> list[dict[str, Any]]:
|
||
positions = row.get("positions") or []
|
||
if isinstance(positions, dict):
|
||
positions = positions.get("rows") or []
|
||
return [item for item in positions if isinstance(item, dict)]
|
||
|
||
|
||
def _assortment_external_id(position: dict[str, Any]) -> str:
|
||
assortment = position.get("assortment") or {}
|
||
if not isinstance(assortment, dict):
|
||
return ""
|
||
return _reference_external_id(assortment)
|
||
|
||
|
||
def _document_currency(row: dict[str, Any]) -> str:
|
||
rate = row.get("rate") if isinstance(row.get("rate"), dict) else {}
|
||
currency = rate.get("currency") if isinstance(rate.get("currency"), dict) else {}
|
||
value = str(
|
||
currency.get("isoCode") or currency.get("code") or currency.get("name") or "RUB"
|
||
).upper()
|
||
return value[:3] if len(value) >= 3 else "RUB"
|
||
|
||
|
||
def _document_channel(row: dict[str, Any], channel_names: dict[str, str] | None) -> str:
|
||
sales_channel = (
|
||
row.get("salesChannel") if isinstance(row.get("salesChannel"), dict) else {}
|
||
)
|
||
name = str(sales_channel.get("name") or "").strip()
|
||
if not name and channel_names:
|
||
name = channel_names.get(_reference_external_id(sales_channel), "")
|
||
return (name or "Не указан")[:64]
|
||
|
||
|
||
def _position_cost(position: dict[str, Any], quantity: Decimal) -> tuple[Decimal, str]:
|
||
position_cost = _decimal(position.get("cost")) / Decimal("100")
|
||
if position_cost > 0:
|
||
return (
|
||
(quantity * position_cost).quantize(
|
||
Decimal("0.01"), rounding=ROUND_HALF_UP
|
||
),
|
||
"position.cost",
|
||
)
|
||
assortment = (
|
||
position.get("assortment")
|
||
if isinstance(position.get("assortment"), dict)
|
||
else {}
|
||
)
|
||
buy_price = assortment.get("buyPrice")
|
||
unit_cost = (
|
||
_decimal(buy_price.get("value")) / Decimal("100")
|
||
if isinstance(buy_price, dict)
|
||
else Decimal("0")
|
||
)
|
||
return (quantity * unit_cost).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), (
|
||
"assortment.buyPrice" if unit_cost > 0 else "unavailable"
|
||
)
|
||
|
||
|
||
def _historical_product_from_document(
|
||
*, source, assortment: dict[str, Any], external_id: str
|
||
) -> Product:
|
||
existing = (
|
||
ProductBinding.objects.select_related("product")
|
||
.filter(
|
||
source=source,
|
||
external_id=external_id,
|
||
)
|
||
.first()
|
||
)
|
||
if existing:
|
||
return existing.product
|
||
external_sku = str(
|
||
assortment.get("article") or assortment.get("code") or ""
|
||
).strip()
|
||
product = Product.objects.create(
|
||
sku=_candidate_sku(assortment, external_id),
|
||
name=str(assortment.get("name") or external_sku or external_id)[:255],
|
||
barcode=_barcode(assortment)[:64],
|
||
description=str(assortment.get("description") or ""),
|
||
kind=_product_kind(assortment),
|
||
is_active=False,
|
||
)
|
||
metadata = _binding_metadata(assortment)
|
||
metadata["historical_from_document"] = True
|
||
ProductBinding.objects.create(
|
||
product=product,
|
||
source=source,
|
||
external_id=external_id,
|
||
external_sku=external_sku[:128],
|
||
metadata=metadata,
|
||
is_primary=True,
|
||
)
|
||
return product
|
||
|
||
|
||
@transaction.atomic
|
||
def normalize_moysklad_document(
|
||
*,
|
||
connection: IntegrationConnection,
|
||
row: dict[str, Any],
|
||
document_type: str,
|
||
channel_names: dict[str, str] | None = None,
|
||
) -> int:
|
||
from commerce.services.moysklad_projects import project_reference
|
||
|
||
source = connection.source
|
||
source_id = _external_id(row)
|
||
if not source_id:
|
||
return 0
|
||
external_id = f"{document_type}:{source_id}"
|
||
event_type = (
|
||
SaleEventType.RETURN if document_type == "salesreturn" else SaleEventType.SALE
|
||
)
|
||
source_result = ingest_source_record(
|
||
source=source,
|
||
entity_type=document_type,
|
||
external_id=source_id,
|
||
payload=row,
|
||
source_updated_at=_source_updated_at(row),
|
||
)
|
||
occurred_at = _source_updated_at({"updated": row.get("moment")}) or timezone.now()
|
||
state = row.get("state") if isinstance(row.get("state"), dict) else {}
|
||
agent = row.get("agent") if isinstance(row.get("agent"), dict) else {}
|
||
sale, _ = Sale.objects.update_or_create(
|
||
source=source,
|
||
external_id=external_id,
|
||
defaults={
|
||
"source_record": source_result.record,
|
||
"event_type": event_type,
|
||
"number": str(row.get("name") or "")[:128],
|
||
"channel": _document_channel(row, channel_names),
|
||
"customer_external_id": _reference_external_id(agent)[:255],
|
||
"customer_name": str(agent.get("name") or "")[:255],
|
||
"project": project_reference(source, row.get("project")),
|
||
"status": str(state.get("name") or "")[:128],
|
||
"counts_as_sale": bool(row.get("applicable", True)),
|
||
"currency": _document_currency(row),
|
||
"occurred_at": occurred_at,
|
||
"source_updated_at": _source_updated_at(row),
|
||
},
|
||
)
|
||
retained_line_ids = []
|
||
for index, position in enumerate(_document_positions(row), start=1):
|
||
position_id = _external_id(position) or str(index)
|
||
product_external_id = _assortment_external_id(position)
|
||
assortment = position.get("assortment") or {}
|
||
if not isinstance(assortment, dict):
|
||
assortment = {}
|
||
external_sku = str(
|
||
assortment.get("article") or assortment.get("code") or ""
|
||
).strip()
|
||
barcode = _barcode(assortment)
|
||
match = match_product(
|
||
source=source,
|
||
external_id=product_external_id,
|
||
external_sku=external_sku,
|
||
barcode=barcode,
|
||
)
|
||
matched_product = match.product
|
||
if matched_product is None and product_external_id:
|
||
matched_product = _historical_product_from_document(
|
||
source=source,
|
||
assortment=assortment,
|
||
external_id=product_external_id,
|
||
)
|
||
quantity = _decimal(position.get("quantity"))
|
||
gross, discount, vat_amount = _position_revenue(row, position)
|
||
cost_amount, cost_source = _position_cost(position, quantity)
|
||
line, _ = SaleLine.objects.update_or_create(
|
||
sale=sale,
|
||
external_id=position_id,
|
||
defaults={
|
||
"product": matched_product,
|
||
"external_product_id": product_external_id[:255],
|
||
"external_sku": external_sku[:128],
|
||
"name": str(assortment.get("name") or "")[:255],
|
||
"quantity": quantity,
|
||
"gross_amount": gross,
|
||
"discount_amount": min(discount, gross),
|
||
"cost_amount": cost_amount,
|
||
"metadata": {
|
||
"provider": "moysklad",
|
||
"cost_source": cost_source,
|
||
"vat": position.get("vat"),
|
||
"vat_enabled": bool(position.get("vatEnabled")),
|
||
"vat_amount": str(vat_amount),
|
||
"overhead": str(_money(position.get("overhead"))),
|
||
},
|
||
},
|
||
)
|
||
retained_line_ids.append(line.pk)
|
||
sale.lines.exclude(pk__in=retained_line_ids).delete()
|
||
return len(retained_line_ids)
|
||
|
||
|
||
def _sync_catalog_stage(connection, client, stats, captured_at, report):
|
||
image_rows = []
|
||
report(IntegrationSyncStage.CATALOG, 0)
|
||
for row in client.iter_assortment():
|
||
created, matched, _has_stock = normalize_moysklad_assortment_row(
|
||
connection=connection,
|
||
row=row,
|
||
captured_at=captured_at,
|
||
)
|
||
stats.processed += 1
|
||
stats.created_products += int(created)
|
||
stats.matched_products += int(matched)
|
||
image_rows.append(
|
||
{
|
||
"id": row.get("id"),
|
||
"meta": row.get("meta") or {},
|
||
"images": row.get("images") or {},
|
||
}
|
||
)
|
||
if stats.processed % 25 == 0:
|
||
report(IntegrationSyncStage.CATALOG, stats.processed)
|
||
report(IntegrationSyncStage.CATALOG, stats.processed, stats.processed)
|
||
return image_rows
|
||
|
||
|
||
def _image_failure_logger(*, connection, row, stats, run_id, task_id, stage_started):
|
||
def image_failed(remote, error):
|
||
stats.image_errors += 1
|
||
logger.warning(
|
||
"MoySklad product image could not be synchronized",
|
||
extra={
|
||
"event": "commerce.product.image_sync_failed",
|
||
"run_id": run_id,
|
||
"task_id": task_id,
|
||
"connection_id": str(connection.pk),
|
||
"provider": connection.provider,
|
||
"stage": IntegrationSyncStage.IMAGES,
|
||
"external_product_id": str(row.get("id") or ""),
|
||
"external_image_id": str(getattr(remote, "external_id", "") or ""),
|
||
"error_type": type(error).__name__,
|
||
"failure_reason": str(error)[:500],
|
||
"processed": stats.processed,
|
||
"created_products": stats.created_products,
|
||
"matched_products": stats.matched_products,
|
||
"stock_snapshots": stats.stock_snapshots,
|
||
"images": stats.images,
|
||
"image_errors": stats.image_errors,
|
||
"sales": stats.sales,
|
||
"duration_ms": round((monotonic() - stage_started) * 1000, 2),
|
||
},
|
||
)
|
||
|
||
return image_failed
|
||
|
||
|
||
def _sync_images_stage(
|
||
connection,
|
||
client,
|
||
image_rows,
|
||
stats,
|
||
report,
|
||
*,
|
||
run_id,
|
||
task_id,
|
||
):
|
||
stage_started = monotonic()
|
||
report(IntegrationSyncStage.IMAGES, 0, len(image_rows))
|
||
for image_index, row in enumerate(image_rows, start=1):
|
||
try:
|
||
stats.images += sync_moysklad_product_images(
|
||
connection=connection,
|
||
row=row,
|
||
client=client,
|
||
error_callback=_image_failure_logger(
|
||
connection=connection,
|
||
row=row,
|
||
stats=stats,
|
||
run_id=run_id,
|
||
task_id=task_id,
|
||
stage_started=stage_started,
|
||
),
|
||
)
|
||
except SoftTimeLimitExceeded:
|
||
raise
|
||
except Exception as error: # noqa: BLE001 - one product must not abort facts
|
||
stats.image_errors += 1
|
||
logger.warning(
|
||
"MoySklad product images could not be synchronized",
|
||
extra={
|
||
"event": "commerce.product.images_sync_failed",
|
||
"run_id": run_id,
|
||
"task_id": task_id,
|
||
"connection_id": str(connection.pk),
|
||
"provider": connection.provider,
|
||
"stage": IntegrationSyncStage.IMAGES,
|
||
"external_product_id": str(row.get("id") or ""),
|
||
"entity_type": str((row.get("meta") or {}).get("type") or ""),
|
||
"error_type": type(error).__name__,
|
||
"failure_reason": str(error)[:500],
|
||
"processed": stats.processed,
|
||
"created_products": stats.created_products,
|
||
"matched_products": stats.matched_products,
|
||
"stock_snapshots": stats.stock_snapshots,
|
||
"images": stats.images,
|
||
"image_errors": stats.image_errors,
|
||
"sales": stats.sales,
|
||
"duration_ms": round(
|
||
(monotonic() - stage_started) * 1000,
|
||
2,
|
||
),
|
||
},
|
||
)
|
||
if image_index % 10 == 0 or image_index == len(image_rows):
|
||
report(IntegrationSyncStage.IMAGES, image_index, len(image_rows))
|
||
|
||
|
||
def _sync_variant_links_stage(connection, report):
|
||
total = ProductBinding.objects.filter(source=connection.source).count()
|
||
report(IntegrationSyncStage.VARIANT_LINKS, 0, total)
|
||
reconcile_moysklad_variant_parents(
|
||
connection.source,
|
||
progress_callback=lambda processed, stage_total: report(
|
||
IntegrationSyncStage.VARIANT_LINKS,
|
||
processed,
|
||
stage_total,
|
||
),
|
||
)
|
||
report(IntegrationSyncStage.VARIANT_LINKS, total, total)
|
||
|
||
|
||
def _sync_stocks_stage(connection, client, stats, captured_at, report):
|
||
report(IntegrationSyncStage.STOCKS, 0)
|
||
progress = {"rows": 0}
|
||
|
||
def stock_progress(processed_rows, synchronized):
|
||
progress["rows"] = processed_rows
|
||
stats.stock_snapshots = synchronized
|
||
report(IntegrationSyncStage.STOCKS, processed_rows)
|
||
|
||
stats.stock_snapshots = sync_moysklad_stock_rows(
|
||
connection=connection,
|
||
rows=client.iter_stock_by_store(),
|
||
captured_at=captured_at,
|
||
progress_callback=stock_progress,
|
||
)
|
||
report(IntegrationSyncStage.STOCKS, progress["rows"], progress["rows"])
|
||
StockSnapshot.objects.filter(
|
||
source=connection.source,
|
||
warehouse_external_id="all",
|
||
).delete()
|
||
|
||
|
||
def _sync_documents_stage(connection, client, stats, report):
|
||
channel_names = client.sales_channel_names()
|
||
for document_type, stage in (
|
||
("demand", IntegrationSyncStage.SALES),
|
||
("salesreturn", IntegrationSyncStage.RETURNS),
|
||
):
|
||
stage_processed = 0
|
||
report(stage, 0)
|
||
for row in client.iter_documents(document_type):
|
||
stats.sale_lines += normalize_moysklad_document(
|
||
connection=connection,
|
||
row=row,
|
||
document_type=document_type,
|
||
channel_names=channel_names,
|
||
)
|
||
stats.sales += 1
|
||
stage_processed += 1
|
||
if stage_processed % 25 == 0:
|
||
report(stage, stage_processed)
|
||
report(stage, stage_processed, stage_processed)
|
||
|
||
|
||
def sync_moysklad_connection(
|
||
connection: IntegrationConnection,
|
||
*,
|
||
progress_callback: Callable[..., None] | None = None,
|
||
captured_at: datetime | None = None,
|
||
run_id: str = "",
|
||
task_id: str = "",
|
||
) -> MoySkladSyncStats:
|
||
from commerce.services.moysklad_projects import (
|
||
sync_project_catalog,
|
||
sync_project_payments,
|
||
)
|
||
|
||
credentials = decrypt_credentials(connection)
|
||
stats = MoySkladSyncStats()
|
||
captured_at = captured_at or timezone.now()
|
||
|
||
def report(stage, processed, total=0):
|
||
if progress_callback:
|
||
progress_callback(
|
||
stats,
|
||
stage=stage,
|
||
stage_processed=processed,
|
||
stage_total=total,
|
||
)
|
||
|
||
with MoySkladClient(
|
||
auth_type=connection.auth_type, credentials=credentials
|
||
) as client:
|
||
report(IntegrationSyncStage.PROJECTS, 0)
|
||
stats.projects = sync_project_catalog(connection.source, client)
|
||
report(IntegrationSyncStage.PROJECTS, stats.projects, stats.projects)
|
||
report(IntegrationSyncStage.PROJECT_PAYMENTS, 0)
|
||
stats.project_payments = sync_project_payments(
|
||
connection.source,
|
||
client,
|
||
progress=lambda count: report(IntegrationSyncStage.PROJECT_PAYMENTS, count),
|
||
)
|
||
report(
|
||
IntegrationSyncStage.PROJECT_PAYMENTS,
|
||
stats.project_payments,
|
||
stats.project_payments,
|
||
)
|
||
image_rows = _sync_catalog_stage(
|
||
connection,
|
||
client,
|
||
stats,
|
||
captured_at,
|
||
report,
|
||
)
|
||
_sync_images_stage(
|
||
connection,
|
||
client,
|
||
image_rows,
|
||
stats,
|
||
report,
|
||
run_id=run_id,
|
||
task_id=task_id,
|
||
)
|
||
_sync_variant_links_stage(connection, report)
|
||
_sync_stocks_stage(connection, client, stats, captured_at, report)
|
||
_sync_documents_stage(connection, client, stats, report)
|
||
report(IntegrationSyncStage.FINALIZING, 0, 1)
|
||
report(IntegrationSyncStage.FINALIZING, 1, 1)
|
||
return stats
|
||
|
||
|
||
def test_moysklad_connection(connection: IntegrationConnection) -> dict[str, str]:
|
||
credentials = decrypt_credentials(connection)
|
||
with MoySkladClient(
|
||
auth_type=connection.auth_type, credentials=credentials
|
||
) as client:
|
||
employee = client.employee_context()
|
||
return {
|
||
"account_id": str(employee.get("accountId") or ""),
|
||
"account_name": str(
|
||
employee.get("fullName")
|
||
or employee.get("name")
|
||
or employee.get("uid")
|
||
or ""
|
||
),
|
||
}
|