diff --git a/backend/api/v1/commerce/serializers.py b/backend/api/v1/commerce/serializers.py index 401a50a5..88fb5e8c 100644 --- a/backend/api/v1/commerce/serializers.py +++ b/backend/api/v1/commerce/serializers.py @@ -1,3 +1,6 @@ +from datetime import timedelta + +from django.utils import timezone from rest_framework import serializers from commerce.models import ( @@ -81,6 +84,9 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer): "last_sync_finished_at", "last_stats", "task_id", + "auto_sync", + "sync_interval_minutes", + "next_sync_at", "recent_runs", "created_at", "updated_at", @@ -94,6 +100,7 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer): "last_sync_finished_at", "last_stats", "task_id", + "next_sync_at", "created_at", "updated_at", ] @@ -163,13 +170,22 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer): ) def update(self, instance, validated_data): + previous_auth_type = instance.auth_type credentials = self._credentials(validated_data) if credentials: instance.encrypted_credentials = encrypt_credentials(credentials) for field, value in validated_data.items(): setattr(instance, field, value) - instance.status = "disconnected" - instance.last_error = "" + credentials_changed = credentials or instance.auth_type != previous_auth_type + if credentials_changed: + instance.status = "disconnected" + instance.last_error = "" + if "auto_sync" in validated_data or "sync_interval_minutes" in validated_data: + instance.next_sync_at = ( + timezone.now() + timedelta(minutes=instance.sync_interval_minutes) + if instance.auto_sync + else None + ) instance.save() return instance diff --git a/backend/api/v1/commerce/viewsets.py b/backend/api/v1/commerce/viewsets.py index d7e3c48f..4f78cc82 100644 --- a/backend/api/v1/commerce/viewsets.py +++ b/backend/api/v1/commerce/viewsets.py @@ -1,4 +1,6 @@ from django.utils import timezone +from datetime import timedelta + from django.db.models import Count from django_filters.rest_framework import DjangoFilterBackend from rest_framework import filters, viewsets @@ -25,7 +27,7 @@ from commerce.services import ( ) from commerce.services.credentials import CredentialError from commerce.integrations.moysklad import MoySkladError -from commerce.tasks import sync_moysklad_run +from commerce.tasks import enqueue_connection_sync from .permissions import CommerceAccessPermission from .serializers import ( @@ -71,6 +73,11 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet): connection.account_name = context["account_name"] connection.last_error = "" connection.last_tested_at = timezone.now() + connection.next_sync_at = ( + timezone.now() + timedelta(minutes=connection.sync_interval_minutes) + if connection.auto_sync + else None + ) connection.source.metadata = { **connection.source.metadata, "account_id": context["account_id"], @@ -82,6 +89,7 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet): "account_name", "last_error", "last_tested_at", + "next_sync_at", "updated_at", ] ) @@ -95,16 +103,12 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet): {"detail": "Синхронизация уже выполняется."}, status=status.HTTP_409_CONFLICT, ) - run = IntegrationSyncRun.objects.create(connection=connection) - connection.status = IntegrationStatus.QUEUED - connection.task_id = "" - connection.last_error = "" - connection.save(update_fields=["status", "task_id", "last_error", "updated_at"]) - async_result = sync_moysklad_run.delay(str(run.pk)) - run.task_id = async_result.id or "" - run.save(update_fields=["task_id"]) - connection.task_id = run.task_id - connection.save(update_fields=["task_id", "updated_at"]) + run = enqueue_connection_sync(connection) + if run is None: + return Response( + {"detail": "Синхронизация уже выполняется."}, + status=status.HTTP_409_CONFLICT, + ) return Response( IntegrationSyncRunSerializer(run).data, status=status.HTTP_202_ACCEPTED, diff --git a/backend/commerce/README.md b/backend/commerce/README.md index 8358e372..ce9d8a6e 100644 --- a/backend/commerce/README.md +++ b/backend/commerce/README.md @@ -33,6 +33,10 @@ created by employees. The first production connector is `moysklad`: it scans the complete paginated assortment, aggregate stock, shipments and customer returns. Raw API payloads are retained in `SourceRecord` before normalization. Repeated scans update the same source identities and append stock observations. +Celery Beat checks due connections every minute. Each connection stores whether +automatic synchronization is enabled, its interval (15 minutes by default) and +the next planned run; row locking and connection status prevent overlapping +full scans. Both a Celery worker and Celery Beat must be running. Imported cards are matched by an existing source binding, SKU or barcode. A card that cannot be matched safely is created separately and can later be diff --git a/backend/commerce/migrations/0003_integrationconnection_auto_sync_and_more.py b/backend/commerce/migrations/0003_integrationconnection_auto_sync_and_more.py new file mode 100644 index 00000000..f3960777 --- /dev/null +++ b/backend/commerce/migrations/0003_integrationconnection_auto_sync_and_more.py @@ -0,0 +1,29 @@ +# Generated by Django 5.2.4 on 2026-08-17 16:26 + +import django.core.validators +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('commerce', '0002_integrationconnection_integrationsyncrun'), + ] + + operations = [ + migrations.AddField( + model_name='integrationconnection', + name='auto_sync', + field=models.BooleanField(default=True, verbose_name='Автоматическая синхронизация'), + ), + migrations.AddField( + model_name='integrationconnection', + name='next_sync_at', + field=models.DateTimeField(blank=True, db_index=True, null=True, verbose_name='Следующая синхронизация'), + ), + migrations.AddField( + model_name='integrationconnection', + name='sync_interval_minutes', + field=models.PositiveIntegerField(default=15, validators=[django.core.validators.MinValueValidator(5), django.core.validators.MaxValueValidator(1440)], verbose_name='Интервал синхронизации, минут'), + ), + ] diff --git a/backend/commerce/models/source.py b/backend/commerce/models/source.py index 7cbdf74e..00df2c71 100644 --- a/backend/commerce/models/source.py +++ b/backend/commerce/models/source.py @@ -1,6 +1,7 @@ import uuid from django.conf import settings +from django.core.validators import MaxValueValidator, MinValueValidator from django.db import models @@ -159,6 +160,15 @@ class IntegrationConnection(models.Model): ) last_stats = models.JSONField("Результат синхронизации", default=dict, blank=True) task_id = models.CharField("ID фоновой задачи", max_length=255, blank=True) + auto_sync = models.BooleanField("Автоматическая синхронизация", default=True) + sync_interval_minutes = models.PositiveIntegerField( + "Интервал синхронизации, минут", + default=15, + validators=[MinValueValidator(5), MaxValueValidator(1440)], + ) + next_sync_at = models.DateTimeField( + "Следующая синхронизация", null=True, blank=True, db_index=True + ) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, diff --git a/backend/commerce/tasks/__init__.py b/backend/commerce/tasks/__init__.py index 113e62bf..34899c9d 100644 --- a/backend/commerce/tasks/__init__.py +++ b/backend/commerce/tasks/__init__.py @@ -1,17 +1,69 @@ import logging +from datetime import timedelta from celery import shared_task +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.models import IntegrationStatus, IntegrationSyncRun +from commerce.models import IntegrationConnection, IntegrationStatus, IntegrationSyncRun from commerce.services import sync_moysklad_connection from commerce.services.credentials import CredentialError logger = logging.getLogger(__name__) +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 enqueue_connection_sync(connection): + """Create and publish one run while preventing overlapping scans.""" + + 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) + locked.status = IntegrationStatus.QUEUED + locked.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", + ] + ) + try: + async_result = sync_moysklad_run.delay(str(run.pk)) + except Exception: + retry_at = timezone.now() + timedelta(minutes=5) + IntegrationSyncRun.objects.filter(pk=run.pk).update( + status="error", + error="Не удалось поставить синхронизацию в очередь.", + finished_at=timezone.now(), + ) + IntegrationConnection.objects.filter(pk=connection.pk).update( + status=IntegrationStatus.ERROR, + last_error="Не удалось поставить синхронизацию в очередь.", + next_sync_at=retry_at, + ) + raise + task_id = async_result.id or "" + IntegrationSyncRun.objects.filter(pk=run.pk).update(task_id=task_id) + IntegrationConnection.objects.filter(pk=connection.pk).update(task_id=task_id) + run.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) @@ -54,11 +106,13 @@ def sync_moysklad_run(self, run_id: str): 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( update_fields=[ "status", "last_error", "last_sync_finished_at", + "next_sync_at", "updated_at", ] ) @@ -85,13 +139,44 @@ def sync_moysklad_run(self, run_id: str): 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", ] ) return stats + + +@shared_task(name="commerce.dispatch_due_connection_syncs") +def dispatch_due_connection_syncs(): + now = timezone.now() + stale_cutoff = now - timedelta(minutes=45) + IntegrationConnection.objects.filter( + auto_sync=True, + status__in=[IntegrationStatus.QUEUED, IntegrationStatus.SYNCING], + updated_at__lte=stale_cutoff, + ).update( + status=IntegrationStatus.ERROR, + last_error="Предыдущая синхронизация была прервана и будет перезапущена.", + next_sync_at=now, + updated_at=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} diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index f1536406..9736c1ab 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -762,6 +762,10 @@ CELERYD_PIDFILE = os.path.join(tempfile.gettempdir(), "celery-worker.pid") # Расписание периодических задач CELERY_BEAT_SCHEDULE = { + "commerce-sync-due-connections": { + "task": "commerce.dispatch_due_connection_syncs", + "schedule": 60.0, + }, "open-monthly-payroll-period": { "task": "finance.tasks.open_monthly_payroll_period", "schedule": crontab(day_of_month="1", hour=0, minute=5), diff --git a/backend/tests/commerce/test_connections.py b/backend/tests/commerce/test_connections.py index df7c1872..1a7e6a1b 100644 --- a/backend/tests/commerce/test_connections.py +++ b/backend/tests/commerce/test_connections.py @@ -1,3 +1,4 @@ +from datetime import timedelta from unittest.mock import Mock, patch import httpx @@ -21,6 +22,7 @@ from commerce.services.moysklad_sync import ( normalize_moysklad_assortment_row, normalize_moysklad_document, ) +from commerce.tasks import dispatch_due_connection_syncs pytestmark = pytest.mark.django_db @@ -100,9 +102,7 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory): ) async_result = Mock(id="celery-task-1") - with patch( - "api.v1.commerce.viewsets.sync_moysklad_run.delay", return_value=async_result - ): + with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result): response = api_client.post( reverse( "api:v1:commerce:connection-sync", @@ -117,6 +117,53 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory): assert connection.task_id == "celery-task-1" +def test_beat_dispatcher_queues_due_connection(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.READY, + auto_sync=True, + sync_interval_minutes=15, + next_sync_at=timezone.now() - timedelta(minutes=1), + ) + async_result = Mock(id="scheduled-task-1") + + with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result): + result = dispatch_due_connection_syncs() + + connection.refresh_from_db() + assert result == {"queued": 1} + assert connection.status == IntegrationStatus.QUEUED + assert connection.task_id == "scheduled-task-1" + assert connection.next_sync_at > timezone.now() + + +def test_schedule_update_does_not_disconnect_connection(api_client, user_factory): + user = user_factory(staff=True) + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="encrypted", + status=IntegrationStatus.READY, + ) + api_client.force_authenticate(user=user) + + response = api_client.patch( + reverse("api:v1:commerce:connection-detail", args=[connection.pk]), + {"auto_sync": True, "sync_interval_minutes": 60}, + format="json", + ) + + assert response.status_code == status.HTTP_200_OK + connection.refresh_from_db() + assert connection.status == IntegrationStatus.READY + assert connection.sync_interval_minutes == 60 + assert connection.next_sync_at is not None + + def test_moysklad_row_creates_product_binding_raw_record_and_stock(): source = SourceSystem.objects.create(code="moysklad", name="МойСклад") connection = IntegrationConnection.objects.create( diff --git a/frontend/src/app/products/connections/page.tsx b/frontend/src/app/products/connections/page.tsx index b36da006..c85b4803 100644 --- a/frontend/src/app/products/connections/page.tsx +++ b/frontend/src/app/products/connections/page.tsx @@ -11,6 +11,7 @@ import { ArrowLeft, CheckCircle2, CloudDownload, + Clock3, KeyRound, Loader2, PackageCheck, @@ -146,6 +147,34 @@ export default function ProductConnectionsPage() { } } + async function updateSchedule( + connection: IntegrationConnection, + changes: Pick< + IntegrationConnectionInput, + "auto_sync" | "sync_interval_minutes" + >, + ) { + setBusy(connection.id); + setError(null); + try { + const updated = await apiClient.updateCommerceConnection( + connection.id, + changes, + ); + setConnections((current) => + current.map((item) => (item.id === updated.id ? updated : item)), + ); + } catch (requestError) { + setError( + requestError instanceof Error + ? requestError.message + : "Не удалось изменить расписание.", + ); + } finally { + setBusy(null); + } + } + async function disconnect(connection: IntegrationConnection) { if (!window.confirm("Отключить сервис? Уже импортированные товары сохранятся.")) { return; @@ -216,6 +245,13 @@ export default function ProductConnectionsPage() {
Последняя проверка
{displayDate(moysklad.last_tested_at)}
Последний импорт
{displayDate(moysklad.last_sync_finished_at)}
+
+ +
+ + Следующий запуск: {moysklad.auto_sync ? displayDate(moysklad.next_sync_at) : "отключён"} +
+
diff --git a/frontend/src/types/commerce.ts b/frontend/src/types/commerce.ts index 0634d4e3..47d89b45 100644 --- a/frontend/src/types/commerce.ts +++ b/frontend/src/types/commerce.ts @@ -64,6 +64,9 @@ export type IntegrationConnection = { sale_lines: number; }>; task_id: string; + auto_sync: boolean; + sync_interval_minutes: number; + next_sync_at: string | null; recent_runs: IntegrationSyncRun[]; created_at: string; updated_at: string; @@ -75,6 +78,8 @@ export type IntegrationConnectionInput = { token?: string; login?: string; password?: string; + auto_sync?: boolean; + sync_interval_minutes?: number; }; export type CommerceProduct = {