fix(parsers): align vacancy sources and procurement counters

2026-05-14 14:45:58 +02:00
parent 89607356b7
commit 6d1ec2e55c
12 changed files with 340 additions and 57 deletions
--- a/src/apps/exchange/state_corp_services.py
+++ b/src/apps/exchange/state_corp_services.py
@@ -19,6 +19,7 @@ from zipfile import ZIP_DEFLATED, ZipFile

 import requests
 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    GenericParserRecord,
    IndustrialProductRecord,
    InspectionRecord,
@@ -583,7 +584,7 @@ class StateCorpExchangeService:
        items: list[dict[str, str | None]] = []
        for record in cls._generic_records(
            allowed_inns,
-            sources=[ParserLoadLog.Source.TRUDVSEM],
+            sources=list(VACANCY_RECORD_SOURCES),
        ):
            payload = cls._record_payload(record)
            published_at = cls._coerce_date(
--- a/src/apps/parsers/migrations/0023_add_job_board_generic_record_sources.py
+++ b/src/apps/parsers/migrations/0023_add_job_board_generic_record_sources.py
@@ -0,0 +1,42 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("parsers", "0022_seed_daily_registry_enrichment_schedule"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="genericparserrecord",
+            name="source",
+            field=models.CharField(
+                choices=[
+                    (
+                        "industrial",
+                        "Сертификаты промышленного производства",
+                    ),
+                    ("industrial_products", "Реестр промышленной продукции"),
+                    ("manufactures", "Реестр производителей"),
+                    ("inspections", "Единый реестр проверок"),
+                    ("procurements", "Единая информационная система закупок"),
+                    ("fns_reports", "Бухгалтерская отчетность ФНС"),
+                    ("procurements_44fz", "Закупки 44-ФЗ"),
+                    ("procurements_223fz", "Закупки 223-ФЗ"),
+                    ("contracts", "Контракты ЕИС"),
+                    ("unfair_suppliers", "Недобросовестные поставщики"),
+                    ("fas_goz", "Уклонение от ГОЗ"),
+                    ("arbitration", "Арбитражные дела"),
+                    ("fedresurs_bankruptcy", "Банкротства Федресурс"),
+                    ("fstec", "Реестры ФСТЭК"),
+                    ("trudvsem", "Вакансии Работа России"),
+                    ("hh", "Вакансии HeadHunter"),
+                    ("superjob", "Вакансии SuperJob"),
+                ],
+                db_index=True,
+                help_text="Источник данных",
+                max_length=50,
+                verbose_name="источник",
+            ),
+        ),
+    ]
--- a/src/apps/parsers/models.py
+++ b/src/apps/parsers/models.py
@@ -8,6 +8,8 @@ from apps.core.mixins import TimestampMixin
 from django.db import models
 from django.utils.translation import gettext_lazy as _

+VACANCY_RECORD_SOURCES = ("trudvsem", "hh", "superjob")
+

 class ParserLoadLog(TimestampMixin, models.Model):
    """
@@ -115,6 +117,13 @@ class ParserBatchSequence(TimestampMixin, models.Model):
        return f"{self.source}: next batch {self.next_batch_id}"


+GENERIC_RECORD_SOURCE_CHOICES = [
+    *ParserLoadLog.Source.choices,
+    ("hh", _("Вакансии HeadHunter")),
+    ("superjob", _("Вакансии SuperJob")),
+]
+
+
 class IndustrialCertificateRecord(TimestampMixin, models.Model):
    """
    Сертификат промышленного производства РФ.
@@ -369,7 +378,7 @@ class GenericParserRecord(TimestampMixin, models.Model):
    source = models.CharField(
        _("источник"),
        max_length=50,
-        choices=ParserLoadLog.Source.choices,
+        choices=GENERIC_RECORD_SOURCE_CHOICES,
        db_index=True,
        help_text=_("Источник данных"),
    )
--- a/src/apps/parsers/serializers.py
+++ b/src/apps/parsers/serializers.py
@@ -850,6 +850,7 @@ class ParserResultRecordSerializer(serializers.Serializer):
    id = serializers.IntegerField()
    load_batch = serializers.IntegerField()
    source = serializers.CharField()
