feat: update finance exchange and source refresh jobs

src/apps/core/models.py

@@ -169,12 +169,16 @@ class BackgroundJob(TimestampMixin, models.Model):
         self.status = JobStatus.SUCCESS
         self.progress = 100
         self.result = result
+        self.error = ""
+        self.traceback = ""
         self.completed_at = timezone.now()
         self.save(
             update_fields=[
                 "status",
                 "progress",
                 "result",
+                "error",
+                "traceback",
                 "completed_at",
                 "updated_at",
             ]

src/apps/core/services.py

@@ -752,21 +752,21 @@ class BackgroundJobService(BaseReadOnlyService):
         cls,
         *,
         max_age_minutes: int,
+        pending_max_age_minutes: int = 24 * 60,
         task_names: set[str] | None = None,
         meta_sources: set[str] | None = None,
     ) -> int:
         """Mark old active jobs as failed after worker restarts or hard kills."""
         from apps.core.models import JobStatus
 
-        cutoff = timezone.now() - timedelta(minutes=max_age_minutes)
-        queryset = (
-            cls.get_queryset()
-            .filter(
-                status__in=[JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY],
-            )
-            .filter(
-                Q(started_at__isnull=False, started_at__lt=cutoff)
-                | Q(started_at__isnull=True, created_at__lt=cutoff)
+        now = timezone.now()
+        cutoff = now - timedelta(minutes=max_age_minutes)
+        pending_cutoff = now - timedelta(minutes=pending_max_age_minutes)
+        queryset = cls.get_queryset().filter(
+            Q(status=JobStatus.PENDING, created_at__lt=pending_cutoff)
+            | Q(
+                status__in=[JobStatus.STARTED, JobStatus.RETRY],
+                updated_at__lt=cutoff,
             )
         )
         if task_names:

src/apps/exchange/state_corp_services.py

@@ -785,6 +785,10 @@ class StateCorpExchangeService:
             if not case_number or decision_date is None:
                 continue
 
+            claim_amount = cls._serialize_decimal(record.amount)
+            if claim_amount is None:
+                claim_amount = str(payload.get("claim_amount") or "").strip() or None
+
             items.append(
                 {
                     "organization_inn": cls._digits(record.inn),
@@ -795,6 +799,7 @@ class StateCorpExchangeService:
                     ).strip(),
                     "status": str(payload.get("status") or record.status or "").strip(),
                     "decision_date": decision_date.isoformat(),
+                    "claim_amount": claim_amount,
                 }
             )
         return items

src/apps/parsers/frontend_compat.py

@@ -8,6 +8,7 @@ from dataclasses import dataclass
 from datetime import timedelta
 from typing import Any
 
+from apps.core.models import JobStatus
 from apps.core.response import api_error_response, api_response
 from apps.core.services import BackgroundJobService
 from apps.parsers.models import (
@@ -46,7 +47,8 @@ SYSTEM_LOGS_TAG = "System Logs"
 ACTIVE_JOB_STATUSES = {"pending", "started", "retry"}
 SUCCESS_LOAD_STATUSES = {"success", "skipped"}
 ERROR_LOAD_STATUSES = {"failed", "failure", "error"}
-STALE_ACTIVE_MAX_AGE_MINUTES = 90
+STALE_ACTIVE_MAX_AGE_MINUTES = 4 * 60
+STALE_PENDING_MAX_AGE_MINUTES = 24 * 60
 
 PARSING_SETTINGS_CACHE_KEY = "parsers:frontend_compat:parsing_settings"
 PARSING_SETTINGS_FIELDS = {
@@ -297,15 +299,34 @@ def _active_tasks_for_definition(
         for source_key in definition.source_keys
         if source_key in PARSER_SOURCES
     ]
+    now = timezone.now()
+    active_cutoff = now - timedelta(
+        minutes=int(
+            getattr(
+                settings,
+                "PARSER_STALE_LOAD_MAX_AGE_MINUTES",
+                STALE_ACTIVE_MAX_AGE_MINUTES,
+            )
+        )
+    )
+    pending_cutoff = now - timedelta(
+        minutes=int(
+            getattr(
+                settings,
+                "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES",
+                STALE_PENDING_MAX_AGE_MINUTES,
+            )
+        )
+    )
     queryset = (
         BackgroundJobService.get_queryset()
+        .filter(task_name__in=task_names)
         .filter(
-            task_name__in=task_names,
-            status__in=ACTIVE_JOB_STATUSES,
-        )
-        .filter(
-            Q(started_at__isnull=False, started_at__gte=_stale_cutoff())
-            | Q(started_at__isnull=True, created_at__gte=_stale_cutoff())
+            Q(status=JobStatus.PENDING, created_at__gte=pending_cutoff)
+            | Q(
+                status__in=[JobStatus.STARTED, JobStatus.RETRY],
+                updated_at__gte=active_cutoff,
+            )
         )
     )
     return [_serialize_active_job(job) for job in queryset.order_by("-created_at")[:10]]

src/apps/parsers/migrations/0030_extend_stale_parser_job_timeouts.py

@@ -0,0 +1,39 @@
+import json
+
+from django.db import migrations
+
+CLEANUP_TASK_NAME = "parser:cleanup-stale-loads"
+ACTIVE_MAX_AGE_MINUTES = 4 * 60
+PENDING_MAX_AGE_MINUTES = 24 * 60
+
+
+def extend_stale_parser_job_timeouts(apps, schema_editor):
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PeriodicTask.objects.filter(name=CLEANUP_TASK_NAME).update(
+        kwargs=json.dumps(
+            {
+                "max_age_minutes": ACTIVE_MAX_AGE_MINUTES,
+                "pending_max_age_minutes": PENDING_MAX_AGE_MINUTES,
+            }
+        )
+    )
+
+
+def restore_stale_parser_job_timeouts(apps, schema_editor):
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PeriodicTask.objects.filter(name=CLEANUP_TASK_NAME).update(
+        kwargs=json.dumps({"max_age_minutes": 90})
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("parsers", "0029_neutralize_external_collection_labels"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            extend_stale_parser_job_timeouts,
+            reverse_code=restore_stale_parser_job_timeouts,
+        ),
+    ]

src/apps/parsers/serializers.py

@@ -945,6 +945,7 @@ class SourceCardItemSerializer(serializers.Serializer):
     """Подисточник внутри агрегированной карточки."""
 
     code = serializers.CharField(read_only=True)
+    refresh_key = serializers.CharField(read_only=True)
     title = serializers.CharField(read_only=True)
     description = serializers.CharField(read_only=True)
     parser_source = serializers.CharField(read_only=True, allow_null=True)

src/apps/parsers/services.py

@@ -419,11 +419,14 @@ class ParserLoadLogService(BaseService[ParserLoadLog]):
         cls,
         *,
         max_age_minutes: int,
+        pending_max_age_minutes: int = 24 * 60,
     ) -> int:
         """Закрыть зависшие in_progress логи без живой свежей BackgroundJob."""
         from apps.core.models import BackgroundJob, JobStatus
 
-        cutoff = timezone.now() - timedelta(minutes=max_age_minutes)
+        now = timezone.now()
+        cutoff = now - timedelta(minutes=max_age_minutes)
+        pending_cutoff = now - timedelta(minutes=pending_max_age_minutes)
         stale_logs = list(
             cls.model.objects.filter(
                 status=ParserLoadLog.Status.IN_PROGRESS,
@@ -442,8 +445,11 @@ class ParserLoadLogService(BaseService[ParserLoadLog]):
                 meta__source=log.source,
             ).filter(Q(meta__batch_id=log.batch_id) | Q(meta__batch_id__isnull=True))
             fresh_jobs = active_jobs.filter(
-                Q(started_at__isnull=False, started_at__gte=cutoff)
-                | Q(started_at__isnull=True, created_at__gte=cutoff)
+                Q(status=JobStatus.PENDING, created_at__gte=pending_cutoff)
+                | Q(
+                    status__in=[JobStatus.STARTED, JobStatus.RETRY],
+                    updated_at__gte=cutoff,
+                )
             )
             if fresh_jobs.exists():
                 continue
@@ -451,8 +457,11 @@ class ParserLoadLogService(BaseService[ParserLoadLog]):
             cls.mark_failed(log, stale_message)
             updated += 1
             stale_jobs = active_jobs.filter(
-                Q(started_at__isnull=False, started_at__lt=cutoff)
-                | Q(started_at__isnull=True, created_at__lt=cutoff)
+                Q(status=JobStatus.PENDING, created_at__lt=pending_cutoff)
+                | Q(
+                    status__in=[JobStatus.STARTED, JobStatus.RETRY],
+                    updated_at__lt=cutoff,
+                )
             )
             for job in stale_jobs.order_by("created_at"):
                 job.fail(error=stale_message)

src/apps/parsers/source_cards.py

@@ -31,7 +31,8 @@ from rest_framework.exceptions import ValidationError
 
 SUCCESSFUL_LOAD_STATUSES = {"success", "skipped"}
 ACTIVE_JOB_STATUSES = [JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY]
-STALE_ACTIVE_MAX_AGE_MINUTES = 90
+STALE_ACTIVE_MAX_AGE_MINUTES = 4 * 60
+STALE_PENDING_MAX_AGE_MINUTES = 24 * 60
 SOURCE_CARD_STATS_CACHE_TIMEOUT_SECONDS = 7 * 24 * 60 * 60
 
 
@@ -55,6 +56,7 @@ class SourceItemDefinition:
     title: str
     description: str
     parser_source: str | None = None
+    refresh_key: str | None = None
 
 
 @dataclass(frozen=True)
@@ -218,6 +220,7 @@ SOURCE_CARD_DEFINITIONS: tuple[SourceCardDefinition, ...] = (
                     "Реестр промышленной продукции, произведенной на территории РФ."
                 ),
                 parser_source=ParserLoadLog.Source.INDUSTRIAL_PRODUCTS,
+                refresh_key="mpt_products",
             ),
             SourceItemDefinition(
                 code="manufactures",
@@ -769,17 +772,12 @@ class SourceCardService:
         active_tasks_by_slug: dict[str, list[dict[str, Any]]] = {
             slug: [] for slug in task_names_by_slug
         }
-        cutoff = cls._stale_cutoff()
         queryset = (
             BackgroundJobService.get_queryset()
             .filter(
                 task_name__in=list(slugs_by_task_name),
-                status__in=ACTIVE_JOB_STATUSES,
-            )
-            .filter(
-                Q(started_at__isnull=False, started_at__gte=cutoff)
-                | Q(started_at__isnull=True, created_at__gte=cutoff)
             )
+            .filter(cls._fresh_active_job_filter())
             .order_by("-created_at")
         )
 
@@ -1157,6 +1155,7 @@ class SourceCardService:
 
         return {
             "code": item.code,
+            "refresh_key": item.refresh_key or item.code,
             "title": item.title,
             "description": item.description,
             "parser_source": item.parser_source,
@@ -1288,17 +1287,12 @@ class SourceCardService:
     def _get_active_tasks(
         cls, definition: SourceCardDefinition
     ) -> list[dict[str, Any]]:
-        cutoff = cls._stale_cutoff()
         queryset = (
             BackgroundJobService.get_queryset()
             .filter(
                 task_name__in=definition.task_names,
-                status__in=ACTIVE_JOB_STATUSES,
-            )
-            .filter(
-                Q(started_at__isnull=False, started_at__gte=cutoff)
-                | Q(started_at__isnull=True, created_at__gte=cutoff)
             )
+            .filter(cls._fresh_active_job_filter())
         )
         return [
             cls._serialize_job(job) for job in queryset.order_by("-created_at")[:10]
@@ -1349,6 +1343,31 @@ class SourceCardService:
         )
         return timezone.now() - timedelta(minutes=max_age_minutes)
 
+    @classmethod
+    def _fresh_active_job_filter(cls) -> Q:
+        now = timezone.now()
+        active_max_age_minutes = int(
+            getattr(
+                settings,
+                "PARSER_STALE_LOAD_MAX_AGE_MINUTES",
+                STALE_ACTIVE_MAX_AGE_MINUTES,
+            )
+        )
+        pending_max_age_minutes = int(
+            getattr(
+                settings,
+                "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES",
+                STALE_PENDING_MAX_AGE_MINUTES,
+            )
+        )
+        return Q(
+            status=JobStatus.PENDING,
+            created_at__gte=now - timedelta(minutes=pending_max_age_minutes),
+        ) | Q(
+            status__in=[JobStatus.STARTED, JobStatus.RETRY],
+            updated_at__gte=now - timedelta(minutes=active_max_age_minutes),
+        )
+
     @classmethod
     def _is_stale_load(cls, latest_load: ParserLoadLog | None) -> bool:
         if latest_load is None or latest_load.status != "in_progress":

src/apps/parsers/tasks.py

@@ -88,7 +88,8 @@ REGISTRY_INSPECTIONS_CHECKO_LIMIT = 1000
 REGISTRY_CONTRACTS_CHECKO_LIMIT = 1000
 REGISTRY_ENRICHMENT_BATCH_SIZE = 250
 FSTEC_CHECKO_IDENTITY_LOOKUP_LIMIT = 1000
-PARSER_STALE_LOAD_MAX_AGE_MINUTES = 90
+PARSER_STALE_LOAD_MAX_AGE_MINUTES = 4 * 60
+PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES = 24 * 60
 PARSER_SOFT_TIME_LIMIT_SECONDS = 15 * 60
 PARSER_TIME_LIMIT_SECONDS = 20 * 60
 INDUSTRIAL_PRODUCTS_SOFT_TIME_LIMIT_SECONDS = 45 * 60
@@ -3604,7 +3605,10 @@ def parse_fstec_registers(
 
 
 @shared_task
-def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict:
+def cleanup_stale_parser_loads(
+    max_age_minutes: int | None = None,
+    pending_max_age_minutes: int | None = None,
+) -> dict:
     """Закрыть stale in_progress загрузки и jobs после рестартов worker/deploy."""
     if max_age_minutes is None:
         max_age_minutes = getattr(
@@ -3612,14 +3616,22 @@ def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict:
             "PARSER_STALE_LOAD_MAX_AGE_MINUTES",
             PARSER_STALE_LOAD_MAX_AGE_MINUTES,
         )
+    if pending_max_age_minutes is None:
+        pending_max_age_minutes = getattr(
+            settings,
+            "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES",
+            PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES,
+        )
     source_values = {descriptor.source for descriptor in PARSER_SOURCES.values()}
     task_names = {descriptor.task_name for descriptor in PARSER_SOURCES.values()}
     task_names.add("apps.parsers.tasks.scan_fns_directory")
     marked_failed = ParserLoadLogService.mark_stale_in_progress_failed(
-        max_age_minutes=int(max_age_minutes)
+        max_age_minutes=int(max_age_minutes),
+        pending_max_age_minutes=int(pending_max_age_minutes),
     )
     marked_jobs_failed = BackgroundJobService.mark_stale_active_jobs_failed(
         max_age_minutes=int(max_age_minutes),
+        pending_max_age_minutes=int(pending_max_age_minutes),
         task_names=task_names,
         meta_sources=source_values,
     )
@@ -3628,6 +3640,7 @@ def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict:
         "marked_failed": marked_failed,
         "marked_jobs_failed": marked_jobs_failed,
         "max_age_minutes": int(max_age_minutes),
+        "pending_max_age_minutes": int(pending_max_age_minutes),
     }
 
 

src/apps/parsers/views.py

@@ -151,6 +151,10 @@ TASKS_BY_NAME = {
     "apps.parsers.tasks.parse_trudvsem_vacancies": tasks.parse_trudvsem_vacancies,
 }
 
+PARSER_SOURCE_ALIASES = {
+    "industrial_products": "mpt_products",
+}
+
 
 class MultipartFormSwaggerAutoSchema(SwaggerAutoSchema):
     """Document mixed JSON/multipart upload endpoints as form-data in Swagger."""
@@ -2642,21 +2646,25 @@ class ParserRunView(APIView):
     permission_classes = [IsAuthenticated]
 
     def post(self, request: Request, source_key: str):
-        descriptor = PARSER_SOURCES.get(source_key)
+        canonical_source_key = PARSER_SOURCE_ALIASES.get(source_key, source_key)
+        descriptor = PARSER_SOURCES.get(canonical_source_key)
         if descriptor is None:
             return _source_not_found_response(source_key)
         serializer = ParserRunRequestSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         task = TASKS_BY_NAME[descriptor.task_name]
         task_kwargs = build_task_kwargs(
-            source_key, serializer.validated_data, request.user.id
+            canonical_source_key, serializer.validated_data, request.user.id
         )
         task_id = str(uuid.uuid4())
         BackgroundJobService.create_job(
             task_id=task_id,
             task_name=descriptor.task_name,
             user_id=request.user.id,
-            meta={"source_key": source_key, "source": descriptor.source},
+            meta={
+                "source_key": canonical_source_key,
+                "source": descriptor.source,
+            },
         )
         async_result = task.apply_async(kwargs=task_kwargs, task_id=task_id)
         return api_response(

src/organizations/test_companies.py

@@ -661,6 +661,7 @@ class TestCompanyDatasetService:
 
     @staticmethod
     def _refresh_financial_lines(*, record, index: int) -> None:
+        report_year = timezone.localdate().year
         expected = {
             ("1", "1600", "Баланс (актив)", 10_000_000 + index * 100_000),
             ("1", "1300", "Капитал и резервы", 4_000_000 + index * 50_000),
@@ -669,12 +670,12 @@ class TestCompanyDatasetService:
         }
         expected_keys = []
         for form_code, line_code, line_name, period_end in expected:
-            expected_keys.append((form_code, line_code, 2025))
+            expected_keys.append((form_code, line_code, report_year))
             OrganizationSourceFinancialLine.objects.update_or_create(
                 source_record=record,
                 form_code=form_code,
                 line_code=line_code,
-                year=2025,
+                year=report_year,
                 defaults={
                     "line_name": line_name,
                     "period_start": period_end - 100_000,