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 "" ), }