feat: expand registry ingestion and demo exchange

src/apps/exchange/state_corp_services.py

@@ -34,7 +34,7 @@ from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 from django.conf import settings
 from django.db.models import Q
 from django.utils import timezone
-from organizations.models import Organization
+from organizations.models import Organization, OrganizationSourceRecord
 
 
 class StateCorpExchangeError(ValueError):
@@ -395,6 +395,42 @@ class StateCorpExchangeService:
                     "ogrn": cls._digits(record.ogrn),
                 }
             )
+        for record in cls._canonical_records(
+            allowed_inns,
+            sources=[ParserLoadLog.Source.INDUSTRIAL],
+        ):
+            payload = cls._record_payload(record)
+            certificate_number = cls._payload_lookup(
+                payload, ["certificate_number", "registry_number"]
+            )
+            if not certificate_number:
+                continue
+            issue_date = cls._coerce_date(
+                None,
+                cls._payload_lookup(payload, ["issue_date_normalized", "issue_date"])
+                or record.record_date,
+            )
+            expiry_date = cls._coerce_date(
+                None,
+                cls._payload_lookup(payload, ["expiry_date_normalized", "expiry_date"]),
+            )
+            items.append(
+                {
+                    "organization_inn": cls._digits(record.inn),
+                    "certificate_number": certificate_number,
+                    "issue_date": issue_date.isoformat() if issue_date else None,
+                    "expiry_date": expiry_date.isoformat() if expiry_date else None,
+                    "certificate_file_url": cls._payload_lookup(
+                        payload, ["certificate_file_url"]
+                    )
+                    or record.url,
+                    "organisation_name": cls._payload_lookup(
+                        payload, ["organisation_name"]
+                    )
+                    or record.registry_organization.name,
+                    "ogrn": cls._digits(record.ogrn),
+                }
+            )
         return items
 
     @classmethod
@@ -405,7 +441,7 @@ class StateCorpExchangeService:
         queryset = ManufacturerRecord.objects.filter(inn__in=allowed_inns).order_by(
             "id"
         )
