diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 105dd0d..877f7de 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -89,6 +89,7 @@ services: volumes: - ./src:/app/src - ./logs:/app/logs + - ./media:/app/media - ./input:/app/input command: ["celery-worker"] diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index deed6e9..8c1c974 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -53,6 +53,7 @@ services: memswap_limit: 3g volumes: - ./logs:/app/logs + - ./media:/app/media - ./input:/app/input command: ["celery-worker"] diff --git a/docker/Dockerfile b/docker/Dockerfile index 390a53c..cfdbcbf 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -63,7 +63,10 @@ RUN mkdir -p logs media staticfiles input src/static \ && chown -R appuser:appgroup /app ENV PATH="/app/.venv/bin:${PATH}" \ - PYTHONPATH=/app/src + PYTHONPATH=/app/src \ + SOURCE_RECORD_EXPORT_DIRECTORY=/app/media/source-record-exports \ + SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE=100000 \ + SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS=300 USER appuser ENTRYPOINT ["/app/docker/scripts/entrypoint.sh"] diff --git a/docs/source-record-export-matrix-ru.md b/docs/source-record-export-matrix-ru.md new file mode 100644 index 0000000..975d3ad --- /dev/null +++ b/docs/source-record-export-matrix-ru.md @@ -0,0 +1,78 @@ +# Матрица файловых выгрузок внешних данных State Corp + +## Пользовательский контракт + +Администраторский frontend отправляет +`POST /api/v2/organization-source-records/export-ticket/` с массивом `sources` +и форматом. Backend проверяет последнее полностью опубликованное поколение и +возвращает короткоживущий одноразовый ticket. Затем frontend передаёт ticket в +теле обычной HTML-формы на +`POST /api/v2/organization-source-records/export-download/`. + +Браузер получает потоковый ZIP напрямую, без многогигабайтного `Blob` в +JavaScript. Ticket не попадает в URL и после первого запроса становится +недействительным. Совместимый администраторский endpoint +`POST /api/v2/organization-source-records/export/` сразу возвращает тот же ZIP +для API-клиентов. + +Во время скачивания таблицы `external_data` не читаются: endpoint упаковывает +готовые файлы последнего ночного поколения. При отсутствии поколения API +возвращает `503` с кодом `source_export_not_ready`. + +## Матрица + +| Группа API | Таблицы State Corp | Файл | CSV | XLSX | JSON | +|---|---|---|:---:|:---:|:---:| +| `financial_indicators` | `FinancialReport`, `FinancialReportLine` | `financial-indicators` | — | — | да | +| `government_procurements` | `PublicProcurement` | `public-procurements` | да | да | да | +| `industrial_production` | `IndustrialProduct`, `IndustrialCertificate`, `ManufacturerRegistryEntry` | `manufacturers-and-products` | да | да | да | +| `planned_inspections` | `ProsecutorCheck` | `planned-inspections` | да | да | да | +| `bankruptcy` | `BankruptcyProcedure` | `bankruptcy-procedures` | да | да | да | +| `defense_suppliers` | `DefenseUnreliableSupplier` | `defense-unreliable-suppliers` | да | да | да | +| `arbitration` | `ArbitrationCase` | `arbitration-cases` | да | да | да | +| `security_registries` | `InformationSecurityRegistryEntry` | `information-security-registries` | да | да | да | +| `vacancies` | `LaborVacancy` | `labor-vacancies` | да | да | да | + +Итого формируется 25 логических артефактов. Финансовые показатели всегда +выгружаются в JSON с вложенным массивом `financial_lines`. Промышленная группа +объединяет три таблицы, а поле `record_type` различает тип строки. Все строки +содержат реквизиты организации, включая ОКПО. + +Физических XLSX-файлов может быть больше: по умолчанию один файл содержит не +более 100 000 строк данных и получает суффикс `-part-001`, `-part-002` и далее. + +## Ночная генерация + +Celery Beat запускает +`apps.external_data.tasks.refresh_source_record_export_artifacts` ежедневно в +`05:30 Europe/Moscow`. + +Генератор: + +1. читает каждую нормализованную таблицу один раз без model-level сортировки; +2. создаёт компактный JSON-массив и переиспользует его как готовый JSON; +3. потоково формирует CSV и write-only XLSX; +4. атомарно публикует `current.json` только после готовности всей матрицы; +5. при ошибке удаляет staging и продолжает отдавать предыдущее поколение; +6. сохраняет текущее и предыдущее поколения по умолчанию. + +Web и Celery worker должны использовать общий read-write volume `/app/media`. + +## Первый запуск и настройки + +Первое поколение можно сформировать вручную: + +```bash +PYTHONPATH=src uv run python src/manage.py build_source_record_exports +``` + +| Настройка | Значение по умолчанию | Назначение | +|---|---:|---| +| `SOURCE_RECORD_EXPORT_DIRECTORY` | `media/source-record-exports` | Общий каталог поколений | +| `SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP` | `2` | Число успешных поколений | +| `SOURCE_RECORD_EXPORT_LOCK_TTL_SECONDS` | `21600` | TTL распределённой блокировки | +| `SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE` | `100000` | Строк данных в одной XLSX-части | +| `SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS` | `300` | Срок действия download-ticket | + +Для атомарной генерации требуется свободное место не меньше +`(GENERATIONS_TO_KEEP + 1) * размер поколения` плюс запас файловой системы. diff --git a/src/apps/external_data/export_serializers.py b/src/apps/external_data/export_serializers.py new file mode 100644 index 0000000..f2b22d1 --- /dev/null +++ b/src/apps/external_data/export_serializers.py @@ -0,0 +1,37 @@ +"""Request serializers for prepared external-data exports.""" + +from apps.external_data.source_record_export import ( + EXPORT_FORMATS, + SOURCE_GROUP_EXPORT_SPECS, +) +from rest_framework import serializers + + +class SourceRecordExportRequestSerializer(serializers.Serializer): + """Validate selected source groups and the requested file format.""" + + sources = serializers.ListField( + child=serializers.ChoiceField( + choices=[ + (source_group, source_group) + for source_group in SOURCE_GROUP_EXPORT_SPECS + ] + ), + allow_empty=False, + ) + format = serializers.ChoiceField( + choices=[ + (export_format, export_format.upper()) for export_format in EXPORT_FORMATS + ] + ) + + def validate_sources(self, value: list[str]) -> list[str]: + if len(value) != len(set(value)): + raise serializers.ValidationError("Источники не должны повторяться.") + return value + + +class SourceRecordExportDownloadSerializer(serializers.Serializer): + """Validate a one-time ticket submitted by a native browser form.""" + + ticket = serializers.CharField(max_length=64, trim_whitespace=False) diff --git a/src/apps/external_data/export_urls.py b/src/apps/external_data/export_urls.py new file mode 100644 index 0000000..fe2a8b9 --- /dev/null +++ b/src/apps/external_data/export_urls.py @@ -0,0 +1,22 @@ +"""URL routes for prepared external-data exports.""" + +from apps.external_data.export_views import ( + SourceRecordExportDownloadView, + SourceRecordExportTicketView, + SourceRecordExportView, +) +from django.urls import path + +app_name = "source_record_exports" + +urlpatterns = [ + path("export/", SourceRecordExportView.as_view(), name="export"), + path( + "export-ticket/", SourceRecordExportTicketView.as_view(), name="export-ticket" + ), + path( + "export-download/", + SourceRecordExportDownloadView.as_view(), + name="export-download", + ), +] diff --git a/src/apps/external_data/export_views.py b/src/apps/external_data/export_views.py new file mode 100644 index 0000000..fa1cd3c --- /dev/null +++ b/src/apps/external_data/export_views.py @@ -0,0 +1,174 @@ +"""HTTP endpoints for prepared external-data exports.""" + +from apps.external_data.export_serializers import ( + SourceRecordExportDownloadSerializer, + SourceRecordExportRequestSerializer, +) +from apps.external_data.source_record_export import ( + SourceRecordExportArchive, + SourceRecordExportArtifactsUnavailable, + SourceRecordExportTicketInvalid, + build_source_records_export_archive, + consume_source_record_export_download_ticket, + create_source_record_export_download_ticket, +) +from django.http import StreamingHttpResponse +from drf_yasg import openapi +from drf_yasg.utils import swagger_auto_schema +from rest_framework import status +from rest_framework.permissions import AllowAny, IsAdminUser +from rest_framework.request import Request +from rest_framework.response import Response +from rest_framework.views import APIView + + +class SourceRecordExportResponseMixin: + """Build the shared streaming ZIP response for export endpoints.""" + + @staticmethod + def source_record_export_response( + package: SourceRecordExportArchive, + ) -> StreamingHttpResponse: + response = StreamingHttpResponse( + package.archive_chunks, + content_type="application/zip", + ) + response[ + "Content-Disposition" + ] = f'attachment; filename="{package.archive_name}"' + response["X-Source-Export-Files"] = str(package.files_count) + response["X-Source-Export-Generated-At"] = package.generated_at + return response + + @staticmethod + def source_record_export_not_ready_response() -> Response: + return Response( + { + "detail": "Готовая ночная выгрузка ещё не сформирована.", + "code": "source_export_not_ready", + }, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + headers={"Retry-After": "3600"}, + ) + + +class SourceRecordExportView(SourceRecordExportResponseMixin, APIView): + """Stream a request-specific ZIP from the current prepared generation.""" + + permission_classes = [IsAdminUser] + + @swagger_auto_schema( + operation_summary="Выгрузить записи внешних источников", + operation_description=( + "Упаковывает в ZIP готовые файлы выбранных групп без повторного " + "чтения таблиц external_data. Финансовые показатели всегда JSON." + ), + request_body=SourceRecordExportRequestSerializer, + responses={ + 200: openapi.Response( + description="Потоковый ZIP-архив.", + schema=openapi.Schema(type=openapi.TYPE_FILE), + ), + 400: "Некорректные параметры.", + 403: "Доступ разрешён только администраторам.", + 503: "Ночная выгрузка ещё не сформирована.", + }, + tags=["Внешние данные"], + ) + def post(self, request: Request) -> StreamingHttpResponse | Response: + serializer = SourceRecordExportRequestSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + try: + package = build_source_records_export_archive( + source_groups=serializer.validated_data["sources"], + export_format=serializer.validated_data["format"], + ) + except SourceRecordExportArtifactsUnavailable: + return self.source_record_export_not_ready_response() + return self.source_record_export_response(package) + + +class SourceRecordExportTicketView(SourceRecordExportResponseMixin, APIView): + """Issue a short-lived ticket for a native browser download.""" + + permission_classes = [IsAdminUser] + + @swagger_auto_schema( + operation_summary="Подготовить нативное скачивание внешних данных", + request_body=SourceRecordExportRequestSerializer, + responses={ + 201: "Одноразовый ticket и имя ZIP-файла.", + 400: "Некорректные параметры.", + 403: "Доступ разрешён только администраторам.", + 503: "Выгрузка или ticket временно недоступны.", + }, + tags=["Внешние данные"], + ) + def post(self, request: Request) -> Response: + serializer = SourceRecordExportRequestSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + try: + download_ticket = create_source_record_export_download_ticket( + source_groups=serializer.validated_data["sources"], + export_format=serializer.validated_data["format"], + ) + except SourceRecordExportArtifactsUnavailable: + return self.source_record_export_not_ready_response() + except RuntimeError: + return Response( + { + "detail": "Не удалось подготовить скачивание. Повторите запрос.", + "code": "source_export_ticket_unavailable", + }, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + headers={"Retry-After": "5"}, + ) + + return Response( + { + "ticket": download_ticket.ticket, + "file_name": download_ticket.archive_name, + "expires_in": download_ticket.expires_in, + }, + status=status.HTTP_201_CREATED, + ) + + +class SourceRecordExportDownloadView(SourceRecordExportResponseMixin, APIView): + """Consume a one-time ticket and stream the prepared ZIP archive.""" + + authentication_classes: list = [] + permission_classes = [AllowAny] + + @swagger_auto_schema( + operation_summary="Скачать готовые внешние данные по ticket", + request_body=SourceRecordExportDownloadSerializer, + responses={ + 200: openapi.Response( + description="Потоковый ZIP-архив.", + schema=openapi.Schema(type=openapi.TYPE_FILE), + ), + 400: "Ticket отсутствует или имеет неверный формат.", + 410: "Ticket истёк или уже использован.", + 503: "Опубликованная выгрузка больше недоступна.", + }, + tags=["Внешние данные"], + ) + def post(self, request: Request) -> StreamingHttpResponse | Response: + serializer = SourceRecordExportDownloadSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + try: + package = consume_source_record_export_download_ticket( + serializer.validated_data["ticket"] + ) + except SourceRecordExportTicketInvalid: + return Response( + { + "detail": "Ticket скачивания истёк или уже использован.", + "code": "source_export_ticket_invalid", + }, + status=status.HTTP_410_GONE, + ) + except SourceRecordExportArtifactsUnavailable: + return self.source_record_export_not_ready_response() + return self.source_record_export_response(package) diff --git a/src/apps/external_data/management/__init__.py b/src/apps/external_data/management/__init__.py new file mode 100644 index 0000000..526fd87 --- /dev/null +++ b/src/apps/external_data/management/__init__.py @@ -0,0 +1 @@ +"""Management package for external-data operations.""" diff --git a/src/apps/external_data/management/commands/__init__.py b/src/apps/external_data/management/commands/__init__.py new file mode 100644 index 0000000..8e6ab52 --- /dev/null +++ b/src/apps/external_data/management/commands/__init__.py @@ -0,0 +1 @@ +"""Management commands for external-data operations.""" diff --git a/src/apps/external_data/management/commands/build_source_record_exports.py b/src/apps/external_data/management/commands/build_source_record_exports.py new file mode 100644 index 0000000..924c4b1 --- /dev/null +++ b/src/apps/external_data/management/commands/build_source_record_exports.py @@ -0,0 +1,32 @@ +"""Build the complete prepared external-data export matrix.""" + +import json + +from apps.core.management.commands.base import BaseAppCommand +from apps.external_data.source_record_export import ( + build_source_record_export_artifacts, +) + + +class Command(BaseAppCommand): + """Build source-record files synchronously for bootstrap and recovery.""" + + help = "Формирует готовые CSV/XLSX/JSON выгрузки внешних источников" + use_transaction = False + + def execute_command(self, *args, **options) -> str: + generation = build_source_record_export_artifacts() + rendered = json.dumps( + { + "generation_id": generation.generation_id, + "generated_at": generation.generated_at, + "artifacts_count": generation.artifacts_count, + "files_count": generation.files_count, + "records_count": generation.records_count, + "total_size": generation.total_size, + }, + ensure_ascii=False, + sort_keys=True, + ) + self.log_success(rendered) + return rendered diff --git a/src/apps/external_data/migrations/0007_seed_nightly_source_record_exports.py b/src/apps/external_data/migrations/0007_seed_nightly_source_record_exports.py new file mode 100644 index 0000000..2ea47e5 --- /dev/null +++ b/src/apps/external_data/migrations/0007_seed_nightly_source_record_exports.py @@ -0,0 +1,61 @@ +import json + +from django.db import migrations + +NIGHTLY_SOURCE_EXPORT_TASK_NAME = "external-data:source-record-exports:nightly-msk" +NIGHTLY_SOURCE_EXPORT_TASK_PATH = ( + "apps.external_data.tasks.refresh_source_record_export_artifacts" +) +NIGHTLY_SOURCE_EXPORT_MSK_CRON = { + "minute": "30", + "hour": "5", + "day_of_week": "*", + "day_of_month": "*", + "month_of_year": "*", + "timezone": "Europe/Moscow", +} + + +def seed_nightly_source_record_export_schedule(apps, schema_editor): + CrontabSchedule = apps.get_model("django_celery_beat", "CrontabSchedule") + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + + crontab, _ = CrontabSchedule.objects.get_or_create(**NIGHTLY_SOURCE_EXPORT_MSK_CRON) + field_names = {field.name for field in PeriodicTask._meta.fields} + 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=NIGHTLY_SOURCE_EXPORT_TASK_NAME, + defaults={ + "task": NIGHTLY_SOURCE_EXPORT_TASK_PATH, + "args": json.dumps([]), + "kwargs": json.dumps({}), + "enabled": True, + "description": ( + "Nightly preparation of State Corp external-data export artifacts." + ), + **schedule_fields, + }, + ) + + +def remove_nightly_source_record_export_schedule(apps, schema_editor): + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + PeriodicTask.objects.filter(name=NIGHTLY_SOURCE_EXPORT_TASK_NAME).delete() + + +class Migration(migrations.Migration): + dependencies = [ + ("django_celery_beat", "0018_improve_crontab_helptext"), + ("external_data", "0006_bankruptcy_procedure_status_length"), + ] + + operations = [ + migrations.RunPython( + seed_nightly_source_record_export_schedule, + reverse_code=remove_nightly_source_record_export_schedule, + ), + ] diff --git a/src/apps/external_data/source_record_export.py b/src/apps/external_data/source_record_export.py new file mode 100644 index 0000000..417cd14 --- /dev/null +++ b/src/apps/external_data/source_record_export.py @@ -0,0 +1,1121 @@ +"""Prepared file exports for State Corp external source data.""" + +from __future__ import annotations + +import csv +import json +import os +import re +import secrets +import shutil +import zipfile +from collections.abc import Iterable, Iterator, Sequence +from dataclasses import dataclass +from datetime import UTC, date, datetime +from decimal import Decimal +from itertools import islice +from pathlib import Path +from tempfile import NamedTemporaryFile +from typing import Any, BinaryIO, cast +from uuid import UUID, uuid4 + +from apps.external_data.models import ( + ArbitrationCase, + BankruptcyProcedure, + DefenseUnreliableSupplier, + FinancialReport, + IndustrialCertificate, + IndustrialProduct, + InformationSecurityRegistryEntry, + LaborVacancy, + ManufacturerRegistryEntry, + ProsecutorCheck, + PublicProcurement, +) +from django.conf import settings +from django.core.cache import cache +from django.db.models import QuerySet, prefetch_related_objects +from django.utils import timezone +from openpyxl import Workbook + +EXPORT_FORMAT_CSV = "csv" +EXPORT_FORMAT_XLSX = "xlsx" +EXPORT_FORMAT_JSON = "json" +EXPORT_FORMATS = (EXPORT_FORMAT_CSV, EXPORT_FORMAT_XLSX, EXPORT_FORMAT_JSON) +FINANCIAL_SOURCE_GROUP = "financial_indicators" +EXPORT_MANIFEST_VERSION = 1 +CURRENT_EXPORT_MANIFEST_FILE_NAME = "current.json" +GENERATION_MANIFEST_FILE_NAME = "manifest.json" +GENERATION_DIRECTORY_NAME = "generations" +SOURCE_RECORD_EXPORT_ITERATOR_CHUNK_SIZE = 1000 +SOURCE_RECORD_EXPORT_ZIP_CHUNK_SIZE = 1024 * 1024 +EXCEL_MAX_DATA_ROWS_PER_SHEET = 1_048_575 +DEFAULT_XLSX_DATA_ROWS_PER_FILE = 100_000 +DEFAULT_DOWNLOAD_TICKET_TTL_SECONDS = 5 * 60 +SOURCE_RECORD_EXPORT_TICKET_CACHE_PREFIX = "external-data:source-record-exports:ticket" +SOURCE_RECORD_EXPORT_TICKET_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}") + +ORGANIZATION_EXPORT_FIELDS = [ + "Наименование", + "ИНН", + "ОГРН", + "КПП", + "ОКПО", + "organization", +] +SOURCE_RECORD_EXPORT_FIELDS = [ + "uid", + "source_group", + "record_type", + "created_at", + "updated_at", +] +FINANCIAL_LINE_FIELDS = [ + "id", + "form_code", + "line_code", + "line_name", + "year", + "period_start", + "period_end", +] + + +@dataclass(frozen=True) +class SourceModelExportSpec: + """Mapping from one normalized model to an exported record type.""" + + model: Any + record_type: str + fields: tuple[str, ...] + prefetch_related: tuple[str, ...] = () + + +@dataclass(frozen=True) +class SourceGroupExportSpec: + """Complete export definition for one frontend source group.""" + + source_group: str + file_stem: str + models: tuple[SourceModelExportSpec, ...] + + +SOURCE_GROUP_EXPORT_SPECS: dict[str, SourceGroupExportSpec] = { + FINANCIAL_SOURCE_GROUP: SourceGroupExportSpec( + source_group=FINANCIAL_SOURCE_GROUP, + file_stem="financial-indicators", + models=( + SourceModelExportSpec( + model=FinancialReport, + record_type="financial_report", + fields=( + "external_id", + "ogrn", + "file_name", + "file_hash", + "load_batch", + "status", + "source", + "error_message", + ), + prefetch_related=("lines",), + ), + ), + ), + "government_procurements": SourceGroupExportSpec( + source_group="government_procurements", + file_stem="public-procurements", + models=( + SourceModelExportSpec( + model=PublicProcurement, + record_type="public_procurement", + fields=( + "purchase_number", + "law_type", + "status", + "contract_amount", + "contract_date", + "execution_start_date", + "execution_end_date", + "purchase_name", + ), + ), + ), + ), + "industrial_production": SourceGroupExportSpec( + source_group="industrial_production", + file_stem="manufacturers-and-products", + models=( + SourceModelExportSpec( + model=IndustrialProduct, + record_type="industrial_product", + fields=( + "product_name", + "product_class", + "okpd2_code", + "tnved_code", + "registry_number", + ), + ), + SourceModelExportSpec( + model=IndustrialCertificate, + record_type="industrial_certificate", + fields=( + "certificate_number", + "issue_date", + "expiry_date", + "certificate_file_url", + "organisation_name", + "ogrn", + ), + ), + SourceModelExportSpec( + model=ManufacturerRegistryEntry, + record_type="manufacturer_registry_entry", + fields=("full_legal_name", "inn", "ogrn", "address"), + ), + ), + ), + "planned_inspections": SourceGroupExportSpec( + source_group="planned_inspections", + file_stem="planned-inspections", + models=( + SourceModelExportSpec( + model=ProsecutorCheck, + record_type="prosecutor_check", + fields=( + "registration_number", + "law_type", + "control_authority", + "prosecutor_office", + "start_date", + "status", + ), + ), + ), + ), + "bankruptcy": SourceGroupExportSpec( + source_group="bankruptcy", + file_stem="bankruptcy-procedures", + models=( + SourceModelExportSpec( + model=BankruptcyProcedure, + record_type="bankruptcy_procedure", + fields=( + "external_id", + "message_type", + "message_date", + "case_number", + "status", + "source_url", + ), + ), + ), + ), + "defense_suppliers": SourceGroupExportSpec( + source_group="defense_suppliers", + file_stem="defense-unreliable-suppliers", + models=( + SourceModelExportSpec( + model=DefenseUnreliableSupplier, + record_type="defense_unreliable_supplier", + fields=( + "external_id", + "registry_source", + "registry_number", + "supplier_name", + "reason", + "included_at", + "status", + "source_url", + ), + ), + ), + ), + "arbitration": SourceGroupExportSpec( + source_group="arbitration", + file_stem="arbitration-cases", + models=( + SourceModelExportSpec( + model=ArbitrationCase, + record_type="arbitration_case", + fields=( + "case_number", + "court_name", + "party_role", + "status", + "decision_date", + ), + ), + ), + ), + "security_registries": SourceGroupExportSpec( + source_group="security_registries", + file_stem="information-security-registries", + models=( + SourceModelExportSpec( + model=InformationSecurityRegistryEntry, + record_type="information_security_registry_entry", + fields=( + "external_id", + "registry_name", + "presence_status", + "entry_number", + "issued_at", + "expires_at", + ), + ), + ), + ), + "vacancies": SourceGroupExportSpec( + source_group="vacancies", + file_stem="labor-vacancies", + models=( + SourceModelExportSpec( + model=LaborVacancy, + record_type="labor_vacancy", + fields=( + "external_id", + "vacancy_source", + "title", + "status", + "published_at", + "salary_amount", + "source_url", + ), + ), + ), + ), +} + + +class SourceRecordExportArtifactsUnavailable(Exception): + """Raised when there is no complete published export generation.""" + + +class SourceRecordExportTicketInvalid(Exception): + """Raised when a native-download ticket is invalid, expired, or consumed.""" + + +@dataclass(frozen=True) +class SourceRecordExportArtifact: + """One prepared source-group file in a published generation.""" + + source_group: str + file_format: str + file_name: str + path: Path + size: int + records_count: int + part_number: int = 1 + parts_count: int = 1 + + +@dataclass(frozen=True) +class SourceRecordExportGeneration: + """Atomically published set of all source-record export files.""" + + generation_id: str + generated_at: str + artifacts: tuple[SourceRecordExportArtifact, ...] + records_count: int + + @property + def artifacts_count(self) -> int: + return len( + { + (artifact.source_group, artifact.file_format) + for artifact in self.artifacts + } + ) + + @property + def files_count(self) -> int: + return len(self.artifacts) + + @property + def total_size(self) -> int: + return sum(artifact.size for artifact in self.artifacts) + + +@dataclass(frozen=True) +class SourceRecordExportArchive: + """A request-specific ZIP streamed only from prepared files.""" + + archive_name: str + archive_chunks: Iterable[bytes] + files_count: int + generated_at: str + + +@dataclass(frozen=True) +class SourceRecordExportDownloadTicket: + """Short-lived capability for one native browser download.""" + + ticket: str + archive_name: str + expires_in: int + + +class _StreamingZipSink: + """Unseekable zipfile target whose written chunks can be drained.""" + + def __init__(self) -> None: + self._offset = 0 + self._chunks: list[bytes] = [] + + def write(self, data: bytes) -> int: + rendered_data = bytes(data) + self._chunks.append(rendered_data) + self._offset += len(rendered_data) + return len(rendered_data) + + def tell(self) -> int: + return self._offset + + def flush(self) -> None: + return None + + def drain(self) -> tuple[bytes, ...]: + chunks = tuple(self._chunks) + self._chunks.clear() + return chunks + + +def build_source_record_export_artifacts( + *, + now: datetime | None = None, + export_directory: str | Path | None = None, +) -> SourceRecordExportGeneration: + """Build all files on disk and atomically publish a new generation.""" + + root_directory = _resolve_export_directory(export_directory) + generations_directory = root_directory / GENERATION_DIRECTORY_NAME + root_directory.mkdir(parents=True, exist_ok=True) + generations_directory.mkdir(parents=True, exist_ok=True) + + generated_at_datetime = _normalize_generation_datetime(now or timezone.now()) + generation_id = ( + f"{generated_at_datetime.strftime('%Y%m%dT%H%M%SZ')}-{uuid4().hex[:8]}" + ) + staging_directory = generations_directory / f".building-{generation_id}" + final_directory = generations_directory / generation_id + staging_directory.mkdir() + + try: + artifacts: list[SourceRecordExportArtifact] = [] + source_record_counts: dict[str, int] = {} + + for source_group, source_spec in SOURCE_GROUP_EXPORT_SPECS.items(): + row_spool_path = staging_directory / f".{source_group}.rows.json" + headers, records_count = _spool_source_group_rows( + source_spec=source_spec, + output_path=row_spool_path, + ) + source_record_counts[source_group] = records_count + + try: + for file_format in _source_group_export_formats(source_group): + file_name = _build_source_group_file_name( + source_group=source_group, + file_format=file_format, + ) + artifact_paths = _render_source_group_artifact( + row_spool_path=row_spool_path, + output_path=staging_directory / file_name, + headers=headers, + file_format=file_format, + records_count=records_count, + ) + parts_count = len(artifact_paths) + artifacts.extend( + SourceRecordExportArtifact( + source_group=source_group, + file_format=file_format, + file_name=artifact_path.name, + path=final_directory / artifact_path.name, + size=artifact_path.stat().st_size, + records_count=records_count, + part_number=part_number, + parts_count=parts_count, + ) + for part_number, artifact_path in enumerate( + artifact_paths, + start=1, + ) + ) + finally: + row_spool_path.unlink(missing_ok=True) + + generation = SourceRecordExportGeneration( + generation_id=generation_id, + generated_at=generated_at_datetime.isoformat(), + artifacts=tuple(artifacts), + records_count=sum(source_record_counts.values()), + ) + manifest_payload = _generation_manifest_payload( + generation, + root_directory=root_directory, + ) + _write_json_file( + staging_directory / GENERATION_MANIFEST_FILE_NAME, + manifest_payload, + ) + os.replace(staging_directory, final_directory) + _write_json_file_atomically( + root_directory / CURRENT_EXPORT_MANIFEST_FILE_NAME, + manifest_payload, + ) + _cleanup_stale_generations( + generations_directory=generations_directory, + current_generation_id=generation_id, + ) + return generation + except Exception: + if staging_directory.exists(): + shutil.rmtree(staging_directory) + raise + + +def load_current_source_record_export_generation( + *, + export_directory: str | Path | None = None, +) -> SourceRecordExportGeneration: + """Load and validate the atomically published current generation.""" + + root_directory = _resolve_export_directory(export_directory) + manifest_path = root_directory / CURRENT_EXPORT_MANIFEST_FILE_NAME + try: + manifest_payload = json.loads(manifest_path.read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError, OSError) as exc: + raise SourceRecordExportArtifactsUnavailable( + "Prepared source-record export is not available." + ) from exc + + return _generation_from_manifest( + manifest_payload, + root_directory=root_directory, + ) + + +def build_source_records_export_archive( + *, + source_groups: Sequence[str], + export_format: str, + export_directory: str | Path | None = None, +) -> SourceRecordExportArchive: + """Package selected prepared files without querying external-data tables.""" + + root_directory = _resolve_export_directory(export_directory) + generation = load_current_source_record_export_generation( + export_directory=root_directory, + ) + artifacts_by_key: dict[ + tuple[str, str], + list[SourceRecordExportArtifact], + ] = {} + for artifact in generation.artifacts: + artifacts_by_key.setdefault( + (artifact.source_group, artifact.file_format), + [], + ).append(artifact) + selected_artifacts: list[SourceRecordExportArtifact] = [] + + for source_group in source_groups: + file_format = _resolve_source_group_export_format( + source_group=source_group, + requested_format=export_format, + ) + artifacts = artifacts_by_key.get((source_group, file_format), []) + if not artifacts or any(not artifact.path.is_file() for artifact in artifacts): + raise SourceRecordExportArtifactsUnavailable( + f"Prepared export artifact is missing: {source_group}/{file_format}." + ) + selected_artifacts.extend( + sorted(artifacts, key=lambda artifact: artifact.part_number) + ) + + generated_at = datetime.fromisoformat(generation.generated_at) + timestamp = generated_at.strftime("%Y%m%d_%H%M%S") + return SourceRecordExportArchive( + archive_name=f"organization_source_records_export_{timestamp}.zip", + archive_chunks=_stream_zip_archive(selected_artifacts), + files_count=len(selected_artifacts), + generated_at=generation.generated_at, + ) + + +def create_source_record_export_download_ticket( + *, + source_groups: Sequence[str], + export_format: str, +) -> SourceRecordExportDownloadTicket: + """Validate prepared files and cache a short-lived download capability.""" + + package = build_source_records_export_archive( + source_groups=source_groups, + export_format=export_format, + ) + expires_in = max( + 1, + int( + getattr( + settings, + "SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS", + DEFAULT_DOWNLOAD_TICKET_TTL_SECONDS, + ) + ), + ) + payload = {"sources": list(source_groups), "format": export_format} + for _attempt in range(3): + ticket = secrets.token_urlsafe(32) + if cache.add( + _source_record_export_ticket_cache_key(ticket), + payload, + timeout=expires_in, + ): + return SourceRecordExportDownloadTicket( + ticket=ticket, + archive_name=package.archive_name, + expires_in=expires_in, + ) + raise RuntimeError("Could not allocate a source-record export download ticket.") + + +def consume_source_record_export_download_ticket( + ticket: str, +) -> SourceRecordExportArchive: + """Consume a download ticket before streaming the prepared archive.""" + + if not SOURCE_RECORD_EXPORT_TICKET_PATTERN.fullmatch(ticket): + raise SourceRecordExportTicketInvalid + + cache_key = _source_record_export_ticket_cache_key(ticket) + payload = cache.get(cache_key) + if payload is None: + raise SourceRecordExportTicketInvalid + cache.delete(cache_key) + + try: + source_groups = payload["sources"] + export_format = payload["format"] + if ( + not isinstance(source_groups, list) + or not source_groups + or any( + not isinstance(source_group, str) + or source_group not in SOURCE_GROUP_EXPORT_SPECS + for source_group in source_groups + ) + or len(source_groups) != len(set(source_groups)) + or export_format not in EXPORT_FORMATS + ): + raise ValueError + except (KeyError, TypeError, ValueError): + raise SourceRecordExportTicketInvalid from None + + return build_source_records_export_archive( + source_groups=source_groups, + export_format=export_format, + ) + + +def _source_record_export_ticket_cache_key(ticket: str) -> str: + return f"{SOURCE_RECORD_EXPORT_TICKET_CACHE_PREFIX}:{ticket}" + + +def _stream_zip_archive( + artifacts: Sequence[SourceRecordExportArtifact], +) -> Iterator[bytes]: + sink = _StreamingZipSink() + with zipfile.ZipFile( + cast(BinaryIO, sink), + mode="w", + compression=zipfile.ZIP_STORED, + allowZip64=True, + ) as archive: + for artifact in artifacts: + with artifact.path.open("rb") as source_file: + with archive.open( + artifact.file_name, + mode="w", + force_zip64=True, + ) as archive_entry: + yield from sink.drain() + while chunk := source_file.read( + SOURCE_RECORD_EXPORT_ZIP_CHUNK_SIZE + ): + archive_entry.write(chunk) + yield from sink.drain() + yield from sink.drain() + yield from sink.drain() + + +def _resolve_export_directory(export_directory: str | Path | None) -> Path: + if export_directory is not None: + return Path(export_directory) + + configured_directory = getattr( + settings, + "SOURCE_RECORD_EXPORT_DIRECTORY", + Path(settings.MEDIA_ROOT) / "source-record-exports", + ) + return Path(str(configured_directory)) + + +def _normalize_generation_datetime(value: datetime) -> datetime: + if timezone.is_naive(value): + value = timezone.make_aware(value, UTC) + return value.astimezone(UTC) + + +def _source_group_export_formats(source_group: str) -> tuple[str, ...]: + if source_group == FINANCIAL_SOURCE_GROUP: + return (EXPORT_FORMAT_JSON,) + return EXPORT_FORMATS + + +def _resolve_source_group_export_format( + *, + source_group: str, + requested_format: str, +) -> str: + if source_group == FINANCIAL_SOURCE_GROUP: + return EXPORT_FORMAT_JSON + return requested_format + + +def _build_source_group_file_name(*, source_group: str, file_format: str) -> str: + return f"{SOURCE_GROUP_EXPORT_SPECS[source_group].file_stem}.{file_format}" + + +def _source_model_queryset(model_spec: SourceModelExportSpec) -> QuerySet: + return cast( + QuerySet, + model_spec.model.objects.select_related("organization").order_by(), + ) + + +def _iter_source_model_records( + model_spec: SourceModelExportSpec, +) -> Iterator[Any]: + iterator = _source_model_queryset(model_spec).iterator( + chunk_size=SOURCE_RECORD_EXPORT_ITERATOR_CHUNK_SIZE + ) + while True: + batch = list(islice(iterator, SOURCE_RECORD_EXPORT_ITERATOR_CHUNK_SIZE)) + if not batch: + return + if model_spec.prefetch_related: + prefetch_related_objects(batch, *model_spec.prefetch_related) + yield from batch + + +def _source_group_headers(source_spec: SourceGroupExportSpec) -> list[str]: + model_fields: list[str] = [] + for model_spec in source_spec.models: + for field_name in model_spec.fields: + if field_name not in model_fields: + model_fields.append(field_name) + if source_spec.source_group == FINANCIAL_SOURCE_GROUP: + model_fields.append("financial_lines") + return [ + *ORGANIZATION_EXPORT_FIELDS, + *SOURCE_RECORD_EXPORT_FIELDS, + *model_fields, + ] + + +def _spool_source_group_rows( + *, + source_spec: SourceGroupExportSpec, + output_path: Path, +) -> tuple[list[str], int]: + records_count = 0 + with output_path.open("w", encoding="utf-8", newline="") as output: + output.write("[") + is_first_row = True + for model_spec in source_spec.models: + for record in _iter_source_model_records(model_spec): + row = _build_record_row( + record, + source_spec=source_spec, + model_spec=model_spec, + ) + if is_first_row: + output.write("\n") + is_first_row = False + else: + output.write(",\n") + output.write(json.dumps(row, ensure_ascii=False, separators=(",", ":"))) + records_count += 1 + if not is_first_row: + output.write("\n") + output.write("]") + return _source_group_headers(source_spec), records_count + + +def _render_source_group_artifact( + *, + row_spool_path: Path, + output_path: Path, + headers: Sequence[str], + file_format: str, + records_count: int, +) -> tuple[Path, ...]: + if file_format == EXPORT_FORMAT_CSV: + _render_csv_file( + row_spool_path=row_spool_path, + output_path=output_path, + headers=headers, + ) + return (output_path,) + if file_format == EXPORT_FORMAT_XLSX: + return _render_xlsx_files( + row_spool_path=row_spool_path, + output_path=output_path, + headers=headers, + records_count=records_count, + ) + os.link(row_spool_path, output_path) + return (output_path,) + + +def _iter_spooled_rows(row_spool_path: Path) -> Iterator[dict[str, Any]]: + with row_spool_path.open("r", encoding="utf-8") as rows_file: + for line in rows_file: + serialized_row = line.strip() + if not serialized_row or serialized_row in {"[", "]", "[]"}: + continue + if serialized_row.endswith(","): + serialized_row = serialized_row[:-1] + yield json.loads(serialized_row) + + +def _render_csv_file( + *, + row_spool_path: Path, + output_path: Path, + headers: Sequence[str], +) -> None: + with output_path.open("w", encoding="utf-8-sig", newline="") as output: + writer = csv.DictWriter(output, fieldnames=list(headers), lineterminator="\n") + writer.writeheader() + for row in _iter_spooled_rows(row_spool_path): + writer.writerow( + {key: _serialize_flat_value(row.get(key)) for key in headers} + ) + + +def _render_xlsx_files( + *, + row_spool_path: Path, + output_path: Path, + headers: Sequence[str], + records_count: int, +) -> tuple[Path, ...]: + rows_per_file = min( + EXCEL_MAX_DATA_ROWS_PER_SHEET, + max( + 1, + int( + getattr( + settings, + "SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE", + DEFAULT_XLSX_DATA_ROWS_PER_FILE, + ) + ), + ), + ) + parts_count = max(1, (records_count + rows_per_file - 1) // rows_per_file) + part_paths = tuple( + _build_xlsx_part_path( + output_path=output_path, + part_number=part_number, + parts_count=parts_count, + ) + for part_number in range(1, parts_count + 1) + ) + part_number = 1 + rows_in_file = 0 + workbook, worksheet = _new_export_workbook(headers) + + for row in _iter_spooled_rows(row_spool_path): + if rows_in_file >= rows_per_file: + workbook.save(part_paths[part_number - 1]) + workbook.close() + part_number += 1 + rows_in_file = 0 + workbook, worksheet = _new_export_workbook(headers) + worksheet.append([_serialize_flat_value(row.get(key)) for key in headers]) + rows_in_file += 1 + + workbook.save(part_paths[part_number - 1]) + workbook.close() + if part_number != parts_count: + raise ValueError("Unexpected XLSX source-record export parts count.") + return part_paths + + +def _new_export_workbook(headers: Sequence[str]): + workbook = Workbook(write_only=True) + worksheet = workbook.create_sheet(title="data") + worksheet.append(list(headers)) + return workbook, worksheet + + +def _build_xlsx_part_path( + *, + output_path: Path, + part_number: int, + parts_count: int, +) -> Path: + if parts_count == 1: + return output_path + return output_path.with_name( + f"{output_path.stem}-part-{part_number:03d}{output_path.suffix}" + ) + + +def _build_record_row( + record: Any, + *, + source_spec: SourceGroupExportSpec, + model_spec: SourceModelExportSpec, +) -> dict[str, Any]: + organization = record.organization + row: dict[str, Any] = { + "Наименование": organization.full_name + or organization.short_name + or organization.name, + "ИНН": organization.inn, + "ОГРН": organization.ogrn, + "КПП": organization.kpp, + "ОКПО": organization.okpo, + "organization": organization.id, + "uid": record.id, + "source_group": source_spec.source_group, + "record_type": model_spec.record_type, + "created_at": record.created_at, + "updated_at": record.updated_at, + **{field_name: getattr(record, field_name) for field_name in model_spec.fields}, + } + if source_spec.source_group == FINANCIAL_SOURCE_GROUP: + row["financial_lines"] = [ + { + field_name: getattr(line, field_name) + for field_name in FINANCIAL_LINE_FIELDS + } + for line in record.lines.all() + ] + return {key: _serialize_json_value(value) for key, value in row.items()} + + +def _serialize_flat_value(value: Any) -> str | int | float | bool: + if value is None: + return "" + if isinstance(value, dict | list | tuple): + return json.dumps(value, ensure_ascii=False, sort_keys=True) + if isinstance(value, str | int | float | bool): + return value + return str(value) + + +def _serialize_json_value(value: Any) -> Any: + if isinstance(value, dict): + return {key: _serialize_json_value(item) for key, item in value.items()} + if isinstance(value, list | tuple): + return [_serialize_json_value(item) for item in value] + if isinstance(value, Decimal | UUID): + return str(value) + if isinstance(value, datetime | date): + return value.isoformat() + return value + + +def _generation_manifest_payload( + generation: SourceRecordExportGeneration, + *, + root_directory: Path, +) -> dict[str, Any]: + return { + "version": EXPORT_MANIFEST_VERSION, + "generation_id": generation.generation_id, + "generated_at": generation.generated_at, + "records_count": generation.records_count, + "artifacts_count": generation.artifacts_count, + "files_count": generation.files_count, + "total_size": generation.total_size, + "artifacts": [ + { + "source_group": artifact.source_group, + "format": artifact.file_format, + "file_name": artifact.file_name, + "relative_path": str(artifact.path.relative_to(root_directory)), + "size": artifact.size, + "records_count": artifact.records_count, + "part_number": artifact.part_number, + "parts_count": artifact.parts_count, + } + for artifact in generation.artifacts + ], + } + + +def _generation_from_manifest( + payload: Any, + *, + root_directory: Path, +) -> SourceRecordExportGeneration: + try: + if payload["version"] != EXPORT_MANIFEST_VERSION: + raise ValueError("Unsupported source-record export manifest version.") + generation_id = str(payload["generation_id"]) + generated_at = str(payload["generated_at"]) + datetime.fromisoformat(generated_at) + records_count = int(payload["records_count"]) + artifact_payloads = payload["artifacts"] + if not isinstance(artifact_payloads, list): + raise TypeError("Manifest artifacts must be a list.") + + root_resolved = root_directory.resolve() + artifacts: list[SourceRecordExportArtifact] = [] + for artifact_payload in artifact_payloads: + artifact_path = root_directory / str(artifact_payload["relative_path"]) + artifact_path.resolve().relative_to(root_resolved) + if not artifact_path.is_file(): + raise FileNotFoundError(artifact_path) + source_group = str(artifact_payload["source_group"]) + file_format = str(artifact_payload["format"]) + file_name = str(artifact_payload["file_name"]) + artifact_size = int(artifact_payload["size"]) + part_number = int(artifact_payload.get("part_number", 1)) + parts_count = int(artifact_payload.get("parts_count", 1)) + expected_file_name = _build_source_group_file_name( + source_group=source_group, + file_format=file_format, + ) + if file_format == EXPORT_FORMAT_XLSX: + expected_file_name = _build_xlsx_part_path( + output_path=Path(expected_file_name), + part_number=part_number, + parts_count=parts_count, + ).name + if file_name != expected_file_name: + raise ValueError("Unexpected source-record export artifact name.") + if artifact_size != artifact_path.stat().st_size: + raise ValueError("Source-record export artifact size mismatch.") + artifacts.append( + SourceRecordExportArtifact( + source_group=source_group, + file_format=file_format, + file_name=file_name, + path=artifact_path, + size=artifact_size, + records_count=int(artifact_payload["records_count"]), + part_number=part_number, + parts_count=parts_count, + ) + ) + + _validate_manifest_artifact_parts(artifacts) + except (KeyError, TypeError, ValueError, OSError) as exc: + raise SourceRecordExportArtifactsUnavailable( + "Prepared source-record export manifest is invalid." + ) from exc + + return SourceRecordExportGeneration( + generation_id=generation_id, + generated_at=generated_at, + artifacts=tuple(artifacts), + records_count=records_count, + ) + + +def _validate_manifest_artifact_parts( + artifacts: Sequence[SourceRecordExportArtifact], +) -> None: + expected_artifact_keys = { + (source_group, file_format) + for source_group in SOURCE_GROUP_EXPORT_SPECS + for file_format in _source_group_export_formats(source_group) + } + artifacts_by_key: dict[ + tuple[str, str], + list[SourceRecordExportArtifact], + ] = {} + for artifact in artifacts: + artifacts_by_key.setdefault( + (artifact.source_group, artifact.file_format), + [], + ).append(artifact) + if set(artifacts_by_key) != expected_artifact_keys: + raise ValueError("Source-record export manifest matrix is incomplete.") + + for artifact_key, artifact_parts in artifacts_by_key.items(): + parts_count = len(artifact_parts) + if ( + {artifact.parts_count for artifact in artifact_parts} != {parts_count} + or {artifact.part_number for artifact in artifact_parts} + != set(range(1, parts_count + 1)) + or (artifact_key[1] != EXPORT_FORMAT_XLSX and parts_count != 1) + ): + raise ValueError("Source-record export artifact parts are invalid.") + + +def _write_json_file(file_path: Path, payload: dict[str, Any]) -> None: + file_path.write_text( + json.dumps(payload, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + +def _write_json_file_atomically(file_path: Path, payload: dict[str, Any]) -> None: + temp_path: Path | None = None + try: + with NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=file_path.parent, + prefix=f".{file_path.name}.", + suffix=".tmp", + delete=False, + ) as temp_file: + json.dump(payload, temp_file, ensure_ascii=False, indent=2) + temp_file.flush() + os.fsync(temp_file.fileno()) + temp_path = Path(temp_file.name) + os.replace(temp_path, file_path) + finally: + if temp_path is not None: + temp_path.unlink(missing_ok=True) + + +def _cleanup_stale_generations( + *, + generations_directory: Path, + current_generation_id: str, +) -> None: + generations_to_keep = max( + 1, + int(getattr(settings, "SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP", 2)), + ) + published_generations = sorted( + ( + path + for path in generations_directory.iterdir() + if path.is_dir() + and not path.name.startswith(".") + and (path / GENERATION_MANIFEST_FILE_NAME).is_file() + ), + key=lambda path: path.name, + reverse=True, + ) + retained_names = {current_generation_id} + retained_names.update( + path.name for path in published_generations[:generations_to_keep] + ) + + for generation_directory in published_generations: + if generation_directory.name not in retained_names: + shutil.rmtree(generation_directory) diff --git a/src/apps/external_data/tasks.py b/src/apps/external_data/tasks.py new file mode 100644 index 0000000..861125b --- /dev/null +++ b/src/apps/external_data/tasks.py @@ -0,0 +1,46 @@ +"""Celery tasks for prepared external-data exports.""" + +import logging + +from apps.core.tasks import PeriodicTask as CorePeriodicTask +from apps.external_data.source_record_export import ( + build_source_record_export_artifacts, +) +from celery import shared_task +from django.conf import settings +from django.core.cache import cache + +logger = logging.getLogger(__name__) + + +@shared_task(bind=True, base=CorePeriodicTask) +def refresh_source_record_export_artifacts(self) -> dict: # noqa: ARG001 + """Build and atomically publish the nightly external-data export matrix.""" + + lock_key = getattr( + settings, + "SOURCE_RECORD_EXPORT_LOCK_KEY", + "external-data:source-record-exports:lock", + ) + lock_ttl = int( + getattr(settings, "SOURCE_RECORD_EXPORT_LOCK_TTL_SECONDS", 6 * 60 * 60) + ) + if not cache.add(lock_key, "1", timeout=lock_ttl): + logger.info("Source-record export generation skipped: lock is already held") + return {"status": "skipped", "reason": "locked"} + + try: + generation = build_source_record_export_artifacts() + result = { + "status": "success", + "generation_id": generation.generation_id, + "generated_at": generation.generated_at, + "artifacts_count": generation.artifacts_count, + "files_count": generation.files_count, + "records_count": generation.records_count, + "total_size": generation.total_size, + } + logger.info("Source-record export generation published: %s", result) + return result + finally: + cache.delete(lock_key) diff --git a/src/core/api_v2_urls.py b/src/core/api_v2_urls.py new file mode 100644 index 0000000..792da40 --- /dev/null +++ b/src/core/api_v2_urls.py @@ -0,0 +1,12 @@ +"""API v2 routes shared with the Mostovik administrative frontend contract.""" + +from django.urls import include, path + +app_name = "api_v2" + +urlpatterns = [ + path( + "organization-source-records/", + include("apps.external_data.export_urls"), + ), +] diff --git a/src/core/urls.py b/src/core/urls.py index 4e71e67..9dfcd68 100644 --- a/src/core/urls.py +++ b/src/core/urls.py @@ -40,6 +40,7 @@ urlpatterns = [ path("admin/", admin.site.urls), path("health/", include("apps.core.urls")), path("api/v1/", include("core.api_v1_urls", namespace="api_v1")), + path("api/v2/", include("core.api_v2_urls", namespace="api_v2")), path("auth/", include("rest_framework.urls")), ] diff --git a/src/settings/base.py b/src/settings/base.py index 2599a80..7344c2b 100644 --- a/src/settings/base.py +++ b/src/settings/base.py @@ -231,6 +231,26 @@ STATICFILES_STORAGE = "whitenoise.storage.CompressedManifestStaticFilesStorage" MEDIA_URL = "/media/" MEDIA_ROOT = PROJECT_ROOT / "media" +SOURCE_RECORD_EXPORT_DIRECTORY = os.getenv( + "SOURCE_RECORD_EXPORT_DIRECTORY", + str(PROJECT_ROOT / "media" / "source-record-exports"), +) +SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP = int( + os.getenv("SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP", "2") +) +SOURCE_RECORD_EXPORT_LOCK_KEY = os.getenv( + "SOURCE_RECORD_EXPORT_LOCK_KEY", + "external-data:source-record-exports:lock", +) +SOURCE_RECORD_EXPORT_LOCK_TTL_SECONDS = int( + os.getenv("SOURCE_RECORD_EXPORT_LOCK_TTL_SECONDS", str(6 * 60 * 60)) +) +SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE = int( + os.getenv("SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE", "100000") +) +SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS = int( + os.getenv("SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS", "300") +) DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" AUTH_USER_MODEL = "user.User" diff --git a/tests/apps/external_data/test_export_tasks.py b/tests/apps/external_data/test_export_tasks.py new file mode 100644 index 0000000..899ef2b --- /dev/null +++ b/tests/apps/external_data/test_export_tasks.py @@ -0,0 +1,70 @@ +"""Tests for the external-data export task and schedule.""" + +from importlib import import_module +from tempfile import TemporaryDirectory + +from apps.external_data.tasks import refresh_source_record_export_artifacts +from django.apps import apps as django_apps +from django.conf import settings +from django.core.cache import cache +from django.test import TestCase, override_settings +from django_celery_beat.models import PeriodicTask + + +class SourceRecordExportArtifactsTaskTest(TestCase): + """Check nightly artifact generation and its distributed lock.""" + + def setUp(self): + cache.clear() + self.export_directory = TemporaryDirectory() + self.settings_override = override_settings( + SOURCE_RECORD_EXPORT_DIRECTORY=self.export_directory.name, + SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP=2, + SOURCE_RECORD_EXPORT_LOCK_KEY="test:state-corp-source-exports:lock", + SOURCE_RECORD_EXPORT_LOCK_TTL_SECONDS=300, + ) + self.settings_override.enable() + + def tearDown(self): + self.settings_override.disable() + self.export_directory.cleanup() + cache.clear() + super().tearDown() + + def test_refresh_task_builds_matrix_and_releases_lock(self): + result = refresh_source_record_export_artifacts() + + self.assertEqual(result["status"], "success") + self.assertEqual(result["artifacts_count"], 25) + self.assertIsNone(cache.get(settings.SOURCE_RECORD_EXPORT_LOCK_KEY)) + + def test_refresh_task_skips_when_generation_lock_is_held(self): + cache.set(settings.SOURCE_RECORD_EXPORT_LOCK_KEY, "busy", timeout=300) + + result = refresh_source_record_export_artifacts() + + self.assertEqual(result, {"status": "skipped", "reason": "locked"}) + + +class SourceRecordExportScheduleMigrationTest(TestCase): + """Check the nightly Celery Beat schedule for prepared exports.""" + + def test_migration_seeds_nightly_export_task_idempotently(self): + migration = import_module( + "apps.external_data.migrations.0007_seed_nightly_source_record_exports" + ) + + migration.seed_nightly_source_record_export_schedule(django_apps, None) + migration.seed_nightly_source_record_export_schedule(django_apps, None) + + task = PeriodicTask.objects.get(name=migration.NIGHTLY_SOURCE_EXPORT_TASK_NAME) + self.assertEqual( + task.task, + "apps.external_data.tasks.refresh_source_record_export_artifacts", + ) + self.assertTrue(task.enabled) + self.assertEqual(task.args, "[]") + self.assertEqual(task.kwargs, "{}") + self.assertEqual(task.crontab.minute, "30") + self.assertEqual(task.crontab.hour, "5") + self.assertEqual(str(task.crontab.timezone), "Europe/Moscow") diff --git a/tests/apps/external_data/test_source_record_export.py b/tests/apps/external_data/test_source_record_export.py new file mode 100644 index 0000000..ce253d7 --- /dev/null +++ b/tests/apps/external_data/test_source_record_export.py @@ -0,0 +1,225 @@ +"""Tests for prepared State Corp external-data exports.""" + +import json +import zipfile +from io import BytesIO, StringIO +from tempfile import TemporaryDirectory + +from apps.external_data.source_record_export import ( + build_source_record_export_artifacts, + load_current_source_record_export_generation, +) +from django.core.management import call_command +from django.test import override_settings +from openpyxl import load_workbook +from rest_framework import status +from rest_framework.test import APITestCase + +from tests.apps.external_data.factories import ( + FinancialReportFactory, + FinancialReportLineFactory, + IndustrialCertificateFactory, + IndustrialProductFactory, + ManufacturerRegistryEntryFactory, + ProsecutorCheckFactory, +) +from tests.apps.organization.factories import OrganizationFactory +from tests.apps.user.factories import UserFactory + + +class SourceRecordExportApiTest(APITestCase): + """Check admin access and zero-query delivery of prepared files.""" + + export_url = "/api/v2/organization-source-records/export/" + ticket_url = "/api/v2/organization-source-records/export-ticket/" + download_url = "/api/v2/organization-source-records/export-download/" + + def setUp(self): + self.export_directory = TemporaryDirectory() + self.settings_override = override_settings( + SOURCE_RECORD_EXPORT_DIRECTORY=self.export_directory.name, + SOURCE_RECORD_EXPORT_GENERATIONS_TO_KEEP=2, + ) + self.settings_override.enable() + + def tearDown(self): + self.settings_override.disable() + self.export_directory.cleanup() + super().tearDown() + + @staticmethod + def _response_body(response) -> bytes: + return b"".join(response.streaming_content) + + def test_export_is_unavailable_before_first_generation(self): + self.client.force_authenticate(UserFactory.create_superuser()) + + response = self.client.post( + self.export_url, + {"sources": ["planned_inspections"], "format": "json"}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE) + self.assertEqual(response.data["code"], "source_export_not_ready") + self.assertEqual(response["Retry-After"], "3600") + + def test_generation_builds_full_matrix_from_normalized_tables(self): + organization = OrganizationFactory.create( + full_name='Акционерное общество "Экспорт"', + okpo="12345678", + ) + IndustrialProductFactory.create(organization=organization) + IndustrialCertificateFactory.create(organization=organization) + ManufacturerRegistryEntryFactory.create(organization=organization) + ProsecutorCheckFactory.create(organization=organization) + report = FinancialReportFactory.create(organization=organization) + FinancialReportLineFactory.create(report=report, line_code="1600") + + generation = build_source_record_export_artifacts() + + self.assertEqual(generation.artifacts_count, 25) + self.assertEqual(generation.files_count, 25) + self.assertEqual(generation.records_count, 5) + industrial_path = next( + artifact.path + for artifact in generation.artifacts + if artifact.source_group == "industrial_production" + and artifact.file_format == "json" + ) + industrial_rows = json.loads(industrial_path.read_text(encoding="utf-8")) + self.assertEqual( + {row["record_type"] for row in industrial_rows}, + { + "industrial_certificate", + "industrial_product", + "manufacturer_registry_entry", + }, + ) + self.assertEqual({row["ОКПО"] for row in industrial_rows}, {"12345678"}) + + financial_path = next( + artifact.path + for artifact in generation.artifacts + if artifact.source_group == "financial_indicators" + ) + financial_rows = json.loads(financial_path.read_text(encoding="utf-8")) + self.assertEqual(financial_rows[0]["financial_lines"][0]["line_code"], "1600") + + def test_management_command_bootstraps_first_generation(self): + command_output = StringIO() + + call_command("build_source_record_exports", stdout=command_output) + + generation = load_current_source_record_export_generation() + self.assertEqual(generation.artifacts_count, 25) + self.assertIn('"artifacts_count": 25', command_output.getvalue()) + + def test_admin_streams_selected_prepared_files_without_database_queries(self): + self.client.force_authenticate(UserFactory.create_superuser()) + organization = OrganizationFactory.create(okpo="87654321") + ProsecutorCheckFactory.create(organization=organization) + report = FinancialReportFactory.create(organization=organization) + FinancialReportLineFactory.create(report=report) + generation = build_source_record_export_artifacts() + + with self.assertNumQueries(0): + response = self.client.post( + self.export_url, + { + "sources": ["planned_inspections", "financial_indicators"], + "format": "xlsx", + }, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertTrue(response.streaming) + self.assertEqual(response["Content-Type"], "application/zip") + self.assertEqual( + response["X-Source-Export-Generated-At"], generation.generated_at + ) + self.assertNotIn("Content-Length", response) + + with zipfile.ZipFile(BytesIO(self._response_body(response))) as archive: + self.assertEqual( + set(archive.namelist()), + {"planned-inspections.xlsx", "financial-indicators.json"}, + ) + workbook = load_workbook( + BytesIO(archive.read("planned-inspections.xlsx")), + read_only=True, + ) + rows = list(workbook["data"].iter_rows(values_only=True)) + self.assertEqual( + rows[0][:6], + ("Наименование", "ИНН", "ОГРН", "КПП", "ОКПО", "organization"), + ) + self.assertEqual(rows[1][4], "87654321") + + def test_ticket_is_admin_only_and_can_be_consumed_once_without_auth(self): + build_source_record_export_artifacts() + regular_user = UserFactory.create_user() + self.client.force_authenticate(regular_user) + forbidden_response = self.client.post( + self.ticket_url, + {"sources": ["bankruptcy"], "format": "json"}, + format="json", + ) + self.assertEqual(forbidden_response.status_code, status.HTTP_403_FORBIDDEN) + + self.client.force_authenticate(UserFactory.create_superuser()) + with self.assertNumQueries(0): + ticket_response = self.client.post( + self.ticket_url, + {"sources": ["bankruptcy"], "format": "json"}, + format="json", + ) + self.assertEqual(ticket_response.status_code, status.HTTP_201_CREATED) + self.assertRegex(ticket_response.data["ticket"], r"^[A-Za-z0-9_-]{43}$") + self.assertEqual(ticket_response.data["expires_in"], 300) + + self.client.force_authenticate(user=None) + with self.assertNumQueries(0): + download_response = self.client.post( + self.download_url, + {"ticket": ticket_response.data["ticket"]}, + format="multipart", + ) + self.assertEqual(download_response.status_code, status.HTTP_200_OK) + with zipfile.ZipFile( + BytesIO(self._response_body(download_response)) + ) as archive: + self.assertEqual(archive.namelist(), ["bankruptcy-procedures.json"]) + + consumed_response = self.client.post( + self.download_url, + {"ticket": ticket_response.data["ticket"]}, + format="multipart", + ) + self.assertEqual(consumed_response.status_code, status.HTTP_410_GONE) + self.assertEqual(consumed_response.data["code"], "source_export_ticket_invalid") + + @override_settings(SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE=2) + def test_xlsx_is_split_into_bounded_workbook_parts(self): + organization = OrganizationFactory.create() + ProsecutorCheckFactory.create_batch(3, organization=organization) + + generation = build_source_record_export_artifacts() + + inspection_parts = [ + artifact + for artifact in generation.artifacts + if artifact.source_group == "planned_inspections" + and artifact.file_format == "xlsx" + ] + self.assertEqual( + [artifact.part_number for artifact in inspection_parts], [1, 2] + ) + self.assertEqual( + [artifact.file_name for artifact in inspection_parts], + [ + "planned-inspections-part-001.xlsx", + "planned-inspections-part-002.xlsx", + ], + )