From bd515fd6d5017a5014a925da07505cb3bdc0057b Mon Sep 17 00:00:00 2001 From: lizerk Date: Tue, 18 Aug 2026 18:48:00 +0300 Subject: [PATCH] Harden commerce import reliability --- .env.example | 4 +- backend/Dockerfile | 2 +- backend/api/v1/commerce/serializers.py | 4 + backend/commerce/admin.py | 13 +- .../commands/reconcile_commerce_sync_runs.py | 43 ++ backend/commerce/media.py | 299 ++++++--- ...1_integrationsyncrun_progress_heartbeat.py | 70 ++ backend/commerce/models/__init__.py | 2 + backend/commerce/models/source.py | 34 +- backend/commerce/realtime.py | 4 + backend/commerce/services/__init__.py | 2 + backend/commerce/services/moysklad_sync.py | 484 ++++++++++---- backend/commerce/services/sync_runs.py | 119 ++++ backend/commerce/services/wildberries_sync.py | 21 +- backend/commerce/tasks/__init__.py | 602 +++++++++++++----- backend/eusrr_backend/celery.py | 32 +- backend/eusrr_backend/logging_utils.py | 14 +- backend/eusrr_backend/settings.py | 21 +- backend/tests/commerce/test_connections.py | 386 ++++++++++- backend/tests/commerce/test_wildberries.py | 4 +- backend/tests/test_celery_schedule.py | 11 + backend/tests/test_logging.py | 54 ++ docs/guides/COMMERCE.md | 44 +- .../commerce/ProductConnectionsContent.tsx | 20 + frontend/src/types/commerce.ts | 12 + 25 files changed, 1870 insertions(+), 431 deletions(-) create mode 100644 backend/commerce/management/commands/reconcile_commerce_sync_runs.py create mode 100644 backend/commerce/migrations/0011_integrationsyncrun_progress_heartbeat.py create mode 100644 backend/commerce/services/sync_runs.py diff --git a/.env.example b/.env.example index 619f054e..7ee06e05 100644 --- a/.env.example +++ b/.env.example @@ -98,6 +98,6 @@ PAYROLL_SIMPLE_ADMIN_ACCESS=true # Generate: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" COMMERCE_CREDENTIAL_KEY= # A queued import should normally be picked up within seconds. Celery Beat marks -# lost queue messages as failed after this timeout; running imports use a longer limit. +# lost queue messages as failed after this timeout; running imports use heartbeat. COMMERCE_QUEUED_STALE_SECONDS=120 -COMMERCE_SYNCING_STALE_MINUTES=45 +COMMERCE_RUNNING_STALE_SECONDS=600 diff --git a/backend/Dockerfile b/backend/Dockerfile index 914bd43f..91b8e5fb 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -4,7 +4,7 @@ FROM python:3.13-slim WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 -RUN apt-get update && apt-get install -y --no-install-recommends build-essential libpq-dev curl ffmpeg && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y --no-install-recommends build-essential libpq-dev curl ffmpeg imagemagick && rm -rf /var/lib/apt/lists/* COPY backend/requirements.txt /app/requirements.txt RUN pip install --no-cache-dir -r /app/requirements.txt diff --git a/backend/api/v1/commerce/serializers.py b/backend/api/v1/commerce/serializers.py index 3b5ffddd..8014d69f 100644 --- a/backend/api/v1/commerce/serializers.py +++ b/backend/api/v1/commerce/serializers.py @@ -42,6 +42,10 @@ class IntegrationSyncRunSerializer(serializers.ModelSerializer): "connection", "status", "task_id", + "stage", + "stage_processed", + "stage_total", + "last_progress_at", "processed", "created_products", "matched_products", diff --git a/backend/commerce/admin.py b/backend/commerce/admin.py index ebc95d15..6959f721 100644 --- a/backend/commerce/admin.py +++ b/backend/commerce/admin.py @@ -24,8 +24,17 @@ class IntegrationConnectionAdmin(admin.ModelAdmin): @admin.register(IntegrationSyncRun) class IntegrationSyncRunAdmin(admin.ModelAdmin): - list_display = ["connection", "status", "processed", "created_at", "finished_at"] - list_filter = ["status", "connection__provider"] + list_display = [ + "connection", + "status", + "stage", + "stage_processed", + "stage_total", + "last_progress_at", + "created_at", + "finished_at", + ] + list_filter = ["status", "stage", "connection__provider"] readonly_fields = [field.name for field in IntegrationSyncRun._meta.fields] diff --git a/backend/commerce/management/commands/reconcile_commerce_sync_runs.py b/backend/commerce/management/commands/reconcile_commerce_sync_runs.py new file mode 100644 index 00000000..f82e37e1 --- /dev/null +++ b/backend/commerce/management/commands/reconcile_commerce_sync_runs.py @@ -0,0 +1,43 @@ +from datetime import timedelta + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError + +from commerce.services.sync_runs import reconcile_stale_sync_runs + + +class Command(BaseCommand): + help = ( + "Переводит устаревшие queued/running импорты commerce в error, " + "не удаляя историю и не запуская их повторно." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--queued-seconds", + type=int, + default=settings.COMMERCE_QUEUED_STALE_SECONDS, + ) + parser.add_argument( + "--running-seconds", + type=int, + default=settings.COMMERCE_RUNNING_STALE_SECONDS, + ) + parser.add_argument("--dry-run", action="store_true") + + def handle(self, *args, **options): + if options["queued_seconds"] <= 0 or options["running_seconds"] <= 0: + raise CommandError("Таймауты должны быть положительными числами.") + result = reconcile_stale_sync_runs( + queued_timeout=timedelta(seconds=options["queued_seconds"]), + running_timeout=timedelta(seconds=options["running_seconds"]), + dry_run=options["dry_run"], + ) + mode = "Найдено" if options["dry_run"] else "Восстановлено" + self.stdout.write( + self.style.SUCCESS( + f"{mode}: queued={len(result.queued_run_ids)}, " + f"running={len(result.running_run_ids)}, " + f"connections={len(set(result.connection_ids))}." + ) + ) diff --git a/backend/commerce/media.py b/backend/commerce/media.py index 386b7784..8bbbb497 100644 --- a/backend/commerce/media.py +++ b/backend/commerce/media.py @@ -1,6 +1,14 @@ +from __future__ import annotations + import hashlib import io +import logging +import math from pathlib import Path +import shutil +import subprocess +import tempfile +from typing import TYPE_CHECKING, Callable import warnings from dateutil import parser as date_parser @@ -9,15 +17,25 @@ from django.db import transaction from django.utils import timezone from PIL import Image, ImageOps -from commerce.integrations.moysklad import MoySkladClient from commerce.models import IntegrationConnection, ProductImage +if TYPE_CHECKING: + from commerce.integrations.moysklad import MoySkladClient + + +logger = logging.getLogger(__name__) PRODUCT_THUMBNAIL_SIZE = (480, 480) MAX_PRODUCT_IMAGE_PIXELS = 40_000_000 +MAX_PRODUCT_IMAGE_DIMENSION = 4096 +OVERSIZED_IMAGE_TIMEOUT_SECONDS = 90 IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"} +class OversizedProductImage(ValueError): + pass + + def _source_updated_at(value: str): if not value: return None @@ -30,15 +48,98 @@ def _source_updated_at(value: str): return parsed -def _validated_image_and_thumbnail(content: bytes) -> tuple[bytes, str]: +def _inspect_product_image(content: bytes) -> str: with warnings.catch_warnings(): warnings.simplefilter("error", Image.DecompressionBombWarning) - with Image.open(io.BytesIO(content)) as probe: - if probe.width * probe.height > MAX_PRODUCT_IMAGE_PIXELS: - raise ValueError("Изображение превышает лимит 40 мегапикселей.") - image_format = (probe.format or "").upper() - probe.verify() - with Image.open(io.BytesIO(content)) as source: + try: + with Image.open(io.BytesIO(content)) as probe: + if probe.width * probe.height > MAX_PRODUCT_IMAGE_PIXELS: + raise OversizedProductImage( + "Изображение превышает безопасный лимит разрешения." + ) + image_format = (probe.format or "").upper() + probe.verify() + except (Image.DecompressionBombError, Image.DecompressionBombWarning) as exc: + raise OversizedProductImage( + "Изображение превышает безопасный лимит разрешения." + ) from exc + return image_format + + +def _resize_oversized_image(content: bytes) -> bytes: + magick = shutil.which("magick") or shutil.which("convert") + if magick is None: + raise RuntimeError( + "Для безопасного уменьшения большого изображения нужен ImageMagick." + ) + max_edge = min( + MAX_PRODUCT_IMAGE_DIMENSION, + math.isqrt(MAX_PRODUCT_IMAGE_PIXELS), + ) + with tempfile.TemporaryDirectory(prefix="commerce-image-") as temporary_dir: + input_path = Path(temporary_dir) / "source-image" + output_path = Path(temporary_dir) / "normalized.webp" + input_path.write_bytes(content) + try: + result = subprocess.run( + [ + magick, + "-limit", + "thread", + "1", + "-limit", + "memory", + "128MiB", + "-limit", + "map", + "256MiB", + "-limit", + "disk", + "2GiB", + "-limit", + "time", + str(OVERSIZED_IMAGE_TIMEOUT_SECONDS), + "-define", + f"jpeg:size={max_edge}x{max_edge}", + f"{input_path}[0]", + "-auto-orient", + "-thumbnail", + f"{max_edge}x{max_edge}>", + "-strip", + "-quality", + "84", + str(output_path), + ], + capture_output=True, + check=False, + timeout=OVERSIZED_IMAGE_TIMEOUT_SECONDS + 5, + ) + except subprocess.TimeoutExpired as exc: + raise ValueError( + "Не удалось уменьшить большое изображение за безопасное время." + ) from exc + if result.returncode != 0 or not output_path.exists(): + detail = result.stderr.decode("utf-8", errors="replace").strip()[:500] + raise ValueError( + "Не удалось безопасно уменьшить большое изображение." + + (f" ImageMagick: {detail}" if detail else "") + ) + resized = output_path.read_bytes() + _inspect_product_image(resized) + return resized + + +def _prepare_product_image(content: bytes) -> tuple[bytes, bytes, str, bool]: + try: + image_format = _inspect_product_image(content) + stored_content = content + was_resized = False + except OversizedProductImage: + stored_content = _resize_oversized_image(content) + image_format = "WEBP" + was_resized = True + + with Image.open(io.BytesIO(stored_content)) as source: source.seek(0) image = ImageOps.exif_transpose(source) if image.mode not in {"RGB", "RGBA"}: @@ -46,14 +147,14 @@ def _validated_image_and_thumbnail(content: bytes) -> tuple[bytes, str]: image.thumbnail(PRODUCT_THUMBNAIL_SIZE, Image.Resampling.LANCZOS) output = io.BytesIO() image.save(output, format="WEBP", quality=84, method=5) - suffix = { - "JPEG": ".jpg", - "PNG": ".png", - "WEBP": ".webp", - "GIF": ".gif", - "BMP": ".bmp", - }.get(image_format, ".img") - return output.getvalue(), suffix + suffix = { + "JPEG": ".jpg", + "PNG": ".png", + "WEBP": ".webp", + "GIF": ".gif", + "BMP": ".bmp", + }.get(image_format, ".img") + return stored_content, output.getvalue(), suffix, was_resized def _storage_filename( @@ -61,9 +162,11 @@ def _storage_filename( source_name: str, content: bytes, detected_suffix: str, + *, + force_detected_suffix: bool = False, ) -> str: suffix = Path(source_name).suffix.lower() - if suffix not in IMAGE_SUFFIXES: + if force_detected_suffix or suffix not in IMAGE_SUFFIXES: suffix = detected_suffix digest = hashlib.sha256(content).hexdigest()[:16] return f"{external_id}-{digest}{suffix}" @@ -81,6 +184,7 @@ def sync_moysklad_product_images( connection: IntegrationConnection, row: dict, client: MoySkladClient, + error_callback: Callable[[object, Exception], None] | None = None, ) -> int: external_product_id = str(row.get("id") or "").strip() if not external_product_id: @@ -107,83 +211,110 @@ def sync_moysklad_product_images( else client.product_images(entity_type, external_product_id) ) retained_ids = [] + synchronized = 0 for position, remote in enumerate(remote_images): - source_updated_at = _source_updated_at(remote.updated) - image = ProductImage.objects.filter( - product=product, - source=connection.source, - external_id=remote.external_id, - ).first() - unchanged = bool( - image - and image.source_size == remote.size - and image.source_updated_at == source_updated_at - and image.file - and image.file.storage.exists(image.file.name) - and image.thumbnail - and image.thumbnail.storage.exists(image.thumbnail.name) - ) - if unchanged: + image = None + try: + source_updated_at = _source_updated_at(remote.updated) + image = ProductImage.objects.filter( + product=product, + source=connection.source, + external_id=remote.external_id, + ).first() + unchanged = bool( + image + and image.source_size == remote.size + and image.source_updated_at == source_updated_at + and image.file + and image.file.storage.exists(image.file.name) + and image.thumbnail + and image.thumbnail.storage.exists(image.thumbnail.name) + ) + if unchanged: + image.product = product + image.filename = remote.filename + image.title = remote.title + image.position = position + image.is_primary = position == 0 + image.save( + update_fields=[ + "product", + "filename", + "title", + "position", + "is_primary", + "updated_at", + ] + ) + retained_ids.append(remote.external_id) + synchronized += 1 + continue + + source_content, _content_type = client.download_image(remote.download_href) + content, thumbnail, detected_suffix, was_resized = _prepare_product_image( + source_content + ) + if was_resized: + logger.info( + "Oversized commerce product image resized", + extra={ + "event": "commerce.product_image.resized", + "connection_id": str(connection.pk), + "product_id": str(product.pk), + "external_product_id": external_product_id, + "external_image_id": remote.external_id, + "source_bytes": len(source_content), + "stored_bytes": len(content), + }, + ) + storage_name = _storage_filename( + remote.external_id, + remote.filename, + content, + detected_suffix, + force_detected_suffix=was_resized, + ) + if image is None: + image = ProductImage( + source=connection.source, + external_id=remote.external_id, + ) + old_names = { + "file": image.file.name if image.file else "", + "thumbnail": image.thumbnail.name if image.thumbnail else "", + } image.product = product image.filename = remote.filename image.title = remote.title + image.source_size = remote.size + image.source_updated_at = source_updated_at image.position = position image.is_primary = position == 0 - image.save( - update_fields=[ - "product", - "filename", - "title", - "position", - "is_primary", - "updated_at", - ] + image.file.save(storage_name, ContentFile(content), save=False) + image.thumbnail.save( + f"{remote.external_id}-{hashlib.sha256(content).hexdigest()[:16]}.webp", + ContentFile(thumbnail), + save=False, ) - retained_ids.append(remote.external_id) - continue - - content, _content_type = client.download_image(remote.download_href) - thumbnail, detected_suffix = _validated_image_and_thumbnail(content) - storage_name = _storage_filename( - remote.external_id, - remote.filename, - content, - detected_suffix, - ) - if image is None: - image = ProductImage( - source=connection.source, - external_id=remote.external_id, - ) - old_names = { - "file": image.file.name if image.file else "", - "thumbnail": image.thumbnail.name if image.thumbnail else "", - } - image.product = product - image.filename = remote.filename - image.title = remote.title - image.source_size = remote.size - image.source_updated_at = source_updated_at - image.position = position - image.is_primary = position == 0 - image.file.save(storage_name, ContentFile(content), save=False) - image.thumbnail.save( - f"{remote.external_id}-{hashlib.sha256(content).hexdigest()[:16]}.webp", - ContentFile(thumbnail), - save=False, - ) - with transaction.atomic(): - image.save() - transaction.on_commit( - lambda current=image, previous=old_names: _delete_replaced_files( - current, - previous, + with transaction.atomic(): + image.save() + transaction.on_commit( + lambda current=image, previous=old_names: _delete_replaced_files( + current, + previous, + ) ) - ) - retained_ids.append(remote.external_id) + retained_ids.append(remote.external_id) + synchronized += 1 + except Exception as error: + if image is not None and image.pk: + retained_ids.append(remote.external_id) + if error_callback is None: + raise + error_callback(remote, error) product.images.filter(source=connection.source).exclude( external_id__in=retained_ids ).delete() - return len(retained_ids) + return synchronized diff --git a/backend/commerce/migrations/0011_integrationsyncrun_progress_heartbeat.py b/backend/commerce/migrations/0011_integrationsyncrun_progress_heartbeat.py new file mode 100644 index 00000000..dc078b4c --- /dev/null +++ b/backend/commerce/migrations/0011_integrationsyncrun_progress_heartbeat.py @@ -0,0 +1,70 @@ +# Generated by Django 5.2.4 on 2026-08-18 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("commerce", "0010_wildberries_and_sync_costs"), + ] + + operations = [ + migrations.AddField( + model_name="integrationsyncrun", + name="last_progress_at", + field=models.DateTimeField( + blank=True, + null=True, + verbose_name="Последний прогресс", + ), + ), + migrations.AddField( + model_name="integrationsyncrun", + name="stage", + field=models.CharField( + blank=True, + choices=[ + ("catalog", "Каталог"), + ("images", "Изображения"), + ("variant_links", "Связи вариантов"), + ("stocks", "Остатки"), + ("sales", "Продажи"), + ("returns", "Возвраты"), + ("finalizing", "Завершение"), + ], + default="", + max_length=32, + verbose_name="Этап", + ), + ), + migrations.AddField( + model_name="integrationsyncrun", + name="stage_processed", + field=models.PositiveIntegerField( + default=0, + verbose_name="Обработано на этапе", + ), + ), + migrations.AddField( + model_name="integrationsyncrun", + name="stage_total", + field=models.PositiveIntegerField( + default=0, + verbose_name="Всего на этапе", + ), + ), + migrations.AddIndex( + model_name="integrationsyncrun", + index=models.Index( + fields=["status", "created_at"], + name="commerce_sync_status_created", + ), + ), + migrations.AddIndex( + model_name="integrationsyncrun", + index=models.Index( + fields=["status", "last_progress_at"], + name="commerce_sync_status_progress", + ), + ), + ] diff --git a/backend/commerce/models/__init__.py b/backend/commerce/models/__init__.py index d9b804f5..1b1fbe89 100644 --- a/backend/commerce/models/__init__.py +++ b/backend/commerce/models/__init__.py @@ -5,6 +5,7 @@ from .source import ( IntegrationConnection, IntegrationProvider, IntegrationStatus, + IntegrationSyncStage, IntegrationSyncRun, ProductBinding, SourceRecord, @@ -18,6 +19,7 @@ __all__ = [ "IntegrationConnection", "IntegrationProvider", "IntegrationStatus", + "IntegrationSyncStage", "IntegrationSyncRun", "Product", "ProductBinding", diff --git a/backend/commerce/models/source.py b/backend/commerce/models/source.py index bff5caff..30919be7 100644 --- a/backend/commerce/models/source.py +++ b/backend/commerce/models/source.py @@ -126,6 +126,16 @@ class IntegrationStatus(models.TextChoices): ERROR = "error", "Ошибка" +class IntegrationSyncStage(models.TextChoices): + CATALOG = "catalog", "Каталог" + IMAGES = "images", "Изображения" + VARIANT_LINKS = "variant_links", "Связи вариантов" + STOCKS = "stocks", "Остатки" + SALES = "sales", "Продажи" + RETURNS = "returns", "Возвраты" + FINALIZING = "finalizing", "Завершение" + + class IntegrationConnection(models.Model): """Encrypted credentials and synchronization state for an external source.""" @@ -216,6 +226,20 @@ class IntegrationSyncRun(models.Model): default="queued", ) task_id = models.CharField("ID фоновой задачи", max_length=255, blank=True) + stage = models.CharField( + "Этап", + max_length=32, + choices=IntegrationSyncStage.choices, + blank=True, + default="", + ) + stage_processed = models.PositiveIntegerField("Обработано на этапе", default=0) + stage_total = models.PositiveIntegerField("Всего на этапе", default=0) + last_progress_at = models.DateTimeField( + "Последний прогресс", + null=True, + blank=True, + ) processed = models.PositiveIntegerField("Обработано", default=0) created_products = models.PositiveIntegerField("Создано товаров", default=0) matched_products = models.PositiveIntegerField("Сопоставлено", default=0) @@ -237,7 +261,15 @@ class IntegrationSyncRun(models.Model): models.Index( fields=["connection", "-created_at"], name="commerce_sync_connection_date", - ) + ), + models.Index( + fields=["status", "created_at"], + name="commerce_sync_status_created", + ), + models.Index( + fields=["status", "last_progress_at"], + name="commerce_sync_status_progress", + ), ] def __str__(self): diff --git a/backend/commerce/realtime.py b/backend/commerce/realtime.py index 20d7d330..170e16a5 100644 --- a/backend/commerce/realtime.py +++ b/backend/commerce/realtime.py @@ -19,6 +19,10 @@ def _run_payload(run: IntegrationSyncRun | None): "connection": str(run.connection_id), "status": run.status, "task_id": run.task_id, + "stage": run.stage, + "stage_processed": run.stage_processed, + "stage_total": run.stage_total, + "last_progress_at": _iso(run.last_progress_at), "processed": run.processed, "created_products": run.created_products, "matched_products": run.matched_products, diff --git a/backend/commerce/services/__init__.py b/backend/commerce/services/__init__.py index 139fcb03..5cdf16c0 100644 --- a/backend/commerce/services/__init__.py +++ b/backend/commerce/services/__init__.py @@ -5,6 +5,7 @@ from .matching import ProductMatch, match_product from .metrics import get_product_metrics from .moysklad_sync import sync_moysklad_connection, test_moysklad_connection from .product_merge import ProductMergeResult, merge_products +from .sync_runs import reconcile_stale_sync_runs from .wildberries_sync import ( sync_wildberries_connection, test_wildberries_connection, @@ -18,6 +19,7 @@ __all__ = [ "ingest_source_record", "match_product", "merge_products", + "reconcile_stale_sync_runs", "sync_moysklad_connection", "test_moysklad_connection", "sync_wildberries_connection", diff --git a/backend/commerce/services/moysklad_sync.py b/backend/commerce/services/moysklad_sync.py index b25e526a..68e9e2c1 100644 --- a/backend/commerce/services/moysklad_sync.py +++ b/backend/commerce/services/moysklad_sync.py @@ -3,6 +3,7 @@ import logging 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 @@ -14,6 +15,7 @@ from commerce.integrations.moysklad import MoySkladClient from commerce.media import sync_moysklad_product_images from commerce.models import ( IntegrationConnection, + IntegrationSyncStage, Product, ProductBinding, ProductKind, @@ -97,12 +99,8 @@ def _position_revenue( 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 - ) + 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 @@ -270,10 +268,13 @@ def normalize_moysklad_assortment_row( external_sku=external_sku, barcode=barcode, ) - if candidate.product and not ProductBinding.objects.filter( - product=candidate.product, - source=source, - ).exists(): + if ( + candidate.product + and not ProductBinding.objects.filter( + product=candidate.product, + source=source, + ).exists() + ): product = candidate.product created = False matched = product is not None @@ -314,7 +315,12 @@ def normalize_moysklad_assortment_row( return created, matched, has_stock and product.kind != ProductKind.SERVICE -def reconcile_moysklad_variant_parents(source) -> int: +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 @@ -322,55 +328,143 @@ def reconcile_moysklad_variant_parents(source) -> int: "product" ) } - for binding in bindings.values(): + 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 not parent_binding or parent_binding.product_id == binding.product_id: - continue - if binding.product.parent_id != parent_binding.product_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.save(update_fields=["parent", "updated_at"]) + 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 -@transaction.atomic def normalize_moysklad_stock_row( *, connection: IntegrationConnection, row: dict[str, Any], captured_at: datetime ) -> int: - source = connection.source - external_product_id = _external_id(row) - binding = ProductBinding.objects.filter( - source=source, - external_id=external_product_id, - ).first() - if not binding: + 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 - created = 0 - for store_stock in row.get("stockByStore") or []: - if not isinstance(store_stock, dict): - continue - warehouse_external_id = _reference_external_id(store_stock) - if not warehouse_external_id: - continue - StockSnapshot.objects.update_or_create( - product=binding.product, - source=source, - warehouse_external_id=warehouse_external_id, - captured_at=captured_at, - defaults={ - "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, - }, - }, + 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", + ], ) - created += 1 - return created + 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]]: @@ -389,21 +483,14 @@ def _assortment_external_id(position: dict[str, Any]) -> str: 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 {} - ) + 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" + 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: +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 {} ) @@ -433,9 +520,7 @@ def _position_cost(position: dict[str, Any], quantity: Decimal) -> tuple[Decimal if isinstance(buy_price, dict) else Decimal("0") ) - return (quantity * unit_cost).quantize( - Decimal("0.01"), rounding=ROUND_HALF_UP - ), ( + return (quantity * unit_cost).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), ( "assortment.buyPrice" if unit_cost > 0 else "unavailable" ) @@ -443,10 +528,14 @@ def _position_cost(position: dict[str, Any], quantity: Decimal) -> tuple[Decimal 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() + existing = ( + ProductBinding.objects.select_related("product") + .filter( + source=source, + external_id=external_id, + ) + .first() + ) if existing: return existing.product external_sku = str( @@ -567,72 +656,223 @@ def normalize_moysklad_document( 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 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[[MoySkladSyncStats], None] | None = None, + progress_callback: Callable[..., None] | None = None, + captured_at: datetime | None = None, + run_id: str = "", + task_id: str = "", ) -> MoySkladSyncStats: credentials = decrypt_credentials(connection) stats = MoySkladSyncStats() - captured_at = timezone.now() + 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: - 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) - try: - stats.images += sync_moysklad_product_images( - connection=connection, - row=row, - client=client, - ) - except Exception as error: # noqa: BLE001 - one image must not abort facts - stats.image_errors += 1 - logger.warning( - "MoySklad product images could not be synchronized", - extra={ - "event": "commerce.product.images_sync_failed", - "connection_id": str(connection.pk), - "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], - }, - ) - if progress_callback and stats.processed % 25 == 0: - progress_callback(stats) - reconcile_moysklad_variant_parents(connection.source) - for stock_row in client.iter_stock_by_store(): - stats.stock_snapshots += normalize_moysklad_stock_row( - connection=connection, - row=stock_row, - captured_at=captured_at, - ) - StockSnapshot.objects.filter( - source=connection.source, - warehouse_external_id="all", - ).delete() - channel_names = client.sales_channel_names() - for document_type in ("demand", "salesreturn"): - 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 - if progress_callback and stats.sales % 25 == 0: - progress_callback(stats) - if progress_callback: - progress_callback(stats) + 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 diff --git a/backend/commerce/services/sync_runs.py b/backend/commerce/services/sync_runs.py new file mode 100644 index 00000000..1fb3eeb6 --- /dev/null +++ b/backend/commerce/services/sync_runs.py @@ -0,0 +1,119 @@ +from dataclasses import dataclass, field +from datetime import timedelta + +from django.db import transaction +from django.db.models import Q +from django.utils import timezone + +from commerce.models import IntegrationStatus, IntegrationSyncRun + + +STALE_QUEUED_ERROR = "Задача не была получена обработчиком и снята с очереди." +STALE_RUNNING_ERROR = ( + "Синхронизация перестала передавать прогресс и была завершена как ошибочная." +) + + +@dataclass +class SyncRunReconciliationResult: + queued_run_ids: list[str] = field(default_factory=list) + running_run_ids: list[str] = field(default_factory=list) + connection_ids: list[str] = field(default_factory=list) + + @property + def recovered(self) -> int: + return len(self.queued_run_ids) + len(self.running_run_ids) + + +def reconcile_stale_sync_runs( + *, + queued_timeout: timedelta, + running_timeout: timedelta, + now=None, + dry_run: bool = False, +) -> SyncRunReconciliationResult: + """Mark stale runs as errors without deleting history or silently retrying them.""" + + now = now or timezone.now() + queued_cutoff = now - queued_timeout + running_cutoff = now - running_timeout + queued_ids = list( + IntegrationSyncRun.objects.filter( + status="queued", + created_at__lte=queued_cutoff, + ).values_list("pk", flat=True) + ) + running_ids = list( + IntegrationSyncRun.objects.filter(status="running") + .filter( + Q(last_progress_at__lte=running_cutoff) + | Q(last_progress_at__isnull=True, started_at__lte=running_cutoff) + | Q( + last_progress_at__isnull=True, + started_at__isnull=True, + created_at__lte=running_cutoff, + ) + ) + .values_list("pk", flat=True) + ) + candidates = [ + *( + (run_id, "queued", queued_cutoff, STALE_QUEUED_ERROR, "queued") + for run_id in queued_ids + ), + *( + (run_id, "running", running_cutoff, STALE_RUNNING_ERROR, "running") + for run_id in running_ids + ), + ] + result = SyncRunReconciliationResult() + for run_id, expected_status, cutoff, message, target in candidates: + with transaction.atomic(): + run = ( + IntegrationSyncRun.objects.select_for_update() + .select_related("connection") + .get(pk=run_id) + ) + heartbeat = run.created_at + if expected_status == "running": + heartbeat = run.last_progress_at or run.started_at or run.created_at + if run.status != expected_status or heartbeat > cutoff: + continue + getattr(result, f"{target}_run_ids").append(str(run.pk)) + connection = run.connection + owns_connection = ( + connection.task_id == run.task_id + and connection.status + in { + IntegrationStatus.QUEUED, + IntegrationStatus.SYNCING, + } + ) + if owns_connection: + result.connection_ids.append(str(connection.pk)) + if dry_run: + continue + + run.status = "error" + run.error = message + run.finished_at = now + run.save(update_fields=["status", "error", "finished_at"]) + if owns_connection: + connection.status = IntegrationStatus.ERROR + connection.last_error = message + connection.last_sync_finished_at = now + connection.next_sync_at = ( + now + timedelta(minutes=connection.sync_interval_minutes) + if connection.auto_sync + else None + ) + connection.save( + update_fields=[ + "status", + "last_error", + "last_sync_finished_at", + "next_sync_at", + "updated_at", + ] + ) + return result diff --git a/backend/commerce/services/wildberries_sync.py b/backend/commerce/services/wildberries_sync.py index 89d61503..2e81754d 100644 --- a/backend/commerce/services/wildberries_sync.py +++ b/backend/commerce/services/wildberries_sync.py @@ -13,6 +13,7 @@ from commerce.models import ( Cost, CostType, IntegrationConnection, + IntegrationSyncStage, ProductBinding, Sale, SaleEventType, @@ -321,7 +322,7 @@ def normalize_wildberries_finance_row( def sync_wildberries_connection( connection: IntegrationConnection, *, - progress_callback: Callable[[WildberriesSyncStats], None] | None = None, + progress_callback: Callable[..., None] | None = None, ) -> WildberriesSyncStats: credentials = decrypt_credentials(connection) metadata = connection.source.metadata @@ -340,7 +341,17 @@ def sync_wildberries_connection( date_to = date.today().isoformat() stats = WildberriesSyncStats() + def report(stage, processed, total=0): + if progress_callback: + progress_callback( + stats, + stage=stage, + stage_processed=processed, + stage_total=total, + ) + with WildberriesClient(credentials=credentials) as client: + report(IntegrationSyncStage.SALES, 0) rows = client.financial_report( date_from=date_from, date_to=date_to, @@ -359,8 +370,8 @@ def sync_wildberries_connection( stats.costs += costs stats.matched_products += int(matched) stats.unmatched_products += int(unmatched) - if progress_callback and stats.processed % 250 == 0: - progress_callback(stats) + if stats.processed % 250 == 0: + report(IntegrationSyncStage.SALES, stats.processed) # WB explicitly requires following rrdId pages until the endpoint returns # 204. A short 200 page is not documented as an end-of-report signal. @@ -384,10 +395,10 @@ def sync_wildberries_connection( updated_metadata.pop("finance_cursor_date_from", None) updated_metadata.pop("finance_cursor_date_to", None) updated_metadata["finance_initial_complete"] = True + report(IntegrationSyncStage.SALES, stats.processed, stats.processed) connection.source.metadata = updated_metadata connection.source.save(update_fields=["metadata", "updated_at"]) - if progress_callback: - progress_callback(stats) + report(IntegrationSyncStage.FINALIZING, 1, 1) return stats diff --git a/backend/commerce/tasks/__init__.py b/backend/commerce/tasks/__init__.py index be7d0280..04056603 100644 --- a/backend/commerce/tasks/__init__.py +++ b/backend/commerce/tasks/__init__.py @@ -1,5 +1,6 @@ import logging from datetime import timedelta +from time import monotonic from celery import shared_task from celery.utils import uuid as celery_uuid @@ -11,17 +12,28 @@ from django.utils import timezone from commerce.integrations import sync_registered_source from commerce.integrations.moysklad import MoySkladError from commerce.integrations.wildberries import WildberriesError -from commerce.models import IntegrationConnection, IntegrationStatus, IntegrationSyncRun +from commerce.models import ( + IntegrationConnection, + IntegrationStatus, + IntegrationSyncRun, + IntegrationSyncStage, +) from commerce.realtime import send_connection_update from commerce.services import sync_moysklad_connection, sync_wildberries_connection from commerce.services.credentials import CredentialError +from commerce.services.sync_runs import ( + STALE_QUEUED_ERROR as STALE_QUEUED_ERROR, + reconcile_stale_sync_runs, +) logger = logging.getLogger(__name__) QUEUE_PUBLISH_ERROR = "Не удалось поставить синхронизацию в очередь." MISSING_RUN_ERROR = "Запуск синхронизации не найден. Повторите импорт." -STALE_QUEUED_ERROR = "Задача не была получена обработчиком и снята с очереди." -STALE_SYNCING_ERROR = "Предыдущая синхронизация была прервана." + + +class SyncRunNoLongerActive(RuntimeError): + pass def _next_sync_at(connection, *, now=None): @@ -104,6 +116,11 @@ def _publish_connection_sync(*, connection_id, run_id, task_id): "connection_id": str(connection_id), "run_id": str(run_id), "task_id": task_id, + "provider": connection.provider, + "stage": run.stage, + "processed": run.processed, + "duration_ms": 0, + "failure_reason": QUEUE_PUBLISH_ERROR, }, ) raise @@ -151,62 +168,179 @@ def sync_source(source_code: str, entity_type: str): } -@shared_task(bind=True, name="commerce.sync_connection_run") -def sync_connection_run(self, run_id: str, connection_id: str | None = None): - try: - run = IntegrationSyncRun.objects.select_related("connection__source").get( - pk=run_id +def _missing_sync_run_result(*, run_id, connection_id, task_id): + connection_query = IntegrationConnection.objects.filter( + status__in=[IntegrationStatus.QUEUED, IntegrationStatus.SYNCING] + ) + connection = connection_query.filter(task_id=task_id).first() if task_id else None + if connection is None and connection_id is not None: + connection = connection_query.filter(pk=connection_id).first() + if connection is not None and (not task_id or connection.task_id == task_id): + finished_at = timezone.now() + connection.status = IntegrationStatus.ERROR + connection.last_error = MISSING_RUN_ERROR + connection.next_sync_at = ( + finished_at + timedelta(minutes=5) if connection.auto_sync else None ) - except IntegrationSyncRun.DoesNotExist: - task_id = self.request.id or "" - connection = None - if connection_id is not None: - connection = IntegrationConnection.objects.filter(pk=connection_id).first() - if connection is not None and (not task_id or connection.task_id == task_id): - finished_at = timezone.now() - connection.status = IntegrationStatus.ERROR - connection.last_error = MISSING_RUN_ERROR - connection.next_sync_at = ( - finished_at + timedelta(minutes=5) if connection.auto_sync else None + connection.save( + update_fields=["status", "last_error", "next_sync_at", "updated_at"] + ) + _safe_send_connection_update(connection) + logger.error( + "Commerce synchronization run does not exist", + extra={ + "event": "commerce.connection.run_missing", + "connection_id": str(connection.pk) + if connection is not None + else connection_id, + "run_id": run_id, + "task_id": task_id, + "provider": connection.provider if connection is not None else None, + "stage": None, + "processed": 0, + "duration_ms": 0, + "failure_reason": MISSING_RUN_ERROR, + }, + ) + return {"status": "missing", "run_id": run_id} + + +def _ignored_delivery(run, connection, *, task_id, reason): + logger.warning( + "Commerce synchronization delivery ignored", + extra={ + "event": "commerce.connection.delivery_ignored", + "run_id": str(run.pk), + "task_id": task_id or run.task_id, + "connection_id": str(connection.pk), + "provider": connection.provider, + "stage": run.stage, + "run_status": run.status, + "ignore_reason": reason, + "processed": run.processed, + "duration_ms": 0, + }, + ) + if run.status == "success": + return connection.last_stats + return {"status": "ignored", "reason": reason, "run_status": run.status} + + +def _claim_sync_run(*, run_id, connection_id, task_id): + try: + with transaction.atomic(): + run = ( + IntegrationSyncRun.objects.select_for_update() + .select_related("connection__source") + .get(pk=run_id) ) + connection = run.connection + if task_id and run.task_id and run.task_id != task_id: + return ( + None, + None, + _ignored_delivery( + run, + connection, + task_id=task_id, + reason="task_id_mismatch", + ), + ) + if run.status != "queued": + return ( + None, + None, + _ignored_delivery( + run, + connection, + task_id=task_id, + reason="duplicate_delivery_ignored", + ), + ) + owns_connection = ( + connection.task_id == run.task_id + and connection.status == IntegrationStatus.QUEUED + ) + if not owns_connection: + return ( + None, + None, + _ignored_delivery( + run, + connection, + task_id=task_id, + reason="ownership_lost", + ), + ) + + started_at = timezone.now() + run.status = "running" + run.stage = IntegrationSyncStage.CATALOG + run.stage_processed = 0 + run.stage_total = 0 + run.started_at = started_at + run.last_progress_at = started_at + run.task_id = task_id or run.task_id + run.error = "" + run.save( + update_fields=[ + "status", + "stage", + "stage_processed", + "stage_total", + "started_at", + "last_progress_at", + "task_id", + "error", + ] + ) + connection.status = IntegrationStatus.SYNCING + connection.last_sync_started_at = started_at + connection.last_error = "" connection.save( update_fields=[ "status", + "last_sync_started_at", "last_error", - "next_sync_at", "updated_at", ] ) - _safe_send_connection_update(connection) - logger.error( - "Commerce synchronization run does not exist", - extra={ - "event": "commerce.connection.run_missing", - "connection_id": connection_id, - "run_id": run_id, - "task_id": task_id, - }, + except IntegrationSyncRun.DoesNotExist: + return ( + None, + None, + _missing_sync_run_result( + run_id=run_id, + connection_id=connection_id, + task_id=task_id, + ), ) - return {"status": "missing", "run_id": run_id} + return run, connection, None - connection = run.connection - started_at = timezone.now() - run.status = "running" - run.started_at = started_at - run.task_id = self.request.id or run.task_id - run.error = "" - run.save(update_fields=["status", "started_at", "task_id", "error"]) - connection.status = IntegrationStatus.SYNCING - connection.last_sync_started_at = started_at - connection.last_error = "" - connection.save( - update_fields=["status", "last_sync_started_at", "last_error", "updated_at"] - ) - _safe_send_connection_update(connection, run=run) - def report_progress(current_stats): +def _progress_reporter(run, connection, *, task_started): + task_id = run.task_id + + def report_progress( + current_stats, + *, + stage=None, + stage_processed=None, + stage_total=None, + ): now = timezone.now() - IntegrationSyncRun.objects.filter(pk=run.pk).update( + current_stage = stage or run.stage or IntegrationSyncStage.CATALOG + updated = IntegrationSyncRun.objects.filter( + pk=run.pk, + status="running", + task_id=task_id, + ).update( + stage=current_stage, + stage_processed=( + stage_processed if stage_processed is not None else run.stage_processed + ), + stage_total=stage_total if stage_total is not None else run.stage_total, + last_progress_at=now, processed=current_stats.processed, created_products=current_stats.created_products, matched_products=current_stats.matched_products, @@ -216,94 +350,235 @@ def sync_connection_run(self, run_id: str, connection_id: str | None = None): sales=current_stats.sales, costs=getattr(current_stats, "costs", 0), ) + if not updated: + raise SyncRunNoLongerActive( + "Запуск синхронизации уже завершён механизмом восстановления." + ) IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=now) run.refresh_from_db() connection.refresh_from_db() - _safe_send_connection_update(connection, run=run) - - try: - if connection.provider == "wildberries": - synchronizer = sync_wildberries_connection - elif connection.provider == "moysklad": - synchronizer = sync_moysklad_connection - else: - raise ValueError(f"Unsupported commerce provider: {connection.provider}") - stats = synchronizer(connection, progress_callback=report_progress).to_dict() - except Exception as exc: - logger.exception( - "Commerce connection synchronization failed", + logger.info( + "Commerce synchronization progress", extra={ - "event": "commerce.connection.sync_failed", - "run_id": run_id, + "event": "commerce.connection.sync_progress", + "run_id": str(run.pk), + "task_id": task_id, "connection_id": str(connection.pk), "provider": connection.provider, - "error_type": type(exc).__name__, + "stage": run.stage, + "stage_processed": run.stage_processed, + "stage_total": run.stage_total, + "processed": run.processed, + "created_products": run.created_products, + "matched_products": run.matched_products, + "stock_snapshots": run.stock_snapshots, + "images": run.images, + "image_errors": run.image_errors, + "sales": run.sales, + "costs": run.costs, + "duration_ms": round((monotonic() - task_started) * 1000, 2), }, ) - if isinstance(exc, (MoySkladError, WildberriesError, CredentialError)): - message = str(exc) - else: - message = "Внутренняя ошибка синхронизации." - finished_at = timezone.now() - run.status = "error" - run.error = message - run.finished_at = finished_at - run.save(update_fields=["status", "error", "finished_at"]) - connection.status = IntegrationStatus.ERROR - connection.last_error = message - connection.last_sync_finished_at = finished_at - connection.next_sync_at = _next_sync_at(connection, now=finished_at) - connection.save( + _safe_send_connection_update(connection, run=run) + + return report_progress + + +def _execute_sync(run, connection, *, task_started): + report_progress = _progress_reporter( + run, + connection, + task_started=task_started, + ) + if connection.provider == "wildberries": + return sync_wildberries_connection( + connection, + progress_callback=report_progress, + ).to_dict() + if connection.provider == "moysklad": + return sync_moysklad_connection( + connection, + progress_callback=report_progress, + captured_at=run.started_at, + run_id=str(run.pk), + task_id=run.task_id, + ).to_dict() + raise ValueError(f"Unsupported commerce provider: {connection.provider}") + + +def _mark_sync_failed(run, connection, exc, *, task_started): + logger.exception( + "Commerce connection synchronization failed", + extra={ + "event": "commerce.connection.sync_failed", + "run_id": str(run.pk), + "task_id": run.task_id, + "connection_id": str(connection.pk), + "provider": connection.provider, + "stage": run.stage, + "error_type": type(exc).__name__, + "failure_reason": str(exc)[:500], + "processed": run.processed, + "created_products": run.created_products, + "matched_products": run.matched_products, + "stock_snapshots": run.stock_snapshots, + "images": run.images, + "image_errors": run.image_errors, + "sales": run.sales, + "costs": run.costs, + "duration_ms": round((monotonic() - task_started) * 1000, 2), + }, + ) + message = ( + str(exc) + if isinstance(exc, (MoySkladError, WildberriesError, CredentialError)) + else "Внутренняя ошибка синхронизации." + ) + finished_at = timezone.now() + IntegrationSyncRun.objects.filter( + pk=run.pk, + status="running", + task_id=run.task_id, + ).update(status="error", error=message, finished_at=finished_at) + IntegrationConnection.objects.filter( + pk=connection.pk, + task_id=run.task_id, + status=IntegrationStatus.SYNCING, + ).update( + status=IntegrationStatus.ERROR, + last_error=message, + last_sync_finished_at=finished_at, + next_sync_at=_next_sync_at(connection, now=finished_at), + updated_at=finished_at, + ) + run.refresh_from_db() + connection.refresh_from_db() + _safe_send_connection_update(connection, run=run) + + +def _complete_sync(run, connection, stats): + task_id = run.task_id + finished_at = timezone.now() + with transaction.atomic(): + completed = IntegrationSyncRun.objects.select_for_update().get(pk=run.pk) + if completed.status != "running" or completed.task_id != task_id: + raise SyncRunNoLongerActive( + "Запуск синхронизации завершён механизмом восстановления." + ) + completed.status = "success" + completed.stage = IntegrationSyncStage.FINALIZING + completed.stage_processed = 1 + completed.stage_total = 1 + completed.last_progress_at = finished_at + completed.processed = stats["processed"] + completed.created_products = stats["created_products"] + completed.matched_products = stats["matched_products"] + completed.stock_snapshots = stats["stock_snapshots"] + completed.images = stats["images"] + completed.image_errors = stats["image_errors"] + completed.sales = stats["sales"] + completed.costs = stats.get("costs", 0) + completed.finished_at = finished_at + completed.save( update_fields=[ "status", - "last_error", - "last_sync_finished_at", - "next_sync_at", - "updated_at", + "stage", + "stage_processed", + "stage_total", + "last_progress_at", + "processed", + "created_products", + "matched_products", + "stock_snapshots", + "images", + "image_errors", + "sales", + "costs", + "finished_at", ] ) - _safe_send_connection_update(connection, run=run) + locked_connection = IntegrationConnection.objects.select_for_update().get( + pk=connection.pk + ) + if locked_connection.task_id == task_id: + locked_connection.status = IntegrationStatus.READY + locked_connection.last_stats = stats + locked_connection.last_error = "" + locked_connection.last_sync_finished_at = finished_at + locked_connection.next_sync_at = _next_sync_at( + locked_connection, + now=finished_at, + ) + locked_connection.save( + update_fields=[ + "status", + "last_stats", + "last_error", + "last_sync_finished_at", + "next_sync_at", + "updated_at", + ] + ) + return completed, locked_connection + + +@shared_task( + bind=True, + name="commerce.sync_connection_run", + reject_on_worker_lost=False, +) +def sync_connection_run(self, run_id: str, connection_id: str | None = None): + task_id = self.request.id or "" + run, connection, immediate_result = _claim_sync_run( + run_id=run_id, + connection_id=connection_id, + task_id=task_id, + ) + if immediate_result is not None: + return immediate_result + + task_id = run.task_id + task_started = monotonic() + logger.info( + "Commerce connection synchronization started", + extra={ + "event": "commerce.connection.sync_started", + "run_id": str(run.pk), + "task_id": task_id, + "connection_id": str(connection.pk), + "provider": connection.provider, + "stage": run.stage, + "processed": run.processed, + "duration_ms": 0, + }, + ) + _safe_send_connection_update(connection, run=run) + try: + stats = _execute_sync(run, connection, task_started=task_started) + except Exception as exc: + _mark_sync_failed(run, connection, exc, task_started=task_started) raise - finished_at = timezone.now() - run.status = "success" - run.processed = stats["processed"] - run.created_products = stats["created_products"] - run.matched_products = stats["matched_products"] - run.stock_snapshots = stats["stock_snapshots"] - run.images = stats["images"] - run.image_errors = stats["image_errors"] - run.sales = stats["sales"] - run.costs = stats.get("costs", 0) - run.finished_at = finished_at - run.save( - update_fields=[ - "status", - "processed", - "created_products", - "matched_products", - "stock_snapshots", - "images", - "image_errors", - "sales", - "costs", - "finished_at", - ] - ) - connection.status = IntegrationStatus.READY - connection.last_stats = stats - connection.last_error = "" - connection.last_sync_finished_at = finished_at - connection.next_sync_at = _next_sync_at(connection, now=finished_at) - connection.save( - update_fields=[ - "status", - "last_stats", - "last_error", - "last_sync_finished_at", - "next_sync_at", - "updated_at", - ] + run, connection = _complete_sync(run, connection, stats) + logger.info( + "Commerce connection synchronization completed", + extra={ + "event": "commerce.connection.sync_succeeded", + "run_id": str(run.pk), + "task_id": task_id, + "connection_id": str(connection.pk), + "provider": connection.provider, + "stage": run.stage, + "processed": run.processed, + "created_products": run.created_products, + "matched_products": run.matched_products, + "stock_snapshots": run.stock_snapshots, + "images": run.images, + "image_errors": run.image_errors, + "sales": run.sales, + "costs": run.costs, + "duration_ms": round((monotonic() - task_started) * 1000, 2), + }, ) _safe_send_connection_update(connection, run=run) return stats @@ -315,69 +590,48 @@ sync_moysklad_run = sync_connection_run def _recover_stale_connections(*, now): - queued_cutoff = now - timedelta( - seconds=getattr(settings, "COMMERCE_QUEUED_STALE_SECONDS", 120) - ) - syncing_cutoff = now - timedelta( - minutes=getattr(settings, "COMMERCE_SYNCING_STALE_MINUTES", 45) - ) - recovered = 0 - recovery_rules = ( - (IntegrationStatus.QUEUED, queued_cutoff, ("queued",), STALE_QUEUED_ERROR), - ( - IntegrationStatus.SYNCING, - syncing_cutoff, - ("queued", "running"), - STALE_SYNCING_ERROR, + result = reconcile_stale_sync_runs( + queued_timeout=timedelta( + seconds=getattr(settings, "COMMERCE_QUEUED_STALE_SECONDS", 120) ), + running_timeout=timedelta( + seconds=getattr(settings, "COMMERCE_RUNNING_STALE_SECONDS", 600) + ), + now=now, ) - for connection_status, cutoff, run_statuses, message in recovery_rules: - candidate_ids = list( - IntegrationConnection.objects.filter( - status=connection_status, - updated_at__lte=cutoff, - ).values_list("pk", flat=True) - ) - for connection_id in candidate_ids: - with transaction.atomic(): - connection = IntegrationConnection.objects.select_for_update().get( - pk=connection_id - ) - if connection.status != connection_status or connection.updated_at > cutoff: - continue - run = ( - connection.sync_runs.select_for_update() - .filter(status__in=run_statuses) - .order_by("-created_at") - .first() - ) - if run is not None: - run.status = "error" - run.error = message - run.finished_at = now - run.save(update_fields=["status", "error", "finished_at"]) - connection.status = IntegrationStatus.ERROR - connection.last_error = message - connection.next_sync_at = now if connection.auto_sync else None - connection.save( - update_fields=[ - "status", - "last_error", - "next_sync_at", - "updated_at", - ] - ) + recovered_run_ids = [*result.queued_run_ids, *result.running_run_ids] + recovered_runs = IntegrationSyncRun.objects.filter( + pk__in=recovered_run_ids, + ).select_related("connection") + for run in recovered_runs: + connection = run.connection + if str(connection.pk) in result.connection_ids: _safe_send_connection_update(connection, run=run) - recovered += 1 - if recovered: + logger.warning( + "Recovered stale commerce synchronization", + extra={ + "event": "commerce.connection.stale_run_recovered", + "run_id": str(run.pk), + "task_id": run.task_id, + "connection_id": str(connection.pk), + "provider": connection.provider, + "stage": run.stage, + "processed": run.processed, + "failure_reason": run.error, + "duration_ms": None, + }, + ) + if result.recovered: logger.warning( "Recovered stale commerce synchronizations", extra={ "event": "commerce.connection.stale_runs_recovered", - "recovered": recovered, + "recovered": result.recovered, + "queued": len(result.queued_run_ids), + "running": len(result.running_run_ids), }, ) - return recovered + return result.recovered @shared_task(name="commerce.dispatch_due_connection_syncs") diff --git a/backend/eusrr_backend/celery.py b/backend/eusrr_backend/celery.py index aabe7414..a910030b 100644 --- a/backend/eusrr_backend/celery.py +++ b/backend/eusrr_backend/celery.py @@ -5,6 +5,7 @@ import os from inspect import signature from threading import Lock from time import monotonic +from types import TracebackType from celery import Celery from celery.signals import before_task_publish, task_failure, task_postrun, task_prerun @@ -24,6 +25,32 @@ _task_state_lock = Lock() _TASK_LOG_CONTEXT_HEADER = "eusrr_log_context" _PROPAGATED_LOG_FIELDS = ("request_id", "user_id", "client_ip") + +def _traceback_like(value) -> bool: + return isinstance(value, TracebackType) or all( + hasattr(value, attribute) for attribute in ("tb_frame", "tb_lineno", "tb_next") + ) + + +def _normalize_failure_traceback(exception, traceback_value): + """Return valid logging exc_info plus a serializable fallback traceback.""" + + if exception is None: + return None, None + candidate = traceback_value + fallback = None + if isinstance(candidate, tuple) and len(candidate) == 3: + candidate = candidate[2] + elif hasattr(candidate, "tb"): + fallback = getattr(candidate, "traceback", None) + candidate = getattr(candidate, "tb", None) + if _traceback_like(candidate): + return (type(exception), exception, candidate), fallback + if traceback_value: + fallback = fallback or str(traceback_value) + return None, fallback + + # Устанавливаем настройки Django по умолчанию os.environ.setdefault("DJANGO_SETTINGS_MODULE", "eusrr_backend.settings") @@ -152,6 +179,7 @@ def log_task_failed( sender=None, task_id=None, exception=None, traceback=None, **kwargs ): task_name = getattr(sender, "name", None) + exc_info, failure_traceback = _normalize_failure_traceback(exception, traceback) logger.error( "Celery task failed", extra={ @@ -159,8 +187,10 @@ def log_task_failed( "task_id": task_id, "task_name": task_name, "error_type": type(exception).__name__ if exception else None, + "failure_reason": str(exception) if exception else None, + "failure_traceback": failure_traceback, }, - exc_info=(type(exception), exception, traceback) if exception else None, + exc_info=exc_info, ) diff --git a/backend/eusrr_backend/logging_utils.py b/backend/eusrr_backend/logging_utils.py index 32e765b7..e6f16d9e 100644 --- a/backend/eusrr_backend/logging_utils.py +++ b/backend/eusrr_backend/logging_utils.py @@ -226,7 +226,19 @@ def _record_fields(record: logging.LogRecord) -> dict[str, Any]: def _exception_text(formatter: logging.Formatter, record: logging.LogRecord): if not record.exc_info: return None - return redact(formatter.formatException(record.exc_info)) + try: + return redact(formatter.formatException(record.exc_info)) + except (AttributeError, TypeError, ValueError): + # Celery/billiard may occasionally provide a serialized traceback where + # stdlib logging expects a traceback object. Never hide the real failure + # behind a formatter exception. + if isinstance(record.exc_info, tuple): + error_type, error, traceback_value = (*record.exc_info, None, None)[:3] + return redact( + f"{getattr(error_type, '__name__', error_type)}: {error}; " + f"traceback={traceback_value}" + ) + return redact(record.exc_info) class ContextFilter(logging.Filter): diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index 40b4d01d..1f6acefd 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -757,20 +757,23 @@ COMMERCE_CREDENTIAL_KEY = os.getenv("COMMERCE_CREDENTIAL_KEY", "") COMMERCE_QUEUED_STALE_SECONDS = int( os.getenv("COMMERCE_QUEUED_STALE_SECONDS", "120") ) -COMMERCE_SYNCING_STALE_MINUTES = int( - os.getenv("COMMERCE_SYNCING_STALE_MINUTES", "45") +COMMERCE_RUNNING_STALE_SECONDS = int( + os.getenv("COMMERCE_RUNNING_STALE_SECONDS", "600") ) MOYSKLAD_API_BASE_URL = os.getenv( "MOYSKLAD_API_BASE_URL", "https://api.moysklad.ru/api/remap/1.2" ) -# Приоритеты очередей (закомментировано для упрощения в development) -# В production можно раскомментировать для масштабирования -# CELERY_TASK_ROUTES = { -# 'notifications.*': {'queue': 'notifications'}, -# 'documents.tasks.*': {'queue': 'default'}, -# 'employees.tasks.*': {'queue': 'default'}, -# } +# Долгий импорт commerce изолирован от коротких фоновых задач. Worker должен +# слушать очередь commerce отдельно либо вместе с default: -Q celery,commerce. +CELERY_TASK_ROUTES = { + "commerce.sync_connection_run": {"queue": "commerce"}, +} +CELERY_TASK_ANNOTATIONS = { + # Requeue after a hard worker loss stays disabled until every provider has + # an explicit, end-to-end idempotency contract. + "commerce.sync_connection_run": {"reject_on_worker_lost": False}, +} # Настройки для периодических задач (Celery Beat) CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler" diff --git a/backend/tests/commerce/test_connections.py b/backend/tests/commerce/test_connections.py index 2987318a..cec9e13d 100644 --- a/backend/tests/commerce/test_connections.py +++ b/backend/tests/commerce/test_connections.py @@ -7,7 +7,8 @@ from unittest.mock import Mock, patch import httpx import pytest from PIL import Image -from django.db import transaction +from django.db import connection as db_connection, transaction +from django.test.utils import CaptureQueriesContext from django.urls import reverse from django.utils import timezone from rest_framework import status @@ -15,6 +16,7 @@ from rest_framework import status from commerce.models import ( IntegrationConnection, IntegrationStatus, + IntegrationSyncStage, IntegrationSyncRun, Product, ProductBinding, @@ -30,7 +32,7 @@ from commerce.integrations.moysklad import ( MoySkladRemoteImage, ) from commerce.media import ( - _validated_image_and_thumbnail, + _prepare_product_image, sync_moysklad_product_images, ) from commerce.realtime import COMMERCE_CONNECTIONS_GROUP, send_connection_update @@ -42,7 +44,9 @@ from commerce.services.moysklad_sync import ( normalize_moysklad_document, normalize_moysklad_stock_row, reconcile_moysklad_variant_parents, + sync_moysklad_stock_rows, ) +from commerce.services.sync_runs import STALE_RUNNING_ERROR from commerce.tasks import ( STALE_QUEUED_ERROR, dispatch_due_connection_syncs, @@ -59,11 +63,20 @@ def _jpeg_bytes(): return output.getvalue() -def test_product_thumbnail_rejects_decompression_bomb(monkeypatch): +def test_product_thumbnail_safely_resizes_decompression_bomb(monkeypatch): monkeypatch.setattr(Image, "MAX_IMAGE_PIXELS", 100) + monkeypatch.setattr("commerce.media.MAX_PRODUCT_IMAGE_PIXELS", 100) - with pytest.raises((Image.DecompressionBombError, Image.DecompressionBombWarning)): - _validated_image_and_thumbnail(_jpeg_bytes()) + stored, thumbnail, suffix, was_resized = _prepare_product_image(_jpeg_bytes()) + + with Image.open(io.BytesIO(stored)) as resized: + assert resized.width * resized.height <= 100 + assert resized.format == "WEBP" + with Image.open(io.BytesIO(thumbnail)) as preview: + assert preview.width * preview.height <= 100 + assert preview.format == "WEBP" + assert suffix == ".webp" + assert was_resized is True def test_moysklad_client_keeps_api_prefix(settings): @@ -295,12 +308,87 @@ def test_product_images_are_stored_deduplicated_and_removed_with_source_state(): assert client.download_image.call_count == 1 client.product_images.return_value = [] - assert sync_moysklad_product_images( + assert ( + sync_moysklad_product_images( + connection=connection, + row=row, + client=client, + ) + == 0 + ) + assert not ProductImage.objects.filter(pk=image.pk).exists() + + +def test_one_broken_product_image_does_not_abort_remaining_images(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + ) + row = { + "id": "external-product-images", + "name": "Товар с частично повреждёнными фото", + "article": "PHOTO-PARTIAL", + "meta": {"type": "product"}, + "images": {"meta": {"size": 2}}, + } + normalize_moysklad_assortment_row( + connection=connection, + row=row, + captured_at=timezone.now(), + ) + remotes = [ + MoySkladRemoteImage( + external_id="broken-image", + filename="broken.jpg", + title="Повреждённое", + size=100, + updated="2026-08-17 12:00:00.000", + download_href="https://api.example/download/broken", + ), + MoySkladRemoteImage( + external_id="valid-image", + filename="valid.jpg", + title="Рабочее", + size=len(_jpeg_bytes()), + updated="2026-08-17 12:00:00.000", + download_href="https://api.example/download/valid", + ), + ] + client = Mock() + client.product_images.return_value = remotes + + def download(download_href): + if download_href.endswith("/broken"): + raise OSError("image cannot be decoded") + return _jpeg_bytes(), "image/jpeg" + + client.download_image.side_effect = download + failures = [] + + synchronized = sync_moysklad_product_images( connection=connection, row=row, client=client, - ) == 0 - assert not ProductImage.objects.filter(pk=image.pk).exists() + error_callback=lambda remote, error: failures.append( + (remote.external_id, str(error)) + ), + ) + + assert synchronized == 1 + assert failures == [("broken-image", "image cannot be decoded")] + assert ( + ProductImage.objects.filter( + source=source, + external_id="valid-image", + ).count() + == 1 + ) + assert not ProductImage.objects.filter( + source=source, + external_id="broken-image", + ).exists() def test_moysklad_client_includes_safe_upstream_error_detail(settings): @@ -477,6 +565,32 @@ def test_connection_sync_is_published_only_after_outer_transaction_commit(): ] +@pytest.mark.django_db(transaction=True) +def test_connection_sync_is_not_published_after_transaction_rollback(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.READY, + ) + + with ( + patch("commerce.tasks.celery_uuid", return_value="rolled-back-task"), + patch("commerce.tasks.sync_connection_run.apply_async") as apply_async, + pytest.raises(RuntimeError, match="force rollback"), + ): + with transaction.atomic(): + enqueue_connection_sync(connection) + raise RuntimeError("force rollback") + + apply_async.assert_not_called() + assert not IntegrationSyncRun.objects.filter(task_id="rolled-back-task").exists() + connection.refresh_from_db() + assert connection.status == IntegrationStatus.READY + assert connection.task_id == "" + + def test_enqueue_failure_transitions_run_and_connection_to_error( django_capture_on_commit_callbacks, ): @@ -519,7 +633,7 @@ def test_missing_sync_run_does_not_leave_connection_queued(): with patch("commerce.tasks.send_connection_update"): result = sync_moysklad_run.apply( - args=[str(uuid.uuid4()), str(connection.pk)], + args=[str(uuid.uuid4()), str(uuid.uuid4())], task_id="missing-run-task", ).get() @@ -535,10 +649,13 @@ def test_sync_run_persists_product_image_progress(): source=source, provider="moysklad", encrypted_credentials="unused", + status=IntegrationStatus.QUEUED, + task_id="sync-task", ) - run = connection.sync_runs.create() + run = connection.sync_runs.create(task_id="sync-task") + observed_progress = [] - def synchronize(_connection, *, progress_callback): + def synchronize(_connection, *, progress_callback, run_id, **_kwargs): stats = MoySkladSyncStats( processed=25, created_products=4, @@ -546,7 +663,31 @@ def test_sync_run_persists_product_image_progress(): images=12, sales=3, ) - progress_callback(stats) + for stage, processed, total in ( + (IntegrationSyncStage.CATALOG, 25, 25), + (IntegrationSyncStage.IMAGES, 12, 12), + (IntegrationSyncStage.VARIANT_LINKS, 25, 25), + (IntegrationSyncStage.STOCKS, 20, 20), + (IntegrationSyncStage.SALES, 3, 3), + (IntegrationSyncStage.RETURNS, 1, 1), + (IntegrationSyncStage.FINALIZING, 1, 1), + ): + progress_callback( + stats, + stage=stage, + stage_processed=processed, + stage_total=total, + ) + progress = IntegrationSyncRun.objects.get(pk=run_id) + observed_progress.append( + ( + progress.stage, + progress.stage_processed, + progress.stage_total, + progress.stock_snapshots, + progress.last_progress_at, + ) + ) return stats with ( @@ -563,6 +704,21 @@ def test_sync_run_persists_product_image_progress(): assert run.images == 12 assert run.image_errors == 0 assert run.sales == 3 + assert run.stage == IntegrationSyncStage.FINALIZING + assert run.stage_processed == run.stage_total == 1 + assert run.last_progress_at is not None + assert [item[0] for item in observed_progress] == [ + IntegrationSyncStage.CATALOG, + IntegrationSyncStage.IMAGES, + IntegrationSyncStage.VARIANT_LINKS, + IntegrationSyncStage.STOCKS, + IntegrationSyncStage.SALES, + IntegrationSyncStage.RETURNS, + IntegrationSyncStage.FINALIZING, + ] + assert observed_progress[3][1:4] == (20, 20, 20) + assert all(item[4] is not None for item in observed_progress) + assert connection.status == IntegrationStatus.READY assert connection.last_stats["images"] == 12 @@ -605,7 +761,7 @@ def test_dispatcher_recovers_old_queued_run_before_syncing_timeout(settings): ) run = connection.sync_runs.create(status="queued", task_id="lost-task") stale_at = timezone.now() - timedelta(minutes=2) - IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=stale_at) + IntegrationSyncRun.objects.filter(pk=run.pk).update(created_at=stale_at) with patch("commerce.tasks.send_connection_update"): result = dispatch_due_connection_syncs() @@ -620,6 +776,68 @@ def test_dispatcher_recovers_old_queued_run_before_syncing_timeout(settings): assert run.finished_at is not None +def test_dispatcher_recovers_stale_running_run_by_heartbeat(settings): + settings.COMMERCE_RUNNING_STALE_SECONDS = 60 + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.SYNCING, + task_id="lost-running-task", + auto_sync=False, + ) + stale_at = timezone.now() - timedelta(minutes=2) + run = connection.sync_runs.create( + status="running", + task_id="lost-running-task", + stage=IntegrationSyncStage.STOCKS, + started_at=stale_at, + last_progress_at=stale_at, + ) + + with patch("commerce.tasks.send_connection_update"): + result = dispatch_due_connection_syncs() + + connection.refresh_from_db() + run.refresh_from_db() + assert result == {"queued": 0} + assert connection.status == IntegrationStatus.ERROR + assert connection.last_error == STALE_RUNNING_ERROR + assert run.status == "error" + assert run.error == STALE_RUNNING_ERROR + assert run.stage == IntegrationSyncStage.STOCKS + assert run.finished_at is not None + + +def test_dispatcher_keeps_running_run_with_fresh_heartbeat(settings): + settings.COMMERCE_RUNNING_STALE_SECONDS = 60 + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.SYNCING, + task_id="active-running-task", + auto_sync=False, + ) + run = connection.sync_runs.create( + status="running", + task_id="active-running-task", + stage=IntegrationSyncStage.IMAGES, + started_at=timezone.now() - timedelta(hours=1), + last_progress_at=timezone.now(), + ) + + result = dispatch_due_connection_syncs() + + connection.refresh_from_db() + run.refresh_from_db() + assert result == {"queued": 0} + assert connection.status == IntegrationStatus.SYNCING + assert run.status == "running" + + def test_schedule_update_does_not_disconnect_connection(api_client, user_factory): user = user_factory(staff=True) source = SourceSystem.objects.create(code="moysklad", name="МойСклад") @@ -742,6 +960,134 @@ def test_moysklad_stock_report_creates_separate_warehouse_snapshots(): assert first.reserved == 2 +def test_moysklad_stock_sync_is_batched_idempotent_and_reports_progress(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + ) + rows = [] + for index in range(3): + external_id = f"stock-product-{index}" + product = Product.objects.create(sku=f"STOCK-{index}", name=f"Товар {index}") + ProductBinding.objects.create( + source=source, + product=product, + external_id=external_id, + ) + rows.append( + { + "meta": { + "href": ( + "https://api.moysklad.ru/api/remap/1.2/entity/product/" + f"{external_id}" + ) + }, + "stockByStore": [ + { + "name": "Основной склад", + "stock": index + 1, + "reserve": 0, + "inTransit": 0, + "meta": { + "href": ( + "https://api.moysklad.ru/api/remap/1.2/entity/store/" + "main-store" + ) + }, + } + ], + } + ) + captured_at = timezone.now() + progress = [] + + with CaptureQueriesContext(db_connection) as queries: + first_count = sync_moysklad_stock_rows( + connection=connection, + rows=iter(rows), + captured_at=captured_at, + batch_size=2, + progress_callback=lambda processed, synchronized: progress.append( + (processed, synchronized) + ), + ) + second_count = sync_moysklad_stock_rows( + connection=connection, + rows=iter(rows), + captured_at=captured_at, + batch_size=2, + ) + + binding_selects = [ + query["sql"] + for query in queries.captured_queries + if "commerce_productbinding" in query["sql"].lower() + and query["sql"].lstrip().upper().startswith("SELECT") + ] + assert first_count == second_count == 3 + assert progress == [(2, 2), (3, 3)] + assert len(binding_selects) == 1 + assert StockSnapshot.objects.filter(source=source).count() == 3 + + +def test_repeated_moysklad_documents_do_not_duplicate_sales_or_returns(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + ) + product = Product.objects.create(sku="REPEAT-1", name="Повторяемый товар") + ProductBinding.objects.create( + source=source, + product=product, + external_id="repeat-product", + ) + base_document = { + "id": "repeat-document", + "name": "R-1", + "moment": "2026-08-17 13:00:00.000", + "positions": { + "rows": [ + { + "id": "repeat-position", + "quantity": 1, + "price": 10000, + "assortment": { + "id": "repeat-product", + "name": "Повторяемый товар", + "article": "REPEAT-1", + }, + } + ] + }, + } + + for document_type in ("demand", "salesreturn"): + normalize_moysklad_document( + connection=connection, + row=base_document, + document_type=document_type, + ) + normalize_moysklad_document( + connection=connection, + row=base_document, + document_type=document_type, + ) + + assert Sale.objects.filter(source=source).count() == 2 + assert sum(sale.lines.count() for sale in Sale.objects.filter(source=source)) == 2 + assert ( + SourceRecord.objects.filter( + source=source, + external_id="repeat-document", + ).count() + == 2 + ) + + def test_moysklad_row_matches_existing_product_by_sku(): source = SourceSystem.objects.create(code="moysklad", name="МойСклад") connection = IntegrationConnection.objects.create( @@ -822,9 +1168,7 @@ def test_moysklad_variant_is_not_merged_with_parent_and_parent_is_reconciled(): ) assert variant.product != parent.product assert variant.product.parent == parent.product - assert variant.metadata["characteristics"] == [ - {"name": "Размер", "value": "XL"} - ] + assert variant.metadata["characteristics"] == [{"name": "Размер", "value": "XL"}] def test_moysklad_existing_binding_refreshes_product_fields(): @@ -871,6 +1215,16 @@ def test_moysklad_existing_binding_refreshes_product_fields(): assert product.description == "Новое описание" assert product.barcode == "460000000002" assert product.is_active is False + assert Product.objects.filter(sku="REFRESH-1").count() == 1 + assert ProductBinding.objects.filter( + source=source, + external_id="product-1", + ).count() == 1 + assert SourceRecord.objects.filter( + source=source, + entity_type="product", + external_id="product-1", + ).count() == 1 def test_moysklad_demand_creates_sale_and_matches_lines(): diff --git a/backend/tests/commerce/test_wildberries.py b/backend/tests/commerce/test_wildberries.py index b3c8085f..171cf109 100644 --- a/backend/tests/commerce/test_wildberries.py +++ b/backend/tests/commerce/test_wildberries.py @@ -293,8 +293,10 @@ def test_background_run_dispatches_wildberries_synchronizer(): source=source, provider="wildberries", encrypted_credentials=encrypt_credentials({"token": "test"}), + status=IntegrationStatus.QUEUED, + task_id="wildberries-task", ) - run = connection.sync_runs.create() + run = connection.sync_runs.create(task_id="wildberries-task") stats = Mock() stats.to_dict.return_value = { "processed": 5, diff --git a/backend/tests/test_celery_schedule.py b/backend/tests/test_celery_schedule.py index 98aa943d..fdca0e1c 100644 --- a/backend/tests/test_celery_schedule.py +++ b/backend/tests/test_celery_schedule.py @@ -1,6 +1,7 @@ from django.conf import settings from eusrr_backend.celery import app +from commerce.tasks import sync_connection_run def test_due_personnel_actions_are_scheduled(): @@ -20,3 +21,13 @@ def test_scheduled_tasks_are_registered(): ] assert missing == [] + + +def test_commerce_import_has_dedicated_queue_without_automatic_redelivery(): + assert settings.CELERY_TASK_ROUTES["commerce.sync_connection_run"] == { + "queue": "commerce" + } + assert settings.CELERY_TASK_ANNOTATIONS["commerce.sync_connection_run"] == { + "reject_on_worker_lost": False + } + assert sync_connection_run.reject_on_worker_lost is False diff --git a/backend/tests/test_logging.py b/backend/tests/test_logging.py index d2327ceb..0692c33f 100644 --- a/backend/tests/test_logging.py +++ b/backend/tests/test_logging.py @@ -8,6 +8,8 @@ from pathlib import Path from types import SimpleNamespace import pytest +from billiard.einfo import ExceptionInfo +from billiard.exceptions import WorkerLostError from django.conf import settings from django.http import HttpResponse from django.test import RequestFactory @@ -16,7 +18,9 @@ from django.test.utils import override_settings from api.auth.authentication import SessionAwareJWTAuthentication from eusrr_backend.auth_backends import EmailOrPhoneBackend from eusrr_backend.celery import ( + _normalize_failure_traceback, inject_task_log_context, + log_task_failed, log_task_completed, log_task_started, ) @@ -210,6 +214,56 @@ def test_formatter_serializes_exception_without_losing_event(): assert "ValueError" in payload["exception"] +def test_worker_lost_string_traceback_does_not_break_celery_logging(monkeypatch): + captured = {} + + def capture_error(message, **kwargs): + captured.update(message=message, **kwargs) + + monkeypatch.setattr("eusrr_backend.celery.logger.error", capture_error) + failure = WorkerLostError("worker exited with SIGTERM") + + log_task_failed( + sender=SimpleNamespace(name="commerce.sync_connection_run"), + task_id="lost-task", + exception=failure, + traceback="serialized billiard traceback", + ) + + assert captured["exc_info"] is None + assert captured["extra"]["failure_traceback"] == ("serialized billiard traceback") + record = _record(**captured["extra"]) + record.exc_info = captured["exc_info"] + assert "WorkerLostError" in JsonFormatter().format(record) + + +def test_billiard_exception_info_is_normalized_to_real_traceback(): + try: + raise RuntimeError("original worker failure") + except RuntimeError as error: + billiard_info = ExceptionInfo() + exc_info, fallback = _normalize_failure_traceback(error, billiard_info) + + assert exc_info is not None + assert exc_info[2].tb_frame is not None + assert fallback + record = _record(event="celery.task.failed") + record.exc_info = exc_info + payload = json.loads(JsonFormatter().format(record)) + assert "original worker failure" in payload["exception"] + + +@pytest.mark.parametrize("formatter", [ConsoleFormatter(), JsonFormatter()]) +def test_formatter_survives_malformed_string_exc_info(formatter): + failure = WorkerLostError("worker disappeared") + record = _record(event="celery.task.failed") + record.exc_info = (WorkerLostError, failure, "not-a-traceback-object") + + output = formatter.format(record) + + assert "worker disappeared" in output + + def test_formatter_sanitizes_django_request_extra(): request = RequestFactory().get( "/api/v1/documents/?token=query-secret&search=private" diff --git a/docs/guides/COMMERCE.md b/docs/guides/COMMERCE.md index 808b3d6c..306c8360 100644 --- a/docs/guides/COMMERCE.md +++ b/docs/guides/COMMERCE.md @@ -10,21 +10,37 @@ Commerce-контур хранит нормализованные товары, Синхронизация любого подключения запускается Celery-задачей `commerce.sync_connection_run`; сервис выбирается по `connection.provider`. -Для работы ручного импорта нужен Celery worker, -для расписания — также Celery Beat. +Задача публикуется только после commit транзакции, а заранее сформированный +`task_id` до публикации сохраняется и в подключении, и в истории запуска. +Долгие импорты направляются в очередь `commerce`: worker должен слушать её +отдельно либо вместе с очередью `celery`. Для работы ручного импорта нужен +Celery worker, для расписания и восстановления зависших запусков — также Celery +Beat. Импорт МойСклад выполняет следующие этапы: -1. Загружает ассортимент страницами по 100 позиций с `expand=images`. -2. Нормализует товары, модификации, комплекты, привязки и остатки. -3. Скачивает оригиналы изображений по авторизованному `downloadHref`, проверяет - формат и лимит 25 МБ, создаёт локальную WebP-миниатюру. -4. Импортирует отгрузки и возвраты вместе с позициями. -5. Сохраняет итоговую статистику и публикует состояние через общий WebSocket. +1. `catalog` — загружает и идемпотентно нормализует ассортимент. +2. `images` — синхронизирует локальные изображения. +3. `variant_links` — восстанавливает связи модификаций с родителями. +4. `stocks` — пакетно обновляет остатки по складам. +5. `sales` и `returns` — импортирует отгрузки и возвраты с позициями. +6. `finalizing` — сохраняет итоговую статистику и состояние подключения. -Прогресс запуска сохраняется и отправляется frontend каждые 25 товаров или -документов. Ошибка отдельного изображения учитывается в `image_errors`, но не -отменяет импорт товарных и финансовых данных. +На каждом длительном этапе сохраняются `stage`, `stage_processed`, +`stage_total` и `last_progress_at`; обновление отправляется frontend через +общий WebSocket. Ошибка отдельного изображения учитывается в `image_errors`, +но не отменяет импорт товарных и финансовых данных. + +Celery Beat раз в минуту выполняет reconciliation. Запуски, слишком долго +оставшиеся в `queued`, и `running` без свежего heartbeat переводятся в `error` +без удаления истории и без автоматического повторного выполнения. Таймауты +задаются `COMMERCE_QUEUED_STALE_SECONDS` и +`COMMERCE_RUNNING_STALE_SECONDS`. Ручная безопасная проверка: + +```bash +python manage.py reconcile_commerce_sync_runs --dry-run +python manage.py reconcile_commerce_sync_runs +``` ## Wildberries @@ -61,6 +77,10 @@ Commerce-контур хранит нормализованные товары, публичные URL. Повторная синхронизация сравнивает внешний размер и время изменения, поэтому неизменённое изображение не скачивается повторно. Удалённые в источнике изображения удаляются локально после успешного чтения списка. +Изображения сверх безопасного лимита декодирования уменьшаются изолированным +процессом ImageMagick до допустимого размера, не увеличивая лимит Pillow в +процессе worker. Ошибка одного файла оставляет остальные изображения товара и +весь импорт работоспособными. REST `ProductSerializer` возвращает `images[]` с локальными `url` и `thumbnail_url`. Каталог использует миниатюру, карточка товара — оригинал для @@ -69,7 +89,7 @@ REST `ProductSerializer` возвращает `images[]` с локальными ## Realtime и fallback Событие `commerce_connection_updated` содержит состояние подключения и -последнего запуска, включая `processed`, `images`, `sales` и `costs`. При рабочем +последнего запуска, включая этап, heartbeat, прогресс и счётчики. При рабочем WebSocket frontend не опрашивает connections endpoint. Если WebSocket недоступен во время активного импорта, включается REST fallback раз в 30 секунд. diff --git a/frontend/src/components/commerce/ProductConnectionsContent.tsx b/frontend/src/components/commerce/ProductConnectionsContent.tsx index e2de4e0c..2804c0be 100644 --- a/frontend/src/components/commerce/ProductConnectionsContent.tsx +++ b/frontend/src/components/commerce/ProductConnectionsContent.tsx @@ -10,6 +10,7 @@ import wsManager from "@/lib/websocketManager"; import type { IntegrationConnection, IntegrationConnectionInput, + IntegrationSyncRun, } from "@/types/commerce"; import { ArrowLeft, @@ -37,6 +38,16 @@ const statusLabels: Record = { error: "Ошибка", }; +const stageLabels: Record, string> = { + catalog: "Каталог", + images: "Изображения", + variant_links: "Связи вариантов", + stocks: "Остатки", + sales: "Продажи", + returns: "Возвраты", + finalizing: "Завершение", +}; + function displayDate(value?: string | null) { if (!value) return "—"; return new Intl.DateTimeFormat("ru-RU", { @@ -359,6 +370,15 @@ export function ProductConnectionsContent({ + {currentRun?.stage ? ( +
+ Этап: {stageLabels[currentRun.stage]} + + {currentRun.stage_processed} + {currentRun.stage_total > 0 ? ` / ${currentRun.stage_total}` : ""} + +
+ ) : null} {(visibleStats?.image_errors ?? 0) > 0 ? (
diff --git a/frontend/src/types/commerce.ts b/frontend/src/types/commerce.ts index 82a42a22..d40ba323 100644 --- a/frontend/src/types/commerce.ts +++ b/frontend/src/types/commerce.ts @@ -33,6 +33,18 @@ export type IntegrationSyncRun = { connection: string; status: "queued" | "running" | "success" | "error"; task_id: string; + stage: + | "" + | "catalog" + | "images" + | "variant_links" + | "stocks" + | "sales" + | "returns" + | "finalizing"; + stage_processed: number; + stage_total: number; + last_progress_at: string | null; processed: number; created_products: number; matched_products: number;