-        return [
+        items = [
             {
                 "organization_inn": cls._digits(record.inn),
                 "full_legal_name": record.full_legal_name,
@@ -415,6 +451,26 @@ class StateCorpExchangeService:
             }
             for record in queryset
         ]
+        for record in cls._canonical_records(
+            allowed_inns,
+            sources=[ParserLoadLog.Source.MANUFACTURES],
+        ):
+            payload = cls._record_payload(record)
+            items.append(
+                {
+                    "organization_inn": cls._digits(record.inn),
+                    "full_legal_name": cls._payload_lookup(
+                        payload, ["full_legal_name", "organisation_name"]
+                    )
+                    or record.registry_organization.full_name
+                    or record.registry_organization.name,
+                    "inn": cls._digits(record.inn),
+                    "ogrn": cls._digits(record.ogrn),
+                    "address": cls._payload_lookup(payload, ["address"])
+                    or record.registry_organization.legal_address,
+                }
+            )
+        return items
 
     @classmethod
     def _serialize_industrial_products(
@@ -424,7 +480,7 @@ class StateCorpExchangeService:
         queryset = IndustrialProductRecord.objects.filter(
             inn__in=allowed_inns
         ).order_by("id")
-        return [
+        items = [
             {
                 "organization_inn": cls._digits(record.inn),
                 "product_name": record.product_name,
@@ -437,6 +493,27 @@ class StateCorpExchangeService:
             }
             for record in queryset
         ]
+        for record in cls._canonical_records(
+            allowed_inns,
+            sources=[ParserLoadLog.Source.INDUSTRIAL_PRODUCTS],
+        ):
+            payload = cls._record_payload(record)
+            product_name = cls._payload_lookup(payload, ["product_name"])
+            items.append(
+                {
+                    "organization_inn": cls._digits(record.inn),
+                    "product_name": product_name or record.title,
+                    "product_class": cls._payload_lookup(
+                        payload, ["product_model", "regulatory_document"]
+                    )
+                    or "Промышленная продукция",
+                    "okpd2_code": cls._payload_lookup(payload, ["okpd2_code"]),
+                    "tnved_code": cls._payload_lookup(payload, ["tnved_code"]),
+                    "registry_number": cls._payload_lookup(payload, ["registry_number"])
+                    or record.external_id,
+                }
+            )
+        return items
 
     @classmethod
     def _serialize_prosecutor_checks(
@@ -463,6 +540,36 @@ class StateCorpExchangeService:
                     "status": record.status,
                 }
             )
+        for record in cls._canonical_records(
+            allowed_inns,
+            sources=[ParserLoadLog.Source.INSPECTIONS],
+        ):
+            payload = cls._record_payload(record)
+            start_date = cls._coerce_date(
+                None,
+                cls._payload_lookup(payload, ["start_date_normalized", "start_date"])
+                or record.record_date,
+            )
+            if start_date is None:
+                continue
+            items.append(
+                {
+                    "organization_inn": cls._digits(record.inn),
+                    "registration_number": cls._payload_lookup(
+                        payload, ["registration_number"]
+                    )
+                    or record.external_id,
+                    "law_type": cls._payload_lookup(payload, ["legal_basis"]),
+                    "control_authority": cls._payload_lookup(
+                        payload, ["control_authority"]
+                    ),
+                    "prosecutor_office": cls._payload_lookup(
+                        payload, ["prosecutor_office"]
+                    ),
+                    "start_date": start_date.isoformat(),
+                    "status": record.status or cls._payload_lookup(payload, ["status"]),
+                }
+            )
         return items
 
     @classmethod
@@ -504,6 +611,7 @@ class StateCorpExchangeService:
         for record in cls._generic_records(
             allowed_inns,
             sources=[
+                ParserLoadLog.Source.PROCUREMENTS,
                 ParserLoadLog.Source.PROCUREMENTS_44FZ,
                 ParserLoadLog.Source.PROCUREMENTS_223FZ,
                 ParserLoadLog.Source.CONTRACTS,
@@ -585,7 +693,7 @@ class StateCorpExchangeService:
             .prefetch_related("lines")
             .order_by("id")
         )
-        return [
+        items = [
             {
                 "organization_inn": allowed_ogrn_to_inn[report.ogrn],
                 "external_id": report.external_id,
@@ -611,6 +719,33 @@ class StateCorpExchangeService:
             for report in queryset
             if report.external_id
         ]
+        for record in cls._canonical_records(
+            set(allowed_ogrn_to_inn.values()),
+            sources=[ParserLoadLog.Source.FNS_REPORTS],
+        ):
+            if not record.external_id:
+                continue
+            payload = cls._record_payload(record)
+            items.append(
+                {
+                    "organization_inn": cls._digits(record.inn),
+                    "external_id": record.external_id,
+                    "ogrn": cls._digits(record.ogrn),
+                    "file_name": cls._payload_lookup(payload, ["file_name"]),
+                    "file_hash": cls._payload_lookup(payload, ["file_hash"]),
+                    "load_batch": record.load_batch,
+                    "status": record.status,
+                    "source": cls._payload_lookup(payload, ["source"]) or record.source,
+                    "error_message": cls._payload_lookup(payload, ["error_message"]),
+                    "lines": [
+                        cls._serialize_financial_report_line(line)
+                        for line in record.financial_lines.all().order_by(
+                            "year", "form_code", "line_code"
+                        )
+                    ],
+                }
+            )
+        return items
 
     @staticmethod
     def _serialize_financial_report_line(
@@ -631,11 +766,10 @@ class StateCorpExchangeService:
         allowed_inns: set[str],
     ) -> list[dict[str, str]]:
         items: list[dict[str, str]] = []
-        queryset = GenericParserRecord.objects.filter(
-            source=ParserLoadLog.Source.ARBITRATION,
-            inn__in=allowed_inns,
-        ).order_by("id")
-        for record in queryset:
+        for record in cls._generic_records(
+            allowed_inns,
+            sources=[ParserLoadLog.Source.ARBITRATION],
+        ):
             payload = record.payload if isinstance(record.payload, dict) else {}
             target = payload.get("target")
             if not isinstance(target, dict):
@@ -843,7 +977,7 @@ class StateCorpExchangeService:
         return items
 
     @staticmethod
-    def _record_payload(record: GenericParserRecord) -> dict[str, Any]:
+    def _record_payload(record: Any) -> dict[str, Any]:
         return record.payload if isinstance(record.payload, dict) else {}
 
     @classmethod
@@ -854,11 +988,35 @@ class StateCorpExchangeService:
         sources: list[str],
     ):
         if not allowed_inns:
-            return GenericParserRecord.objects.none()
-        return GenericParserRecord.objects.filter(
-            source__in=sources,
-            inn__in=allowed_inns,
-        ).order_by("id")
+            return []
+        legacy_records = list(
+            GenericParserRecord.objects.filter(
+                source__in=sources,
+                inn__in=allowed_inns,
+            ).order_by("id")
+        )
+        return legacy_records + list(
+            cls._canonical_records(allowed_inns, sources=sources)
+        )
+
+    @staticmethod
+    def _canonical_records(
+        allowed_inns: set[str],
+        *,
+        sources: list[str],
+    ):
+        """Return canonical-only records, excluding legacy-backed mirrors."""
+        if not allowed_inns:
+            return OrganizationSourceRecord.objects.none()
+        return (
+            OrganizationSourceRecord.objects.filter(
+                source__in=sources,
+                extension__organization__inn__in=allowed_inns,
+                legacy_model="",
+            )
+            .select_related("extension__organization")
+            .order_by("created_at", "uid")
+        )
 
     @staticmethod
     def _payload_lookup(payload: dict[str, Any], candidates: list[str]) -> str: