fix manual parser refresh reliability

src/apps/parsers/clients/gisp/client.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import logging
+import time
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -18,6 +20,8 @@ MAX_PAGE_SIZE = 100
 MAX_PAGES = 100
 MAX_RECORDS = 10_000
 
+logger = logging.getLogger(__name__)
+
 
 class GispProductsClientError(HTTPClientError):
     """Ошибка клиента реестра продукции ГИСП."""
@@ -44,6 +48,8 @@ class GispProductsClient:
     max_records: int = DEFAULT_MAX_RECORDS
     proxies: list[str] | None = None
     timeout: int = 120
+    max_retries: int = 2
+    retry_backoff_seconds: float = 1.0
     http_adapter: BaseAdapter | None = None
     _http_client: BaseHTTPClient | None = field(default=None, repr=False)
 
@@ -54,6 +60,10 @@ class GispProductsClient:
             raise ValueError(f"max_pages must be between 1 and {MAX_PAGES}")
         if not 1 <= self.max_records <= MAX_RECORDS:
             raise ValueError(f"max_records must be between 1 and {MAX_RECORDS}")
+        if self.max_retries < 0:
+            raise ValueError("max_retries must be non-negative")
+        if self.retry_backoff_seconds < 0:
+            raise ValueError("retry_backoff_seconds must be non-negative")
 
     @property
     def http_client(self) -> BaseHTTPClient:
@@ -129,16 +139,33 @@ class GispProductsClient:
         take: int,
         previous_total_count: int | None,
     ) -> tuple[list[Any], int | None]:
-        data = self.http_client.post_json(
-            PRODUCTS_ENDPOINT,
-            payload={
-                "opt": {
-                    "skip": offset,
-                    "take": take,
-                    "requireTotalCount": True,
-                }
-            },
-        )
+        payload = {
+            "opt": {
+                "skip": offset,
+                "take": take,
+                "requireTotalCount": True,
+            }
+        }
+        for attempt in range(self.max_retries + 1):
+            try:
+                data = self.http_client.post_json(
+                    PRODUCTS_ENDPOINT,
+                    payload=payload,
+                )
+                break
+            except HTTPClientError as exc:
+                transient = exc.status_code is None or exc.status_code >= 500
+                if not transient or attempt >= self.max_retries:
+                    raise
+                delay = self.retry_backoff_seconds * (2**attempt)
+                logger.warning(
+                    "GISP request failed temporarily (attempt %d/%d): %s",
+                    attempt + 1,
+                    self.max_retries + 1,
+                    exc,
+                )
+                if delay:
+                    time.sleep(delay)
         if data.get("ok") is False:
             raise GispProductsClientError("GISP products API returned ok=false")
         items = data.get("items")

src/apps/parsers/clients/proverki/client.py

@@ -693,7 +693,7 @@ class ProverkiClient:
 
         return inspections
 