+    vacancy_source = serializers.CharField(allow_blank=True, required=False)
    external_id = serializers.CharField(allow_blank=True)
    inn = serializers.CharField(allow_blank=True)
    ogrn = serializers.CharField(allow_blank=True)
--- a/src/apps/parsers/services.py
+++ b/src/apps/parsers/services.py
@@ -25,6 +25,7 @@ from apps.parsers.clients.proverki.schemas import Inspection
 from apps.parsers.clients.proxy_tools import ProxyToolsClient, ProxyToolsClientError
 from apps.parsers.clients.zakupki.schemas import Procurement
 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    FinancialReport,
    FinancialReportLine,
    GenericParserRecord,
@@ -780,6 +781,23 @@ class GenericParserRecordService(BulkOperationsMixin, BaseService[GenericParserR
    """Сервис для универсальных записей новых источников."""

    model = GenericParserRecord
+    vacancy_record_sources = set(VACANCY_RECORD_SOURCES)
+
+    @classmethod
+    def _storage_source_for_record(
+        cls,
+        record: GenericParserItem,
+        *,
+        default_source: str,
+    ) -> str:
+        """Return DB source for a record without changing the parser load source."""
+        if default_source != ParserLoadLog.Source.TRUDVSEM:
+            return default_source
+        payload = record.payload if isinstance(record.payload, dict) else {}
+        vacancy_source = str(payload.get("vacancy_source") or "").strip()
+        if vacancy_source in cls.vacancy_record_sources:
+            return vacancy_source
+        return default_source

    @classmethod
    def _create_with_exact_count(
@@ -837,15 +855,21 @@ class GenericParserRecordService(BulkOperationsMixin, BaseService[GenericParserR
            logger.warning("No generic parser records to save (source=%s)", source)
            return 0

-        unique_records = {}
+        unique_records: dict[tuple[str, str], GenericParserItem] = {}
        for record in records:
-            unique_records.setdefault(record.external_id, record)
+            record_source = cls._storage_source_for_record(
+                record,
+                default_source=source,
+            )
+            unique_records.setdefault((record_source, record.external_id), record)

-        existing_external_ids = set(
+        source_values = {record_source for record_source, _ in unique_records}
+        external_ids = {external_id for _, external_id in unique_records}
+        existing_keys = set(
            cls.model.objects.filter(
-                source=source,
-                external_id__in=unique_records.keys(),
-            ).values_list("external_id", flat=True)
+                source__in=source_values,
+                external_id__in=external_ids,
+            ).values_list("source", "external_id")
        )
        registry_lookup = RegistryOrganizationResolver.build_lookup(
            [(record.inn, record.ogrn) for record in unique_records.values()]
@@ -853,7 +877,7 @@ class GenericParserRecordService(BulkOperationsMixin, BaseService[GenericParserR
        instances = [
            cls.model(
                load_batch=batch_id,
-                source=source,
+                source=record_source,
                external_id=record.external_id,
                inn=record.inn,
                ogrn=record.ogrn,
@@ -870,8 +894,8 @@ class GenericParserRecordService(BulkOperationsMixin, BaseService[GenericParserR
                    ogrn=record.ogrn,
                ),
            )
-            for external_id, record in unique_records.items()
-            if external_id not in existing_external_ids
+            for (record_source, external_id), record in unique_records.items()
+            if (record_source, external_id) not in existing_keys
        ]
        if not instances:
            logger.info("No new generic records to save (source=%s)", source)
--- a/src/apps/parsers/source_cards.py
+++ b/src/apps/parsers/source_cards.py
@@ -11,6 +11,7 @@ from typing import Any
 from apps.core.models import JobStatus
 from apps.core.services import BackgroundJobService
 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    FinancialReport,
    FinancialReportLine,
    GenericParserRecord,
@@ -22,7 +23,8 @@ from apps.parsers.models import (
    ProcurementRecord,
 )
 from django.conf import settings
-from django.db.models import Max, Q
+from django.db.models import CharField, F, Max, Q, Value
+from django.db.models.functions import Coalesce, NullIf
 from django.http import Http404
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
@@ -326,6 +328,11 @@ GENERIC_RECORD_SOURCES_BY_ITEM_CODE = {
    "fstec": ParserLoadLog.Source.FSTEC,
    "trudvsem": ParserLoadLog.Source.TRUDVSEM,
 }
+PROCUREMENT_BUYER_ITEM_CODES = {
+    "procurements_44fz",
+    "procurements_223fz",
+    "contracts",
+}


 class SourceCardService:
@@ -772,9 +779,11 @@ class SourceCardService:

    @classmethod
    def _get_source_records_count(cls, item_code: str) -> int:
-        generic_source = GENERIC_RECORD_SOURCES_BY_ITEM_CODE.get(item_code)
-        if generic_source:
-            return GenericParserRecord.objects.filter(source=generic_source).count()
+        generic_sources = cls._get_generic_sources_for_item_code(item_code)
+        if generic_sources:
+            return GenericParserRecord.objects.filter(
+                source__in=generic_sources
+            ).count()
        if item_code == "fns_reports":
            return FinancialReportLine.objects.count()
        if item_code == "industrial":
@@ -791,10 +800,14 @@ class SourceCardService:

    @classmethod
    def _get_source_organizations_count(cls, item_code: str) -> int:
-        generic_source = GENERIC_RECORD_SOURCES_BY_ITEM_CODE.get(item_code)
-        if generic_source:
+        generic_sources = cls._get_generic_sources_for_item_code(item_code)
+        if generic_sources:
+            if item_code in PROCUREMENT_BUYER_ITEM_CODES:
+                return cls._get_generic_procurement_buyer_identities(
+                    generic_sources
+                ).count()
            return (
-                GenericParserRecord.objects.filter(source=generic_source)
+                GenericParserRecord.objects.filter(source__in=generic_sources)
                .exclude(inn="")
                .values("inn")
                .distinct()
@@ -846,11 +859,11 @@ class SourceCardService:

    @classmethod
    def _get_source_data_timestamp(cls, item_code: str):
-        generic_source = GENERIC_RECORD_SOURCES_BY_ITEM_CODE.get(item_code)
-        if generic_source:
-            return GenericParserRecord.objects.filter(source=generic_source).aggregate(
-                last_updated=Max("updated_at")
-            )["last_updated"]
+        generic_sources = cls._get_generic_sources_for_item_code(item_code)
+        if generic_sources:
+            return GenericParserRecord.objects.filter(
+                source__in=generic_sources
+            ).aggregate(last_updated=Max("updated_at"))["last_updated"]
        if item_code == "fns_reports":
            return FinancialReport.objects.aggregate(last_updated=Max("updated_at"))[
                "last_updated"
@@ -887,16 +900,13 @@ class SourceCardService:
            generic_sources = cls._get_generic_sources_for_definition(definition)
            legacy_inns = (
                ProcurementRecord.objects.exclude(customer_inn="")
+                .annotate(buyer_identity=F("customer_inn"))
                .order_by()
-                .values_list("customer_inn", flat=True)
+                .values_list("buyer_identity", flat=True)
                .distinct()
            )
-            generic_inns = (
-                GenericParserRecord.objects.filter(source__in=generic_sources)
-                .exclude(inn="")
-                .order_by()
-                .values_list("inn", flat=True)
-                .distinct()
+            generic_inns = cls._get_generic_procurement_buyer_identities(
+                generic_sources
            )
            return legacy_inns.union(generic_inns).count()

@@ -932,15 +942,44 @@ class SourceCardService:
        )
        return industrial_inns.union(manufacturer_inns, product_inns).count()

+    @staticmethod
+    def _get_generic_sources_for_item_code(item_code: str) -> list[str]:
+        generic_source = GENERIC_RECORD_SOURCES_BY_ITEM_CODE.get(item_code)
+        if not generic_source:
+            return []
+        if generic_source == ParserLoadLog.Source.TRUDVSEM:
+            return list(VACANCY_RECORD_SOURCES)
+        return [generic_source]
+
+    @staticmethod
+    def _get_generic_procurement_buyer_identities(generic_sources: list[str]):
+        return (
+            GenericParserRecord.objects.filter(source__in=generic_sources)
+            .annotate(
+                buyer_identity=Coalesce(
+                    NullIf(F("inn"), Value("")),
+                    NullIf(F("organisation_name"), Value("")),
+                    output_field=CharField(),
+                )
+            )
+            .exclude(buyer_identity__isnull=True)
+            .order_by()
+            .values_list("buyer_identity", flat=True)
+            .distinct()
+        )
+
    @staticmethod
    def _get_generic_sources_for_definition(
        definition: SourceCardDefinition,
    ) -> list[str]:
        return list(
            dict.fromkeys(
-                GENERIC_RECORD_SOURCES_BY_ITEM_CODE[item.code]
+                source
                for item in definition.source_items
                if item.code in GENERIC_RECORD_SOURCES_BY_ITEM_CODE
+                for source in SourceCardService._get_generic_sources_for_item_code(
+                    item.code
+                )
            )
        )

--- a/src/apps/parsers/views.py
+++ b/src/apps/parsers/views.py
@@ -17,6 +17,7 @@ from apps.core.services import BackgroundJobService
 from apps.parsers import tasks
 from apps.parsers.fns_upload import FNSUploadService
 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    FinancialReport,
    GenericParserRecord,
    IndustrialCertificateRecord,
@@ -157,6 +158,7 @@ class MultipartFormSwaggerAutoSchema(SwaggerAutoSchema):


 PARSER_TASK_NAMES = set(TASKS_BY_NAME)
+VACANCY_RESULT_SOURCES = set(VACANCY_RECORD_SOURCES)
 NATIVE_RECORD_MODELS = {
    ParserLoadLog.Source.INDUSTRIAL: IndustrialCertificateRecord,
    ParserLoadLog.Source.INDUSTRIAL_PRODUCTS: IndustrialProductRecord,
@@ -1554,10 +1556,23 @@ def _generic_record_to_result(
    *,
    include_payload: bool = True,
 ) -> dict:
+    response_source = record.source
+    vacancy_source = ""
+    if record.source in VACANCY_RESULT_SOURCES:
+        payload = record.payload if isinstance(record.payload, dict) else {}
+        payload_source = str(payload.get("vacancy_source") or "").strip()
+        vacancy_source = (
+            payload_source
+            if payload_source in VACANCY_RESULT_SOURCES
+            else record.source
+        )
+        response_source = vacancy_source
+
    return {
        "id": record.id,
        "load_batch": record.load_batch,
-        "source": record.source,
+        "source": response_source,
+        "vacancy_source": vacancy_source,
        "external_id": record.external_id,
        "inn": record.inn,
        "ogrn": record.ogrn,
@@ -2208,6 +2223,8 @@ def _apply_native_search(queryset, source: str, search: str):


 def _route_model_sources(descriptor) -> set[str]:
+    if descriptor.source == ParserLoadLog.Source.TRUDVSEM:
+        return set(VACANCY_RECORD_SOURCES)
    return {
        item.source
        for item in PARSER_SOURCES.values()
@@ -2220,6 +2237,11 @@ def _result_sources_for_request(descriptor, params: dict) -> set[str]:
    requested_source = params.get("source")
    if not requested_source:
        return route_sources
+    if (
+        descriptor.source == ParserLoadLog.Source.TRUDVSEM
+        and requested_source in VACANCY_RESULT_SOURCES
+    ):
+        return route_sources
    if requested_source in route_sources:
        return {requested_source}
    requested_descriptor = PARSER_SOURCES.get(requested_source)
@@ -2252,10 +2274,27 @@ def _filter_native_result_queryset(source: str, params: dict, sources: set[str])
    return queryset.order_by(*(ordering or ["-created_at"]))


-def _filter_generic_result_queryset(sources: set[str], params: dict):
+def _filter_generic_result_queryset(
+    sources: set[str],
+    params: dict,
+    *,
+    route_source: str,
+):
    queryset = GenericParserRecord.objects.filter(source__in=sources)
    if not sources:
        queryset = queryset.none()
+    requested_source = params.get("source")
+    if (
+        route_source == ParserLoadLog.Source.TRUDVSEM
+        and requested_source in VACANCY_RESULT_SOURCES
+    ):
+        if requested_source == ParserLoadLog.Source.TRUDVSEM:
+            queryset = queryset.filter(
+                Q(payload__vacancy_source=ParserLoadLog.Source.TRUDVSEM)
+                | ~Q(payload__has_key="vacancy_source")
+            )
+        else:
+            queryset = queryset.filter(payload__vacancy_source=requested_source)
    for field in ("id", "external_id", "inn", "ogrn", "load_batch", "status"):
        value = params.get(field)
        if value not in ("", None):
@@ -2295,7 +2334,11 @@ def _filter_result_queryset(source_key: str, params: dict):
    source = descriptor.source
    if source in NATIVE_RECORD_MODELS:
        return descriptor, _filter_native_result_queryset(source, params, sources)
-    return descriptor, _filter_generic_result_queryset(sources, params)
+    return descriptor, _filter_generic_result_queryset(
+        sources,
+        params,
+        route_source=source,
+    )


 def _result_record_to_dict(source: str, record, *, include_payload: bool) -> dict:
--- a/src/organizations/api_enrichment.py
+++ b/src/organizations/api_enrichment.py
@@ -7,6 +7,7 @@ from datetime import date, datetime
 from typing import Any

 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    FinancialReport,
    FinancialReportLine,
    GenericParserRecord,
@@ -148,6 +149,12 @@ def _source_matches(source: str) -> dict[str, set[str]]:
                FinancialReport.objects.values_list("ogrn", flat=True).distinct()
            ),
        }
+    if source == ParserLoadLog.Source.TRUDVSEM:
+        return OrganizationApiEnrichmentService._matching_identifiers_for_all(
+            GenericParserRecord.objects.filter(source__in=VACANCY_RECORD_SOURCES),
+            inn_field="inn",
+            ogrn_field="ogrn",
+        )
    if source in GENERIC_SOURCES:
        return OrganizationApiEnrichmentService._matching_identifiers_for_all(
            GenericParserRecord.objects.filter(source=source),
@@ -395,6 +402,24 @@ class OrganizationApiEnrichmentService:
                items.append(item)
            presence[str(organization.uid)][to_api_data_source(source)] = items

+    @staticmethod
+    def _generic_query_sources(
+        selected_sources: list[str],
+    ) -> tuple[list[str], dict[str, str]]:
+        query_sources: list[str] = []
+        source_bucket_by_record_source: dict[str, str] = {}
+        for source in selected_sources:
+            source_key = str(source)
+            expanded_sources = (
+                VACANCY_RECORD_SOURCES
+                if source == ParserLoadLog.Source.TRUDVSEM
+                else (source_key,)
+            )
+            for expanded_source in expanded_sources:
+                query_sources.append(str(expanded_source))
+                source_bucket_by_record_source[str(expanded_source)] = source_key
+        return query_sources, source_bucket_by_record_source
+
    @classmethod
    def _attach_generic_records(
        cls,
@@ -411,6 +436,10 @@ class OrganizationApiEnrichmentService:
        if identity_filter is None:
            return

+        query_sources, source_bucket_by_record_source = cls._generic_query_sources(
+            selected_sources
+        )
+
        records_by_source_and_inn: dict[str, dict[str, list[dict[str, Any]]]] = {
            str(source): {} for source in selected_sources
        }
@@ -419,13 +448,13 @@ class OrganizationApiEnrichmentService:
        }

        records = (
-            GenericParserRecord.objects.filter(source__in=selected_sources)
+            GenericParserRecord.objects.filter(source__in=query_sources)
            .filter(identity_filter)
            .order_by("source", "-created_at", "-id")
        )
        for record in records:
            item = cls._serialize_generic_record(record)
-            source = str(record.source)
+            source = source_bucket_by_record_source[str(record.source)]
            if record.inn:
                records_by_source_and_inn[source].setdefault(record.inn, []).append(
                    item
--- a/src/organizations/services.py
+++ b/src/organizations/services.py
@@ -7,6 +7,7 @@ from collections.abc import Iterable
 from dataclasses import dataclass

 from apps.parsers.models import (
+    VACANCY_RECORD_SOURCES,
    FinancialReport,
    GenericParserRecord,
    IndustrialCertificateRecord,
@@ -273,8 +274,16 @@ class OrganizationDataSnapshotRefreshService:
            ParserLoadLog.Source.FSTEC,
            ParserLoadLog.Source.TRUDVSEM,
        }:
+            sources = (
+                VACANCY_RECORD_SOURCES
+                if source == ParserLoadLog.Source.TRUDVSEM
+                else (source,)
+            )
            return _identity_values(
-                GenericParserRecord.objects.filter(source=source, load_batch=batch_id),
+                GenericParserRecord.objects.filter(
+                    source__in=sources,
+                    load_batch=batch_id,
+                ),
                inn_field="inn",
                ogrn_field="ogrn",
            )