Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden commerce import reliability

Browse Source
This commit is contained in:
lizerk 2026-08-18 18:48:00 +03:00
parent 214323760b
commit bd515fd6d5
25 changed files with 1870 additions and 431 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@ -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

2
backend/Dockerfile
View File

@ -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

4
backend/api/v1/commerce/serializers.py
View File

@ -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",

13
backend/commerce/admin.py
View File

@ -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]

43
backend/commerce/management/commands/reconcile_commerce_sync_runs.py Normal file
View File

@ -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))}."
)
)

299
backend/commerce/media.py
View File

@ -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

70
backend/commerce/migrations/0011_integrationsyncrun_progress_heartbeat.py Normal file
View File

@ -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",
),
),
]

2
backend/commerce/models/__init__.py
View File

@ -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",

34
backend/commerce/models/source.py
View File

@ -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):

4
backend/commerce/realtime.py
View File

@ -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,

2
backend/commerce/services/__init__.py
View File

@ -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",

484
backend/commerce/services/moysklad_sync.py
View File

@ -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

119
backend/commerce/services/sync_runs.py Normal file
View File

@ -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

21
backend/commerce/services/wildberries_sync.py
View File

@ -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

602
backend/commerce/tasks/__init__.py
View File

@ -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")

32
backend/eusrr_backend/celery.py
View File

@ -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,
)

14
backend/eusrr_backend/logging_utils.py
View File

@ -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):

21
backend/eusrr_backend/settings.py
View File

@ -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"

386
backend/tests/commerce/test_connections.py
View File

@ -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():

4
backend/tests/commerce/test_wildberries.py
View File

@ -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,

11
backend/tests/test_celery_schedule.py
View File

@ -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

54
backend/tests/test_logging.py
View File

@ -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"

44
docs/guides/COMMERCE.md
View File

@ -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 секунд.

20
frontend/src/components/commerce/ProductConnectionsContent.tsx
View File

@ -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<IntegrationConnection["status"], string> = {
error: "Ошибка",
};
const stageLabels: Record<Exclude<IntegrationSyncRun["stage"], "">, 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({
<Metric icon={Store} label="Остатки" value={visibleStats?.stock_snapshots ?? 0} />
<Metric icon={CloudDownload} label="Продаж и возвратов" value={visibleStats?.sales ?? 0} />
</div>
{currentRun?.stage ? (
<div className="rounded-xl bg-[var(--surface-secondary)] px-4 py-3 text-sm">
<span className="font-medium">Этап: {stageLabels[currentRun.stage]}</span>
<span className="ml-2 app-text-muted">
{currentRun.stage_processed}
{currentRun.stage_total > 0 ? ` / ${currentRun.stage_total}` : ""}
</span>
</div>
) : null}
{(visibleStats?.image_errors ?? 0) > 0 ? (
<div className="app-feedback-warning flex gap-2 rounded-xl p-4 text-sm">
<TriangleAlert className="mt-0.5 h-4 w-4 shrink-0" />

12
frontend/src/types/commerce.ts
View File

@ -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;