-    def _parse_xml_streaming(
+    def _parse_xml_streaming(  # noqa: C901
         self,
         content: bytes,
         progress_callback: Callable[[int, str], None] | None = None,
@@ -704,52 +704,53 @@ class ProverkiClient:
         Использует iterparse для обработки файла по элементам,
         не загружая весь файл в память.
         """
-        inspections = []
-
-        # Декодируем и создаём поток
-        for encoding in ["utf-8", "windows-1251", "cp1251"]:
-            try:
-                xml_str = content.decode(encoding)
-                break
-            except UnicodeDecodeError:
-                continue
-        else:
-            xml_str = content.decode("utf-8", errors="replace")
-
-        xml_str = self._sanitize_xml(xml_str)
-
-        # Используем iterparse для потоковой обработки
         import io
 
-        xml_stream = io.StringIO(xml_str)
-
-        # Определяем теги, которые нас интересуют
+        inspections = []
         target_tags = {"INSPECTION", "inspection", "check", "КНМ"}
-
         count = 0
-        try:
+
+        def parse_stream(xml_stream: io.BytesIO | io.StringIO) -> None:
+            nonlocal count
             for _event, elem in ET.iterparse(xml_stream, events=["end"]):  # noqa: S314
-                # Извлекаем имя тега без namespace
                 tag_name = elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag
+                if tag_name not in target_tags:
+                    continue
+                inspection = self._parse_xml_record(elem)
+                if inspection:
+                    inspections.append(inspection)
+                    count += 1
+                    if count % 10000 == 0:
+                        logger.info("Streaming parsed %d inspections...", count)
+                elem.clear()
 
-                if tag_name in target_tags:
-                    inspection = self._parse_xml_record(elem)
-                    if inspection:
-                        inspections.append(inspection)
-                        count += 1
-
-                        if count % 10000 == 0:
-                            logger.info("Streaming parsed %d inspections...", count)
-
-                    # Очищаем элемент для освобождения памяти
-                    elem.clear()
-
+        try:
+            # ElementTree сам учитывает encoding из XML declaration. Для больших
+            # файлов это исключает несколько полноразмерных Unicode-копий XML.
+            parse_stream(io.BytesIO(content))
         except ET.ParseError as e:
-            logger.error("XML streaming parse error at %d records: %s", count, e)
             if inspections:
+                logger.error("XML streaming parse error at %d records: %s", count, e)
                 logger.info(
                     "Returning %d successfully parsed records", len(inspections)
                 )
+            elif len(content) <= self.STREAMING_THRESHOLD_BYTES:
+                # Совместимость с небольшими файлами без корректной декларации
+                # кодировки и с историческим поведением очистки XML.
+                for encoding in ["utf-8", "windows-1251", "cp1251"]:
+                    try:
+                        xml_str = content.decode(encoding)
+                        break
+                    except UnicodeDecodeError:
+                        continue
+                else:
+                    xml_str = content.decode("utf-8", errors="replace")
+                try:
+                    parse_stream(io.StringIO(self._sanitize_xml(xml_str)))
+                except ET.ParseError as fallback_error:
+                    raise ProverkiClientError(
+                        f"Failed to parse XML: {fallback_error}"
+                    ) from fallback_error
             else:
                 raise ProverkiClientError(f"Failed to parse XML: {e}") from e
 

src/apps/parsers/views.py

@@ -9,8 +9,10 @@ import csv
 import json
 import uuid
 from collections import defaultdict
+from datetime import timedelta
 
 from apps.core.filters import BaseFilterSet
+from apps.core.models import JobStatus
 from apps.core.openapi import CommonResponses, ErrorResponses, swagger_tag
 from apps.core.response import api_error_response, api_response
 from apps.core.serializers import BackgroundJobListSerializer
@@ -72,6 +74,7 @@ from django.core.paginator import Paginator
 from django.db.models import CharField, Count, Q
 from django.db.models.functions import Cast, Lower
 from django.http import HttpResponse
+from django.utils import timezone
 from django.utils.text import get_valid_filename
 from django_celery_beat.models import CrontabSchedule, IntervalSchedule, PeriodicTask
 from django_filters import rest_framework as filters
@@ -2656,6 +2659,33 @@ class ParserRunView(APIView):
         task_kwargs = build_task_kwargs(
             canonical_source_key, serializer.validated_data, request.user.id
         )
+        now = timezone.now()
+        active_job = (
+            BackgroundJobService.get_queryset()
+            .filter(task_name=descriptor.task_name)
+            .filter(
+                Q(
+                    status=JobStatus.PENDING,
+                    created_at__gte=now - timedelta(hours=24),
+                )
+                | Q(
+                    status__in=[JobStatus.STARTED, JobStatus.RETRY],
+                    updated_at__gte=now - timedelta(hours=4),
+                )
+            )
+            .order_by("-created_at")
+            .first()
+        )
+        if active_job is not None:
+            return api_response(
+                {
+                    "task_id": active_job.task_id,
+                    "source": descriptor.source,
+                    "task_name": descriptor.task_name,
+                    "already_running": True,
+                },
+                status_code=status.HTTP_202_ACCEPTED,
+            )
         task_id = str(uuid.uuid4())
         BackgroundJobService.create_job(
             task_id=task_id,

tests/apps/parsers/test_gisp_client.py

@@ -49,6 +49,35 @@ class GispProductsClientTest(SimpleTestCase):
         ):
             GispProductsClient(max_records=10_001)
 
+    def test_retries_transient_server_error(self):
+        calls = 0
+
+        def handle_page(_request, _body: bytes) -> Response:
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                return Response(status=504, body=b"gateway timeout")
+            return Response(
+                body=b'{"ok":true,"total_count":1,"items":[]}',
+                headers={"Content-Type": "application/json"},
+            )
+
+        with HTTPTestServer() as server:
+            server.add_route("POST", "/pp719v2/pub/prod/b/", handle_page)
+            client = GispProductsClient(
+                base_url=server.base_url,
+                max_pages=1,
+                retry_backoff_seconds=0,
+                http_adapter=server.adapter,
+            )
+            with self.assertRaisesMessage(
+                GispProductsClientError,
+                "empty page before total_count",
+            ):
+                client.fetch_products()
+
+        self.assertEqual(calls, 2)
+
     def test_fetch_products_follows_skip_pagination_and_maps_items(self):
         requests: list[dict] = []
 

tests/apps/parsers/test_proverki_client.py

@@ -9,6 +9,7 @@ import tempfile
 import types
 from asyncio import events as asyncio_events
 from pathlib import Path
+from unittest.mock import patch
 from xml.etree import ElementPath as element_path
 from xml.etree import ElementTree as ET
 
@@ -563,6 +564,18 @@ class ProverkiParseXMLTest(SimpleTestCase):
         inspections = client._parse_xml_content(xml)
         self.assertEqual(len(inspections), 1)
 
+    def test_large_xml_streaming_does_not_build_sanitized_string(self):
+        xml = _xml_with_tag("INSPECTION", _inspection_attrs())
+        client = ProverkiClient()
+        client.STREAMING_THRESHOLD_BYTES = 1
+        with patch.object(
+            client,
+            "_sanitize_xml",
+            side_effect=AssertionError("large XML must stay as bytes"),
+        ):
+            inspections = client._parse_xml_content(xml)
+        self.assertEqual(len(inspections), 1)
+
     def test_parse_xml_record_missing_fields_returns_none(self):
         element = ET.fromstring("<INSPECTION />")  # noqa: S314
         client = ProverkiClient()

tests/apps/parsers/test_views.py

@@ -11,7 +11,7 @@ import zipfile
 from datetime import date
 from unittest.mock import Mock, patch
 
-from apps.core.models import BackgroundJob
+from apps.core.models import BackgroundJob, JobStatus
 from apps.parsers.models import (
     FinancialReport,
     FinancialReportLine,
@@ -1428,3 +1428,22 @@ class ParsersViewSetTest(APITestCase):
         queued_task_id = apply_async_mock.call_args.kwargs["task_id"]
         job = BackgroundJob.objects.get(task_id=queued_task_id)
         self.assertEqual(job.meta["source_key"], "mpt_products")
+
+    def test_run_parser_reuses_fresh_active_job(self):
+        self.client.force_authenticate(self.user)
+        existing = BackgroundJob.objects.create(
+            task_id="active-products",
+            task_name="apps.parsers.tasks.parse_industrial_products",
+            status=JobStatus.STARTED,
+        )
+        url = reverse("api_v1:parsers:run-parser", args=["industrial_products"])
+
+        with patch(
+            "apps.parsers.views.tasks.parse_industrial_products.apply_async"
+        ) as apply_async_mock:
+            response = self.client.post(url, {}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+        self.assertEqual(response.data["data"]["task_id"], existing.task_id)
+        self.assertTrue(response.data["data"]["already_running"])
+        apply_async_mock.assert_not_called()