"""Build and deliver exchange packages for state-corp-backend.""" from __future__ import annotations import base64 import hashlib import json import secrets import struct import zlib from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal from io import BytesIO from pathlib import Path from typing import Any from uuid import uuid4 from zipfile import ZIP_DEFLATED, ZipFile import requests from apps.parsers.models import ( VACANCY_RECORD_SOURCES, FinancialReport, FinancialReportLine, GenericParserRecord, IndustrialCertificateRecord, IndustrialProductRecord, InspectionRecord, ManufacturerRecord, ParserLoadLog, ProcurementRecord, ) 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, OrganizationSourceRecord class StateCorpExchangeError(ValueError): """Raised when building or sending state-corp exchange packages fails.""" @dataclass(frozen=True) class StateCorpExchangePackage: """Built archive ready for upload or manual delivery.""" package_id: str archive_name: str bin_name: str archive_bytes: bytes payload_counts: dict[str, int] produced_at: str class StateCorpExchangeService: """Create and send packages compatible with state-corp exchange import.""" MAGIC = b"EXCH" AAD = b"state-corp-exchange-v1" PAYLOAD_FORMAT = "state-corp-exchange-payload" BIN_FORMAT = "state-corp-exchange-bin" SCHEMA_VERSION = 3 ROSATOM_ROSCOSMOS_GK_CODE_VALUES = ("rosatom", "roscosmos", "roskosmos") ROSATOM_ROSCOSMOS_GK_NAME_KEYWORDS = ("Росатом", "Роскосмос") @classmethod def build_package( cls, *, package_id: str | None = None, source_system: str = "mostovik", organization_inns: list[str] | None = None, actual_date: date | str | None = None, ) -> StateCorpExchangePackage: """Build encrypted archive for state-corp import.""" produced_at = timezone.now() snapshot_date = cls._coerce_actual_date(actual_date) normalized_inns = cls._normalize_inn_list(organization_inns) organizations = ( cls._get_organizations(normalized_inns) if normalized_inns else cls._get_rosatom_roscosmos_organizations() ) allowed_inns = {str(item.inn) for item in organizations if item.inn} allowed_ogrn_to_inn = { str(item.ogrn): str(item.inn) for item in organizations if item.ogrn and item.inn } data = { "organizations": cls._serialize_organizations(organizations), "industrial_certificates": cls._serialize_industrial_certificates( allowed_inns ), "manufacturers": cls._serialize_manufacturers(allowed_inns), "industrial_products": cls._serialize_industrial_products(allowed_inns), "prosecutor_checks": cls._serialize_prosecutor_checks(allowed_inns), "public_procurements": cls._serialize_public_procurements(allowed_inns), "financial_reports": cls._serialize_financial_reports(allowed_ogrn_to_inn), "arbitration_cases": cls._serialize_arbitration_cases(allowed_inns), "bankruptcy_procedures": cls._serialize_bankruptcy_procedures(allowed_inns), "defense_unreliable_suppliers": ( cls._serialize_defense_unreliable_suppliers(allowed_inns) ), "information_security_registries": ( cls._serialize_information_security_registries(allowed_inns) ), "labor_vacancies": cls._serialize_labor_vacancies(allowed_inns), } payload_counts = {key: len(value) for key, value in data.items()} package_id = package_id or cls._build_package_id() payload = { "format": cls.PAYLOAD_FORMAT, "schema_version": cls.SCHEMA_VERSION, "manifest": { "package_id": package_id, "source_system": source_system, "produced_at": produced_at.isoformat(), "actual_date": snapshot_date.isoformat(), "schema_version": cls.SCHEMA_VERSION, "sections": [key for key, items in data.items() if items], }, "data": data, } archive_name, bin_name, archive_bytes = cls._render_archive( package_id=package_id, payload=payload, produced_at=produced_at, ) return StateCorpExchangePackage( package_id=package_id, archive_name=archive_name, bin_name=bin_name, archive_bytes=archive_bytes, payload_counts=payload_counts, produced_at=produced_at.isoformat(), ) @classmethod def save_package( cls, *, package: StateCorpExchangePackage, output_path: str | Path, ) -> Path: """Persist package to disk for manual transfer.""" target_path = Path(output_path).expanduser().resolve() target_path.parent.mkdir(parents=True, exist_ok=True) target_path.write_bytes(package.archive_bytes) return target_path @classmethod def send_package( cls, *, package: StateCorpExchangePackage, target_url: str | None = None, token: str | None = None, timeout_seconds: int | None = None, ) -> dict[str, Any]: """Send package to state-corp dev endpoint.""" resolved_url = str(target_url or settings.STATE_CORP_EXCHANGE_URL).strip() resolved_token = str(token or settings.STATE_CORP_EXCHANGE_TOKEN).strip() timeout_seconds = timeout_seconds or int( getattr(settings, "STATE_CORP_EXCHANGE_TIMEOUT_SECONDS", 60) ) if not resolved_url: raise StateCorpExchangeError("STATE_CORP_EXCHANGE_URL не настроен") if not resolved_token: raise StateCorpExchangeError("STATE_CORP_EXCHANGE_TOKEN не настроен") try: response = requests.post( resolved_url, files={ "file": ( package.archive_name, package.archive_bytes, "application/zip", ) }, headers={"X-Exchange-Token": resolved_token}, timeout=timeout_seconds, ) except requests.RequestException as exc: raise StateCorpExchangeError( f"Не удалось отправить пакет в state-corp: {exc}" ) from exc try: body = response.json() except ValueError: body = {"raw": response.text} if response.status_code >= 400: raise StateCorpExchangeError( f"state-corp вернул HTTP {response.status_code}: {body}" ) return { "status_code": response.status_code, "response": body, "package_id": package.package_id, "archive_name": package.archive_name, } @classmethod def _build_package_id(cls) -> str: return f"mostovik-{timezone.now():%Y%m%d%H%M%S}-{uuid4().hex[:8]}" @classmethod def _render_archive( cls, *, package_id: str, payload: dict[str, Any], produced_at, ) -> tuple[str, str, bytes]: token = str(settings.STATE_CORP_EXCHANGE_TOKEN or "").strip() if not token: raise StateCorpExchangeError("STATE_CORP_EXCHANGE_TOKEN не настроен") payload_bytes = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") compressed_payload = zlib.compress(payload_bytes, level=9) nonce = secrets.token_bytes(12) encrypted_payload = AESGCM( hashlib.sha256(token.encode("utf-8")).digest() ).encrypt( nonce, compressed_payload, cls.AAD, ) header = { "format": cls.BIN_FORMAT, "version": 1, "key_id": settings.STATE_CORP_EXCHANGE_KEY_ID, "nonce": cls._b64url(nonce), "aad": cls._b64url(cls.AAD), "package_id": package_id, "schema_version": cls.SCHEMA_VERSION, "plaintext_sha256": hashlib.sha256(payload_bytes).hexdigest(), "compressed_sha256": hashlib.sha256(compressed_payload).hexdigest(), "ciphertext_sha256": hashlib.sha256(encrypted_payload).hexdigest(), } header_bytes = json.dumps( header, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") bin_bytes = ( cls.MAGIC + bytes([1]) + struct.pack(">I", len(header_bytes)) + header_bytes + encrypted_payload ) timestamp = produced_at.strftime("%Y%m%d_%H%M%S") archive_name = f"state_corp_exchange_{timestamp}.zip" bin_name = f"state_corp_exchange_{timestamp}.bin" archive_bytes = BytesIO() with ZipFile(archive_bytes, "w", compression=ZIP_DEFLATED) as archive: archive.writestr(bin_name, bin_bytes) archive.writestr( f"{bin_name}.sha256", f"{hashlib.sha256(bin_bytes).hexdigest()} {bin_name}\n", ) return archive_name, bin_name, archive_bytes.getvalue() @classmethod def _get_organizations(cls, organization_inns: list[str]) -> list[Organization]: queryset = cls._rosatom_roscosmos_queryset() if organization_inns: queryset = queryset.filter(inn__in=organization_inns) return list(queryset) @classmethod def _get_rosatom_roscosmos_organizations(cls) -> list[Organization]: return list(cls._rosatom_roscosmos_queryset()) @classmethod def _rosatom_roscosmos_queryset(cls): queryset = Organization.objects.exclude(inn="") name_query = Q() for keyword in cls.ROSATOM_ROSCOSMOS_GK_NAME_KEYWORDS: name_query |= Q(gk_name__icontains=keyword) code_query = Q() for code in cls.ROSATOM_ROSCOSMOS_GK_CODE_VALUES: code_query |= Q(gk_code__iexact=code) return queryset.filter(name_query | code_query).order_by("rn", "uid") @classmethod def _serialize_organizations( cls, organizations: list[Organization], ) -> list[dict[str, str | int | bool | None]]: return [ { "mostovik_uid": str(item.uid), "rn": item.rn, "name": item.name, "full_name": item.full_name, "short_name": item.short_name, "pn_name": item.pn_name, "pn_name_en": item.pn_name_en, "inn": item.inn, "kpp": item.kpp, "ogrn": item.ogrn, "ogrip": item.ogrip, "okpo": item.okpo, "identity_status": item.identity_status, "primary_identity": item.primary_identity, "gk_code": item.gk_code, "gk_name": item.gk_name, "in_korp_code": item.in_korp_code, "in_korp_name": item.in_korp_name, "filial": item.filial, "is_branch": item.is_branch, "re_za": item.re_za, "re_zasf": item.re_zasf, "goz_participation": item.goz_participation, "opk_registry_membership": item.opk_registry_membership, "ropk_num": item.ropk_num, "ropk_razdel_num": item.ropk_razdel_num, "ropk_razdel_name": item.ropk_razdel_name, "registration_date": ( item.registration_date.isoformat() if item.registration_date else None ), "create_date": item.create_date, "organizational_legal_form": item.organizational_legal_form, "organizational_legal_form1": item.organizational_legal_form1, "ownership_form": item.ownership_form, "ownership_form1": item.ownership_form1, "authorized_capital": cls._serialize_decimal(item.authorized_capital), "legal_address": item.legal_address, "business_act_cod": item.business_act_cod, "business_activity": item.business_activity, "general_director": item.general_director, "general_director_tax_id": item.general_director_tax_id, "uk": item.uk, "inn_uk": item.inn_uk, "appointment_date": ( item.appointment_date.isoformat() if item.appointment_date else None ), "cf_fl_rn": item.cf_fl_rn, "akc_fs": item.akc_fs, "akc_sf": item.akc_sf, "min": item.min, "dep": item.dep, "otr": item.otr, "integrated_structure": item.integrated_structure, "state_sector_code": item.state_sector_code, "state_sector_name": item.state_sector_name, } for item in organizations ] @classmethod def _serialize_industrial_certificates( cls, allowed_inns: set[str], ) -> list[dict[str, str | None]]: queryset = IndustrialCertificateRecord.objects.filter( inn__in=allowed_inns ).order_by("id") items: list[dict[str, str | None]] = [] for record in queryset: if not record.certificate_number: continue issue_date = cls._coerce_date( record.issue_date_normalized, record.issue_date ) expiry_date = cls._coerce_date( record.expiry_date_normalized, record.expiry_date ) items.append( { "organization_inn": cls._digits(record.inn), "certificate_number": record.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": record.certificate_file_url, "organisation_name": record.organisation_name, "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 def _serialize_manufacturers( cls, allowed_inns: set[str], ) -> list[dict[str, str]]: queryset = ManufacturerRecord.objects.filter(inn__in=allowed_inns).order_by( "id" ) items = [ { "organization_inn": cls._digits(record.inn), "full_legal_name": record.full_legal_name, "inn": cls._digits(record.inn), "ogrn": cls._digits(record.ogrn), "address": record.address, } 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( cls, allowed_inns: set[str], ) -> list[dict[str, str]]: queryset = IndustrialProductRecord.objects.filter( inn__in=allowed_inns ).order_by("id") items = [ { "organization_inn": cls._digits(record.inn), "product_name": record.product_name, "product_class": record.product_model or record.regulatory_document or "Промышленная продукция", "okpd2_code": record.okpd2_code, "tnved_code": record.tnved_code, "registry_number": record.registry_number, } 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( cls, allowed_inns: set[str], ) -> list[dict[str, str]]: items: list[dict[str, str]] = [] queryset = InspectionRecord.objects.filter(inn__in=allowed_inns).order_by("id") for record in queryset: start_date = cls._coerce_date( record.start_date_normalized, record.start_date ) if start_date is None: continue items.append( { "organization_inn": cls._digits(record.inn), "registration_number": record.registration_number, "law_type": record.legal_basis or ("248-ФЗ" if record.is_federal_law_248 else "294-ФЗ"), "control_authority": record.control_authority, "prosecutor_office": "", "start_date": start_date.isoformat(), "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 def _serialize_public_procurements( cls, allowed_inns: set[str], ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] queryset = ProcurementRecord.objects.filter( customer_inn__in=allowed_inns ).order_by("id") for record in queryset: contract_date = cls._coerce_date( record.publish_date_normalized, record.publish_date, ) if contract_date is None: continue execution_end_date = cls._coerce_date( record.end_date_normalized, record.end_date, ) items.append( { "organization_inn": cls._digits(record.customer_inn), "purchase_number": record.purchase_number, "law_type": record.law_type, "status": record.status, "contract_amount": cls._serialize_decimal(record.max_price_amount), "contract_date": contract_date.isoformat(), "execution_start_date": contract_date.isoformat(), "execution_end_date": ( execution_end_date.isoformat() if execution_end_date else None ), "purchase_name": record.purchase_name, } ) for record in cls._generic_records( allowed_inns, sources=[ ParserLoadLog.Source.PROCUREMENTS, ParserLoadLog.Source.PROCUREMENTS_44FZ, ParserLoadLog.Source.PROCUREMENTS_223FZ, ], ): item = cls._serialize_generic_public_procurement(record) if item is not None: items.append(item) return items @classmethod def _serialize_generic_public_procurement( cls, record: GenericParserRecord, ) -> dict[str, Any] | None: payload = cls._record_payload(record) purchase_number = ( cls._payload_lookup(payload, ["registry_number", "number"]) or record.external_id ) contract_date = cls._coerce_date( None, record.record_date or cls._payload_lookup( payload, ["contract_date", "publish_date", "Размещено", "Обновлено"], ), ) if not purchase_number or contract_date is None: return None execution_end_date = cls._coerce_date( None, cls._payload_lookup( payload, ["execution_end_date", "end_date", "Окончание подачи заявок"], ), ) return { "organization_inn": cls._digits(record.inn), "purchase_number": purchase_number, "law_type": cls._payload_lookup(payload, ["law"]) or cls._law_type_for_generic_procurement_source(record.source), "status": record.status or cls._payload_lookup(payload, ["status", "Статус"]), "contract_amount": cls._serialize_decimal(record.amount), "contract_date": contract_date.isoformat(), "execution_start_date": contract_date.isoformat(), "execution_end_date": ( execution_end_date.isoformat() if execution_end_date else None ), "purchase_name": record.title or cls._payload_lookup( payload, ["purchase_name", "Объект закупки", "Объекты закупки"], ), } @staticmethod def _law_type_for_generic_procurement_source(source: str) -> str: if source == ParserLoadLog.Source.PROCUREMENTS_44FZ: return "44-ФЗ" if source == ParserLoadLog.Source.PROCUREMENTS_223FZ: return "223-ФЗ" return "" @classmethod def _serialize_financial_reports( cls, allowed_ogrn_to_inn: dict[str, str], ) -> list[dict[str, Any]]: if not allowed_ogrn_to_inn: return [] queryset = ( FinancialReport.objects.filter(ogrn__in=allowed_ogrn_to_inn) .prefetch_related("lines") .order_by("id") ) items = [ { "organization_inn": allowed_ogrn_to_inn[report.ogrn], "external_id": report.external_id, "ogrn": cls._digits(report.ogrn), "file_name": report.file_name, "file_hash": report.file_hash, "load_batch": report.load_batch, "status": report.status, "source": report.source, "error_message": report.error_message, "lines": [ cls._serialize_financial_report_line(line) for line in sorted( report.lines.all(), key=lambda line: ( line.year, line.form_code, line.line_code, ), ) ], } 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( line: FinancialReportLine, ) -> dict[str, Any]: return { "form_code": line.form_code, "line_code": line.line_code, "line_name": line.line_name, "year": line.year, "period_start": line.period_start, "period_end": line.period_end, } @classmethod def _serialize_arbitration_cases( cls, allowed_inns: set[str], ) -> list[dict[str, str]]: items: list[dict[str, str]] = [] 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): target = {} decision_date = cls._coerce_date( None, payload.get("result_date") or payload.get("decision_date") or payload.get("filing_date") or record.record_date, ) case_number = str( payload.get("case_number") or record.title or record.external_id or "" ).strip() if not case_number or decision_date is None: continue items.append( { "organization_inn": cls._digits(record.inn), "case_number": case_number, "court_name": str(payload.get("court_name") or "").strip(), "party_role": str( target.get("role") or payload.get("party_role") or "" ).strip(), "status": str(payload.get("status") or record.status or "").strip(), "decision_date": decision_date.isoformat(), } ) return items @classmethod def _serialize_bankruptcy_procedures( cls, allowed_inns: set[str], ) -> list[dict[str, str | None]]: items: list[dict[str, str | None]] = [] for record in cls._generic_records( allowed_inns, sources=[ParserLoadLog.Source.FEDRESURS_BANKRUPTCY], ): payload = cls._record_payload(record) message_date = cls._coerce_date( None, cls._payload_lookup(payload, ["message_date", "date", "Дата"]) or record.record_date, ) message_type = cls._payload_lookup(payload, ["message_type", "type"]) items.append( { "organization_inn": cls._digits(record.inn), "external_id": record.external_id, "message_type": message_type or record.title or record.status, "message_date": message_date.isoformat() if message_date else None, "case_number": cls._payload_lookup(payload, ["case_number"]), "status": record.status, "source_url": record.url, } ) return items @classmethod def _serialize_defense_unreliable_suppliers( cls, allowed_inns: set[str], ) -> list[dict[str, str | None]]: items: list[dict[str, str | None]] = [] for record in cls._generic_records( allowed_inns, sources=[ ParserLoadLog.Source.UNFAIR_SUPPLIERS, ParserLoadLog.Source.FAS_GOZ, ], ): payload = cls._record_payload(record) included_at = cls._coerce_date( None, cls._payload_lookup( payload, [ "included_at", "date", "Дата вступления постановления", "Включено", ], ) or record.record_date, ) items.append( { "organization_inn": cls._digits(record.inn), "external_id": record.external_id, "registry_source": record.source, "registry_number": cls._payload_lookup( payload, [ "registry_number", "number", "Номер реестровой записи", "Номер реестровой записи в ЕРУЗ", ], ), "supplier_name": cls._payload_lookup( payload, [ "supplier_name", "organisation_name", "Полное наименование лица", "Фирменное наименование лица", "Наименование ФИО недобросовестного поставщика", ], ) or record.organisation_name, "reason": record.title, "included_at": included_at.isoformat() if included_at else None, "status": record.status, "source_url": record.url, } ) return items @classmethod def _serialize_information_security_registries( cls, allowed_inns: set[str], ) -> list[dict[str, str | None]]: items: list[dict[str, str | None]] = [] for record in cls._generic_records( allowed_inns, sources=[ParserLoadLog.Source.FSTEC], ): payload = cls._record_payload(record) issued_at = cls._coerce_date( None, cls._payload_lookup( payload, ["issued_at", "date", "Дата предоставления лицензии"], ) or record.record_date, ) expires_at = cls._coerce_date( None, cls._payload_lookup( payload, ["expires_at", "Срок действия сертификата", "res_valid_till"], ), ) items.append( { "organization_inn": cls._digits(record.inn), "external_id": record.external_id, "registry_name": cls._payload_lookup(payload, ["registry_name"]) or record.title or "Реестр ФСТЭК", "presence_status": "present", "entry_number": cls._payload_lookup( payload, [ "entry_number", "registration_number", "Регистрационный номер лицензии", "№ сертификата", ], ) or record.external_id, "issued_at": issued_at.isoformat() if issued_at else None, "expires_at": expires_at.isoformat() if expires_at else None, } ) return items @classmethod def _serialize_labor_vacancies( cls, allowed_inns: set[str], ) -> list[dict[str, str | None]]: items: list[dict[str, str | None]] = [] for record in cls._generic_records( allowed_inns, sources=list(VACANCY_RECORD_SOURCES), ): payload = cls._record_payload(record) published_at = cls._coerce_date( None, cls._payload_lookup( payload, ["published_at", "creation-date", "date", "published_at"], ) or record.record_date, ) items.append( { "organization_inn": cls._digits(record.inn), "external_id": record.external_id, "vacancy_source": cls._payload_lookup(payload, ["vacancy_source"]) or "trudvsem", "title": record.title, "status": record.status, "published_at": published_at.isoformat() if published_at else None, "salary_amount": cls._serialize_decimal(record.amount), "source_url": record.url, } ) return items @staticmethod def _record_payload(record: Any) -> dict[str, Any]: return record.payload if isinstance(record.payload, dict) else {} @classmethod def _generic_records( cls, allowed_inns: set[str], *, sources: list[str], ): if not allowed_inns: 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: for key in candidates: value = payload.get(key) if value not in (None, ""): return str(value).strip() return "" @staticmethod def _normalize_inn_list(organization_inns: list[str] | None) -> list[str]: normalized_items: list[str] = [] for item in organization_inns or []: normalized = "".join(char for char in str(item).strip() if char.isdigit()) if normalized: normalized_items.append(normalized) return normalized_items @staticmethod def _coerce_actual_date(value: date | str | None) -> date: if value is None: return timezone.localdate() if isinstance(value, date): return value raw_text = str(value).strip() try: return date.fromisoformat(raw_text) except ValueError as exc: raise StateCorpExchangeError( "actual_date должен быть датой в формате YYYY-MM-DD" ) from exc @staticmethod def _digits(value: str | int) -> str: return "".join(char for char in str(value).strip() if char.isdigit()) @staticmethod def _b64url(data: bytes) -> str: return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") @classmethod def _coerce_date( cls, normalized: date | None, raw_value: str | None ) -> date | None: if normalized is not None: return normalized raw_text = str(raw_value or "").strip() if not raw_text: return None try: return date.fromisoformat(raw_text) except ValueError: pass for date_format in ("%d.%m.%Y", "%d.%m.%y"): try: return datetime.strptime(raw_text, date_format).date() except ValueError: continue return None @staticmethod def _serialize_decimal(value: Decimal | None) -> str | None: if value is None: return None return format(value, "f")