diff --git a/src/apps/exchange/state_corp_services.py b/src/apps/exchange/state_corp_services.py index b02cdd4..a6ecf75 100644 --- a/src/apps/exchange/state_corp_services.py +++ b/src/apps/exchange/state_corp_services.py @@ -614,7 +614,6 @@ class StateCorpExchangeService: ParserLoadLog.Source.PROCUREMENTS, ParserLoadLog.Source.PROCUREMENTS_44FZ, ParserLoadLog.Source.PROCUREMENTS_223FZ, - ParserLoadLog.Source.CONTRACTS, ], ): item = cls._serialize_generic_public_procurement(record) @@ -676,8 +675,6 @@ class StateCorpExchangeService: return "44-ФЗ" if source == ParserLoadLog.Source.PROCUREMENTS_223FZ: return "223-ФЗ" - if source == ParserLoadLog.Source.CONTRACTS: - return "contract" return "" @classmethod diff --git a/src/apps/parsers/checko_collection.py b/src/apps/parsers/checko_collection.py new file mode 100644 index 0000000..ab04a0b --- /dev/null +++ b/src/apps/parsers/checko_collection.py @@ -0,0 +1,56 @@ +"""Quota-safe monthly claims for organization lookups in Checko.""" + +from datetime import date, datetime + +from apps.parsers.models import CheckoCollectionAttempt +from django.utils import timezone + + +def collection_period_month(at: date | datetime | None = None) -> date: + if at is None: + local_date = timezone.localdate() + elif isinstance(at, datetime): + local_date = timezone.localtime(at).date() if timezone.is_aware(at) else at.date() + else: + local_date = at + return local_date.replace(day=1) + + +def claim_monthly_collection( + *, + organization_id: str, + source: str, + at: date | datetime | None = None, +) -> CheckoCollectionAttempt | None: + """Atomically claim this month's only allowed Checko collection attempt.""" + attempt, created = CheckoCollectionAttempt.objects.get_or_create( + organization_id=organization_id, + source=source, + period_month=collection_period_month(at), + defaults={"status": CheckoCollectionAttempt.Status.IN_PROGRESS}, + ) + return attempt if created else None + + +def finish_collection( + attempt: CheckoCollectionAttempt, + *, + records_count: int, + error: Exception | str | None = None, +) -> None: + """Persist the outcome without releasing the monthly claim.""" + attempt.records_count = max(int(records_count), 0) + attempt.error_message = str(error or "") + attempt.status = ( + CheckoCollectionAttempt.Status.FAILED + if error is not None + else CheckoCollectionAttempt.Status.SUCCESS + ) + attempt.save( + update_fields=[ + "records_count", + "error_message", + "status", + "updated_at", + ] + ) diff --git a/src/apps/parsers/migrations/0027_checkocollectionattempt.py b/src/apps/parsers/migrations/0027_checkocollectionattempt.py new file mode 100644 index 0000000..2f8ab58 --- /dev/null +++ b/src/apps/parsers/migrations/0027_checkocollectionattempt.py @@ -0,0 +1,116 @@ +# Generated by Django 5.2.9 on 2026-07-19 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("organizations", "0007_auto_20260607_1017"), + ("parsers", "0026_update_fns_financial_schedule"), + ] + + operations = [ + migrations.CreateModel( + name="CheckoCollectionAttempt", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "created_at", + models.DateTimeField( + auto_now_add=True, + db_index=True, + help_text="Дата и время создания записи", + verbose_name="создано", + ), + ), + ( + "updated_at", + models.DateTimeField( + auto_now=True, + help_text="Дата и время последнего обновления", + verbose_name="обновлено", + ), + ), + ( + "source", + models.CharField( + choices=[ + ("arbitration", "Арбитражные дела"), + ("bankruptcy", "Банкротства"), + ("contracts", "Контракты"), + ("inspections", "Проверки"), + ], + max_length=32, + verbose_name="источник", + ), + ), + ( + "period_month", + models.DateField( + help_text="Первый день календарного месяца, за который занята попытка", + verbose_name="месяц сбора", + ), + ), + ( + "status", + models.CharField( + choices=[ + ("in_progress", "В процессе"), + ("success", "Успешно"), + ("failed", "Ошибка"), + ], + default="in_progress", + max_length=20, + verbose_name="статус", + ), + ), + ( + "records_count", + models.PositiveIntegerField( + default=0, + verbose_name="количество записей", + ), + ), + ( + "error_message", + models.TextField(blank=True, verbose_name="сообщение об ошибке"), + ), + ( + "organization", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="checko_collection_attempts", + to="organizations.organization", + verbose_name="организация", + ), + ), + ], + options={ + "verbose_name": "попытка сбора Checko", + "verbose_name_plural": "попытки сбора Checko", + "db_table": "parsers_checko_collection_attempt", + "ordering": ["-period_month", "source", "organization_id"], + "indexes": [ + models.Index( + fields=["source", "period_month"], + name="parsers_che_source_0ca426_idx", + ) + ], + "constraints": [ + models.UniqueConstraint( + fields=("organization", "source", "period_month"), + name="unique_checko_collection_per_org_source_month", + ) + ], + }, + ), + ] diff --git a/src/apps/parsers/models.py b/src/apps/parsers/models.py index 70433ca..771f669 100644 --- a/src/apps/parsers/models.py +++ b/src/apps/parsers/models.py @@ -117,6 +117,63 @@ class ParserBatchSequence(TimestampMixin, models.Model): return f"{self.source}: next batch {self.next_batch_id}" +class CheckoCollectionAttempt(TimestampMixin, models.Model): + """Monthly Checko collection claim for one organization and source.""" + + class Source(models.TextChoices): + ARBITRATION = "arbitration", _("Арбитражные дела") + BANKRUPTCY = "bankruptcy", _("Банкротства") + CONTRACTS = "contracts", _("Контракты") + INSPECTIONS = "inspections", _("Проверки") + + class Status(models.TextChoices): + IN_PROGRESS = "in_progress", _("В процессе") + SUCCESS = "success", _("Успешно") + FAILED = "failed", _("Ошибка") + + organization = models.ForeignKey( + "organizations.Organization", + on_delete=models.CASCADE, + related_name="checko_collection_attempts", + verbose_name=_("организация"), + ) + source = models.CharField( + _("источник"), + max_length=32, + choices=Source.choices, + ) + period_month = models.DateField( + _("месяц сбора"), + help_text=_("Первый день календарного месяца, за который занята попытка"), + ) + status = models.CharField( + _("статус"), + max_length=20, + choices=Status.choices, + default=Status.IN_PROGRESS, + ) + records_count = models.PositiveIntegerField(_("количество записей"), default=0) + error_message = models.TextField(_("сообщение об ошибке"), blank=True) + + class Meta: + db_table = "parsers_checko_collection_attempt" + verbose_name = _("попытка сбора Checko") + verbose_name_plural = _("попытки сбора Checko") + ordering = ["-period_month", "source", "organization_id"] + constraints = [ + models.UniqueConstraint( + fields=["organization", "source", "period_month"], + name="unique_checko_collection_per_org_source_month", + ), + ] + indexes = [ + models.Index(fields=["source", "period_month"]), + ] + + def __str__(self) -> str: + return f"{self.organization_id}: {self.source} ({self.period_month:%Y-%m})" + + GENERIC_RECORD_SOURCE_CHOICES = [ *ParserLoadLog.Source.choices, ("hh", _("Вакансии HeadHunter")), diff --git a/src/apps/parsers/tasks.py b/src/apps/parsers/tasks.py index ff33f75..4c1dda9 100644 --- a/src/apps/parsers/tasks.py +++ b/src/apps/parsers/tasks.py @@ -18,6 +18,11 @@ from pathlib import Path from apps.core.services import BackgroundJobService from apps.core.tasks import PeriodicTask as CorePeriodicTask +from apps.parsers.checko_collection import ( + claim_monthly_collection, + collection_period_month, + finish_collection, +) from apps.parsers.clients.base import HTTPClientError from apps.parsers.clients.checko import ( CheckoClient, @@ -43,7 +48,7 @@ from apps.parsers.clients.proverki import ProverkiClient from apps.parsers.clients.proverki.schemas import Inspection as ProverkiInspection from apps.parsers.clients.vacancies import VacanciesClient from apps.parsers.clients.zakupki import ZakupkiClient -from apps.parsers.models import ParserLoadLog +from apps.parsers.models import CheckoCollectionAttempt, ParserLoadLog from apps.parsers.services import ( FNSReportOrganizationResolutionSkipped, FNSReportService, @@ -171,11 +176,17 @@ def _resolve_lookup_limit( def _active_registry_lookup_targets( *, limit: int | None = None, + checko_source: str | None = None, ) -> list[RegistryLookupTarget]: """Вернуть организации, которые сейчас состоят хотя бы в одном реестре.""" + queryset = SourceOrganization.objects.filter(opk_registry_membership=True) + if checko_source is not None: + queryset = queryset.exclude( + checko_collection_attempts__source=checko_source, + checko_collection_attempts__period_month=collection_period_month(), + ) queryset = ( - SourceOrganization.objects.filter(opk_registry_membership=True) - .order_by("inn", "ogrn", "uid") + queryset.order_by("inn", "ogrn", "uid") .values( "uid", "inn", @@ -817,9 +828,14 @@ def _fetch_checko_bankruptcy_records( if limit <= 0: logger.info("Fedresurs Checko fallback is disabled by limit=%s", limit) return [] - targets = _active_registry_lookup_targets(limit=limit) + targets = _active_registry_lookup_targets( + limit=limit, + checko_source=CheckoCollectionAttempt.Source.BANKRUPTCY, + ) if not targets: - logger.info("No active registry organizations found for Fedresurs fallback") + logger.info( + "No active registry organizations are due for monthly Fedresurs fallback" + ) return [] checko_proxies = ( @@ -828,11 +844,19 @@ def _fetch_checko_bankruptcy_records( client = CheckoClient(api_key=api_key, proxies=checko_proxies, timeout=30) records: list[GenericParserItem] = [] for target in targets: + attempt = claim_monthly_collection( + organization_id=target.organization_id, + source=CheckoCollectionAttempt.Source.BANKRUPTCY, + ) + if attempt is None: + continue + records_before = len(records) try: response = client.get_company( CompanyRequest(inn=target.inn or None, ogrn=target.ogrn or None) ) except CheckoRateLimitError as exc: + finish_collection(attempt, records_count=0, error=exc) logger.warning( "Checko bankruptcy fallback stopped: quota/rate limit reached " "(status_code=%s)", @@ -840,6 +864,7 @@ def _fetch_checko_bankruptcy_records( ) break except CheckoError as exc: + finish_collection(attempt, records_count=0, error=exc) logger.info( "Checko bankruptcy lookup skipped for target=%s: %s", target.inn or target.ogrn, @@ -847,15 +872,18 @@ def _fetch_checko_bankruptcy_records( ) continue company = response.data - if company is None: - continue - records.extend( - _checko_bankruptcy_items( - company=company, - fallback_inn=target.inn, - fallback_ogrn=target.ogrn, - fallback_name=target.name, + if company is not None: + records.extend( + _checko_bankruptcy_items( + company=company, + fallback_inn=target.inn, + fallback_ogrn=target.ogrn, + fallback_name=target.name, + ) ) + finish_collection( + attempt, + records_count=len(records) - records_before, ) logger.info("Fetched %d bankruptcy records through Checko fallback", len(records)) return records @@ -990,7 +1018,10 @@ def _add_arbitration_subject( def _arbitration_subjects(limit: int) -> list[ArbitrationSubject]: """Собрать активные организации из реестров для арбитражного lookup.""" subjects: dict[str, ArbitrationSubject] = {} - for target in _active_registry_lookup_targets(limit=limit): + for target in _active_registry_lookup_targets( + limit=limit, + checko_source=CheckoCollectionAttempt.Source.ARBITRATION, + ): _add_arbitration_subject( subjects, inn=target.inn, @@ -1031,7 +1062,7 @@ def _fetch_checko_arbitration_records( subjects = _arbitration_subjects(resolved_limit) if not subjects: raise ParserSourceSkipped( - "no active registry organizations found for arbitration" + "no active registry organizations are due for monthly arbitration lookup" ) checko_proxies = ( @@ -1040,7 +1071,16 @@ def _fetch_checko_arbitration_records( client = CheckoClient(api_key=api_key, proxies=checko_proxies, timeout=30) records: list[GenericParserItem] = [] failed_lookups = 0 + attempted_lookups = 0 for subject in subjects: + attempt = claim_monthly_collection( + organization_id=subject.registry_organization_id, + source=CheckoCollectionAttempt.Source.ARBITRATION, + ) + if attempt is None: + continue + attempted_lookups += 1 + records_before = len(records) try: request = LegalCasesRequest( inn=subject.inn or None, @@ -1051,13 +1091,19 @@ def _fetch_checko_arbitration_records( records.append(_checko_arbitration_item(legal_case, subject=subject)) except CheckoError as exc: failed_lookups += 1 + finish_collection(attempt, records_count=0, error=exc) logger.info( "Checko arbitration lookup skipped for subject=%s: %s", _arbitration_subject_key(subject), exc, ) + else: + finish_collection( + attempt, + records_count=len(records) - records_before, + ) - if failed_lookups == len(subjects) and not records: + if attempted_lookups and failed_lookups == attempted_lookups and not records: raise ParserSourceSkipped("Checko arbitration lookups failed for all subjects") logger.info( @@ -1273,10 +1319,13 @@ def _fetch_checko_registry_inspections( logger.info("Registry inspections Checko parser is disabled by limit=%s", limit) return [] - targets = _active_registry_lookup_targets(limit=resolved_limit) + targets = _active_registry_lookup_targets( + limit=resolved_limit, + checko_source=CheckoCollectionAttempt.Source.INSPECTIONS, + ) if not targets: raise ParserSourceSkipped( - "no active registry organizations found for registry inspections" + "no active registry organizations are due for monthly inspections lookup" ) checko_proxies = ( @@ -1285,7 +1334,16 @@ def _fetch_checko_registry_inspections( client = CheckoClient(api_key=api_key, proxies=checko_proxies, timeout=30) records: list[ProverkiInspection] = [] failed_lookups = 0 + attempted_lookups = 0 for target in targets: + attempt = claim_monthly_collection( + organization_id=target.organization_id, + source=CheckoCollectionAttempt.Source.INSPECTIONS, + ) + if attempt is None: + continue + attempted_lookups += 1 + records_before = len(records) try: request = InspectionsRequest( inn=target.inn or None, @@ -1296,13 +1354,19 @@ def _fetch_checko_registry_inspections( records.append(_checko_inspection_item(inspection, target=target)) except CheckoError as exc: failed_lookups += 1 + finish_collection(attempt, records_count=0, error=exc) logger.info( "Checko inspections lookup skipped for target=%s: %s", target.inn or target.ogrn, exc, ) + else: + finish_collection( + attempt, + records_count=len(records) - records_before, + ) - if failed_lookups == len(targets) and not records: + if attempted_lookups and failed_lookups == attempted_lookups and not records: raise ParserSourceSkipped("Checko inspections lookups failed for all targets") logger.info( @@ -1429,10 +1493,13 @@ def _fetch_checko_registry_contract_records( logger.info("Registry contracts Checko parser is disabled by limit=%s", limit) return [] - targets = _active_registry_lookup_targets(limit=resolved_limit) + targets = _active_registry_lookup_targets( + limit=resolved_limit, + checko_source=CheckoCollectionAttempt.Source.CONTRACTS, + ) if not targets: raise ParserSourceSkipped( - "no active registry organizations found for registry contracts" + "no active registry organizations are due for monthly contracts lookup" ) checko_proxies = ( @@ -1441,7 +1508,17 @@ def _fetch_checko_registry_contract_records( client = CheckoClient(api_key=api_key, proxies=checko_proxies, timeout=30) records: list[GenericParserItem] = [] failed_lookups = 0 + attempted_lookups = 0 for target in targets: + attempt = claim_monthly_collection( + organization_id=target.organization_id, + source=CheckoCollectionAttempt.Source.CONTRACTS, + ) + if attempt is None: + continue + attempted_lookups += 1 + records_before = len(records) + target_failures: list[CheckoError] = [] for law in (ContractLaw.FZ44, ContractLaw.FZ223): try: request = ContractsRequest( @@ -1456,15 +1533,21 @@ def _fetch_checko_registry_contract_records( ) except CheckoError as exc: failed_lookups += 1 + target_failures.append(exc) logger.info( "Checko contracts lookup skipped for target=%s law=%s: %s", target.inn or target.ogrn, law.value, exc, ) + finish_collection( + attempt, + records_count=len(records) - records_before, + error=target_failures[0] if len(target_failures) == 2 else None, + ) - expected_lookups = len(targets) * 2 - if failed_lookups == expected_lookups and not records: + expected_lookups = attempted_lookups * 2 + if expected_lookups and failed_lookups == expected_lookups and not records: raise ParserSourceSkipped("Checko contracts lookups failed for all targets") logger.info( @@ -2242,6 +2325,9 @@ def parse_all_sources( proxies=proxies ).id, "contracts": parse_contracts.delay(proxies=proxies).id, + "registry_contracts": parse_registry_contracts.delay( + proxies=proxies + ).id, "unfair_suppliers": parse_unfair_suppliers.delay(proxies=proxies).id, "fas_goz": parse_fas_goz_evasion.delay(proxies=proxies).id, "fns_financial": sync_fns_financial_reports.delay(proxies=proxies).id, @@ -2249,6 +2335,9 @@ def parse_all_sources( "fedresurs_bankruptcy": parse_fedresurs_bankruptcy.delay( proxies=proxies ).id, + "registry_inspections": parse_registry_inspections.delay( + proxies=proxies + ).id, "fstec": parse_fstec_registers.delay(proxies=proxies).id, "trudvsem": parse_trudvsem_vacancies.delay(proxies=proxies).id, } diff --git a/src/organizations/directory_import.py b/src/organizations/directory_import.py index e213219..dcab28c 100644 --- a/src/organizations/directory_import.py +++ b/src/organizations/directory_import.py @@ -14,6 +14,7 @@ from django.db import transaction from django.utils import timezone from openpyxl import load_workbook +from organizations.cache import invalidate_organization_api_cache from organizations.models import Organization DIRECTORY_SHEET = "Лист1" @@ -142,12 +143,14 @@ class OrganizationDirectoryImportService: else: updated += 1 - return OrganizationDirectoryImportResult( + result = OrganizationDirectoryImportResult( scanned=scanned, created=created, updated=updated, skipped=skipped, ) + transaction.on_commit(invalidate_organization_api_cache) + return result @classmethod def _organization_defaults( diff --git a/src/organizations/test_companies.py b/src/organizations/test_companies.py index 1fea671..8a48023 100644 --- a/src/organizations/test_companies.py +++ b/src/organizations/test_companies.py @@ -8,7 +8,16 @@ from decimal import Decimal from hashlib import sha256 from uuid import UUID, uuid5 -from apps.parsers.models import ParserLoadLog +from apps.parsers.models import ( + FinancialReport, + GenericParserRecord, + IndustrialCertificateRecord, + IndustrialProductRecord, + InspectionRecord, + ManufacturerRecord, + ParserLoadLog, + ProcurementRecord, +) from django.db.models import Count, Max, Min from django.utils import timezone @@ -105,6 +114,21 @@ class TestCompanyDatasetService: **record_defaults, }, ) + legacy_record = cls._sync_parser_result_record( + source=source, + external_id=external_id, + organization=organization, + defaults=record_defaults, + index=index, + ) + legacy_module = legacy_record.__class__.__module__.removesuffix( + ".models" + ) + record.legacy_model = ( + f"{legacy_module}.{legacy_record.__class__.__name__}" + ) + record.legacy_pk = str(legacy_record.pk) + record.save(update_fields=["legacy_model", "legacy_pk", "updated_at"]) if source == ParserLoadLog.Source.FNS_REPORTS: cls._refresh_financial_lines(record=record, index=index) @@ -129,6 +153,7 @@ class TestCompanyDatasetService: company_uids = cls.company_uids() queryset = Organization.objects.filter(uid__in=company_uids) deleted_count = queryset.count() + cls._delete_parser_result_records(company_uids) extension_models = { descriptor.extension_model for source, descriptor in SOURCE_GROUP_DESCRIPTORS.items() @@ -140,6 +165,172 @@ class TestCompanyDatasetService: cls._invalidate_caches() return TestCompanyDatasetResult(organizations_deleted=deleted_count) + @classmethod + def _sync_parser_result_record( + cls, + *, + source: str, + external_id: str, + organization: Organization, + defaults: dict, + index: int, + ): + """Mirror demo rows into the parser tables served by public source APIs.""" + payload = defaults.get("payload", {}) + load_batch = 9_000_000 + index + common = { + "load_batch": load_batch, + "registry_organization": organization, + } + + if source == ParserLoadLog.Source.INDUSTRIAL: + legacy_record, _ = IndustrialCertificateRecord.objects.update_or_create( + certificate_number=payload["certificate_number"], + defaults={ + **common, + "issue_date": payload["issue_date"], + "issue_date_normalized": date.fromisoformat( + payload["issue_date_normalized"] + ), + "expiry_date": payload["expiry_date"], + "expiry_date_normalized": date.fromisoformat( + payload["expiry_date_normalized"] + ), + "certificate_file_url": payload["certificate_file_url"], + "organisation_name": organization.name, + "inn": organization.inn, + "ogrn": organization.ogrn, + }, + ) + return legacy_record + + if source == ParserLoadLog.Source.INDUSTRIAL_PRODUCTS: + legacy_record, _ = IndustrialProductRecord.objects.update_or_create( + registry_number=payload["registry_number"], + defaults={ + **common, + "full_organisation_name": payload["full_organisation_name"], + "ogrn": organization.ogrn, + "inn": organization.inn, + "product_name": payload["product_name"], + "product_model": payload["product_model"], + "okpd2_code": payload["okpd2_code"], + "tnved_code": payload["tnved_code"], + "regulatory_document": payload["regulatory_document"], + }, + ) + return legacy_record + + if source == ParserLoadLog.Source.MANUFACTURES: + legacy_record, _ = ManufacturerRecord.objects.update_or_create( + inn=organization.inn, + defaults={ + **common, + "full_legal_name": payload["full_legal_name"], + "ogrn": organization.ogrn, + "address": payload["address"], + }, + ) + return legacy_record + + if source == ParserLoadLog.Source.INSPECTIONS: + legacy_record, _ = InspectionRecord.objects.update_or_create( + registration_number=payload["registration_number"], + defaults={ + **common, + "inn": organization.inn, + "ogrn": organization.ogrn, + "organisation_name": organization.name, + "control_authority": payload["control_authority"], + "inspection_type": payload["inspection_type"], + "inspection_form": payload["inspection_form"], + "start_date": payload["start_date"], + "start_date_normalized": date.fromisoformat( + payload["start_date_normalized"] + ), + "end_date": payload["end_date"], + "end_date_normalized": date.fromisoformat( + payload["end_date_normalized"] + ), + "status": payload["status"], + "legal_basis": payload["legal_basis"], + "is_federal_law_248": True, + "data_year": payload["data_year"], + "data_month": payload["data_month"], + }, + ) + return legacy_record + + if source == ParserLoadLog.Source.PROCUREMENTS: + legacy_record, _ = ProcurementRecord.objects.update_or_create( + purchase_number=payload["purchase_number"], + defaults={ + **common, + "purchase_name": payload["subject"], + "customer_inn": organization.inn, + "customer_kpp": organization.kpp, + "customer_ogrn": organization.ogrn, + "customer_name": organization.name, + "max_price": payload["price"], + "max_price_amount": defaults["amount"], + "publish_date": payload["published_at"], + "publish_date_normalized": date(2026, 2, 1), + "status": payload["status"], + "law_type": payload["law"], + "purchase_object_info": payload["subject"], + "href": payload["url"], + "region_code": payload["region_code"], + "data_year": payload["data_year"], + "data_month": payload["data_month"], + }, + ) + return legacy_record + + if source == ParserLoadLog.Source.FNS_REPORTS: + legacy_record, _ = FinancialReport.objects.update_or_create( + external_id=f"test-company-{index:02d}", + defaults={ + **common, + "ogrn": organization.ogrn, + "file_name": payload["file_name"], + "file_hash": payload["file_hash"], + "status": FinancialReport.Status.SUCCESS, + "source": FinancialReport.SourceType.API, + }, + ) + return legacy_record + + legacy_record, _ = GenericParserRecord.objects.update_or_create( + source=source, + external_id=external_id, + defaults={ + **common, + "inn": organization.inn, + "ogrn": organization.ogrn, + "organisation_name": organization.name, + "title": defaults.get("title", ""), + "record_date": defaults.get("record_date", ""), + "amount": defaults.get("amount"), + "status": defaults.get("status", ""), + "url": defaults.get("url", ""), + "payload": payload, + }, + ) + return legacy_record + + @staticmethod + def _delete_parser_result_records(company_uids: list[UUID]) -> None: + for model in ( + GenericParserRecord, + IndustrialCertificateRecord, + IndustrialProductRecord, + ManufacturerRecord, + InspectionRecord, + ProcurementRecord, + FinancialReport, + ): + model.objects.filter(registry_organization_id__in=company_uids).delete() + @staticmethod def _descriptors_by_group(): descriptors = {} diff --git a/src/organizations/views.py b/src/organizations/views.py index 851a978..3dcfbf1 100644 --- a/src/organizations/views.py +++ b/src/organizations/views.py @@ -32,7 +32,6 @@ from organizations.cache import ( ORGANIZATION_API_CACHE_CONTRACT_VERSION, ORGANIZATION_API_CACHE_PREFIX, get_organization_api_cache_version, - invalidate_organization_api_cache, ) from organizations.directory_import import ( OrganizationDirectoryImportError, @@ -395,7 +394,6 @@ class OrganizationViewSet(CachedReadOnlyMixin, ReadOnlyModelViewSet): with suppress(FileNotFoundError): os.unlink(temp_path) - invalidate_organization_api_cache() return Response( { "success": True, diff --git a/src/settings/base.py b/src/settings/base.py index 0d1e19c..9322919 100644 --- a/src/settings/base.py +++ b/src/settings/base.py @@ -334,8 +334,10 @@ REST_FRAMEWORK = { SIMPLE_JWT = { "ACCESS_TOKEN_LIFETIME": timedelta(minutes=60), "REFRESH_TOKEN_LIFETIME": timedelta(days=7), - "ROTATE_REFRESH_TOKENS": True, - "BLACKLIST_AFTER_ROTATION": True, + # The frontend can issue concurrent refresh requests. Reusing one refresh token + # avoids a race where the first response blacklists the token used by the second. + "ROTATE_REFRESH_TOKENS": False, + "BLACKLIST_AFTER_ROTATION": False, "UPDATE_LAST_LOGIN": True, "ALGORITHM": "HS256", "VERIFYING_KEY": None, diff --git a/src/settings/test.py b/src/settings/test.py index c1bef6d..b8a2693 100644 --- a/src/settings/test.py +++ b/src/settings/test.py @@ -108,5 +108,6 @@ SIMPLE_JWT = { **globals().get("SIMPLE_JWT", {}), "ACCESS_TOKEN_LIFETIME": timedelta(minutes=5), "REFRESH_TOKEN_LIFETIME": timedelta(days=1), - "ROTATE_REFRESH_TOKENS": True, + "ROTATE_REFRESH_TOKENS": False, + "BLACKLIST_AFTER_ROTATION": False, } diff --git a/src/user/views.py b/src/user/views.py index 84b4a30..5642eb2 100644 --- a/src/user/views.py +++ b/src/user/views.py @@ -13,6 +13,7 @@ from rest_framework.exceptions import ValidationError from rest_framework.permissions import AllowAny, IsAdminUser, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView +from rest_framework_simplejwt.serializers import TokenRefreshSerializer from rest_framework_simplejwt.views import TokenRefreshView as SimpleJWTTokenRefreshView from rest_framework_simplejwt.views import TokenVerifyView as SimpleJWTTokenVerifyView @@ -646,10 +647,20 @@ def user_profile_detail(request): return Response(profile_data) +class ReusableTokenRefreshSerializer(TokenRefreshSerializer): + """Return the unchanged refresh token for concurrent frontend requests.""" + + def validate(self, attrs): + payload = super().validate(attrs) + payload["refresh"] = attrs["refresh"] + return payload + + class TokenRefreshView(SimpleJWTTokenRefreshView): """Обновление access токена через refresh токен.""" permission_classes = [AllowAny] + serializer_class = ReusableTokenRefreshSerializer @swagger_auto_schema( tags=[AUTH_TAG], diff --git a/tests/apps/exchange/test_state_corp_services.py b/tests/apps/exchange/test_state_corp_services.py index 1498163..3720266 100644 --- a/tests/apps/exchange/test_state_corp_services.py +++ b/tests/apps/exchange/test_state_corp_services.py @@ -201,6 +201,18 @@ class StateCorpExchangeServiceTest(TestCase): "Окончание подачи заявок": "20.02.2026", }, ) + GenericParserRecord.objects.create( + source=ParserLoadLog.Source.CONTRACTS, + load_batch=1, + external_id="contract-not-a-procurement", + inn=organization.inn, + ogrn=organization.ogrn, + title="Исполненный контракт", + record_date="16.02.2026", + amount="1000000.00", + status="completed", + payload={"registry_number": "contract-not-a-procurement"}, + ) GenericParserRecord.objects.create( source=ParserLoadLog.Source.ARBITRATION, load_batch=1, @@ -294,6 +306,10 @@ class StateCorpExchangeServiceTest(TestCase): self.assertEqual(package.payload_counts["labor_vacancies"], 1) payload = _decode_package_payload(package) + self.assertNotIn( + "contract-not-a-procurement", + {row["purchase_number"] for row in payload["data"]["public_procurements"]}, + ) self.assertEqual(payload["format"], StateCorpExchangeService.PAYLOAD_FORMAT) self.assertEqual(payload["schema_version"], 3) diff --git a/tests/apps/organizations/test_directory_import.py b/tests/apps/organizations/test_directory_import.py index 739abb3..6fd3ce9 100644 --- a/tests/apps/organizations/test_directory_import.py +++ b/tests/apps/organizations/test_directory_import.py @@ -2,6 +2,7 @@ from decimal import Decimal from tempfile import NamedTemporaryFile +from unittest.mock import patch from django.test import TestCase from openpyxl import Workbook @@ -61,6 +62,19 @@ class OrganizationDirectoryImportServiceTest(TestCase): self.assertTrue(organization.goz_participation) self.assertFalse(organization.opk_registry_membership) + def test_import_xlsx_invalidates_organization_api_cache_after_commit(self): + path = self._workbook_path([self._row(rn="10")]) + + with ( + patch( + "organizations.directory_import.invalidate_organization_api_cache" + ) as invalidate_cache, + self.captureOnCommitCallbacks(execute=True), + ): + OrganizationDirectoryImportService.import_xlsx(path) + + invalidate_cache.assert_called_once_with() + def test_import_xlsx_rejects_missing_required_column(self): path = self._workbook_path([], headers=SOURCE_HEADERS[:-1]) @@ -111,9 +125,7 @@ class OrganizationDirectoryImportServiceTest(TestCase): ] embedded_value = "\t".join(str(value or "") for value in first_tail) embedded_value = ( - embedded_value - + "_x000D_\n" - + "\t".join(["21", "0", "0", 'ООО "Вторая"']) + embedded_value + "_x000D_\n" + "\t".join(["21", "0", "0", 'ООО "Вторая"']) ) second_row = self._row( diff --git a/tests/apps/organizations/test_test_companies_commands.py b/tests/apps/organizations/test_test_companies_commands.py index 65d525e..6848af0 100644 --- a/tests/apps/organizations/test_test_companies_commands.py +++ b/tests/apps/organizations/test_test_companies_commands.py @@ -3,7 +3,16 @@ from io import StringIO from apps.exchange.state_corp_services import StateCorpExchangeService -from apps.parsers.models import ParserLoadLog +from apps.parsers.models import ( + FinancialReport, + GenericParserRecord, + IndustrialCertificateRecord, + IndustrialProductRecord, + InspectionRecord, + ManufacturerRecord, + ParserLoadLog, + ProcurementRecord, +) from django.core.management import call_command from django.test import TestCase, override_settings from organizations.models import ( @@ -79,6 +88,13 @@ class TestCompaniesCommandsTest(TestCase): ).count(), 20 * 4, ) + self.assertEqual(IndustrialCertificateRecord.objects.count(), 20) + self.assertEqual(IndustrialProductRecord.objects.count(), 20) + self.assertEqual(ManufacturerRecord.objects.count(), 20) + self.assertEqual(InspectionRecord.objects.count(), 20) + self.assertEqual(ProcurementRecord.objects.count(), 20) + self.assertEqual(FinancialReport.objects.count(), 20) + self.assertEqual(GenericParserRecord.objects.count(), 20 * 9) def test_create_updates_the_fixed_dataset_without_duplicates(self): call_command("create_test_companies", stdout=StringIO()) @@ -125,9 +141,7 @@ class TestCompaniesCommandsTest(TestCase): .order_by("rn") .values_list("inn", flat=True) ) - package = StateCorpExchangeService.build_package( - organization_inns=company_inns - ) + package = StateCorpExchangeService.build_package(organization_inns=company_inns) self.assertEqual( package.payload_counts, @@ -137,7 +151,7 @@ class TestCompaniesCommandsTest(TestCase): "manufacturers": 20, "industrial_products": 20, "prosecutor_checks": 20, - "public_procurements": 80, + "public_procurements": 60, "financial_reports": 20, "arbitration_cases": 20, "bankruptcy_procedures": 20, @@ -166,3 +180,6 @@ class TestCompaniesCommandsTest(TestCase): self.assertTrue(Organization.objects.filter(uid=untouched.uid).exists()) self.assertEqual(OrganizationSourceExtension.objects.count(), 0) self.assertEqual(OrganizationSourceRecord.objects.count(), 0) + self.assertEqual(GenericParserRecord.objects.count(), 0) + self.assertEqual(IndustrialProductRecord.objects.count(), 0) + self.assertEqual(FinancialReport.objects.count(), 0) diff --git a/tests/apps/parsers/test_checko_collection.py b/tests/apps/parsers/test_checko_collection.py new file mode 100644 index 0000000..b6b55e7 --- /dev/null +++ b/tests/apps/parsers/test_checko_collection.py @@ -0,0 +1,92 @@ +from datetime import date + +from apps.parsers import tasks as parser_tasks +from apps.parsers.checko_collection import claim_monthly_collection, finish_collection +from apps.parsers.models import CheckoCollectionAttempt +from django.test import TestCase + +from tests.apps.parsers.organization_helpers import create_directory_organization + + +class CheckoCollectionClaimTest(TestCase): + def setUp(self): + self.organization = create_directory_organization( + pn_name='ООО "Месячный лимит"', + mn_ogrn=1027700000199, + mn_inn=7701000199, + in_kpp=770101001, + mn_okpo="12345678", + ) + + def test_claim_is_unique_per_organization_source_and_calendar_month(self): + first = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.ARBITRATION, + at=date(2026, 7, 1), + ) + + duplicate = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.ARBITRATION, + at=date(2026, 7, 31), + ) + other_source = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.INSPECTIONS, + at=date(2026, 7, 31), + ) + next_month = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.ARBITRATION, + at=date(2026, 8, 1), + ) + + self.assertIsNotNone(first) + self.assertIsNone(duplicate) + self.assertIsNotNone(other_source) + self.assertIsNotNone(next_month) + + def test_failed_attempt_keeps_monthly_claim(self): + attempt = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.CONTRACTS, + at=date(2026, 7, 19), + ) + assert attempt is not None + finish_collection(attempt, records_count=0, error="quota exhausted") + + repeated = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.CONTRACTS, + at=date(2026, 7, 20), + ) + + attempt.refresh_from_db() + self.assertIsNone(repeated) + self.assertEqual(attempt.status, CheckoCollectionAttempt.Status.FAILED) + self.assertEqual(attempt.error_message, "quota exhausted") + + def test_due_filter_is_applied_before_lookup_limit(self): + self.organization.opk_registry_membership = True + self.organization.save(update_fields=["opk_registry_membership"]) + second = create_directory_organization( + pn_name='ООО "Следующая организация"', + mn_ogrn=1027700000299, + mn_inn=7701000299, + in_kpp=770101001, + mn_okpo="87654321", + ) + second.opk_registry_membership = True + second.save(update_fields=["opk_registry_membership"]) + claimed = claim_monthly_collection( + organization_id=str(self.organization.id), + source=CheckoCollectionAttempt.Source.INSPECTIONS, + ) + self.assertIsNotNone(claimed) + + targets = parser_tasks._active_registry_lookup_targets( + limit=1, + checko_source=CheckoCollectionAttempt.Source.INSPECTIONS, + ) + + self.assertEqual([target.organization_id for target in targets], [str(second.id)]) diff --git a/tests/apps/parsers/test_tasks.py b/tests/apps/parsers/test_tasks.py index 05286de..38b19d0 100644 --- a/tests/apps/parsers/test_tasks.py +++ b/tests/apps/parsers/test_tasks.py @@ -33,6 +33,7 @@ from apps.parsers.clients.minpromtorg.manufactures import ( from apps.parsers.clients.proverki.client import ProverkiClientError from apps.parsers.clients.zakupki import ZakupkiClientError from apps.parsers.models import ( + CheckoCollectionAttempt, FinancialReport, ParserLoadLog, ) @@ -316,6 +317,12 @@ class CheckoFedresursFallbackControlFlowTestCase(SimpleTestCase): return_value=targets, ), patch.object(parser_tasks, "CheckoClient", _RateLimitedCheckoClient), + patch.object( + parser_tasks, + "claim_monthly_collection", + return_value=SimpleNamespace(), + ), + patch.object(parser_tasks, "finish_collection"), ): records = parser_tasks._fetch_checko_bankruptcy_records(proxies=None) @@ -801,6 +808,15 @@ class GenericSourceFetchTestCase(TestCase): [request.inn for request in _CheckoClient.instances[0].requests], [str(organization.mn_inn)], ) + second_result = parser_tasks.parse_arbitration_cases(limit=10, proxies=[]) + self.assertEqual(second_result["status"], "skipped") + self.assertEqual(len(_CheckoClient.instances), 1) + attempt = CheckoCollectionAttempt.objects.get( + organization_id=organization.id, + source=CheckoCollectionAttempt.Source.ARBITRATION, + ) + self.assertEqual(attempt.status, CheckoCollectionAttempt.Status.SUCCESS) + self.assertEqual(attempt.records_count, 1) @override_settings(CHECKO_API_KEY="test-key") def test_arbitration_skips_when_no_active_registry_organizations(self): @@ -1842,11 +1858,13 @@ class MinpromtorgTasksTestCase(TestCase): "parse_procurements_44fz", "parse_procurements_223fz", "parse_contracts", + "parse_registry_contracts", "parse_unfair_suppliers", "parse_fas_goz_evasion", "sync_fns_financial_reports", "parse_arbitration_cases", "parse_fedresurs_bankruptcy", + "parse_registry_inspections", "parse_fstec_registers", "parse_trudvsem_vacancies", ) @@ -1869,6 +1887,8 @@ class MinpromtorgTasksTestCase(TestCase): "sync_fns_financial_reports-id", ) delays["sync_fns_financial_reports"].assert_called_once_with(proxies=[]) + delays["parse_registry_contracts"].assert_called_once_with(proxies=[]) + delays["parse_registry_inspections"].assert_called_once_with(proxies=[]) def test_parse_all_minpromtorg_without_adapter(self): with TestHTTPServer() as server: diff --git a/tests/apps/user/test_views.py b/tests/apps/user/test_views.py index c668ffd..c8638c1 100644 --- a/tests/apps/user/test_views.py +++ b/tests/apps/user/test_views.py @@ -595,6 +595,17 @@ class TokenRefreshViewTest(APITestCase): # New refresh token should be different # Refresh token may be the same or different depending on implementation + def test_same_refresh_token_can_handle_parallel_refresh_requests(self): + data = {"refresh": self.tokens["refresh"]} + + first_response = self.client.post(self.refresh_url, data, format="json") + second_response = self.client.post(self.refresh_url, data, format="json") + + self.assertEqual(first_response.status_code, status.HTTP_200_OK) + self.assertEqual(second_response.status_code, status.HTTP_200_OK) + self.assertEqual(first_response.data["refresh"], self.tokens["refresh"]) + self.assertEqual(second_response.data["refresh"], self.tokens["refresh"]) + def test_refresh_token_invalid(self): """Test token refresh fails with invalid refresh token""" data = {"refresh": fake.pystr(min_chars=20, max_chars=50)}