From 1c2dd985d4868abbe3d1d635a75bd18cb9bce887 Mon Sep 17 00:00:00 2001 From: Aleksandr Meshchriakov Date: Mon, 10 Aug 2026 15:02:00 +0200 Subject: [PATCH] fix manual parser refresh reliability --- src/apps/parsers/clients/gisp/client.py | 47 +++++++++++--- src/apps/parsers/clients/proverki/client.py | 71 +++++++++++---------- src/apps/parsers/views.py | 30 +++++++++ tests/apps/parsers/test_gisp_client.py | 29 +++++++++ tests/apps/parsers/test_proverki_client.py | 13 ++++ tests/apps/parsers/test_views.py | 21 +++++- 6 files changed, 165 insertions(+), 46 deletions(-) diff --git a/src/apps/parsers/clients/gisp/client.py b/src/apps/parsers/clients/gisp/client.py index 3b5da88..204ada9 100644 --- a/src/apps/parsers/clients/gisp/client.py +++ b/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") diff --git a/src/apps/parsers/clients/proverki/client.py b/src/apps/parsers/clients/proverki/client.py index bfc0df7..1d64593 100644 --- a/src/apps/parsers/clients/proverki/client.py +++ b/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 diff --git a/src/apps/parsers/views.py b/src/apps/parsers/views.py index 89fff37..3702954 100644 --- a/src/apps/parsers/views.py +++ b/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, diff --git a/tests/apps/parsers/test_gisp_client.py b/tests/apps/parsers/test_gisp_client.py index 0c59d8f..fbf4931 100644 --- a/tests/apps/parsers/test_gisp_client.py +++ b/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] = [] diff --git a/tests/apps/parsers/test_proverki_client.py b/tests/apps/parsers/test_proverki_client.py index 2bd8a65..9829b09 100644 --- a/tests/apps/parsers/test_proverki_client.py +++ b/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("") # noqa: S314 client = ProverkiClient() diff --git a/tests/apps/parsers/test_views.py b/tests/apps/parsers/test_views.py index 06c51ba..f3836a0 100644 --- a/tests/apps/parsers/test_views.py +++ b/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()