diff --git a/src/apps/exchange/state_corp_services.py b/src/apps/exchange/state_corp_services.py
index 3a7a904..207d440 100644
--- a/src/apps/exchange/state_corp_services.py
+++ b/src/apps/exchange/state_corp_services.py
@@ -60,7 +60,7 @@ class StateCorpExchangeService:
AAD = b"state-corp-exchange-v1"
PAYLOAD_FORMAT = "state-corp-exchange-payload"
BIN_FORMAT = "state-corp-exchange-bin"
- SCHEMA_VERSION = 3
+ SCHEMA_VERSION = 4
ROSATOM_ROSCOSMOS_GK_CODE_VALUES = ("rosatom", "roscosmos", "roskosmos")
ROSATOM_ROSCOSMOS_GK_NAME_KEYWORDS = ("Росатом", "Роскосмос")
@@ -88,7 +88,7 @@ class StateCorpExchangeService:
for item in organizations
if item.ogrn and item.inn
}
- data = {
+ data: dict[str, list[dict[str, Any]]] = {
"organizations": cls._serialize_organizations(organizations),
"industrial_certificates": cls._serialize_industrial_certificates(
allowed_inns
@@ -107,6 +107,10 @@ class StateCorpExchangeService:
cls._serialize_information_security_registries(allowed_inns)
),
"labor_vacancies": cls._serialize_labor_vacancies(allowed_inns),
+ "electronic_document_exchange": (
+ cls._serialize_electronic_document_exchange(allowed_inns)
+ ),
+ "media_mentions": cls._serialize_media_mentions(allowed_inns),
}
payload_counts = {key: len(value) for key, value in data.items()}
package_id = package_id or cls._build_package_id()
@@ -119,7 +123,7 @@ class StateCorpExchangeService:
"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],
+ "sections": list(data),
},
"data": data,
}
@@ -161,8 +165,12 @@ class StateCorpExchangeService:
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()
+ resolved_url = str(
+ target_url or getattr(settings, "STATE_CORP_EXCHANGE_URL", "")
+ ).strip()
+ resolved_token = str(
+ token or getattr(settings, "STATE_CORP_EXCHANGE_TOKEN", "")
+ ).strip()
timeout_seconds = timeout_seconds or int(
getattr(settings, "STATE_CORP_EXCHANGE_TIMEOUT_SECONDS", 300)
)
@@ -219,7 +227,7 @@ class StateCorpExchangeService:
payload: dict[str, Any],
produced_at,
) -> tuple[str, str, bytes]:
- token = str(settings.STATE_CORP_EXCHANGE_TOKEN or "").strip()
+ token = str(getattr(settings, "STATE_CORP_EXCHANGE_TOKEN", "") or "").strip()
if not token:
raise StateCorpExchangeError("STATE_CORP_EXCHANGE_TOKEN не настроен")
@@ -240,7 +248,7 @@ class StateCorpExchangeService:
header = {
"format": cls.BIN_FORMAT,
"version": 1,
- "key_id": settings.STATE_CORP_EXCHANGE_KEY_ID,
+ "key_id": getattr(settings, "STATE_CORP_EXCHANGE_KEY_ID", "default"),
"nonce": cls._b64url(nonce),
"aad": cls._b64url(cls.AAD),
"package_id": package_id,
@@ -761,8 +769,8 @@ class StateCorpExchangeService:
def _serialize_arbitration_cases(
cls,
allowed_inns: set[str],
- ) -> list[dict[str, str]]:
- items: list[dict[str, str]] = []
+ ) -> list[dict[str, str | None]]:
+ items: list[dict[str, str | None]] = []
for record in cls._generic_records(
allowed_inns,
sources=[ParserLoadLog.Source.ARBITRATION],
@@ -978,6 +986,65 @@ class StateCorpExchangeService:
)
return items
+ @classmethod
+ def _serialize_electronic_document_exchange(
+ cls,
+ allowed_inns: set[str],
+ ) -> list[dict[str, Any]]:
+ items: list[dict[str, Any]] = []
+ for record in cls._canonical_records(
+ allowed_inns,
+ sources=[ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY],
+ ):
+ payload = cls._record_payload(record)
+ organization = record.extension.organization
+ items.append(
+ {
+ "organization_inn": cls._digits(organization.inn),
+ "external_id": record.external_id,
+ "record_type": record.record_type,
+ "title": record.title,
+ "status": record.status,
+ "record_date": record.record_date or None,
+ "source_url": record.url or None,
+ "load_batch": record.load_batch,
+ "organization_name": organization.name,
+ "organization_ogrn": organization.ogrn,
+ "organization_okpo": organization.okpo,
+ "organization_kpp": organization.kpp or None,
+ **payload,
+ }
+ )
+ return items
+
+ @classmethod
+ def _serialize_media_mentions(
+ cls,
+ allowed_inns: set[str],
+ ) -> list[dict[str, Any]]:
+ items: list[dict[str, Any]] = []
+ for record in cls._canonical_records(
+ allowed_inns,
+ sources=[ParserLoadLog.Source.MEDIA_NEWS],
+ ):
+ payload = cls._record_payload(record)
+ items.append(
+ {
+ "organization_inn": cls._digits(record.inn),
+ "external_id": record.external_id,
+ "news_source": payload.get("news_source") or "",
+ "title": record.title,
+ "sentiment": payload.get("sentiment") or record.status,
+ "published_at": payload.get("published_at") or record.record_date,
+ "source_url": payload.get("url") or record.url or None,
+ "excerpt_lines": payload.get("excerpt_lines") or [],
+ "full_text": payload.get("full_text") or "",
+ "okpo": payload.get("okpo") or "",
+ "load_batch": record.load_batch,
+ }
+ )
+ return items
+
@staticmethod
def _record_payload(record: Any) -> dict[str, Any]:
return record.payload if isinstance(record.payload, dict) else {}
diff --git a/src/apps/parsers/admin.py b/src/apps/parsers/admin.py
index a5c6fef..ace0a52 100644
--- a/src/apps/parsers/admin.py
+++ b/src/apps/parsers/admin.py
@@ -11,6 +11,8 @@ from apps.parsers.models import (
InspectionRecord,
ManufacturerRecord,
ParserLoadLog,
+ ParserSourceArtifact,
+ ParserStagedRecord,
ProcurementRecord,
Proxy,
)
@@ -227,6 +229,59 @@ class ParserLoadLogAdmin(admin.ModelAdmin):
return False
+@admin.register(ParserSourceArtifact)
+class ParserSourceArtifactAdmin(admin.ModelAdmin):
+ """Read-only audit trail for downloaded and uploaded source files."""
+
+ list_display = [
+ "uid",
+ "source",
+ "status",
+ "version",
+ "load_batch",
+ "parsed_count",
+ "published_count",
+ "quarantined_count",
+ "created_at",
+ ]
+ list_filter = ["source", "status", "created_at"]
+ search_fields = ["uid", "version", "sha256", "original_name"]
+ readonly_fields = [field.name for field in ParserSourceArtifact._meta.fields]
+ ordering = ["-created_at"]
+
+ def has_add_permission(self, request):
+ return False
+
+ def has_delete_permission(self, request, obj=None):
+ return False
+
+
+@admin.register(ParserStagedRecord)
+class ParserStagedRecordAdmin(admin.ModelAdmin):
+ """Read-only staging and quarantine rows retained with their artifact."""
+
+ list_display = [
+ "uid",
+ "artifact",
+ "row_number",
+ "record_type",
+ "external_id",
+ "disposition",
+ "reason",
+ "organization",
+ ]
+ list_filter = ["disposition", "record_type", "reason", "created_at"]
+ search_fields = ["external_id", "reason", "organization__name"]
+ readonly_fields = [field.name for field in ParserStagedRecord._meta.fields]
+ ordering = ["-created_at"]
+
+ def has_add_permission(self, request):
+ return False
+
+ def has_delete_permission(self, request, obj=None):
+ return False
+
+
class HasCertificateNumberFilter(admin.SimpleListFilter):
"""Фильтр по наличию номера сертификата."""
diff --git a/src/apps/parsers/gosedo.py b/src/apps/parsers/gosedo.py
new file mode 100644
index 0000000..45aaea5
--- /dev/null
+++ b/src/apps/parsers/gosedo.py
@@ -0,0 +1,650 @@
+"""Secure download, parsing and atomic publication of the Gosedo directory."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import tempfile
+import uuid
+from collections import Counter, defaultdict
+from dataclasses import dataclass, field
+from datetime import date
+from pathlib import Path
+from typing import BinaryIO, cast
+from urllib.parse import urljoin, urlparse
+
+import requests
+from apps.parsers.models import ParserSourceArtifact, ParserStagedRecord
+from django.core.files import File
+from django.db import transaction
+from lxml import etree
+from organizations.models import Organization, OrganizationSourceRecord
+from organizations.name_normalization import normalize_organization_name
+from organizations.resolver import OrganizationDirectoryResolver
+from organizations.source_cache import invalidate_source_data_cache
+from organizations.source_ingestion import (
+ OrganizationSourceIngestionService,
+ SourceRecordInput,
+)
+
+GOSEDO_SOURCE = "gosedo_address_directory"
+GOSEDO_URL = "https://gosedo.ru/wp-content/uploads/files/addresseesActual.html"
+GOSEDO_ALLOWED_HOSTS = frozenset({"gosedo.ru", "www.gosedo.ru"})
+GOSEDO_MAX_BYTES = 100 * 1024 * 1024
+GOSEDO_UUID_NAMESPACE = uuid.UUID("ccb8702c-c68b-4bf4-9511-c377306434b0")
+SUMMARY_CLASSES = {
+ "pt-on": ("participant", "active"),
+ "pt-off": ("participant", "inactive"),
+ "oper": ("operator", ""),
+ "org": ("organizer", ""),
+}
+KNOWN_DETAIL_FIELDS = {
+ "УЧАСТНИК МЭДО": "full_name",
+ "ОПЕРАТОР МЭДО": "full_name",
+ "ОРГАНИЗАТОР": "full_name",
+ "Ответственное лицо": "responsible_person",
+ "Контакты ответственного": "responsible_contacts_raw",
+ "Статус участника": "participant_status_raw",
+ "Рег.номер": "registration_raw",
+ "Название": "short_name",
+ "Юр.адрес": "legal_address_raw",
+ "Контакты": "organization_contacts",
+}
+PLACEHOLDERS = {"", "—", "-", "данные не предоставлены"}
+EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}", re.IGNORECASE)
+PHONE_RE = re.compile(r"(?:\+?7|8)?[\s(\-]*\d{3}[\s)\-]*\d{3}[\s\-]*\d{2}[\s\-]*\d{2}")
+VERSION_RE = re.compile(r"Версия:\s*([^\s]+)\s+от\s+(.+)", re.IGNORECASE)
+RUSSIAN_MONTHS = {
+ "января": 1,
+ "февраля": 2,
+ "марта": 3,
+ "апреля": 4,
+ "мая": 5,
+ "июня": 6,
+ "июля": 7,
+ "августа": 8,
+ "сентября": 9,
+ "октября": 10,
+ "ноября": 11,
+ "декабря": 12,
+}
+
+
+class GosedoNotModified(Exception):
+ """The upstream returned HTTP 304 for a conditional request."""
+
+
+class GosedoValidationError(ValueError):
+ """The downloaded document is unsafe or structurally incompatible."""
+
+
+@dataclass(frozen=True)
+class GosedoDownload:
+ handle: BinaryIO
+ sha256: str
+ size_bytes: int
+ content_type: str
+ etag: str
+ last_modified: str
+
+
+@dataclass
+class GosedoRow:
+ row_number: int
+ external_id: str = ""
+ record_type: str = ""
+ status: str = ""
+ raw_data: dict = field(default_factory=dict)
+ normalized_data: dict = field(default_factory=dict)
+ error: str = ""
+
+
+@dataclass(frozen=True)
+class GosedoParseResult:
+ version: str
+ published_at: date
+ rows: list[GosedoRow]
+ unknown_fields: int
+
+
+@dataclass(frozen=True)
+class GosedoPublishResult:
+ parsed: int
+ published: int
+ quarantined: int
+ reasons: dict[str, int]
+
+
+def stable_gosedo_uid(record_type: str, external_id: str) -> uuid.UUID:
+ return uuid.uuid5(
+ GOSEDO_UUID_NAMESPACE,
+ f"{GOSEDO_SOURCE}:{record_type}:{external_id}",
+ )
+
+
+def _clean_text(value: object) -> str:
+ return " ".join(str(value or "").replace("\xa0", " ").split())
+
+
+def _nullable(value: object) -> str | None:
+ cleaned = _clean_text(value)
+ return None if cleaned.casefold() in PLACEHOLDERS else cleaned
+
+
+def _validate_url(url: str) -> None:
+ parsed = urlparse(url)
+ if parsed.scheme != "https" or parsed.hostname not in GOSEDO_ALLOWED_HOSTS:
+ raise GosedoValidationError("unsafe_source_url")
+ if parsed.username or parsed.password or parsed.port not in (None, 443):
+ raise GosedoValidationError("unsafe_source_url")
+
+
+def download_gosedo( # noqa: C901
+ *,
+ url: str = GOSEDO_URL,
+ timeout: tuple[int, int] = (15, 120),
+ session: requests.Session | None = None,
+) -> GosedoDownload:
+ """Download the allowlisted source with conditional headers and hard limits."""
+ latest = (
+ ParserSourceArtifact.objects.filter(
+ source=GOSEDO_SOURCE,
+ status=ParserSourceArtifact.Status.PUBLISHED,
+ )
+ .order_by("-created_at")
+ .first()
+ )
+ headers = {"Accept": "text/html,application/xhtml+xml"}
+ if latest and latest.etag:
+ headers["If-None-Match"] = latest.etag
+ if latest and latest.last_modified:
+ headers["If-Modified-Since"] = latest.last_modified
+
+ client = session or requests.Session()
+ current_url = url
+ response = None
+ for redirect_count in range(4):
+ _validate_url(current_url)
+ response = client.get(
+ current_url,
+ headers=headers,
+ stream=True,
+ allow_redirects=False,
+ timeout=timeout,
+ )
+ if response.status_code == 304:
+ response.close()
+ raise GosedoNotModified("Источник не изменился (HTTP 304)")
+ if response.status_code in {301, 302, 303, 307, 308}:
+ if redirect_count >= 3:
+ response.close()
+ raise GosedoValidationError("too_many_redirects")
+ location = response.headers.get("Location", "")
+ response.close()
+ if not location:
+ raise GosedoValidationError("redirect_without_location")
+ current_url = urljoin(current_url, location)
+ continue
+ break
+
+ if response is None or response.status_code != 200:
+ status_code = response.status_code if response is not None else "unknown"
+ if response is not None:
+ response.close()
+ raise GosedoValidationError(f"unexpected_http_status:{status_code}")
+
+ content_type = response.headers.get("Content-Type", "")
+ mime = content_type.split(";", 1)[0].strip().lower()
+ if mime not in {"text/html", "application/xhtml+xml"}:
+ response.close()
+ raise GosedoValidationError("invalid_content_type")
+ content_length = response.headers.get("Content-Length")
+ if content_length:
+ try:
+ declared_size = int(content_length)
+ except ValueError as exc:
+ response.close()
+ raise GosedoValidationError("invalid_content_length") from exc
+ if declared_size > GOSEDO_MAX_BYTES:
+ response.close()
+ raise GosedoValidationError("source_too_large")
+
+ handle = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024, mode="w+b")
+ digest = hashlib.sha256()
+ size = 0
+ try:
+ for chunk in response.iter_content(chunk_size=128 * 1024):
+ if not chunk:
+ continue
+ size += len(chunk)
+ if size > GOSEDO_MAX_BYTES:
+ raise GosedoValidationError("source_too_large")
+ digest.update(chunk)
+ handle.write(chunk)
+ except Exception:
+ handle.close()
+ raise
+ finally:
+ response.close()
+ handle.seek(0)
+ try:
+ handle.read().decode("utf-8", errors="strict")
+ except UnicodeDecodeError as exc:
+ handle.close()
+ raise GosedoValidationError("invalid_utf8") from exc
+ handle.seek(0)
+ return GosedoDownload(
+ handle=cast(BinaryIO, handle),
+ sha256=digest.hexdigest(),
+ size_bytes=size,
+ content_type=content_type,
+ etag=response.headers.get("ETag", ""),
+ last_modified=response.headers.get("Last-Modified", ""),
+ )
+
+
+def _parse_version(text: str) -> tuple[str, date] | None:
+ match = VERSION_RE.search(_clean_text(text))
+ if not match:
+ return None
+ date_parts = match.group(2).casefold().split()
+ if len(date_parts) != 3 or date_parts[1] not in RUSSIAN_MONTHS:
+ return None
+ try:
+ published_at = date(
+ int(date_parts[2]),
+ RUSSIAN_MONTHS[date_parts[1]],
+ int(date_parts[0]),
+ )
+ except ValueError:
+ return None
+ return match.group(1), published_at
+
+
+def _element_text(element) -> str:
+ return _clean_text(" ".join(element.itertext()))
+
+
+def _parse_detail_fields(row) -> tuple[dict[str, str], dict[str, str]]:
+ mapped: dict[str, str] = {}
+ extra: dict[str, str] = {}
+ for dt in row.xpath(".//dt"):
+ label = _element_text(dt).rstrip(":")
+ dd = dt.getnext()
+ while dd is not None and str(dd.tag).lower() != "dd":
+ dd = dd.getnext()
+ value = _element_text(dd) if dd is not None else ""
+ target = KNOWN_DETAIL_FIELDS.get(label)
+ if target:
+ mapped[target] = value
+ elif label:
+ extra[label] = value
+ return mapped, extra
+
+
+def _normalize_gosedo_row(summary: dict, details: dict, extra: dict) -> dict:
+ contacts = _nullable(details.get("responsible_contacts_raw"))
+ emails = EMAIL_RE.findall(contacts or "")
+ phones = [_clean_text(value) for value in PHONE_RE.findall(contacts or "")]
+ status_raw = _nullable(details.get("participant_status_raw"))
+ status_folded = (status_raw or "").casefold()
+ attestation = None
+ if "не аттестован" in status_folded:
+ attestation = "not_attested"
+ elif "аттестован" in status_folded:
+ attestation = "attested"
+ registration_raw = _clean_text(details.get("registration_raw"))
+ registration_match = re.search(r"ОГРН\s*:\s*(\d{13})", registration_raw, re.I)
+ return {
+ "operator_uid": _nullable(summary.get("operator_uid")),
+ "medo_address": _nullable(summary.get("medo_address")),
+ "short_name": _clean_text(
+ details.get("short_name") or summary.get("short_name")
+ ),
+ "full_name": _nullable(details.get("full_name")),
+ "responsible_person": _nullable(details.get("responsible_person")),
+ "responsible_phone": phones[0] if phones else None,
+ "responsible_email": emails[0] if emails else None,
+ "responsible_phones": phones,
+ "responsible_emails": emails,
+ "responsible_contacts_raw": contacts,
+ "participant_status": summary.get("status") or None,
+ "participant_status_raw": status_raw,
+ "attestation_status": attestation,
+ "attestation_status_raw": status_raw,
+ "registration_type": "ogrn" if registration_match else None,
+ "registration_number": registration_match.group(1)
+ if registration_match
+ else None,
+ "legal_address_raw": _nullable(details.get("legal_address_raw")),
+ "organization_contacts": _nullable(details.get("organization_contacts")),
+ "source_row_class": summary["row_class"],
+ "extra_fields": extra,
+ }
+
+
+def parse_gosedo(handle: BinaryIO) -> GosedoParseResult: # noqa: C901
+ """Stream summary/detail row pairs without building a full document DOM."""
+ handle.seek(0)
+ version: str | None = None
+ published_at: date | None = None
+ found_table = False
+ rows: list[GosedoRow] = []
+ pending: tuple[int, dict] | None = None
+ row_number = 0
+ unknown_fields = 0
+ try:
+ context = etree.iterparse(
+ handle,
+ events=("end",),
+ tag=("p", "table", "tr"),
+ html=True,
+ recover=False,
+ encoding="utf-8",
+ )
+ for _, element in context:
+ tag = str(element.tag).lower()
+ if tag == "p" and version is None:
+ parsed_version = _parse_version(_element_text(element))
+ if parsed_version:
+ version, published_at = parsed_version
+ elif tag == "table" and element.get("id") == "dataTable":
+ found_table = True
+ elif tag == "tr":
+ row_number += 1
+ classes = set((element.get("class") or "").split())
+ row_class = next(
+ (item for item in SUMMARY_CLASSES if item in classes), None
+ )
+ if row_class:
+ if pending is not None:
+ pending_number, pending_summary = pending
+ rows.append(
+ GosedoRow(
+ row_number=pending_number,
+ external_id=pending_summary["external_id"],
+ record_type=pending_summary["record_type"],
+ raw_data=pending_summary,
+ error="invalid_row_pair",
+ )
+ )
+ cells = element.xpath("./td")
+ record_type, status = SUMMARY_CLASSES[row_class]
+ summary = {
+ "external_id": _element_text(cells[0]) if cells else "",
+ "short_name": _element_text(cells[1]) if len(cells) > 1 else "",
+ "medo_address": _element_text(cells[2])
+ if len(cells) > 2
+ else "",
+ "operator_uid": element.get("operatorUid")
+ or element.get("operatoruid")
+ or "",
+ "record_type": record_type,
+ "status": status,
+ "row_class": row_class,
+ }
+ pending = (row_number, summary)
+ elif "details" in classes:
+ if pending is None:
+ rows.append(
+ GosedoRow(row_number=row_number, error="invalid_row_pair")
+ )
+ else:
+ pending_number, summary = pending
+ details, extra = _parse_detail_fields(element)
+ unknown_fields += len(extra)
+ normalized = _normalize_gosedo_row(summary, details, extra)
+ error = ""
+ if not summary["external_id"] or not normalized["short_name"]:
+ error = "missing_required_source_fields"
+ rows.append(
+ GosedoRow(
+ row_number=pending_number,
+ external_id=summary["external_id"],
+ record_type=summary["record_type"],
+ status=summary["status"],
+ raw_data={"summary": summary, "details": details},
+ normalized_data=normalized,
+ error=error,
+ )
+ )
+ pending = None
+ elif pending is not None:
+ pending_number, pending_summary = pending
+ rows.append(
+ GosedoRow(
+ row_number=pending_number,
+ external_id=pending_summary["external_id"],
+ record_type=pending_summary["record_type"],
+ raw_data=pending_summary,
+ error="invalid_row_pair",
+ )
+ )
+ pending = None
+ element.clear()
+ while element.getprevious() is not None:
+ del element.getparent()[0]
+ except etree.XMLSyntaxError as exc:
+ raise GosedoValidationError("invalid_html") from exc
+ if pending is not None:
+ pending_number, summary = pending
+ rows.append(
+ GosedoRow(
+ row_number=pending_number,
+ external_id=summary["external_id"],
+ record_type=summary["record_type"],
+ raw_data=summary,
+ error="invalid_row_pair",
+ )
+ )
+ if not found_table or not version or not published_at or not rows:
+ raise GosedoValidationError("source_structure_changed")
+ return GosedoParseResult(version, published_at, rows, unknown_fields)
+
+
+def _organization_name_index() -> dict[str, list[Organization]]:
+ index: dict[str, list[Organization]] = defaultdict(list)
+ queryset = OrganizationDirectoryResolver._directory_queryset().exclude(name="")
+ for organization in queryset:
+ names = {
+ normalize_organization_name(value)
+ for value in (
+ organization.name,
+ organization.full_name,
+ organization.short_name,
+ organization.pn_name,
+ )
+ if value
+ }
+ for name in names:
+ if name:
+ index[name].append(organization)
+ return index
+
+
+def _resolve_row_organization(
+ row: GosedoRow,
+ name_index: dict[str, list[Organization]],
+) -> tuple[Organization | None, str]:
+ if row.error:
+ return None, row.error
+ data = row.normalized_data
+ if row.record_type == "participant":
+ ogrn = str(data.get("registration_number") or "")
+ if len(ogrn) != 13:
+ return None, "invalid_ogrn"
+ result = OrganizationDirectoryResolver.resolve(
+ OrganizationDirectoryResolver.identity(ogrn=ogrn)
+ )
+ if result.organization is None:
+ return None, (
+ "ambiguous_organization"
+ if result.status == "ambiguous"
+ else "organization_not_found"
+ )
+ organization = result.organization
+ else:
+ normalized_name = normalize_organization_name(
+ data.get("full_name") or data.get("short_name") or ""
+ )
+ candidates = name_index.get(normalized_name, [])
+ if not candidates:
+ return None, "organization_not_found"
+ if len(candidates) != 1:
+ return None, "ambiguous_organization"
+ organization = candidates[0]
+ if not all(
+ (organization.name, organization.inn, organization.ogrn, organization.okpo)
+ ):
+ return None, "incomplete_organization_identity"
+ return organization, ""
+
+
+def publish_gosedo_snapshot(
+ *,
+ artifact: ParserSourceArtifact,
+ parsed: GosedoParseResult,
+ load_batch: int,
+) -> GosedoPublishResult:
+ """Resolve rows, retain quarantine and atomically replace published records."""
+ name_index = _organization_name_index()
+ reasons: Counter[str] = Counter()
+ staged_instances: list[ParserStagedRecord] = []
+ inputs: list[SourceRecordInput] = []
+ for row in parsed.rows:
+ organization, reason = _resolve_row_organization(row, name_index)
+ disposition = ParserStagedRecord.Disposition.STAGED
+ if organization is None:
+ disposition = ParserStagedRecord.Disposition.QUARANTINED
+ reasons[reason] += 1
+ else:
+ data = {
+ **row.normalized_data,
+ "source_version": parsed.version,
+ "source_published_at": parsed.published_at.isoformat(),
+ "inn": organization.inn,
+ "kpp": organization.kpp or None,
+ "ogrn": organization.ogrn,
+ "okpo": organization.okpo,
+ "artifact_id": str(artifact.uid),
+ }
+ inputs.append(
+ SourceRecordInput(
+ uid=stable_gosedo_uid(row.record_type, row.external_id),
+ external_id=row.external_id,
+ record_type=row.record_type,
+ title=data.get("short_name") or data.get("full_name") or "",
+ organization_name=organization.name,
+ inn=organization.inn,
+ kpp=organization.kpp,
+ ogrn=organization.ogrn,
+ record_date=parsed.published_at.isoformat(),
+ status=row.status,
+ url=GOSEDO_URL,
+ payload=data,
+ )
+ )
+ staged_instances.append(
+ ParserStagedRecord(
+ artifact=artifact,
+ row_number=row.row_number,
+ external_id=row.external_id,
+ record_type=row.record_type,
+ raw_data=row.raw_data,
+ normalized_data=row.normalized_data,
+ organization=organization,
+ disposition=disposition,
+ reason=reason,
+ )
+ )
+ ParserStagedRecord.objects.bulk_create(staged_instances, batch_size=500)
+
+ keep_uids = [record.uid for record in inputs if record.uid]
+ with transaction.atomic():
+ result = OrganizationSourceIngestionService.save_records(
+ source=GOSEDO_SOURCE,
+ load_batch=load_batch,
+ records=inputs,
+ )
+ if result.unresolved:
+ raise GosedoValidationError("organization_resolution_changed")
+ OrganizationSourceRecord.objects.filter(source=GOSEDO_SOURCE).exclude(
+ uid__in=keep_uids
+ ).delete()
+ from organizations.models import ElectronicDocumentExchangeExtension
+
+ ElectronicDocumentExchangeExtension.objects.filter(
+ records__isnull=True
+ ).delete()
+ ParserStagedRecord.objects.filter(
+ artifact=artifact,
+ disposition=ParserStagedRecord.Disposition.STAGED,
+ ).update(disposition=ParserStagedRecord.Disposition.PUBLISHED)
+ invalidate_source_data_cache()
+ return GosedoPublishResult(
+ parsed=len(parsed.rows),
+ published=len(inputs),
+ quarantined=sum(reasons.values()),
+ reasons=dict(reasons),
+ )
+
+
+def refresh_gosedo(
+ *, load_batch: int, uploaded_by_id: int | None = None
+) -> tuple[ParserSourceArtifact, GosedoPublishResult]:
+ download = download_gosedo()
+ artifact = ParserSourceArtifact.objects.create(
+ source=GOSEDO_SOURCE,
+ sha256=download.sha256,
+ etag=download.etag,
+ last_modified=download.last_modified,
+ content_type=download.content_type,
+ size_bytes=download.size_bytes,
+ original_name=Path(urlparse(GOSEDO_URL).path).name,
+ load_batch=load_batch,
+ uploaded_by_id=uploaded_by_id,
+ )
+ try:
+ artifact.file.save(artifact.original_name, File(download.handle), save=True)
+ download.handle.seek(0)
+ parsed = parse_gosedo(download.handle)
+ artifact.version = parsed.version
+ artifact.source_published_at = parsed.published_at
+ artifact.status = ParserSourceArtifact.Status.PARSED
+ artifact.metadata = {"unknown_detail_fields": parsed.unknown_fields}
+ artifact.save(
+ update_fields=[
+ "version",
+ "source_published_at",
+ "status",
+ "metadata",
+ "updated_at",
+ ]
+ )
+ result = publish_gosedo_snapshot(
+ artifact=artifact,
+ parsed=parsed,
+ load_batch=load_batch,
+ )
+ artifact.status = ParserSourceArtifact.Status.PUBLISHED
+ artifact.parsed_count = result.parsed
+ artifact.published_count = result.published
+ artifact.quarantined_count = result.quarantined
+ artifact.rejection_reasons = result.reasons
+ artifact.save(
+ update_fields=[
+ "status",
+ "parsed_count",
+ "published_count",
+ "quarantined_count",
+ "rejection_reasons",
+ "updated_at",
+ ]
+ )
+ return artifact, result
+ except Exception:
+ artifact.status = ParserSourceArtifact.Status.REJECTED
+ artifact.save(update_fields=["status", "updated_at"])
+ raise
+ finally:
+ download.handle.close()
diff --git a/src/apps/parsers/media_news.py b/src/apps/parsers/media_news.py
new file mode 100644
index 0000000..c5c1649
--- /dev/null
+++ b/src/apps/parsers/media_news.py
@@ -0,0 +1,311 @@
+"""Validated Excel ingestion for organization media mentions."""
+
+from __future__ import annotations
+
+import hashlib
+from collections import Counter
+from dataclasses import dataclass
+from datetime import date, datetime
+from pathlib import Path
+from typing import BinaryIO
+
+from apps.parsers.models import ParserSourceArtifact, ParserStagedRecord
+from django.core.files import File
+from django.db import transaction
+from django.utils.dateparse import parse_date
+from openpyxl import load_workbook
+from organizations.models import Organization
+from organizations.resolver import OrganizationDirectoryResolver
+from organizations.source_ingestion import (
+ OrganizationSourceIngestionService,
+ SourceRecordInput,
+)
+
+MEDIA_NEWS_SOURCE = "media_news"
+MEDIA_NEWS_RECORD_TYPE = "media_mention"
+DATE_HEADERS = {
+ "Дата акутальности новости",
+ "Дата актуальности новости",
+}
+REQUIRED_HEADERS = {
+ "okpo": {"ОКПО"},
+ "inn": {"ИНН"},
+ "published_at": DATE_HEADERS,
+ "news_source": {"Источник"},
+ "url": {"URL"},
+ "full_text": {"Текст"},
+ "sentiment": {"Оценка"},
+}
+SENTIMENTS = {
+ "положительная": "positive",
+ "отрицательная": "negative",
+}
+MEDIA_NEWS_MAX_BYTES = 25 * 1024 * 1024
+
+
+@dataclass(frozen=True)
+class MediaNewsImportResult:
+ parsed: int
+ published: int
+ quarantined: int
+ reasons: dict[str, int]
+
+
+def normalize_news_text(value: object) -> tuple[str, list[str]]:
+ text = (
+ str(value or "")
+ .replace("_x000D_", "")
+ .replace("\r\n", "\n")
+ .replace("\r", "\n")
+ )
+ normalized_lines = [line.strip() for line in text.split("\n")]
+ full_text = "\n".join(normalized_lines).strip()
+ excerpt_lines = [line for line in normalized_lines if line][:4]
+ return full_text, excerpt_lines
+
+
+def _identifier(cell) -> str:
+ value = cell.value
+ if value is None or isinstance(value, bool):
+ return ""
+ if isinstance(value, float):
+ if not value.is_integer():
+ return ""
+ value = int(value)
+ text = str(value).strip()
+ if not text.isdigit():
+ return ""
+ number_format = str(cell.number_format or "")
+ if set(number_format) <= {"0"} and len(number_format) > len(text):
+ text = text.zfill(len(number_format))
+ return text
+
+
+def _published_date(value: object) -> date | None:
+ if isinstance(value, datetime):
+ return value.date()
+ if isinstance(value, date):
+ return value
+ text = str(value or "").strip()
+ parsed = parse_date(text)
+ if parsed:
+ return parsed
+ for format_string in ("%d.%m.%Y", "%d/%m/%Y"):
+ try:
+ return datetime.strptime(text, format_string).date()
+ except ValueError:
+ continue
+ return None
+
+
+def stable_media_external_id(
+ *,
+ inn: str,
+ okpo: str,
+ published_at: date,
+ news_source: str,
+ url: str,
+ full_text: str,
+) -> str:
+ url_or_text_hash = (
+ url.strip() or hashlib.sha256(full_text.encode("utf-8")).hexdigest()
+ )
+ raw = "\x1f".join(
+ (inn, okpo, published_at.isoformat(), news_source.strip(), url_or_text_hash)
+ )
+ return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def _resolve_media_organization(inn: str, okpo: str) -> tuple[Organization | None, str]:
+ directory = OrganizationDirectoryResolver._directory_queryset()
+ by_inn = list(directory.filter(inn=inn)[:2])
+ by_okpo = list(directory.filter(okpo=okpo)[:2])
+ if len(by_inn) > 1 or len(by_okpo) > 1:
+ return None, "ambiguous_organization"
+ if not by_inn or not by_okpo:
+ return None, "organization_not_found"
+ if by_inn[0].uid != by_okpo[0].uid:
+ return None, "inn_okpo_conflict"
+ return by_inn[0], ""
+
+
+def _header_map(sheet) -> dict[str, int]:
+ values = [
+ str(cell.value or "").strip()
+ for cell in next(sheet.iter_rows(min_row=1, max_row=1))
+ ]
+ result: dict[str, int] = {}
+ for field_name, aliases in REQUIRED_HEADERS.items():
+ matches = [index for index, value in enumerate(values) if value in aliases]
+ if len(matches) != 1:
+ raise ValueError(f"missing_or_duplicate_column:{field_name}")
+ result[field_name] = matches[0]
+ return result
+
+
+def import_media_news( # noqa: C901
+ *,
+ handle: BinaryIO,
+ original_name: str,
+ load_batch: int,
+ uploaded_by_id: int | None,
+) -> tuple[ParserSourceArtifact, MediaNewsImportResult]:
+ handle.seek(0, 2)
+ size_bytes = handle.tell()
+ if size_bytes > MEDIA_NEWS_MAX_BYTES:
+ raise ValueError("source_too_large")
+ handle.seek(0)
+ digest = hashlib.sha256()
+ for chunk in iter(lambda: handle.read(128 * 1024), b""):
+ digest.update(chunk)
+ handle.seek(0)
+ sha256 = digest.hexdigest()
+ safe_original_name = Path(str(original_name).replace("\\", "/")).name
+ artifact = ParserSourceArtifact.objects.create(
+ source=MEDIA_NEWS_SOURCE,
+ version=sha256,
+ sha256=sha256,
+ content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+ size_bytes=size_bytes,
+ original_name=safe_original_name,
+ load_batch=load_batch,
+ uploaded_by_id=uploaded_by_id,
+ )
+ reasons: Counter[str] = Counter()
+ inputs: list[SourceRecordInput] = []
+ staged: list[ParserStagedRecord] = []
+ workbook = None
+ try:
+ artifact.file.save(artifact.original_name, File(handle), save=True)
+ handle.seek(0)
+ workbook = load_workbook(handle, read_only=True, data_only=False)
+ sheet = workbook.active
+ columns = _header_map(sheet)
+ for row_number, cells in enumerate(sheet.iter_rows(min_row=2), start=2):
+ values = {key: cells[index] for key, index in columns.items()}
+ reason = ""
+ if any(cell.data_type == "f" for cell in values.values()):
+ reason = "formula_not_allowed"
+ inn = _identifier(values["inn"])
+ okpo = _identifier(values["okpo"])
+ if not reason and (not inn or not okpo):
+ reason = "invalid_identifier"
+ published_at = _published_date(values["published_at"].value)
+ if not reason and published_at is None:
+ reason = "invalid_date"
+ sentiment_raw = str(values["sentiment"].value or "").strip()
+ sentiment = SENTIMENTS.get(sentiment_raw.casefold())
+ if not reason and sentiment is None:
+ reason = "invalid_sentiment"
+ news_source = str(values["news_source"].value or "").strip()
+ url = str(values["url"].value or "").strip()
+ full_text, excerpt_lines = normalize_news_text(values["full_text"].value)
+ if not reason and (not news_source or not full_text):
+ reason = "missing_required_value"
+ organization = None
+ if not reason:
+ organization, reason = _resolve_media_organization(inn, okpo)
+ external_id = ""
+ if published_at is not None and news_source and full_text and inn and okpo:
+ external_id = stable_media_external_id(
+ inn=inn,
+ okpo=okpo,
+ published_at=published_at,
+ news_source=news_source,
+ url=url,
+ full_text=full_text,
+ )
+ normalized = {
+ "inn": inn,
+ "okpo": okpo,
+ "published_at": published_at.isoformat() if published_at else None,
+ "news_source": news_source,
+ "url": url or None,
+ "full_text": full_text,
+ "excerpt_lines": excerpt_lines,
+ "sentiment": sentiment,
+ }
+ if reason:
+ reasons[reason] += 1
+ else:
+ assert organization is not None
+ assert published_at is not None
+ assert sentiment is not None
+ title = excerpt_lines[1] if len(excerpt_lines) > 1 else excerpt_lines[0]
+ payload = {**normalized, "artifact_id": str(artifact.uid)}
+ inputs.append(
+ SourceRecordInput(
+ external_id=external_id,
+ record_type=MEDIA_NEWS_RECORD_TYPE,
+ title=title,
+ organization_name=organization.name,
+ inn=organization.inn,
+ kpp=organization.kpp,
+ ogrn=organization.ogrn,
+ record_date=published_at.isoformat(),
+ status=sentiment,
+ url=url,
+ payload=payload,
+ )
+ )
+ staged.append(
+ ParserStagedRecord(
+ artifact=artifact,
+ row_number=row_number,
+ external_id=external_id,
+ record_type=MEDIA_NEWS_RECORD_TYPE,
+ raw_data={
+ key: str(cell.value or "") for key, cell in values.items()
+ },
+ normalized_data=normalized,
+ organization=organization,
+ disposition=(
+ ParserStagedRecord.Disposition.QUARANTINED
+ if reason
+ else ParserStagedRecord.Disposition.STAGED
+ ),
+ reason=reason,
+ )
+ )
+ ParserStagedRecord.objects.bulk_create(staged, batch_size=500)
+ with transaction.atomic():
+ ingestion = OrganizationSourceIngestionService.save_records(
+ source=MEDIA_NEWS_SOURCE,
+ load_batch=load_batch,
+ records=inputs,
+ )
+ if ingestion.unresolved:
+ raise ValueError("organization_resolution_changed")
+ ParserStagedRecord.objects.filter(
+ artifact=artifact,
+ disposition=ParserStagedRecord.Disposition.STAGED,
+ ).update(disposition=ParserStagedRecord.Disposition.PUBLISHED)
+ result = MediaNewsImportResult(
+ parsed=len(staged),
+ published=len(inputs),
+ quarantined=sum(reasons.values()),
+ reasons=dict(reasons),
+ )
+ artifact.status = ParserSourceArtifact.Status.PUBLISHED
+ artifact.parsed_count = result.parsed
+ artifact.published_count = result.published
+ artifact.quarantined_count = result.quarantined
+ artifact.rejection_reasons = result.reasons
+ artifact.save(
+ update_fields=[
+ "status",
+ "parsed_count",
+ "published_count",
+ "quarantined_count",
+ "rejection_reasons",
+ "updated_at",
+ ]
+ )
+ return artifact, result
+ except Exception:
+ artifact.status = ParserSourceArtifact.Status.REJECTED
+ artifact.save(update_fields=["status", "updated_at"])
+ raise
+ finally:
+ if workbook is not None:
+ workbook.close()
diff --git a/src/apps/parsers/migrations/0031_source_artifacts_and_new_sources.py b/src/apps/parsers/migrations/0031_source_artifacts_and_new_sources.py
new file mode 100644
index 0000000..df6fa2e
--- /dev/null
+++ b/src/apps/parsers/migrations/0031_source_artifacts_and_new_sources.py
@@ -0,0 +1,281 @@
+import uuid
+
+import apps.parsers.models
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+SOURCE_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", "Вакансии Работа России"),
+ ("gosedo_address_directory", "Глобальный адресный справочник ГосЭДО"),
+ ("media_news", "Новости СМИ"),
+]
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("organizations", "0010_gosedo_media_source_groups"),
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ("parsers", "0030_extend_stale_parser_job_timeouts"),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name="ParserSourceArtifact",
+ fields=[
+ (
+ "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="обновлено",
+ ),
+ ),
+ (
+ "uid",
+ models.UUIDField(
+ default=uuid.uuid4,
+ editable=False,
+ primary_key=True,
+ serialize=False,
+ ),
+ ),
+ (
+ "source",
+ models.CharField(
+ choices=SOURCE_CHOICES, db_index=True, max_length=50
+ ),
+ ),
+ (
+ "status",
+ models.CharField(
+ choices=[
+ ("downloaded", "Загружен"),
+ ("parsed", "Разобран"),
+ ("published", "Опубликован"),
+ ("rejected", "Отклонён"),
+ ("skipped", "Пропущен"),
+ ],
+ db_index=True,
+ default="downloaded",
+ max_length=20,
+ ),
+ ),
+ (
+ "version",
+ models.CharField(blank=True, db_index=True, max_length=255),
+ ),
+ (
+ "source_published_at",
+ models.DateField(blank=True, db_index=True, null=True),
+ ),
+ ("sha256", models.CharField(blank=True, db_index=True, max_length=64)),
+ ("etag", models.CharField(blank=True, max_length=512)),
+ ("last_modified", models.CharField(blank=True, max_length=512)),
+ ("content_type", models.CharField(blank=True, max_length=255)),
+ ("size_bytes", models.PositiveBigIntegerField(default=0)),
+ ("original_name", models.CharField(blank=True, max_length=512)),
+ (
+ "file",
+ models.FileField(
+ blank=True,
+ max_length=1024,
+ upload_to=apps.parsers.models.parser_artifact_upload_to,
+ ),
+ ),
+ (
+ "load_batch",
+ models.PositiveIntegerField(blank=True, db_index=True, null=True),
+ ),
+ ("parsed_count", models.PositiveIntegerField(default=0)),
+ ("published_count", models.PositiveIntegerField(default=0)),
+ ("quarantined_count", models.PositiveIntegerField(default=0)),
+ ("rejection_reasons", models.JSONField(blank=True, default=dict)),
+ ("metadata", models.JSONField(blank=True, default=dict)),
+ (
+ "uploaded_by",
+ models.ForeignKey(
+ blank=True,
+ null=True,
+ on_delete=django.db.models.deletion.SET_NULL,
+ related_name="parser_source_artifacts",
+ to=settings.AUTH_USER_MODEL,
+ ),
+ ),
+ ],
+ options={
+ "db_table": "parsers_source_artifact",
+ "ordering": ["-created_at"],
+ },
+ ),
+ migrations.AlterField(
+ model_name="genericparserrecord",
+ name="source",
+ field=models.CharField(
+ choices=[
+ *SOURCE_CHOICES,
+ ("hh", "Вакансии HeadHunter"),
+ ("superjob", "Вакансии SuperJob"),
+ ],
+ db_index=True,
+ help_text="Источник данных",
+ max_length=50,
+ verbose_name="источник",
+ ),
+ ),
+ migrations.AlterField(
+ model_name="parserbatchsequence",
+ name="source",
+ field=models.CharField(
+ choices=SOURCE_CHOICES,
+ help_text="Источник данных",
+ max_length=50,
+ unique=True,
+ verbose_name="источник",
+ ),
+ ),
+ migrations.AlterField(
+ model_name="parserloadlog",
+ name="source",
+ field=models.CharField(
+ choices=SOURCE_CHOICES,
+ db_index=True,
+ help_text="Источник данных",
+ max_length=50,
+ verbose_name="источник",
+ ),
+ ),
+ migrations.CreateModel(
+ name="ParserStagedRecord",
+ fields=[
+ (
+ "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="обновлено",
+ ),
+ ),
+ (
+ "uid",
+ models.UUIDField(
+ default=uuid.uuid4,
+ editable=False,
+ primary_key=True,
+ serialize=False,
+ ),
+ ),
+ ("row_number", models.PositiveIntegerField(db_index=True)),
+ (
+ "external_id",
+ models.CharField(blank=True, db_index=True, max_length=255),
+ ),
+ (
+ "record_type",
+ models.CharField(blank=True, db_index=True, max_length=64),
+ ),
+ ("raw_data", models.JSONField(blank=True, default=dict)),
+ ("normalized_data", models.JSONField(blank=True, default=dict)),
+ (
+ "disposition",
+ models.CharField(
+ choices=[
+ ("staged", "Подготовлена"),
+ ("published", "Опубликована"),
+ ("quarantined", "Карантин"),
+ ],
+ db_index=True,
+ default="staged",
+ max_length=20,
+ ),
+ ),
+ ("reason", models.CharField(blank=True, db_index=True, max_length=255)),
+ (
+ "artifact",
+ models.ForeignKey(
+ on_delete=django.db.models.deletion.CASCADE,
+ related_name="staged_records",
+ to="parsers.parsersourceartifact",
+ ),
+ ),
+ (
+ "organization",
+ models.ForeignKey(
+ blank=True,
+ null=True,
+ on_delete=django.db.models.deletion.SET_NULL,
+ related_name="parser_staged_records",
+ to="organizations.organization",
+ ),
+ ),
+ ],
+ options={
+ "db_table": "parsers_staged_record",
+ "ordering": ["artifact_id", "row_number"],
+ },
+ ),
+ migrations.AddIndex(
+ model_name="parserstagedrecord",
+ index=models.Index(
+ fields=["artifact", "disposition"],
+ name="parsers_sta_artifac_e8f8d3_idx",
+ ),
+ ),
+ migrations.AddIndex(
+ model_name="parserstagedrecord",
+ index=models.Index(
+ fields=["record_type", "external_id"],
+ name="parsers_sta_record__d80428_idx",
+ ),
+ ),
+ migrations.AddConstraint(
+ model_name="parserstagedrecord",
+ constraint=models.UniqueConstraint(
+ fields=("artifact", "row_number"), name="unique_parser_artifact_row"
+ ),
+ ),
+ migrations.AddIndex(
+ model_name="parsersourceartifact",
+ index=models.Index(
+ fields=["source", "-created_at"], name="parsers_sou_source_b35ffe_idx"
+ ),
+ ),
+ migrations.AddIndex(
+ model_name="parsersourceartifact",
+ index=models.Index(
+ fields=["source", "sha256"], name="parsers_sou_source_f2088c_idx"
+ ),
+ ),
+ ]
diff --git a/src/apps/parsers/migrations/0032_seed_gosedo_and_artifact_schedules.py b/src/apps/parsers/migrations/0032_seed_gosedo_and_artifact_schedules.py
new file mode 100644
index 0000000..4b853fe
--- /dev/null
+++ b/src/apps/parsers/migrations/0032_seed_gosedo_and_artifact_schedules.py
@@ -0,0 +1,64 @@
+import json
+
+from django.db import migrations
+
+TASKS = (
+ (
+ "parser:gosedo-address-directory:daily-msk",
+ "apps.parsers.tasks.parse_gosedo_address_directory",
+ "0",
+ "3",
+ "Daily conditional refresh of the Gosedo address directory.",
+ ),
+ (
+ "parser:source-artifacts:cleanup-daily-msk",
+ "apps.parsers.tasks.cleanup_source_artifacts",
+ "30",
+ "6",
+ "Daily parser artifact cleanup (90 days, minimum 10 versions).",
+ ),
+)
+
+
+def seed_schedules(apps, schema_editor):
+ CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule")
+ PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+ field_names = {field.name for field in PeriodicTask._meta.fields}
+ for name, task, minute, hour, description in TASKS:
+ crontab, _ = CrontabSchedule.objects.get_or_create(
+ minute=minute,
+ hour=hour,
+ day_of_week="*",
+ day_of_month="*",
+ month_of_year="*",
+ timezone="Europe/Moscow",
+ )
+ schedule_fields = {"crontab": crontab}
+ for field_name in ("interval", "solar", "clocked"):
+ if field_name in field_names:
+ schedule_fields[field_name] = None
+ PeriodicTask.objects.update_or_create(
+ name=name,
+ defaults={
+ "task": task,
+ "args": json.dumps([]),
+ "kwargs": json.dumps({}),
+ "enabled": True,
+ "description": description,
+ **schedule_fields,
+ },
+ )
+
+
+def remove_schedules(apps, schema_editor):
+ PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+ PeriodicTask.objects.filter(name__in=[item[0] for item in TASKS]).delete()
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("django_celery_beat", "0018_improve_crontab_helptext"),
+ ("parsers", "0031_source_artifacts_and_new_sources"),
+ ]
+
+ operations = [migrations.RunPython(seed_schedules, reverse_code=remove_schedules)]
diff --git a/src/apps/parsers/models.py b/src/apps/parsers/models.py
index 161bacd..1aa126b 100644
--- a/src/apps/parsers/models.py
+++ b/src/apps/parsers/models.py
@@ -4,7 +4,10 @@
Используют миксины из apps.core для стандартных полей и поведения.
"""
+import uuid
+
from apps.core.mixins import TimestampMixin
+from django.conf import settings
from django.db import models
from django.utils.translation import gettext_lazy as _
@@ -34,6 +37,11 @@ class ParserLoadLog(TimestampMixin, models.Model):
FEDRESURS_BANKRUPTCY = "fedresurs_bankruptcy", _("Банкротства Федресурс")
FSTEC = "fstec", _("Реестры ФСТЭК")
TRUDVSEM = "trudvsem", _("Вакансии Работа России")
+ GOSEDO_ADDRESS_DIRECTORY = (
+ "gosedo_address_directory",
+ _("Глобальный адресный справочник ГосЭДО"),
+ )
+ MEDIA_NEWS = "media_news", _("Новости СМИ")
class Status(models.TextChoices):
SUCCESS = "success", _("Успешно")
@@ -117,6 +125,118 @@ class ParserBatchSequence(TimestampMixin, models.Model):
return f"{self.source}: next batch {self.next_batch_id}"
+def parser_artifact_upload_to(instance, filename: str) -> str:
+ """Keep parser artifacts grouped by source and immutable artifact id."""
+ safe_name = filename.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
+ return f"parser-artifacts/{instance.source}/{instance.uid}/{safe_name}"
+
+
+class ParserSourceArtifact(TimestampMixin, models.Model):
+ """Immutable raw file and processing counters for one parser run."""
+
+ class Status(models.TextChoices):
+ DOWNLOADED = "downloaded", _("Загружен")
+ PARSED = "parsed", _("Разобран")
+ PUBLISHED = "published", _("Опубликован")
+ REJECTED = "rejected", _("Отклонён")
+ SKIPPED = "skipped", _("Пропущен")
+
+ uid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ source = models.CharField(
+ max_length=50, choices=ParserLoadLog.Source.choices, db_index=True
+ )
+ status = models.CharField(
+ max_length=20, choices=Status.choices, default=Status.DOWNLOADED, db_index=True
+ )
+ version = models.CharField(max_length=255, blank=True, db_index=True)
+ source_published_at = models.DateField(null=True, blank=True, db_index=True)
+ sha256 = models.CharField(max_length=64, blank=True, db_index=True)
+ etag = models.CharField(max_length=512, blank=True)
+ last_modified = models.CharField(max_length=512, blank=True)
+ content_type = models.CharField(max_length=255, blank=True)
+ size_bytes = models.PositiveBigIntegerField(default=0)
+ original_name = models.CharField(max_length=512, blank=True)
+ file = models.FileField(
+ upload_to=parser_artifact_upload_to, max_length=1024, blank=True
+ )
+ load_batch = models.PositiveIntegerField(null=True, blank=True, db_index=True)
+ uploaded_by = models.ForeignKey(
+ settings.AUTH_USER_MODEL,
+ on_delete=models.SET_NULL,
+ null=True,
+ blank=True,
+ related_name="parser_source_artifacts",
+ )
+ parsed_count = models.PositiveIntegerField(default=0)
+ published_count = models.PositiveIntegerField(default=0)
+ quarantined_count = models.PositiveIntegerField(default=0)
+ rejection_reasons = models.JSONField(default=dict, blank=True)
+ metadata = models.JSONField(default=dict, blank=True)
+
+ class Meta:
+ db_table = "parsers_source_artifact"
+ ordering = ["-created_at"]
+ indexes = [
+ models.Index(fields=["source", "-created_at"]),
+ models.Index(fields=["source", "sha256"]),
+ ]
+
+ def __str__(self) -> str:
+ return f"{self.source} artifact {self.uid}"
+
+
+class ParserStagedRecord(TimestampMixin, models.Model):
+ """Normalized source row, including rows held in quarantine."""
+
+ class Disposition(models.TextChoices):
+ STAGED = "staged", _("Подготовлена")
+ PUBLISHED = "published", _("Опубликована")
+ QUARANTINED = "quarantined", _("Карантин")
+
+ uid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ artifact = models.ForeignKey(
+ ParserSourceArtifact,
+ on_delete=models.CASCADE,
+ related_name="staged_records",
+ )
+ row_number = models.PositiveIntegerField(db_index=True)
+ external_id = models.CharField(max_length=255, blank=True, db_index=True)
+ record_type = models.CharField(max_length=64, blank=True, db_index=True)
+ raw_data = models.JSONField(default=dict, blank=True)
+ normalized_data = models.JSONField(default=dict, blank=True)
+ organization = models.ForeignKey(
+ "organizations.Organization",
+ on_delete=models.SET_NULL,
+ null=True,
+ blank=True,
+ related_name="parser_staged_records",
+ )
+ disposition = models.CharField(
+ max_length=20,
+ choices=Disposition.choices,
+ default=Disposition.STAGED,
+ db_index=True,
+ )
+ reason = models.CharField(max_length=255, blank=True, db_index=True)
+
+ class Meta:
+ db_table = "parsers_staged_record"
+ ordering = ["artifact_id", "row_number"]
+ constraints = [
+ models.UniqueConstraint(
+ fields=["artifact", "row_number"],
+ name="unique_parser_artifact_row",
+ )
+ ]
+ indexes = [
+ models.Index(fields=["artifact", "disposition"]),
+ models.Index(fields=["record_type", "external_id"]),
+ ]
+
+ def __str__(self) -> str:
+ return f"{self.artifact_id} row {self.row_number}: {self.disposition}"
+
+
class CheckoCollectionAttempt(TimestampMixin, models.Model):
"""Monthly external collection claim for one organization and source."""
diff --git a/src/apps/parsers/serializers.py b/src/apps/parsers/serializers.py
index 092c831..7a195b2 100644
--- a/src/apps/parsers/serializers.py
+++ b/src/apps/parsers/serializers.py
@@ -19,6 +19,7 @@ from apps.parsers.models import (
InspectionRecord,
ManufacturerRecord,
ParserLoadLog,
+ ParserSourceArtifact,
ParsingSettings,
ProcurementRecord,
)
@@ -524,6 +525,8 @@ class ParserSourceSerializer(serializers.Serializer):
result_list_url = serializers.CharField()
result_detail_url = serializers.CharField()
upload_url = serializers.CharField(allow_blank=True)
+ supports_refresh = serializers.BooleanField()
+ admin_only = serializers.BooleanField()
class ParserRunRequestSerializer(serializers.Serializer):
@@ -729,6 +732,7 @@ class ParserRunResponseSerializer(serializers.Serializer):
"""Ответ API на запуск задачи."""
task_id = serializers.CharField()
+ task_ids = serializers.ListField(child=serializers.CharField())
source = serializers.CharField()
task_name = serializers.CharField()
@@ -747,6 +751,11 @@ class ParserLoadLogSerializer(serializers.ModelSerializer):
source_display = serializers.CharField(source="get_source_display", read_only=True)
organizations_count = serializers.SerializerMethodField()
+ artifact_uid = serializers.SerializerMethodField()
+ parsed_count = serializers.SerializerMethodField()
+ published_count = serializers.SerializerMethodField()
+ quarantined_count = serializers.SerializerMethodField()
+ rejection_reasons = serializers.SerializerMethodField()
class Meta:
model = ParserLoadLog
@@ -757,6 +766,11 @@ class ParserLoadLogSerializer(serializers.ModelSerializer):
"source_display",
"records_count",
"organizations_count",
+ "artifact_uid",
+ "parsed_count",
+ "published_count",
+ "quarantined_count",
+ "rejection_reasons",
"status",
"error_message",
"created_at",
@@ -765,6 +779,17 @@ class ParserLoadLogSerializer(serializers.ModelSerializer):
read_only_fields = fields
def get_organizations_count(self, obj) -> int:
+ artifact = self._get_artifact(obj)
+ if artifact is not None:
+ return (
+ artifact.staged_records.filter(
+ disposition="published",
+ organization__isnull=False,
+ )
+ .values("organization_id")
+ .distinct()
+ .count()
+ )
if obj.source == ParserLoadLog.Source.FNS_REPORTS:
return (
FinancialReport.objects.filter(load_batch=obj.batch_id)
@@ -815,6 +840,41 @@ class ParserLoadLogSerializer(serializers.ModelSerializer):
)
return 0
+ @staticmethod
+ def _get_artifact(obj) -> ParserSourceArtifact | None:
+ if hasattr(obj, "_source_artifact_cache"):
+ return obj._source_artifact_cache
+ artifact = (
+ ParserSourceArtifact.objects.filter(
+ source=obj.source,
+ load_batch=obj.batch_id,
+ )
+ .order_by("-created_at")
+ .first()
+ )
+ obj._source_artifact_cache = artifact
+ return artifact
+
+ def get_artifact_uid(self, obj) -> str | None:
+ artifact = self._get_artifact(obj)
+ return str(artifact.uid) if artifact is not None else None
+
+ def get_parsed_count(self, obj) -> int:
+ artifact = self._get_artifact(obj)
+ return artifact.parsed_count if artifact is not None else obj.records_count
+
+ def get_published_count(self, obj) -> int:
+ artifact = self._get_artifact(obj)
+ return artifact.published_count if artifact is not None else obj.records_count
+
+ def get_quarantined_count(self, obj) -> int:
+ artifact = self._get_artifact(obj)
+ return artifact.quarantined_count if artifact is not None else 0
+
+ def get_rejection_reasons(self, obj) -> dict:
+ artifact = self._get_artifact(obj)
+ return dict(artifact.rejection_reasons) if artifact is not None else {}
+
class GenericParserRecordSerializer(
CanonicalOrganizationEnrichmentMixin,
@@ -888,6 +948,11 @@ class ParserLoadLogListSerializer(serializers.Serializer):
source_label = serializers.CharField(read_only=True, allow_null=True)
records_count = serializers.IntegerField(read_only=True)
organizations_count = serializers.IntegerField(read_only=True)
+ artifact_uid = serializers.UUIDField(read_only=True, allow_null=True)
+ parsed_count = serializers.IntegerField(read_only=True)
+ published_count = serializers.IntegerField(read_only=True)
+ quarantined_count = serializers.IntegerField(read_only=True)
+ rejection_reasons = serializers.DictField(read_only=True)
status = serializers.CharField(read_only=True)
status_label = serializers.CharField(read_only=True)
error_message = serializers.CharField(read_only=True, allow_blank=True)
@@ -953,6 +1018,7 @@ class SourceCardItemSerializer(serializers.Serializer):
records_count = serializers.IntegerField(read_only=True)
organizations_count = serializers.IntegerField(read_only=True)
last_updated_at = serializers.DateTimeField(read_only=True, allow_null=True)
+ upload_url = serializers.CharField(read_only=True, allow_blank=True)
latest_load = SourceCardLoadSerializer(read_only=True, allow_null=True)
latest_success_load = SourceCardLoadSerializer(read_only=True, allow_null=True)
@@ -975,6 +1041,8 @@ class SourceCardSerializer(serializers.Serializer):
error_message = serializers.CharField(read_only=True)
task_names = serializers.ListField(child=serializers.CharField(), read_only=True)
refresh_requires_params = serializers.BooleanField(read_only=True)
+ supports_refresh = serializers.BooleanField(read_only=True)
+ upload_url = serializers.CharField(read_only=True, allow_blank=True)
refresh_params = SourceCardRefreshParamSerializer(many=True, read_only=True)
@@ -1056,4 +1124,5 @@ class SourceCardRefreshFrontendResponseSerializer(serializers.Serializer):
"""Минимальный ответ запуска обновления карточки по md."""
task_id = serializers.CharField(read_only=True, allow_null=True)
+ task_ids = serializers.ListField(child=serializers.CharField(), read_only=True)
status = serializers.CharField(read_only=True)
diff --git a/src/apps/parsers/source_artifacts.py b/src/apps/parsers/source_artifacts.py
new file mode 100644
index 0000000..ac13da4
--- /dev/null
+++ b/src/apps/parsers/source_artifacts.py
@@ -0,0 +1,37 @@
+"""Retention helpers for immutable parser source artifacts."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from apps.parsers.models import ParserSourceArtifact
+from django.db.models import Q
+from django.utils import timezone
+
+
+def cleanup_parser_source_artifacts(
+ *,
+ retention_days: int = 90,
+ minimum_versions: int = 10,
+) -> int:
+ """Delete expired artifacts while retaining the newest versions per source."""
+ cutoff = timezone.now() - timedelta(days=retention_days)
+ deleted = 0
+ sources = ParserSourceArtifact.objects.values_list("source", flat=True).distinct()
+ for source in sources:
+ protected_ids = list(
+ ParserSourceArtifact.objects.filter(source=source)
+ .order_by("-created_at")
+ .values_list("uid", flat=True)[:minimum_versions]
+ )
+ expired = list(
+ ParserSourceArtifact.objects.filter(source=source, created_at__lt=cutoff)
+ .exclude(Q(uid__in=protected_ids))
+ .only("uid", "file")
+ )
+ for artifact in expired:
+ if artifact.file:
+ artifact.file.delete(save=False)
+ artifact.delete()
+ deleted += 1
+ return deleted
diff --git a/src/apps/parsers/source_cards.py b/src/apps/parsers/source_cards.py
index 8fe2225..44f74ac 100644
--- a/src/apps/parsers/source_cards.py
+++ b/src/apps/parsers/source_cards.py
@@ -72,6 +72,8 @@ class SourceCardDefinition:
refresh_params: tuple[RefreshParamDefinition, ...] = ()
refresh_interval: timedelta | None = None
is_available: bool = True
+ supports_refresh: bool = True
+ upload_url: str = ""
@dataclass(frozen=True)
@@ -338,6 +340,44 @@ SOURCE_CARD_DEFINITIONS: tuple[SourceCardDefinition, ...] = (
),
),
),
+ SourceCardDefinition(
+ slug="gosedo-global-address-directory",
+ title="ГосЭДО: Глобальный адресный справочник",
+ description=(
+ "Участники, операторы и организаторы системы межведомственного "
+ "электронного документооборота."
+ ),
+ order=100,
+ task_names=("apps.parsers.tasks.parse_gosedo_address_directory",),
+ source_items=(
+ SourceItemDefinition(
+ code="gosedo_global_address_directory",
+ title="Глобальный адресный справочник ГосЭДО",
+ description="Официальный адресный справочник участников ГосЭДО.",
+ parser_source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ refresh_key="gosedo_address_directory",
+ ),
+ ),
+ refresh_interval=timedelta(days=1),
+ ),
+ SourceCardDefinition(
+ slug="media-mentions",
+ title="Новости СМИ",
+ description="Загруженные упоминания организаций в СМИ с оценкой тональности.",
+ order=110,
+ task_names=(),
+ source_items=(
+ SourceItemDefinition(
+ code="media_mentions",
+ title="Новости СМИ",
+ description="История упоминаний организаций из Excel-загрузок.",
+ parser_source=ParserLoadLog.Source.MEDIA_NEWS,
+ refresh_key="media_news",
+ ),
+ ),
+ supports_refresh=False,
+ upload_url="/api/v1/parsers/upload/media_news/",
+ ),
)
SOURCE_CARD_BY_SLUG = {item.slug: item for item in SOURCE_CARD_DEFINITIONS}
@@ -501,6 +541,8 @@ class SourceCardService:
"description": definition.description,
"order": definition.order,
"is_available": definition.is_available,
+ "supports_refresh": definition.supports_refresh,
+ "upload_url": definition.upload_url,
"status": status,
"status_label": cls._get_status_label(status),
"progress": progress,
@@ -539,6 +581,10 @@ class SourceCardService:
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
definition = cls.get_definition(slug)
+ if not definition.supports_refresh:
+ raise ValidationError(
+ {"detail": "Обновление для карточки не поддерживается."}
+ )
params = cls._validate_refresh_params(definition, params or {})
cls.clear_cache()
tasks = cls._launch_refresh(
@@ -1048,11 +1094,19 @@ class SourceCardService:
parse_fas_goz_evasion,
parse_fedresurs_bankruptcy,
parse_fstec_registers,
+ parse_gosedo_address_directory,
parse_trudvsem_vacancies,
parse_unfair_suppliers,
)
specs = {
+ "gosedo-global-address-directory": (
+ (
+ parse_gosedo_address_directory,
+ "apps.parsers.tasks.parse_gosedo_address_directory",
+ ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ ),
+ ),
"bankruptcy-procedures": (
(
parse_fedresurs_bankruptcy,
@@ -1169,6 +1223,11 @@ class SourceCardService:
if latest_success_load
else last_updated_at
),
+ "upload_url": (
+ "/api/v1/parsers/upload/media_news/"
+ if item.parser_source == ParserLoadLog.Source.MEDIA_NEWS
+ else ""
+ ),
"latest_load": cls._serialize_load_log(latest_load),
"latest_success_load": cls._serialize_load_log(latest_success_load),
}
diff --git a/src/apps/parsers/source_registry.py b/src/apps/parsers/source_registry.py
index 880ea43..b47a262 100644
--- a/src/apps/parsers/source_registry.py
+++ b/src/apps/parsers/source_registry.py
@@ -27,6 +27,8 @@ class ParserSourceDescriptor:
supports_file_upload: bool = False
api_route: str = ""
upload_route: str = ""
+ supports_refresh: bool = True
+ admin_only: bool = False
@property
def result_list_url(self) -> str:
@@ -41,7 +43,7 @@ class ParserSourceDescriptor:
@property
def upload_url(self) -> str:
"""Frontend/API URL ручной загрузки файла источника."""
- if not self.supports_file_upload or not self.api_route:
+ if not self.supports_file_upload:
return ""
return f"/api/v1/{self.upload_api_route}/"
@@ -295,6 +297,38 @@ PARSER_SOURCES: dict[str, ParserSourceDescriptor] = {
),
api_route="trudvsem/vacancies",
),
+ "gosedo_address_directory": ParserSourceDescriptor(
+ key="gosedo_address_directory",
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ title="ГосЭДО: Глобальный адресный справочник",
+ agency="ГосЭДО",
+ data_scope="Участники, операторы и организаторы электронного документооборота",
+ task_name="apps.parsers.tasks.parse_gosedo_address_directory",
+ mode="official_html",
+ upstream_url="https://gosedo.ru/wp-content/uploads/files/addresseesActual.html",
+ access_method="official_public_file",
+ parser_strategy="streaming_summary_details_html",
+ source_notes="Ежедневный conditional GET; публикация полным атомарным снимком.",
+ admin_only=True,
+ ),
+ "media_news": ParserSourceDescriptor(
+ key="media_news",
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ title="Новости СМИ",
+ agency="Пользовательская загрузка",
+ data_scope="Оцененные упоминания организаций в СМИ",
+ task_name="apps.parsers.tasks.parse_media_news",
+ mode="manual_upload",
+ access_method="admin_upload",
+ parser_strategy="validated_xlsx_upsert",
+ source_notes=(
+ "Повторная загрузка обновляет совпавшие новости и сохраняет историю."
+ ),
+ supports_file_upload=True,
+ upload_route="parsers/upload/media_news",
+ supports_refresh=False,
+ admin_only=True,
+ ),
}
diff --git a/src/apps/parsers/tasks.py b/src/apps/parsers/tasks.py
index 21bada0..27337f9 100644
--- a/src/apps/parsers/tasks.py
+++ b/src/apps/parsers/tasks.py
@@ -52,6 +52,8 @@ 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.gosedo import GosedoNotModified, refresh_gosedo
+from apps.parsers.media_news import import_media_news
from apps.parsers.models import CheckoCollectionAttempt, ParserLoadLog
from apps.parsers.services import (
FNSReportOrganizationResolutionSkipped,
@@ -66,6 +68,7 @@ from apps.parsers.services import (
ProxyService,
ProxyToolsSyncService,
)
+from apps.parsers.source_artifacts import cleanup_parser_source_artifacts
from apps.parsers.source_registry import PARSER_SOURCES
from celery import shared_task
from django.conf import settings
@@ -3604,6 +3607,146 @@ def parse_fstec_registers(
)
+@shared_task(bind=True, soft_time_limit=45 * 60, time_limit=60 * 60)
+def parse_gosedo_address_directory(
+ self,
+ *,
+ requested_by_id: int | None = None,
+) -> dict:
+ """Download and atomically publish the Gosedo directory snapshot."""
+ source = ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY
+ task_name = "apps.parsers.tasks.parse_gosedo_address_directory"
+ load_log, batch_id = ParserLoadLogService.create_load_log_with_next_batch_id(
+ source=source,
+ status=ParserLoadLog.Status.IN_PROGRESS,
+ )
+ task_id = self.request.id or str(uuid.uuid4())
+ job = _get_or_create_background_job(
+ task_id=task_id,
+ task_name=task_name,
+ source=source,
+ batch_id=batch_id,
+ requested_by_id=requested_by_id,
+ meta={"source_key": source},
+ )
+ job.mark_started()
+ job.update_progress(5, "Проверка обновления ГосЭДО...")
+ try:
+ artifact, result = refresh_gosedo(
+ load_batch=batch_id,
+ uploaded_by_id=requested_by_id,
+ )
+ ParserLoadLogService.update(
+ load_log,
+ status=ParserLoadLog.Status.SUCCESS,
+ records_count=result.published,
+ )
+ payload = {
+ "status": "success",
+ "batch_id": batch_id,
+ "artifact_id": str(artifact.uid),
+ "parsed": result.parsed,
+ "published": result.published,
+ "quarantined": result.quarantined,
+ "rejection_reasons": result.reasons,
+ }
+ job.update_progress(100, "Снимок ГосЭДО опубликован")
+ job.complete(result=payload)
+ return payload
+ except GosedoNotModified as exc:
+ ParserLoadLogService.update(
+ load_log,
+ status=ParserLoadLog.Status.SKIPPED,
+ error_message=str(exc),
+ )
+ payload = {
+ "status": "skipped",
+ "batch_id": batch_id,
+ "published": 0,
+ "reason": str(exc),
+ }
+ job.update_progress(100, str(exc))
+ job.complete(result=payload)
+ return payload
+ except Exception as exc:
+ logger.error("Gosedo refresh failed: %s", exc, exc_info=True)
+ ParserLoadLogService.mark_failed(load_log, str(exc))
+ job.fail(error=str(exc))
+ raise
+
+
+@shared_task(bind=True, soft_time_limit=15 * 60, time_limit=20 * 60)
+def parse_media_news(
+ self,
+ *,
+ file_path: str,
+ original_name: str | None = None,
+ requested_by_id: int | None = None,
+) -> dict:
+ """Validate and upsert media mentions from an uploaded XLSX workbook."""
+ from django.core.files.storage import default_storage
+
+ source = ParserLoadLog.Source.MEDIA_NEWS
+ task_name = "apps.parsers.tasks.parse_media_news"
+ load_log, batch_id = ParserLoadLogService.create_load_log_with_next_batch_id(
+ source=source,
+ status=ParserLoadLog.Status.IN_PROGRESS,
+ )
+ task_id = self.request.id or str(uuid.uuid4())
+ job = _get_or_create_background_job(
+ task_id=task_id,
+ task_name=task_name,
+ source=source,
+ batch_id=batch_id,
+ requested_by_id=requested_by_id,
+ meta={"source_key": source},
+ )
+ job.mark_started()
+ job.update_progress(10, "Проверка файла новостей...")
+ try:
+ with default_storage.open(file_path, "rb") as handle:
+ artifact, result = import_media_news(
+ handle=handle,
+ original_name=original_name or Path(file_path).name,
+ load_batch=batch_id,
+ uploaded_by_id=requested_by_id,
+ )
+ ParserLoadLogService.update(
+ load_log,
+ status=ParserLoadLog.Status.SUCCESS,
+ records_count=result.published,
+ )
+ payload = {
+ "status": "success",
+ "batch_id": batch_id,
+ "artifact_id": str(artifact.uid),
+ "parsed": result.parsed,
+ "published": result.published,
+ "quarantined": result.quarantined,
+ "rejection_reasons": result.reasons,
+ }
+ job.update_progress(100, "Новости СМИ загружены")
+ job.complete(result=payload)
+ return payload
+ except Exception as exc:
+ logger.error("Media news import failed: %s", exc, exc_info=True)
+ ParserLoadLogService.mark_failed(load_log, str(exc))
+ job.fail(error=str(exc))
+ raise
+ finally:
+ if file_path.startswith("parser_uploads/") and default_storage.exists(
+ file_path
+ ):
+ default_storage.delete(file_path)
+
+
+@shared_task
+def cleanup_source_artifacts() -> dict:
+ """Apply the 90-day/minimum-ten-versions parser artifact retention policy."""
+ deleted = cleanup_parser_source_artifacts()
+ return {"status": "success", "deleted": deleted}
+
+
@shared_task
def cleanup_stale_parser_loads(
max_age_minutes: int | None = None,
diff --git a/src/apps/parsers/views.py b/src/apps/parsers/views.py
index 3702954..fb806bc 100644
--- a/src/apps/parsers/views.py
+++ b/src/apps/parsers/views.py
@@ -28,6 +28,7 @@ from apps.parsers.models import (
InspectionRecord,
ManufacturerRecord,
ParserLoadLog,
+ ParserSourceArtifact,
ParsingSettings,
ProcurementRecord,
)
@@ -152,6 +153,10 @@ TASKS_BY_NAME = {
),
"apps.parsers.tasks.parse_fstec_registers": tasks.parse_fstec_registers,
"apps.parsers.tasks.parse_trudvsem_vacancies": tasks.parse_trudvsem_vacancies,
+ "apps.parsers.tasks.parse_gosedo_address_directory": (
+ tasks.parse_gosedo_address_directory
+ ),
+ "apps.parsers.tasks.parse_media_news": tasks.parse_media_news,
}
PARSER_SOURCE_ALIASES = {
@@ -223,6 +228,8 @@ EXISTING_TASK_PARAMS = {
"current_month",
},
"fns_financial": {"requested_by_id"},
+ "gosedo_address_directory": {"requested_by_id"},
+ "media_news": {"file_path", "original_name", "requested_by_id"},
}
@@ -589,13 +596,43 @@ def _get_parser_log_organizations_count(log: ParserLoadLog) -> int:
def _serialize_parser_log_row(log: ParserLoadLog) -> dict:
+ artifact = (
+ ParserSourceArtifact.objects.filter(
+ source=log.source,
+ load_batch=log.batch_id,
+ )
+ .order_by("-created_at")
+ .first()
+ )
+ organizations_count = _get_parser_log_organizations_count(log)
+ if artifact is not None:
+ organizations_count = (
+ artifact.staged_records.filter(
+ disposition="published",
+ organization__isnull=False,
+ )
+ .values("organization_id")
+ .distinct()
+ .count()
+ )
return {
"id": log.id,
"batch_id": log.batch_id,
"source": _get_parser_log_source_value(log),
"source_label": _get_parser_log_source_label(log),
"records_count": log.records_count,
- "organizations_count": _get_parser_log_organizations_count(log),
+ "organizations_count": organizations_count,
+ "artifact_uid": artifact.uid if artifact is not None else None,
+ "parsed_count": artifact.parsed_count
+ if artifact is not None
+ else log.records_count,
+ "published_count": (
+ artifact.published_count if artifact is not None else log.records_count
+ ),
+ "quarantined_count": artifact.quarantined_count if artifact is not None else 0,
+ "rejection_reasons": (
+ dict(artifact.rejection_reasons) if artifact is not None else {}
+ ),
"status": log.status,
"status_label": _get_parser_log_status_label(log.status),
"error_message": log.error_message,
@@ -1357,6 +1394,7 @@ class SourceCardRefreshView(APIView):
tasks = output.get("tasks", [])
response_payload = {
"task_id": tasks[0]["task_id"] if tasks else None,
+ "task_ids": [task["task_id"] for task in tasks],
"status": "accepted",
}
return Response(
@@ -1519,6 +1557,11 @@ class ParserLoadLogExportView(APIView):
"source_label",
"records_count",
"organizations_count",
+ "artifact_uid",
+ "parsed_count",
+ "published_count",
+ "quarantined_count",
+ "rejection_reasons",
"status",
"status_label",
"error_message",
@@ -1536,6 +1579,11 @@ class ParserLoadLogExportView(APIView):
row["source_label"],
row["records_count"],
row["organizations_count"],
+ row["artifact_uid"] or "",
+ row["parsed_count"],
+ row["published_count"],
+ row["quarantined_count"],
+ json.dumps(row["rejection_reasons"], ensure_ascii=False),
row["status"],
row["status_label"],
row["error_message"],
@@ -2653,6 +2701,18 @@ class ParserRunView(APIView):
descriptor = PARSER_SOURCES.get(canonical_source_key)
if descriptor is None:
return _source_not_found_response(source_key)
+ if descriptor.admin_only and not request.user.is_staff:
+ return Response(status=status.HTTP_403_FORBIDDEN)
+ if not descriptor.supports_refresh:
+ return api_error_response(
+ [
+ {
+ "code": "run_not_supported",
+ "message": "Источник обновляется только загрузкой файла",
+ }
+ ],
+ status_code=status.HTTP_400_BAD_REQUEST,
+ )
serializer = ParserRunRequestSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
task = TASKS_BY_NAME[descriptor.task_name]
@@ -2680,6 +2740,7 @@ class ParserRunView(APIView):
return api_response(
{
"task_id": active_job.task_id,
+ "task_ids": [active_job.task_id],
"source": descriptor.source,
"task_name": descriptor.task_name,
"already_running": True,
@@ -2700,6 +2761,7 @@ class ParserRunView(APIView):
return api_response(
{
"task_id": async_result.id,
+ "task_ids": [async_result.id],
"source": descriptor.source,
"task_name": descriptor.task_name,
},
@@ -2717,6 +2779,8 @@ class ParserUploadView(APIView):
descriptor = PARSER_SOURCES.get(source_key)
if descriptor is None:
return _source_not_found_response(source_key)
+ if descriptor.admin_only and not request.user.is_staff:
+ return Response(status=status.HTTP_403_FORBIDDEN)
if not descriptor.supports_file_upload:
return api_error_response(
[
@@ -2729,13 +2793,26 @@ class ParserUploadView(APIView):
)
serializer = ParserUploadRequestSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
- file_path = _save_uploaded_parser_file(serializer.validated_data["file"])
+ if source_key == "media_news" and not serializer.validated_data[
+ "file"
+ ].name.lower().endswith(".xlsx"):
+ return api_error_response(
+ [
+ {
+ "code": "invalid_file_type",
+ "message": "Для новостей требуется XLSX",
+ }
+ ],
+ status_code=status.HTTP_400_BAD_REQUEST,
+ )
+ uploaded_file = serializer.validated_data["file"]
+ file_path = _save_uploaded_parser_file(uploaded_file)
run_serializer = ParserRunRequestSerializer(data={"file_path": file_path})
run_serializer.is_valid(raise_exception=True)
task = TASKS_BY_NAME[descriptor.task_name]
task_kwargs = build_task_kwargs(
source_key,
- {"file_path": file_path},
+ {"file_path": file_path, "original_name": uploaded_file.name},
request.user.id,
)
task_id = str(uuid.uuid4())
@@ -2747,12 +2824,14 @@ class ParserUploadView(APIView):
"source_key": source_key,
"source": descriptor.source,
"file_path": file_path,
+ "original_name": uploaded_file.name,
},
)
async_result = task.apply_async(kwargs=task_kwargs, task_id=task_id)
return api_response(
{
"task_id": async_result.id,
+ "task_ids": [async_result.id],
"source": descriptor.source,
"task_name": descriptor.task_name,
},
diff --git a/src/organizations/migrations/0010_gosedo_media_source_groups.py b/src/organizations/migrations/0010_gosedo_media_source_groups.py
new file mode 100644
index 0000000..b6cbbcc
--- /dev/null
+++ b/src/organizations/migrations/0010_gosedo_media_source_groups.py
@@ -0,0 +1,87 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+ dependencies = [("organizations", "0009_source_record_export_year_indexes")]
+
+ operations = [
+ migrations.CreateModel(
+ name="ElectronicDocumentExchangeExtension",
+ fields=[
+ (
+ "organizationsourceextension_ptr",
+ models.OneToOneField(
+ auto_created=True,
+ on_delete=django.db.models.deletion.CASCADE,
+ parent_link=True,
+ primary_key=True,
+ serialize=False,
+ to="organizations.organizationsourceextension",
+ ),
+ )
+ ],
+ options={
+ "verbose_name": "электронный документооборот",
+ "verbose_name_plural": "электронный документооборот",
+ "db_table": "organizations_electronic_document_exchange_extension",
+ },
+ bases=("organizations.organizationsourceextension",),
+ ),
+ migrations.CreateModel(
+ name="MediaMentionExtension",
+ fields=[
+ (
+ "organizationsourceextension_ptr",
+ models.OneToOneField(
+ auto_created=True,
+ on_delete=django.db.models.deletion.CASCADE,
+ parent_link=True,
+ primary_key=True,
+ serialize=False,
+ to="organizations.organizationsourceextension",
+ ),
+ )
+ ],
+ options={
+ "verbose_name": "упоминания в СМИ",
+ "verbose_name_plural": "упоминания в СМИ",
+ "db_table": "organizations_media_mention_extension",
+ },
+ bases=("organizations.organizationsourceextension",),
+ ),
+ migrations.RemoveConstraint(
+ model_name="organizationsourcerecord",
+ name="unique_source_record_external_id",
+ ),
+ migrations.AlterField(
+ model_name="organizationsourceextension",
+ name="source_group",
+ field=models.CharField(
+ choices=[
+ ("financial_indicators", "Финансово-экономические показатели"),
+ ("government_procurements", "Государственные закупки"),
+ ("industrial_production", "Производители и продукция России"),
+ ("planned_inspections", "Плановые проверки"),
+ ("bankruptcy", "Сведения о процедурах банкротства"),
+ ("defense_suppliers", "Недобросовестные поставщики ГОЗ"),
+ ("arbitration", "Арбитражные дела"),
+ ("security_registries", "Реестры по информационной безопасности"),
+ ("vacancies", "Вакансии"),
+ ("electronic_document_exchange", "Электронный документооборот"),
+ ("media_mentions", "Упоминания в СМИ"),
+ ],
+ db_index=True,
+ max_length=64,
+ verbose_name="группа источников",
+ ),
+ ),
+ migrations.AddConstraint(
+ model_name="organizationsourcerecord",
+ constraint=models.UniqueConstraint(
+ condition=~models.Q(external_id=""),
+ fields=("source", "record_type", "external_id"),
+ name="unique_source_record_type_external_id",
+ ),
+ ),
+ ]
diff --git a/src/organizations/models.py b/src/organizations/models.py
index acfa798..9cb8ea9 100644
--- a/src/organizations/models.py
+++ b/src/organizations/models.py
@@ -32,6 +32,11 @@ class SourceGroup(models.TextChoices):
_("Реестры по информационной безопасности"),
)
VACANCIES = "vacancies", _("Вакансии")
+ ELECTRONIC_DOCUMENT_EXCHANGE = (
+ "electronic_document_exchange",
+ _("Электронный документооборот"),
+ )
+ MEDIA_MENTIONS = "media_mentions", _("Упоминания в СМИ")
class SourceExtensionStatus(models.TextChoices):
@@ -664,6 +669,28 @@ class VacancyExtension(OrganizationSourceExtension):
verbose_name_plural = _("вакансии")
+class ElectronicDocumentExchangeExtension(OrganizationSourceExtension):
+ """Gosedo directory records linked to a canonical organization."""
+
+ source_group_value = SourceGroup.ELECTRONIC_DOCUMENT_EXCHANGE
+
+ class Meta:
+ db_table = "organizations_electronic_document_exchange_extension"
+ verbose_name = _("электронный документооборот")
+ verbose_name_plural = _("электронный документооборот")
+
+
+class MediaMentionExtension(OrganizationSourceExtension):
+ """Media mentions linked to a canonical organization."""
+
+ source_group_value = SourceGroup.MEDIA_MENTIONS
+
+ class Meta:
+ db_table = "organizations_media_mention_extension"
+ verbose_name = _("упоминания в СМИ")
+ verbose_name_plural = _("упоминания в СМИ")
+
+
class OrganizationSourceRecord(models.Model):
"""Subordinate source record stored under a source extension."""
@@ -755,9 +782,9 @@ class OrganizationSourceRecord(models.Model):
ordering = ["-created_at"]
constraints = [
models.UniqueConstraint(
- fields=["source", "external_id"],
+ fields=["source", "record_type", "external_id"],
condition=~Q(external_id=""),
- name="unique_source_record_external_id",
+ name="unique_source_record_type_external_id",
),
models.UniqueConstraint(
fields=["legacy_model", "legacy_pk"],
diff --git a/src/organizations/serializers.py b/src/organizations/serializers.py
index 8c30527..76dd1cb 100644
--- a/src/organizations/serializers.py
+++ b/src/organizations/serializers.py
@@ -78,6 +78,134 @@ class OrganizationSourceRecordOrganizationSerializer(serializers.Serializer):
ogrip = serializers.CharField(read_only=True, allow_blank=True)
+class OrganizationSourceRecordPayloadSerializer(serializers.Serializer):
+ """Typed optional fields exposed by source-record list payloads."""
+
+ artifact_id = serializers.UUIDField(read_only=True, allow_null=True)
+ inn = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+ kpp = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+ ogrn = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+ okpo = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+ source_version = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ source_published_at = serializers.DateField(read_only=True, allow_null=True)
+ operator_uid = serializers.CharField(
+ read_only=True, allow_blank=True, allow_null=True
+ )
+ medo_address = serializers.CharField(
+ read_only=True, allow_blank=True, allow_null=True
+ )
+ short_name = serializers.CharField(
+ read_only=True, allow_blank=True, allow_null=True
+ )
+ full_name = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+ responsible_person = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ responsible_phone = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ responsible_email = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ responsible_phones = serializers.ListField(
+ child=serializers.CharField(),
+ read_only=True,
+ )
+ responsible_emails = serializers.ListField(
+ child=serializers.CharField(),
+ read_only=True,
+ )
+ responsible_contacts_raw = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ participant_status = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ participant_status_raw = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ attestation_status = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ attestation_status_raw = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ registration_type = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ registration_number = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ legal_address_raw = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ organization_contacts = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ source_row_class = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ extra_fields = serializers.DictField(read_only=True)
+ published_at = serializers.DateField(read_only=True, allow_null=True)
+ news_source = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ sentiment = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+ excerpt_lines = serializers.ListField(
+ child=serializers.CharField(),
+ read_only=True,
+ )
+ url = serializers.CharField(read_only=True, allow_blank=True, allow_null=True)
+
+
+class OrganizationSourceRecordDetailPayloadSerializer(
+ OrganizationSourceRecordPayloadSerializer
+):
+ """Detail payload including the complete media article text."""
+
+ full_text = serializers.CharField(
+ read_only=True,
+ allow_blank=True,
+ allow_null=True,
+ )
+
+
class OrganizationSourceRecordSerializer(serializers.ModelSerializer):
"""Source record stored under one source extension."""
@@ -124,10 +252,17 @@ class OrganizationSourceRecordSerializer(serializers.ModelSerializer):
return getattr(obj, "canonical_record_date", obj.record_date) or None
@swagger_serializer_method(
- serializer_or_field=serializers.JSONField(read_only=True),
+ serializer_or_field=OrganizationSourceRecordDetailPayloadSerializer,
)
def get_payload(self, obj) -> dict | list | str | int | float | bool | None:
payload = obj.payload
+ if obj.extension.source_group == SourceGroup.MEDIA_MENTIONS:
+ response_payload = dict(payload) if isinstance(payload, dict) else {}
+ request = self.context.get("request")
+ view = self.context.get("view")
+ if request is not None and getattr(view, "action", None) == "list":
+ response_payload.pop("full_text", None)
+ return response_payload
if obj.extension.source_group != SourceGroup.ARBITRATION:
return payload
@@ -164,11 +299,26 @@ class OrganizationSourceRecordSerializer(serializers.ModelSerializer):
}
+class OrganizationSourceRecordListSerializer(OrganizationSourceRecordSerializer):
+ """List record that never exposes the complete media article text."""
+
+ @swagger_serializer_method(
+ serializer_or_field=OrganizationSourceRecordPayloadSerializer,
+ )
+ def get_payload(self, obj) -> dict | list | str | int | float | bool | None:
+ payload = super().get_payload(obj)
+ if obj.extension.source_group != SourceGroup.MEDIA_MENTIONS:
+ return payload
+ response_payload = dict(payload) if isinstance(payload, dict) else {}
+ response_payload.pop("full_text", None)
+ return response_payload
+
+
class OrganizationSourceRecordListResponseSerializer(serializers.Serializer):
"""Paginated source-record list response in unified API format."""
success = serializers.BooleanField(read_only=True)
- data = OrganizationSourceRecordSerializer(many=True, read_only=True)
+ data = OrganizationSourceRecordListSerializer(many=True, read_only=True)
errors = serializers.JSONField(read_only=True, allow_null=True)
meta = serializers.JSONField(read_only=True, allow_null=True)
diff --git a/src/organizations/source_groups.py b/src/organizations/source_groups.py
index e44e664..27b6371 100644
--- a/src/organizations/source_groups.py
+++ b/src/organizations/source_groups.py
@@ -10,9 +10,11 @@ from organizations.models import (
ArbitrationExtension,
BankruptcyExtension,
DefenseSupplierExtension,
+ ElectronicDocumentExchangeExtension,
FinancialIndicatorsExtension,
GovernmentProcurementExtension,
IndustrialProductionExtension,
+ MediaMentionExtension,
OrganizationSourceExtension,
PlannedInspectionExtension,
SecurityRegistryExtension,
@@ -138,6 +140,20 @@ SOURCE_GROUP_DESCRIPTORS: dict[str, SourceGroupDescriptor] = {
title="Вакансии Работа России",
extension_model=VacancyExtension,
),
+ ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY: SourceGroupDescriptor(
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ source_group=SourceGroup.ELECTRONIC_DOCUMENT_EXCHANGE,
+ record_type="participant",
+ title="Электронный документооборот",
+ extension_model=ElectronicDocumentExchangeExtension,
+ ),
+ ParserLoadLog.Source.MEDIA_NEWS: SourceGroupDescriptor(
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ source_group=SourceGroup.MEDIA_MENTIONS,
+ record_type="media_mention",
+ title="Упоминания в СМИ",
+ extension_model=MediaMentionExtension,
+ ),
"hh": SourceGroupDescriptor(
source="hh",
source_group=SourceGroup.VACANCIES,
diff --git a/src/organizations/source_ingestion.py b/src/organizations/source_ingestion.py
index 4a88165..d5c1375 100644
--- a/src/organizations/source_ingestion.py
+++ b/src/organizations/source_ingestion.py
@@ -7,6 +7,7 @@ from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any
+from uuid import UUID
from django.db import transaction
from django.db.models import Count, Max, Min, Q
@@ -48,6 +49,8 @@ class SourceRecordInput:
external_id: str
title: str
organization_name: str
+ record_type: str = ""
+ uid: UUID | None = None
inn: str = ""
kpp: str = ""
ogrn: str = ""
@@ -111,14 +114,14 @@ class OrganizationSourceIngestionService:
def _deduplicate_records(
records: Iterable[SourceRecordInput],
) -> list[SourceRecordInput]:
- by_external_id: dict[str, SourceRecordInput] = {}
+ by_external_id: dict[tuple[str, str], SourceRecordInput] = {}
without_external_id = []
for record in records:
external_id = str(record.external_id or "")
if not external_id:
without_external_id.append(record)
continue
- by_external_id[external_id] = record
+ by_external_id[(str(record.record_type or ""), external_id)] = record
return [*by_external_id.values(), *without_external_id]
@classmethod
@@ -793,17 +796,25 @@ class OrganizationSourceIngestionService:
return {}, 0, 0
source = str(descriptor.source)
- external_ids = [
- str(record_input.external_id or "")
+ identity_keys = {
+ (
+ str(record_input.record_type or descriptor.record_type),
+ str(record_input.external_id or ""),
+ )
for _, record_input, _ in record_inputs_with_extensions
if str(record_input.external_id or "")
- ]
- existing_by_external_id: dict[str, OrganizationSourceRecord] = {}
+ }
+ external_ids = {external_id for _, external_id in identity_keys}
+ record_types = {record_type for record_type, _ in identity_keys}
+ existing_by_external_id: dict[tuple[str, str], OrganizationSourceRecord] = {}
for source_record in OrganizationSourceRecord.objects.filter(
source=source,
- external_id__in=sorted(set(external_ids)),
+ record_type__in=record_types,
+ external_id__in=external_ids,
):
- existing_by_external_id[source_record.external_id] = source_record
+ existing_by_external_id[
+ (source_record.record_type, source_record.external_id)
+ ] = source_record
now = timezone.now()
create_instances: list[OrganizationSourceRecord] = []
@@ -826,9 +837,10 @@ class OrganizationSourceIngestionService:
for index, record_input, extension in record_inputs_with_extensions:
external_id = str(record_input.external_id or "")
+ record_type = str(record_input.record_type or descriptor.record_type)
defaults = {
"extension": extension,
- "record_type": descriptor.record_type,
+ "record_type": record_type,
"title": str(record_input.title or ""),
"record_date": str(record_input.record_date or ""),
"amount": record_input.amount,
@@ -841,9 +853,10 @@ class OrganizationSourceIngestionService:
"updated_at": now,
}
- source_record = existing_by_external_id.get(external_id)
- if source_record is None:
+ existing_record = existing_by_external_id.get((record_type, external_id))
+ if existing_record is None:
source_record = OrganizationSourceRecord(
+ **({"uid": record_input.uid} if record_input.uid else {}),
source=source,
external_id=external_id,
created_at=now,
@@ -852,8 +865,9 @@ class OrganizationSourceIngestionService:
create_instances.append(source_record)
else:
for field_name, value in defaults.items():
- setattr(source_record, field_name, value)
- update_instances.append(source_record)
+ setattr(existing_record, field_name, value)
+ update_instances.append(existing_record)
+ source_record = existing_record
source_records_by_index[index] = source_record
@@ -1014,9 +1028,10 @@ class OrganizationSourceIngestionService:
record_input: SourceRecordInput,
load_batch: int | None,
) -> tuple[OrganizationSourceRecord, bool]:
+ record_type = str(record_input.record_type or descriptor.record_type)
defaults = {
"extension": extension,
- "record_type": descriptor.record_type,
+ "record_type": record_type,
"title": str(record_input.title or ""),
"record_date": str(record_input.record_date or ""),
"amount": record_input.amount,
@@ -1029,13 +1044,23 @@ class OrganizationSourceIngestionService:
}
external_id = str(record_input.external_id or "")
if external_id:
- return OrganizationSourceRecord.objects.update_or_create(
+ create_defaults = dict(defaults)
+ if record_input.uid:
+ create_defaults["uid"] = record_input.uid
+ source_record, created = OrganizationSourceRecord.objects.get_or_create(
source=str(descriptor.source),
+ record_type=record_type,
external_id=external_id,
- defaults=defaults,
+ defaults=create_defaults,
)
+ if not created:
+ for field_name, value in defaults.items():
+ setattr(source_record, field_name, value)
+ source_record.save(update_fields=[*defaults, "updated_at"])
+ return source_record, created
return (
OrganizationSourceRecord.objects.create(
+ **({"uid": record_input.uid} if record_input.uid else {}),
source=str(descriptor.source),
external_id="",
**defaults,
diff --git a/src/organizations/source_record_export.py b/src/organizations/source_record_export.py
index 100491a..8c23c4a 100644
--- a/src/organizations/source_record_export.py
+++ b/src/organizations/source_record_export.py
@@ -38,6 +38,7 @@ EXPORT_FORMAT_XLSX = "xlsx"
EXPORT_FORMAT_JSON = "json"
EXPORT_FORMATS = (EXPORT_FORMAT_CSV, EXPORT_FORMAT_XLSX, EXPORT_FORMAT_JSON)
FINANCIAL_SOURCE_GROUP = SourceGroup.FINANCIAL_INDICATORS.value
+ALL_HISTORY_SOURCE_GROUPS = {SourceGroup.MEDIA_MENTIONS.value}
EXPORT_MANIFEST_VERSION = 2
CURRENT_EXPORT_MANIFEST_FILE_NAME = "current.json"
GENERATION_MANIFEST_FILE_NAME = "manifest.json"
@@ -69,6 +70,8 @@ SOURCE_GROUP_EXPORT_FILE_STEMS: dict[str, str] = {
SourceGroup.ARBITRATION.value: "arbitration-cases",
SourceGroup.SECURITY_REGISTRIES.value: "information-security-registries",
SourceGroup.VACANCIES.value: "labor-vacancies",
+ SourceGroup.ELECTRONIC_DOCUMENT_EXCHANGE.value: "gosedo-address-directory",
+ SourceGroup.MEDIA_MENTIONS.value: "media-mentions",
}
ORGANIZATION_EXPORT_FIELDS = ["Наименование", "ИНН", "ОГРН", "КПП", "ОКПО"]
@@ -555,6 +558,8 @@ def _source_group_queryset(
)
if source_group == FINANCIAL_SOURCE_GROUP:
return queryset.filter(financial_lines__year=export_year).distinct()
+ if source_group in ALL_HISTORY_SOURCE_GROUPS:
+ return queryset
if connection.vendor == "postgresql":
queryset = queryset.annotate(
diff --git a/src/organizations/test_companies.py b/src/organizations/test_companies.py
index b6bfdc2..86989d3 100644
--- a/src/organizations/test_companies.py
+++ b/src/organizations/test_companies.py
@@ -36,6 +36,10 @@ TEST_COMPANY_COUNT = 20
TEST_FINANCIAL_HISTORY_YEARS = 4
TEST_COMPANY_NAMESPACE = UUID("59b36ae9-bcf8-4b7c-b77a-77578f485a01")
TEST_RECORD_PREFIX = "mostovik-test-company"
+CANONICAL_ONLY_TEST_SOURCES = {
+ ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ ParserLoadLog.Source.MEDIA_NEWS,
+}
TEST_BALANCE_LINE_NAMES = {
"1110": "Нематериальные активы",
@@ -186,20 +190,28 @@ 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)
+ if source in CANONICAL_ONLY_TEST_SOURCES:
+ GenericParserRecord.objects.filter(
+ source=source,
+ external_id=external_id,
+ ).delete()
+ record.legacy_model = ""
+ record.legacy_pk = ""
+ else:
+ 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)
legacy_owner = (
OrganizationSourceRecord.objects.filter(
legacy_model=record.legacy_model,
@@ -209,7 +221,7 @@ class TestCompanyDatasetService:
.select_related("extension")
.first()
)
- if legacy_owner is not None:
+ if record.legacy_model and legacy_owner is not None:
if (
legacy_owner.source != source
or legacy_owner.extension.organization_id != organization.pk
@@ -729,6 +741,49 @@ class TestCompanyDatasetService:
"vac_url": url,
},
},
+ ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY: {
+ "title": f"Участник ГосЭДО — {organization.name}",
+ "record_date": "10.08.2026",
+ "status": "active",
+ "url": "https://gosedo.ru/wp-content/uploads/files/addresseesActual.html",
+ "payload": {
+ **common,
+ "full_name": organization.full_name,
+ "short_name": organization.short_name,
+ "record_type": "participant",
+ "participant_status": "active",
+ "registration_number": f"ТЕСТ-МЭДО-{index:05d}",
+ "source_version": "test-v1",
+ "source_published_at": "2026-08-10",
+ },
+ },
+ ParserLoadLog.Source.MEDIA_NEWS: {
+ "title": f"Тестовая новость о компании {index}",
+ "record_date": "01.07.2026",
+ "status": "positive" if index % 2 else "negative",
+ "url": url,
+ "payload": {
+ **common,
+ "okpo": organization.okpo,
+ "published_at": "2026-07-01",
+ "news_source": "Тестовое СМИ",
+ "url": url,
+ "sentiment": "positive" if index % 2 else "negative",
+ "excerpt_lines": [
+ "Тестовая рубрика",
+ f"Тестовая новость о компании {index}",
+ "Первая строка тестового сообщения.",
+ "Вторая строка тестового сообщения.",
+ ],
+ "full_text": (
+ "Тестовая рубрика\n"
+ f"Тестовая новость о компании {index}\n"
+ "Первая строка тестового сообщения.\n"
+ "Вторая строка тестового сообщения.\n"
+ "Полный текст синтетической новости."
+ ),
+ },
+ },
}
return values[source]
diff --git a/src/organizations/views.py b/src/organizations/views.py
index 0cce862..ecb0085 100644
--- a/src/organizations/views.py
+++ b/src/organizations/views.py
@@ -54,6 +54,7 @@ from organizations.serializers import (
OrganizationSourceRecordExportDownloadSerializer,
OrganizationSourceRecordExportRequestSerializer,
OrganizationSourceRecordListResponseSerializer,
+ OrganizationSourceRecordListSerializer,
OrganizationSourceRecordSerializer,
)
from organizations.source_record_export import (
@@ -100,6 +101,8 @@ SOURCE_RECORD_ORDERING_FIELDS = (
"uid",
"extension__organization__inn",
"extension__organization__ogrn",
+ "extension__organization__okpo",
+ "status",
)
SOURCE_RECORD_ORDERING_VALUES = [
value
@@ -192,6 +195,7 @@ SOURCE_RECORD_LIST_PARAMS = [
),
_query_parameter("source", description="Фильтр по legacy source внутри группы."),
_query_parameter("record_type", description="Фильтр по типу записи."),
+ _query_parameter("status", description="Точный фильтр по статусу записи."),
_query_parameter(
"organization",
description="UID организации.",
@@ -496,10 +500,10 @@ class OrganizationSourceExtensionViewSet(ReadOnlyModelViewSet):
)
page = self.paginate_queryset(queryset)
if page is not None:
- serializer = OrganizationSourceRecordSerializer(page, many=True)
+ serializer = OrganizationSourceRecordListSerializer(page, many=True)
return self.get_paginated_response(serializer.data)
- serializer = OrganizationSourceRecordSerializer(queryset, many=True)
+ serializer = OrganizationSourceRecordListSerializer(queryset, many=True)
return Response(serializer.data)
@@ -538,11 +542,17 @@ class OrganizationSourceRecordViewSet(ReadOnlyModelViewSet):
"extension__organization__inn",
"extension__organization__kpp",
"extension__organization__ogrn",
+ "extension__organization__okpo",
"extension__organization__ogrip",
]
ordering_fields = SOURCE_RECORD_ORDERING_FIELDS
ordering = ["-created_at", "-uid"]
+ def get_serializer_class(self):
+ if self.action == "list":
+ return OrganizationSourceRecordListSerializer
+ return OrganizationSourceRecordSerializer
+
def get_permissions(self):
if self.action in {"export", "export_ticket"}:
return [IsAdminUser()]
@@ -594,6 +604,7 @@ class OrganizationSourceRecordViewSet(ReadOnlyModelViewSet):
source_group = params.get("source_group")
source = params.get("source")
record_type = params.get("record_type")
+ record_status = params.get("status")
organization = params.get("organization")
search_terms = SearchFilter().get_search_terms(self.request)
@@ -603,6 +614,8 @@ class OrganizationSourceRecordViewSet(ReadOnlyModelViewSet):
queryset = queryset.filter(source=source)
if record_type:
queryset = queryset.filter(record_type=record_type)
+ if record_status:
+ queryset = queryset.filter(status=record_status)
if organization:
queryset = queryset.filter(extension__organization_id=organization)
if search_terms:
diff --git a/tests/apps/exchange/test_state_corp_services.py b/tests/apps/exchange/test_state_corp_services.py
index 30483a7..7912ade 100644
--- a/tests/apps/exchange/test_state_corp_services.py
+++ b/tests/apps/exchange/test_state_corp_services.py
@@ -22,7 +22,12 @@ from apps.parsers.models import (
)
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from django.test import TestCase, override_settings
-from organizations.models import Organization
+from organizations.models import (
+ ElectronicDocumentExchangeExtension,
+ MediaMentionExtension,
+ Organization,
+ OrganizationSourceRecord,
+)
from organizations.test_companies import TestCompanyDatasetService
from tests.apps.parsers.factories import (
@@ -319,6 +324,54 @@ class StateCorpExchangeServiceTest(TestCase):
url="https://trudvsem.ru/vacancy/001",
payload={"vacancy_source": "trudvsem"},
)
+ gosedo_extension = ElectronicDocumentExchangeExtension.objects.create(
+ organization=organization,
+ title="Электронный документооборот",
+ )
+ OrganizationSourceRecord.objects.create(
+ extension=gosedo_extension,
+ record_type="participant",
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ external_id="gosedo-participant-001",
+ title=organization.name,
+ record_date="2026-08-10",
+ status="active",
+ url="https://gosedo.ru/wp-content/uploads/files/addresseesActual.html",
+ load_batch=10,
+ payload={
+ "source_version": "633",
+ "source_published_at": "2026-08-10",
+ "short_name": organization.short_name,
+ "full_name": organization.full_name,
+ "attestation_status": "attested",
+ "registration_type": "ogrn",
+ "registration_number": organization.ogrn,
+ "operator_uid": "operator-001",
+ },
+ )
+ media_extension = MediaMentionExtension.objects.create(
+ organization=organization,
+ title="Упоминания в СМИ",
+ )
+ OrganizationSourceRecord.objects.create(
+ extension=media_extension,
+ record_type="media_mention",
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ external_id="a" * 64,
+ title="Новый производственный комплекс",
+ record_date="2026-07-15",
+ status="positive",
+ url="https://example.test/news/1",
+ load_batch=11,
+ payload={
+ "news_source": "Тестовое СМИ",
+ "published_at": "2026-07-15",
+ "sentiment": "positive",
+ "excerpt_lines": ["Источник", "Заголовок", "Лид", "Деталь"],
+ "full_text": "Источник\nЗаголовок\nЛид\nДеталь\nПолный текст",
+ "okpo": organization.okpo,
+ },
+ )
package = StateCorpExchangeService.build_package(actual_date="2026-03-15")
self.assertEqual(package.payload_counts["organizations"], 1)
@@ -334,6 +387,8 @@ class StateCorpExchangeServiceTest(TestCase):
self.assertEqual(package.payload_counts["defense_unreliable_suppliers"], 1)
self.assertEqual(package.payload_counts["information_security_registries"], 1)
self.assertEqual(package.payload_counts["labor_vacancies"], 1)
+ self.assertEqual(package.payload_counts["electronic_document_exchange"], 1)
+ self.assertEqual(package.payload_counts["media_mentions"], 1)
payload = _decode_package_payload(package)
self.assertNotIn(
@@ -342,8 +397,23 @@ class StateCorpExchangeServiceTest(TestCase):
)
self.assertEqual(payload["format"], StateCorpExchangeService.PAYLOAD_FORMAT)
- self.assertEqual(payload["schema_version"], 3)
- self.assertEqual(payload["manifest"]["schema_version"], 3)
+ self.assertEqual(payload["schema_version"], 4)
+ self.assertEqual(payload["manifest"]["schema_version"], 4)
+ self.assertEqual(payload["manifest"]["sections"], list(payload["data"]))
+ self.assertIn("electronic_document_exchange", payload["data"])
+ self.assertIn("media_mentions", payload["data"])
+ self.assertEqual(
+ payload["data"]["electronic_document_exchange"][0]["source_version"],
+ "633",
+ )
+ self.assertEqual(
+ payload["data"]["media_mentions"][0]["excerpt_lines"],
+ ["Источник", "Заголовок", "Лид", "Деталь"],
+ )
+ self.assertEqual(
+ payload["data"]["media_mentions"][0]["full_text"],
+ "Источник\nЗаголовок\nЛид\nДеталь\nПолный текст",
+ )
self.assertEqual(payload["manifest"]["source_system"], "mostovik")
self.assertNotIn("registry_memberships", payload["data"])
self.assertNotIn("registry_memberships", payload["manifest"]["sections"])
diff --git a/tests/apps/organizations/test_source_record_export.py b/tests/apps/organizations/test_source_record_export.py
index 71b8c54..53d6a25 100644
--- a/tests/apps/organizations/test_source_record_export.py
+++ b/tests/apps/organizations/test_source_record_export.py
@@ -17,6 +17,7 @@ from django.utils import timezone
from openpyxl import load_workbook
from organizations.models import (
FinancialIndicatorsExtension,
+ MediaMentionExtension,
Organization,
OrganizationSourceFinancialLine,
OrganizationSourceRecord,
@@ -91,9 +92,9 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
call_command("build_source_record_exports", stdout=command_output)
generation = load_current_source_record_export_generation()
- self.assertEqual(generation.artifacts_count, 25)
+ self.assertEqual(generation.artifacts_count, 31)
self.assertEqual(generation.export_year, timezone.localdate().year)
- self.assertIn('"artifacts_count": 25', command_output.getvalue())
+ self.assertIn('"artifacts_count": 31', command_output.getvalue())
def test_generation_contains_only_records_from_its_calendar_year(self):
export_year = 2026
@@ -179,7 +180,6 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
{row["uid"] for row in inspection_rows},
{str(current_record.pk), str(dateless_current_record.pk)},
)
-
financial_path = next(
artifact.path
for artifact in generation.artifacts
@@ -195,6 +195,46 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
{export_year},
)
+ def test_media_mentions_export_keeps_all_history_and_full_text(self):
+ generated_at = datetime(2026, 8, 4, 6, 0, tzinfo=UTC)
+ organization = Organization.objects.create(
+ name="АО История СМИ",
+ inn="7707083802",
+ )
+ extension = MediaMentionExtension.objects.create(
+ organization=organization,
+ title="Новости СМИ",
+ )
+ old_record = OrganizationSourceRecord.objects.create(
+ extension=extension,
+ record_type="media_mention",
+ source="media_news",
+ external_id="old-media-mention",
+ title="Старая публикация",
+ record_date="2023-02-15",
+ payload={
+ "news_source": "Архивное СМИ",
+ "excerpt_lines": ["Строка 1", "Строка 2"],
+ "full_text": "Строка 1\nСтрока 2\nПолная история",
+ },
+ )
+
+ generation = build_source_record_export_artifacts(now=generated_at)
+
+ media_path = next(
+ artifact.path
+ for artifact in generation.artifacts
+ if artifact.source_group == SourceGroup.MEDIA_MENTIONS.value
+ and artifact.file_format == "json"
+ )
+ rows = json.loads(media_path.read_text(encoding="utf-8"))
+ exported = next(row for row in rows if row["uid"] == str(old_record.pk))
+ self.assertEqual(exported["record_date"], "2023-02-15")
+ self.assertEqual(
+ exported["payload.full_text"],
+ "Строка 1\nСтрока 2\nПолная история",
+ )
+
def test_new_calendar_year_requires_a_new_prepared_generation(self):
generated_at = datetime(2026, 12, 31, 23, 59, tzinfo=UTC)
build_source_record_export_artifacts(now=generated_at)
@@ -273,7 +313,7 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
self.assertEqual(
response["X-Source-Export-Generated-At"], generation.generated_at
)
- self.assertEqual(generation.artifacts_count, 25)
+ self.assertEqual(generation.artifacts_count, 31)
self.assertIn(
'filename="planned-inspections__financial-indicators_',
response["Content-Disposition"],
@@ -468,8 +508,8 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
key=lambda item: item.file_name,
)
- self.assertEqual(generation.artifacts_count, 25)
- self.assertEqual(generation.files_count, 26)
+ self.assertEqual(generation.artifacts_count, 31)
+ self.assertEqual(generation.files_count, 32)
self.assertEqual(
[item.file_name for item in artifacts],
[
@@ -667,7 +707,7 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
first_generation = build_source_record_export_artifacts()
current_generation = load_current_source_record_export_generation()
- self.assertEqual(first_generation.artifacts_count, 25)
+ self.assertEqual(first_generation.artifacts_count, 31)
self.assertEqual(
current_generation.generation_id, first_generation.generation_id
)
diff --git a/tests/apps/organizations/test_tasks.py b/tests/apps/organizations/test_tasks.py
index a156a67..23091c7 100644
--- a/tests/apps/organizations/test_tasks.py
+++ b/tests/apps/organizations/test_tasks.py
@@ -145,7 +145,7 @@ class SourceRecordExportArtifactsTaskTest(TestCase):
result = refresh_source_record_export_artifacts()
self.assertEqual(result["status"], "success")
- self.assertEqual(result["artifacts_count"], 25)
+ self.assertEqual(result["artifacts_count"], 31)
self.assertEqual(result["export_year"], timezone.localdate().year)
self.assertIsNone(cache.get(settings.SOURCE_RECORD_EXPORT_LOCK_KEY))
diff --git a/tests/apps/organizations/test_test_companies_commands.py b/tests/apps/organizations/test_test_companies_commands.py
index fd374ec..369e367 100644
--- a/tests/apps/organizations/test_test_companies_commands.py
+++ b/tests/apps/organizations/test_test_companies_commands.py
@@ -214,7 +214,7 @@ class TestCompaniesCommandsTest(TestCase):
year=stale_year,
).exists()
)
- self.assertEqual(OrganizationSourceRecord.objects.count(), 20 * 15)
+ self.assertEqual(OrganizationSourceRecord.objects.count(), 20 * 17)
def test_create_replaces_stale_source_record_for_same_legacy_row(self):
call_command("create_test_companies", stdout=StringIO())
@@ -276,6 +276,8 @@ class TestCompaniesCommandsTest(TestCase):
"defense_unreliable_suppliers": 40,
"information_security_registries": 20,
"labor_vacancies": 20,
+ "electronic_document_exchange": 20,
+ "media_mentions": 20,
},
)
self.assertEqual(len(reports), 20)
diff --git a/tests/apps/parsers/test_gosedo_media_news.py b/tests/apps/parsers/test_gosedo_media_news.py
new file mode 100644
index 0000000..ea30bf6
--- /dev/null
+++ b/tests/apps/parsers/test_gosedo_media_news.py
@@ -0,0 +1,670 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from io import BytesIO
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from apps.parsers.gosedo import (
+ GosedoNotModified,
+ GosedoParseResult,
+ GosedoRow,
+ GosedoValidationError,
+ download_gosedo,
+ parse_gosedo,
+ publish_gosedo_snapshot,
+ stable_gosedo_uid,
+)
+from apps.parsers.media_news import (
+ import_media_news,
+ normalize_news_text,
+ stable_media_external_id,
+)
+from apps.parsers.models import ParserLoadLog, ParserSourceArtifact, ParserStagedRecord
+from apps.parsers.source_artifacts import cleanup_parser_source_artifacts
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase
+from django.urls import reverse
+from django.utils import timezone
+from openpyxl import Workbook
+from organizations.models import Organization, OrganizationSourceRecord
+from organizations.source_ingestion import (
+ OrganizationSourceIngestionService,
+ SourceRecordInput,
+)
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from tests.apps.user.factories import UserFactory
+
+
+def _gosedo_html() -> bytes:
+ rows = []
+ for row_class, external_id, label in (
+ ("pt-on", "participant-on", "УЧАСТНИК МЭДО"),
+ ("pt-off", "participant-off", "УЧАСТНИК МЭДО"),
+ ("oper", "operator-non-rfc-id", "ОПЕРАТОР МЭДО"),
+ ("org", "organizer-1", "ОРГАНИЗАТОР"),
+ ):
+ rows.append(
+ f''
+ f"| {external_id} | Краткое имя | МЭДО-адрес |
"
+ ''
+ f"- {label}
- Полное имя
"
+ "- Рег.номер
- ОГРН: 1027700132195
"
+ "- Статус участника
- АКТИВНЫЙ, не аттестован для ДСП
"
+ "- Новое поле
- Сохранить
"
+ " |
"
+ )
+ return (
+ "Версия: 633 от 10 августа 2026
"
+ '"
+ ).encode("utf-8")
+
+
+def _media_workbook(rows: list[list[object]]) -> BytesIO:
+ workbook = Workbook()
+ sheet = workbook.active
+ sheet.append(
+ [
+ "ОКПО",
+ "ИНН",
+ "Дата актуальности новости",
+ "Источник",
+ "URL",
+ "Текст",
+ "Оценка",
+ ]
+ )
+ for row in rows:
+ sheet.append(row)
+ output = BytesIO()
+ workbook.save(output)
+ workbook.close()
+ output.seek(0)
+ return output
+
+
+class _DownloadResponse:
+ def __init__(self, status_code=200, *, headers=None, body=b""):
+ self.status_code = status_code
+ self.headers = headers or {"Content-Type": "text/html; charset=utf-8"}
+ self.body = body
+
+ def iter_content(self, chunk_size: int):
+ del chunk_size
+ yield self.body
+
+ def close(self):
+ return None
+
+
+class GosedoParserTest(TestCase):
+ def test_parses_all_record_types_unknown_fields_and_statuses(self):
+ parsed = parse_gosedo(BytesIO(_gosedo_html()))
+
+ self.assertEqual(parsed.version, "633")
+ self.assertEqual(parsed.published_at.isoformat(), "2026-08-10")
+ self.assertEqual(
+ [row.record_type for row in parsed.rows],
+ ["participant", "participant", "operator", "organizer"],
+ )
+ self.assertEqual(parsed.rows[0].status, "active")
+ self.assertEqual(parsed.rows[1].status, "inactive")
+ self.assertEqual(
+ parsed.rows[0].normalized_data["attestation_status"],
+ "not_attested",
+ )
+ self.assertEqual(
+ parsed.rows[0].normalized_data["extra_fields"],
+ {"Новое поле": "Сохранить"},
+ )
+ self.assertEqual(parsed.unknown_fields, 4)
+ self.assertEqual(
+ stable_gosedo_uid("operator", "operator-non-rfc-id"),
+ stable_gosedo_uid("operator", "operator-non-rfc-id"),
+ )
+ self.assertNotEqual(
+ stable_gosedo_uid("operator", "same"),
+ stable_gosedo_uid("participant", "same"),
+ )
+
+ def test_missing_details_row_is_quarantinable(self):
+ html = (
+ 'Версия: 1 от 1 января 2026
'
+ '| broken | Имя | |
'
+ '| next | Имя 2 | |
'
+ '- УЧАСТНИК МЭДО
'
+ "- Имя 2
- Рег.номер
- ОГРН: 1027700132195
"
+ "
|
"
+ ).encode()
+
+ parsed = parse_gosedo(BytesIO(html))
+
+ self.assertEqual(parsed.rows[0].error, "invalid_row_pair")
+ self.assertEqual(parsed.rows[1].external_id, "next")
+
+ def test_conditional_get_uses_latest_headers_and_handles_304(self):
+ ParserSourceArtifact.objects.create(
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ status=ParserSourceArtifact.Status.PUBLISHED,
+ etag='"etag-1"',
+ last_modified="Mon, 10 Aug 2026 10:00:00 GMT",
+ )
+ response = SimpleNamespace(status_code=304, headers={}, close=lambda: None)
+ session = SimpleNamespace()
+ session.get = lambda *args, **kwargs: (
+ setattr(session, "request_headers", kwargs["headers"]) or response
+ )
+
+ with self.assertRaises(GosedoNotModified):
+ download_gosedo(session=session)
+
+ self.assertEqual(session.request_headers["If-None-Match"], '"etag-1"')
+ self.assertIn("If-Modified-Since", session.request_headers)
+
+ def test_download_rejects_redirect_outside_allowlist(self):
+ response = _DownloadResponse(
+ 302,
+ headers={"Location": "https://example.test/source.html"},
+ )
+ session = SimpleNamespace(get=lambda *args, **kwargs: response)
+
+ with self.assertRaisesRegex(GosedoValidationError, "unsafe_source_url"):
+ download_gosedo(session=session)
+
+ def test_download_rejects_mime_utf8_and_declared_size(self):
+ cases = (
+ (
+ _DownloadResponse(headers={"Content-Type": "application/json"}),
+ "invalid_content_type",
+ ),
+ (_DownloadResponse(body=b"\xff"), "invalid_utf8"),
+ (
+ _DownloadResponse(
+ headers={
+ "Content-Type": "text/html",
+ "Content-Length": str(100 * 1024 * 1024 + 1),
+ }
+ ),
+ "source_too_large",
+ ),
+ )
+ for response, expected_error in cases:
+ with self.subTest(expected_error=expected_error):
+ session = SimpleNamespace(
+ get=lambda *args, _response=response, **kwargs: _response
+ )
+ with self.assertRaisesRegex(GosedoValidationError, expected_error):
+ download_gosedo(session=session)
+
+ def test_atomic_failure_keeps_previous_snapshot(self):
+ organization = Organization.objects.create(
+ name="АО Тест",
+ inn="7707083893",
+ ogrn="1027700132195",
+ okpo="12345678",
+ directory_imported_at=timezone.now(),
+ )
+ OrganizationSourceIngestionService.save_records(
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ load_batch=1,
+ records=[
+ SourceRecordInput(
+ external_id="old",
+ record_type="participant",
+ title="Старый снимок",
+ organization_name=organization.name,
+ inn=organization.inn,
+ ogrn=organization.ogrn,
+ payload={"okpo": organization.okpo},
+ )
+ ],
+ )
+ artifact = ParserSourceArtifact.objects.create(
+ source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
+ load_batch=2,
+ )
+ row = GosedoRow(
+ row_number=1,
+ external_id="new",
+ record_type="participant",
+ status="active",
+ normalized_data={
+ "registration_number": organization.ogrn,
+ "short_name": organization.name,
+ },
+ )
+ parsed = GosedoParseResult(
+ version="2",
+ published_at=timezone.localdate(),
+ rows=[row],
+ unknown_fields=0,
+ )
+
+ with patch.object(
+ OrganizationSourceIngestionService,
+ "save_records",
+ side_effect=RuntimeError("publish failed"),
+ ), self.assertRaises(RuntimeError):
+ publish_gosedo_snapshot(artifact=artifact, parsed=parsed, load_batch=2)
+
+ self.assertTrue(
+ OrganizationSourceRecord.objects.filter(external_id="old").exists()
+ )
+ self.assertFalse(
+ OrganizationSourceRecord.objects.filter(external_id="new").exists()
+ )
+
+
+class SourceArtifactRetentionTest(TestCase):
+ def test_cleanup_retains_latest_ten_versions_beyond_ninety_days(self):
+ created = []
+ base_time = timezone.now() - timedelta(days=120)
+ for index in range(12):
+ artifact = ParserSourceArtifact.objects.create(
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ version=str(index),
+ )
+ ParserSourceArtifact.objects.filter(pk=artifact.pk).update(
+ created_at=base_time + timedelta(hours=index)
+ )
+ created.append(artifact.uid)
+
+ deleted = cleanup_parser_source_artifacts()
+
+ self.assertEqual(deleted, 2)
+ remaining = set(ParserSourceArtifact.objects.values_list("uid", flat=True))
+ self.assertEqual(remaining, set(created[-10:]))
+
+
+class MediaNewsImportTest(TestCase):
+ def setUp(self):
+ self.organization = Organization.objects.create(
+ name="АО СМИ",
+ inn="0012345678",
+ ogrn="1027700132195",
+ okpo="00123456",
+ directory_imported_at=timezone.now(),
+ )
+
+ def test_normalizes_text_and_stable_id(self):
+ full_text, excerpt = normalize_news_text(
+ "Источник_x000D_\r\n\r\nЗаголовок\nЛид\nСтрока 4\nСтрока 5"
+ )
+ first = stable_media_external_id(
+ inn="0012345678",
+ okpo="00123456",
+ published_at=timezone.localdate(),
+ news_source="СМИ",
+ url="",
+ full_text=full_text,
+ )
+ second = stable_media_external_id(
+ inn="0012345678",
+ okpo="00123456",
+ published_at=timezone.localdate(),
+ news_source="СМИ",
+ url="",
+ full_text=full_text,
+ )
+
+ self.assertEqual(excerpt, ["Источник", "Заголовок", "Лид", "Строка 4"])
+ self.assertEqual(first, second)
+
+ def test_import_upserts_and_quarantines_formula(self):
+ Organization.objects.create(
+ name="АО Другая",
+ inn="0099999999",
+ ogrn="1027700132196",
+ okpo="00999999",
+ directory_imported_at=timezone.now(),
+ )
+ workbook = _media_workbook(
+ [
+ [
+ "00123456",
+ "0012345678",
+ "2026-07-01",
+ "СМИ",
+ "",
+ "Источник\nЗаголовок\nЛид\nСтрока 4\nПолный текст",
+ "Положительная",
+ ],
+ [
+ "00123456",
+ "=12345678",
+ "2026-07-02",
+ "СМИ",
+ "https://example.test/2",
+ "Источник\nЗаголовок\nЛид\nСтрока 4",
+ "Отрицательная",
+ ],
+ [
+ "00999999",
+ "0012345678",
+ "2026-07-03",
+ "СМИ",
+ "https://example.test/3",
+ "Источник\nЗаголовок\nЛид\nСтрока 4",
+ "Отрицательная",
+ ],
+ ]
+ )
+ _, first = import_media_news(
+ handle=workbook,
+ original_name="media.xlsx",
+ load_batch=1,
+ uploaded_by_id=None,
+ )
+ workbook = _media_workbook(
+ [
+ [
+ "00123456",
+ "0012345678",
+ "2026-07-01",
+ "СМИ",
+ "",
+ "Источник\nЗаголовок\nЛид\nСтрока 4\nПолный текст",
+ "Положительная",
+ ]
+ ]
+ )
+ _, second = import_media_news(
+ handle=workbook,
+ original_name="media.xlsx",
+ load_batch=2,
+ uploaded_by_id=None,
+ )
+
+ self.assertEqual(first.published, 1)
+ self.assertEqual(first.quarantined, 2)
+ self.assertEqual(
+ first.reasons,
+ {"formula_not_allowed": 1, "inn_okpo_conflict": 1},
+ )
+ self.assertEqual(second.published, 1)
+ self.assertEqual(OrganizationSourceRecord.objects.count(), 1)
+ record = OrganizationSourceRecord.objects.get()
+ self.assertEqual(record.title, "Заголовок")
+ self.assertEqual(record.status, "positive")
+ self.assertEqual(
+ record.payload["excerpt_lines"],
+ ["Источник", "Заголовок", "Лид", "Строка 4"],
+ )
+ self.assertEqual(
+ ParserStagedRecord.objects.filter(
+ disposition=ParserStagedRecord.Disposition.QUARANTINED
+ ).count(),
+ 2,
+ )
+
+ def test_import_accepts_native_date_typo_header_and_formatted_leading_zeroes(self):
+ workbook = Workbook()
+ sheet = workbook.active
+ sheet.append(
+ [
+ "ОКПО",
+ "ИНН",
+ "Дата акутальности новости",
+ "Источник",
+ "URL",
+ "Текст",
+ "Оценка",
+ ]
+ )
+ sheet.append(
+ [
+ 123456,
+ 12345678,
+ datetime(2026, 7, 4),
+ "СМИ",
+ "https://example.test/native-date",
+ "Источник\nЗаголовок",
+ "Отрицательная",
+ ]
+ )
+ sheet["A2"].number_format = "00000000"
+ sheet["B2"].number_format = "0000000000"
+ output = BytesIO()
+ workbook.save(output)
+ workbook.close()
+ output.seek(0)
+
+ _, result = import_media_news(
+ handle=output,
+ original_name="native.xlsx",
+ load_batch=3,
+ uploaded_by_id=None,
+ )
+
+ self.assertEqual(result.published, 1)
+ record = OrganizationSourceRecord.objects.get()
+ self.assertEqual(record.record_date, "2026-07-04")
+ self.assertEqual(record.payload["inn"], "0012345678")
+ self.assertEqual(record.payload["okpo"], "00123456")
+ self.assertEqual(record.status, "negative")
+
+ def test_same_article_for_two_organizations_stays_separate(self):
+ Organization.objects.create(
+ name="АО СМИ 2",
+ inn="0098765432",
+ ogrn="1027700132196",
+ okpo="00654321",
+ directory_imported_at=timezone.now(),
+ )
+ common = [
+ "2026-07-05",
+ "СМИ",
+ "https://example.test/shared",
+ "Источник\nЗаголовок\nТекст",
+ "Положительная",
+ ]
+ workbook = _media_workbook(
+ [
+ ["00123456", "0012345678", *common],
+ ["00654321", "0098765432", *common],
+ ]
+ )
+
+ _, result = import_media_news(
+ handle=workbook,
+ original_name="shared.xlsx",
+ load_batch=4,
+ uploaded_by_id=None,
+ )
+
+ self.assertEqual(result.published, 2)
+ self.assertEqual(OrganizationSourceRecord.objects.count(), 2)
+ self.assertEqual(
+ OrganizationSourceRecord.objects.values("external_id").distinct().count(),
+ 2,
+ )
+
+ def test_invalid_date_and_sentiment_are_quarantined(self):
+ workbook = _media_workbook(
+ [
+ [
+ "00123456",
+ "0012345678",
+ "31.02.2026",
+ "СМИ",
+ "",
+ "Источник\nЗаголовок",
+ "Положительная",
+ ],
+ [
+ "00123456",
+ "0012345678",
+ "2026-07-06",
+ "СМИ",
+ "",
+ "Источник\nЗаголовок",
+ "Нейтральная",
+ ],
+ ]
+ )
+
+ _, result = import_media_news(
+ handle=workbook,
+ original_name="invalid.xlsx",
+ load_batch=5,
+ uploaded_by_id=None,
+ )
+
+ self.assertEqual(result.published, 0)
+ self.assertEqual(
+ result.reasons,
+ {"invalid_date": 1, "invalid_sentiment": 1},
+ )
+
+
+class MediaNewsPermissionsTest(APITestCase):
+ def setUp(self):
+ self.user = UserFactory.create_user()
+ self.admin = UserFactory.create_user(is_staff=True)
+ self.url = reverse("api_v1:parsers:upload-parser-data", args=["media_news"])
+
+ def test_upload_is_admin_only_and_returns_task_ids(self):
+ self.client.force_authenticate(self.user)
+ response = self.client.post(
+ self.url,
+ {"file": SimpleUploadedFile("media.xlsx", b"not-read-by-worker")},
+ format="multipart",
+ )
+ self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+ self.client.force_authenticate(self.admin)
+ with patch(
+ "apps.parsers.tasks.parse_media_news.apply_async",
+ return_value=SimpleNamespace(id="media-task-1"),
+ ):
+ response = self.client.post(
+ self.url,
+ {"file": SimpleUploadedFile("media.xlsx", b"queued")},
+ format="multipart",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ self.assertEqual(response.data["data"]["task_id"], "media-task-1")
+ self.assertEqual(response.data["data"]["task_ids"], ["media-task-1"])
+
+ def test_gosedo_manual_run_is_admin_only_and_returns_task_ids(self):
+ url = reverse(
+ "api_v1:parsers:run-parser",
+ args=[ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY],
+ )
+ self.client.force_authenticate(self.user)
+ response = self.client.post(url, {}, format="json")
+ self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+ self.client.force_authenticate(self.admin)
+ with patch(
+ "apps.parsers.tasks.parse_gosedo_address_directory.apply_async",
+ return_value=SimpleNamespace(id="gosedo-task-1"),
+ ):
+ response = self.client.post(url, {}, format="json")
+
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ self.assertEqual(response.data["data"]["task_id"], "gosedo-task-1")
+ self.assertEqual(response.data["data"]["task_ids"], ["gosedo-task-1"])
+
+ def test_parser_log_exposes_artifact_quarantine_accounting(self):
+ organization = Organization.objects.create(name="АО Журнал")
+ log = ParserLoadLog.objects.create(
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ batch_id=44,
+ records_count=1,
+ status=ParserLoadLog.Status.SUCCESS,
+ )
+ artifact = ParserSourceArtifact.objects.create(
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ load_batch=44,
+ status=ParserSourceArtifact.Status.PUBLISHED,
+ parsed_count=2,
+ published_count=1,
+ quarantined_count=1,
+ rejection_reasons={"invalid_date": 1},
+ )
+ ParserStagedRecord.objects.create(
+ artifact=artifact,
+ row_number=1,
+ organization=organization,
+ disposition=ParserStagedRecord.Disposition.PUBLISHED,
+ )
+ ParserStagedRecord.objects.create(
+ artifact=artifact,
+ row_number=2,
+ disposition=ParserStagedRecord.Disposition.QUARANTINED,
+ reason="invalid_date",
+ )
+ self.client.force_authenticate(self.admin)
+
+ response = self.client.get(
+ reverse("api_v1:system:parser-logs-list"),
+ {"source": ParserLoadLog.Source.MEDIA_NEWS},
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ row = next(item for item in response.data["results"] if item["id"] == log.id)
+ self.assertEqual(row["artifact_uid"], str(artifact.uid))
+ self.assertEqual(row["parsed_count"], 2)
+ self.assertEqual(row["published_count"], 1)
+ self.assertEqual(row["quarantined_count"], 1)
+ self.assertEqual(row["rejection_reasons"], {"invalid_date": 1})
+ self.assertEqual(row["organizations_count"], 1)
+
+ def test_media_list_omits_full_text_and_detail_includes_it(self):
+ organization = Organization.objects.create(
+ name="АО API СМИ",
+ inn="7707083801",
+ ogrn="1027700132101",
+ okpo="12345671",
+ opk_registry_membership=True,
+ directory_imported_at=timezone.now(),
+ )
+ OrganizationSourceIngestionService.save_records(
+ source=ParserLoadLog.Source.MEDIA_NEWS,
+ load_batch=1,
+ records=[
+ SourceRecordInput(
+ external_id="b" * 64,
+ record_type="media_mention",
+ title="Заголовок",
+ organization_name=organization.name,
+ inn=organization.inn,
+ ogrn=organization.ogrn,
+ record_date="2026-07-01",
+ status="positive",
+ payload={
+ "okpo": organization.okpo,
+ "news_source": "Тестовое СМИ",
+ "sentiment": "positive",
+ "excerpt_lines": ["1", "2", "3", "4"],
+ "full_text": "1\n2\n3\n4\n5",
+ },
+ )
+ ],
+ )
+ record = OrganizationSourceRecord.objects.get()
+ self.client.force_authenticate(self.user)
+
+ list_response = self.client.get(
+ reverse("api_v2:organizations:organization-source-records-list"),
+ {"source_group": "media_mentions", "status": "positive"},
+ )
+ detail_response = self.client.get(
+ reverse(
+ "api_v2:organizations:organization-source-records-detail",
+ args=[record.uid],
+ )
+ )
+
+ self.assertEqual(list_response.status_code, status.HTTP_200_OK)
+ self.assertNotIn("full_text", list_response.data["data"][0]["payload"])
+ self.assertEqual(
+ list_response.data["data"][0]["payload"]["excerpt_lines"],
+ ["1", "2", "3", "4"],
+ )
+ self.assertEqual(detail_response.status_code, status.HTTP_200_OK)
+ self.assertEqual(detail_response.data["payload"]["full_text"], "1\n2\n3\n4\n5")
diff --git a/tests/apps/parsers/test_source_cards_service.py b/tests/apps/parsers/test_source_cards_service.py
index 1026673..4533b41 100644
--- a/tests/apps/parsers/test_source_cards_service.py
+++ b/tests/apps/parsers/test_source_cards_service.py
@@ -84,6 +84,8 @@ class SourceCardServiceUnitTest(SimpleTestCase):
"arbitration-cases",
"information-security-registries",
"labor-vacancies",
+ "gosedo-global-address-directory",
+ "media-mentions",
],
)
self.assertEqual(
@@ -98,6 +100,8 @@ class SourceCardServiceUnitTest(SimpleTestCase):
"Арбитражные дела",
"Реестры по информационной безопасности",
"Вакансии Работа России",
+ "ГосЭДО: Глобальный адресный справочник",
+ "Новости СМИ",
],
)
diff --git a/tests/apps/parsers/test_source_cards_views.py b/tests/apps/parsers/test_source_cards_views.py
index 0d2b98a..aa7e966 100644
--- a/tests/apps/parsers/test_source_cards_views.py
+++ b/tests/apps/parsers/test_source_cards_views.py
@@ -250,7 +250,11 @@ class SourceCardsApiTestCase(APITestCase):
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(response.data["status"], "accepted")
- self.assertEqual(set(response.data.keys()), {"task_id", "status"})
+ self.assertEqual(
+ set(response.data.keys()),
+ {"task_id", "task_ids", "status"},
+ )
+ self.assertEqual(response.data["task_ids"], [response.data["task_id"]])
task_id = response.data["task_id"]
self.assertTrue(
diff --git a/tests/apps/parsers/test_sources_api_e2e.py b/tests/apps/parsers/test_sources_api_e2e.py
index 5fbeb5e..360ce14 100644
--- a/tests/apps/parsers/test_sources_api_e2e.py
+++ b/tests/apps/parsers/test_sources_api_e2e.py
@@ -191,14 +191,13 @@ class SourcesApiE2ETest(APITestCase):
"task-procurements",
],
), patch(
- "apps.parsers.tasks.parse_industrial_production.apply_async",
- return_value=SimpleNamespace(id="task-industrial"),
- ), patch(
- "apps.parsers.tasks.parse_industrial_products.apply_async",
- return_value=SimpleNamespace(id="task-products"),
- ), patch(
- "apps.parsers.tasks.parse_manufactures.apply_async",
- return_value=SimpleNamespace(id="task-manufactures"),
+ "celery.app.task.Task.apply_async",
+ side_effect=[
+ SimpleNamespace(id="task-industrial"),
+ SimpleNamespace(id="task-products"),
+ SimpleNamespace(id="task-manufactures"),
+ SimpleNamespace(id="task-procurements"),
+ ],
):
minprom_response = self.client.post(
reverse(
@@ -209,18 +208,14 @@ class SourcesApiE2ETest(APITestCase):
format="json",
)
- with patch(
- "apps.parsers.tasks.parse_registry_enrichment_sources.apply_async",
- return_value=SimpleNamespace(id="task-procurements"),
- ):
- procurements_response = self.client.post(
- reverse(
- "api_v1:sources:source-cards-refresh",
- kwargs={"slug": "public-procurements"},
- ),
- {"params": {"region_code": "77", "current_year": "2025"}},
- format="json",
- )
+ procurements_response = self.client.post(
+ reverse(
+ "api_v1:sources:source-cards-refresh",
+ kwargs={"slug": "public-procurements"},
+ ),
+ {"params": {"region_code": "77", "current_year": "2025"}},
+ format="json",
+ )
self.assertEqual(minprom_response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(procurements_response.status_code, status.HTTP_202_ACCEPTED)
@@ -228,13 +223,24 @@ class SourcesApiE2ETest(APITestCase):
self.assertEqual(minprom_response.data["status"], "accepted")
self.assertEqual(procurements_response.data["status"], "accepted")
- self.assertEqual(set(minprom_response.data.keys()), {"task_id", "status"})
+ self.assertEqual(
+ set(minprom_response.data.keys()),
+ {"task_id", "task_ids", "status"},
+ )
self.assertEqual(minprom_response.data["task_id"], "task-industrial")
+ self.assertEqual(
+ minprom_response.data["task_ids"],
+ ["task-industrial", "task-products", "task-manufactures"],
+ )
self.assertEqual(
set(procurements_response.data.keys()),
- {"task_id", "status"},
+ {"task_id", "task_ids", "status"},
)
self.assertEqual(procurements_response.data["task_id"], "task-procurements")
+ self.assertEqual(
+ procurements_response.data["task_ids"],
+ ["task-procurements"],
+ )
self.assertEqual(
BackgroundJob.objects.filter(
task_id__in=["task-industrial", "task-products", "task-manufactures"],