661 lines
23 KiB
Python
661 lines
23 KiB
Python
import logging
|
||
from datetime import timedelta
|
||
from time import monotonic
|
||
|
||
from celery import shared_task
|
||
from celery.exceptions import SoftTimeLimitExceeded
|
||
from celery.utils import uuid as celery_uuid
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.db.models import Q
|
||
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,
|
||
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,
|
||
)
|
||
from commerce.services.sync_runs import (
|
||
reconcile_stale_sync_runs,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
QUEUE_PUBLISH_ERROR = "Не удалось поставить синхронизацию в очередь."
|
||
MISSING_RUN_ERROR = "Запуск синхронизации не найден. Повторите импорт."
|
||
SYNC_TIME_LIMIT_ERROR = "Синхронизация превысила допустимое время выполнения."
|
||
|
||
|
||
class SyncRunNoLongerActive(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _next_sync_at(connection, *, now=None):
|
||
if not connection.auto_sync:
|
||
return None
|
||
return (now or timezone.now()) + timedelta(minutes=connection.sync_interval_minutes)
|
||
|
||
|
||
def _safe_send_connection_update(connection, *, run=None):
|
||
"""Realtime is best-effort and must never break the synchronization itself."""
|
||
|
||
try:
|
||
send_connection_update(connection, run=run)
|
||
except Exception:
|
||
logger.exception(
|
||
"Commerce realtime update failed",
|
||
extra={
|
||
"event": "commerce.connection.realtime_update_failed",
|
||
"connection_id": str(connection.pk),
|
||
"run_id": str(run.pk) if run is not None else None,
|
||
},
|
||
)
|
||
|
||
|
||
def _mark_enqueue_failed(*, connection_id, run_id, task_id):
|
||
now = timezone.now()
|
||
with transaction.atomic():
|
||
run = (
|
||
IntegrationSyncRun.objects.select_for_update()
|
||
.filter(pk=run_id, task_id=task_id)
|
||
.first()
|
||
)
|
||
connection = (
|
||
IntegrationConnection.objects.select_for_update()
|
||
.filter(pk=connection_id, task_id=task_id)
|
||
.first()
|
||
)
|
||
if run is not None and run.status == "queued":
|
||
run.status = "error"
|
||
run.error = QUEUE_PUBLISH_ERROR
|
||
run.finished_at = now
|
||
run.save(update_fields=["status", "error", "finished_at"])
|
||
if connection is not None and connection.status == IntegrationStatus.QUEUED:
|
||
connection.status = IntegrationStatus.ERROR
|
||
connection.last_error = QUEUE_PUBLISH_ERROR
|
||
connection.next_sync_at = (
|
||
now + timedelta(minutes=5) if connection.auto_sync else None
|
||
)
|
||
connection.save(
|
||
update_fields=[
|
||
"status",
|
||
"last_error",
|
||
"next_sync_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
if connection is not None:
|
||
_safe_send_connection_update(connection, run=run)
|
||
|
||
|
||
def _publish_connection_sync(*, connection_id, run_id, task_id):
|
||
connection = IntegrationConnection.objects.get(pk=connection_id)
|
||
run = IntegrationSyncRun.objects.get(pk=run_id)
|
||
_safe_send_connection_update(connection, run=run)
|
||
try:
|
||
sync_connection_run.apply_async(
|
||
args=[str(run_id), str(connection_id)],
|
||
task_id=task_id,
|
||
)
|
||
except Exception:
|
||
_mark_enqueue_failed(
|
||
connection_id=connection_id,
|
||
run_id=run_id,
|
||
task_id=task_id,
|
||
)
|
||
logger.exception(
|
||
"Commerce synchronization could not be published",
|
||
extra={
|
||
"event": "commerce.connection.enqueue_failed",
|
||
"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
|
||
|
||
|
||
def enqueue_connection_sync(connection):
|
||
"""Create and publish one run while preventing overlapping scans."""
|
||
|
||
task_id = celery_uuid()
|
||
with transaction.atomic():
|
||
locked = IntegrationConnection.objects.select_for_update().get(pk=connection.pk)
|
||
if locked.status in {IntegrationStatus.QUEUED, IntegrationStatus.SYNCING}:
|
||
return None
|
||
run = IntegrationSyncRun.objects.create(connection=locked, task_id=task_id)
|
||
locked.status = IntegrationStatus.QUEUED
|
||
locked.task_id = task_id
|
||
locked.last_error = ""
|
||
locked.next_sync_at = _next_sync_at(locked)
|
||
locked.save(
|
||
update_fields=[
|
||
"status",
|
||
"task_id",
|
||
"last_error",
|
||
"next_sync_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
transaction.on_commit(
|
||
lambda: _publish_connection_sync(
|
||
connection_id=locked.pk,
|
||
run_id=run.pk,
|
||
task_id=task_id,
|
||
)
|
||
)
|
||
return run
|
||
|
||
|
||
@shared_task(name="commerce.sync_source")
|
||
def sync_source(source_code: str, entity_type: str):
|
||
results = sync_registered_source(source_code, entity_type=entity_type)
|
||
return {
|
||
"processed": len(results),
|
||
"created": sum(result.created for result in results),
|
||
"changed": sum(result.changed for result in results),
|
||
}
|
||
|
||
|
||
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
|
||
)
|
||
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",
|
||
"updated_at",
|
||
]
|
||
)
|
||
except IntegrationSyncRun.DoesNotExist:
|
||
return (
|
||
None,
|
||
None,
|
||
_missing_sync_run_result(
|
||
run_id=run_id,
|
||
connection_id=connection_id,
|
||
task_id=task_id,
|
||
),
|
||
)
|
||
return run, connection, None
|
||
|
||
|
||
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()
|
||
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,
|
||
stock_snapshots=current_stats.stock_snapshots,
|
||
images=current_stats.images,
|
||
image_errors=current_stats.image_errors,
|
||
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()
|
||
logger.info(
|
||
"Commerce synchronization progress",
|
||
extra={
|
||
"event": "commerce.connection.sync_progress",
|
||
"run_id": str(run.pk),
|
||
"task_id": task_id,
|
||
"connection_id": str(connection.pk),
|
||
"provider": connection.provider,
|
||
"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),
|
||
},
|
||
)
|
||
_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),
|
||
},
|
||
)
|
||
if isinstance(exc, SoftTimeLimitExceeded):
|
||
message = SYNC_TIME_LIMIT_ERROR
|
||
elif isinstance(exc, (MoySkladError, WildberriesError, CredentialError)):
|
||
message = str(exc)
|
||
else:
|
||
message = "Внутренняя ошибка синхронизации."
|
||
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",
|
||
"stage",
|
||
"stage_processed",
|
||
"stage_total",
|
||
"last_progress_at",
|
||
"processed",
|
||
"created_products",
|
||
"matched_products",
|
||
"stock_snapshots",
|
||
"images",
|
||
"image_errors",
|
||
"sales",
|
||
"costs",
|
||
"finished_at",
|
||
]
|
||
)
|
||
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,
|
||
soft_time_limit=settings.COMMERCE_TASK_SOFT_TIME_LIMIT,
|
||
time_limit=settings.COMMERCE_TASK_TIME_LIMIT,
|
||
)
|
||
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
|
||
|
||
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
|
||
|
||
|
||
# Compatibility import for code and tests written before provider dispatch was
|
||
# generalized. The registered Celery task itself has the provider-neutral name.
|
||
sync_moysklad_run = sync_connection_run
|
||
|
||
|
||
def _recover_stale_connections(*, now):
|
||
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,
|
||
)
|
||
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)
|
||
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": result.recovered,
|
||
"queued": len(result.queued_run_ids),
|
||
"running": len(result.running_run_ids),
|
||
},
|
||
)
|
||
return result.recovered
|
||
|
||
|
||
@shared_task(name="commerce.dispatch_due_connection_syncs")
|
||
def dispatch_due_connection_syncs():
|
||
now = timezone.now()
|
||
_recover_stale_connections(now=now)
|
||
due_ids = list(
|
||
IntegrationConnection.objects.filter(
|
||
auto_sync=True,
|
||
status__in=[IntegrationStatus.READY, IntegrationStatus.ERROR],
|
||
)
|
||
.filter(Q(next_sync_at__isnull=True) | Q(next_sync_at__lte=now))
|
||
.values_list("pk", flat=True)
|
||
)
|
||
queued = 0
|
||
for connection in IntegrationConnection.objects.filter(pk__in=due_ids):
|
||
if enqueue_connection_sync(connection) is not None:
|
||
queued += 1
|
||
return {"queued": queued}
|