feat: align source data and nightly exports

tests/apps/organizations/test_api_v2_source_extensions.py

@@ -170,6 +170,59 @@ class OrganizationSourceExtensionsApiV2Test(APITestCase):
         self.assertEqual(record["source_group"], "planned_inspections")
         self.assertEqual(record["organization"]["uid"], str(target.uid))
 
+    def test_flat_arbitration_records_expose_frontend_role_labels(self):
+        organization = create_frontend_organization(
+            name='ООО "Arbitration Roles"',
+            inn="7707083899",
+            ogrn="1027700132099",
+        )
+        extension = ArbitrationExtension.objects.create(
+            organization=organization,
+            title="Арбитраж",
+        )
+        expected_roles = {
+            "ROLE-PLAINTIFF": "Истец",
+            "ROLE-DEFENDANT": "Ответчик",
+            "ROLE-THIRD-PARTY": "Третье лицо",
+        }
+        for external_id, provider_role in (
+            ("ROLE-PLAINTIFF", "plaintiff"),
+            ("ROLE-DEFENDANT", "defendant"),
+            ("ROLE-THIRD-PARTY", "third_party"),
+        ):
+            OrganizationSourceRecord.objects.create(
+                extension=extension,
+                record_type="arbitration_case",
+                source="arbitration",
+                external_id=external_id,
+                payload={"target": {"role": provider_role}},
+            )
+
+        response = self.client.get(
+            reverse("api_v2:organizations:organization-source-records-list"),
+            {
+                "source_group": "arbitration",
+                "source": "arbitration",
+                "organization": str(organization.uid),
+            },
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        records_by_external_id = {
+            item["external_id"]: item for item in response.data["data"]
+        }
+        self.assertEqual(
+            {
+                external_id: records_by_external_id[external_id]["payload"]["role"]
+                for external_id in expected_roles
+            },
+            expected_roles,
+        )
+        self.assertNotIn(
+            "role",
+            OrganizationSourceRecord.objects.get(external_id="ROLE-PLAINTIFF").payload,
+        )
+
     def test_flat_source_records_filters_supported_groups_by_canonical_date(self):
         organization = create_frontend_organization(
             name='ООО "Canonical dates"',

tests/apps/organizations/test_source_record_export.py

@@ -2,9 +2,15 @@
 
 import csv
 import json
+import os
 import zipfile
 from io import BytesIO, StringIO
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
 
+from django.core.management import call_command
+from django.test import override_settings
 from django.urls import reverse
 from openpyxl import load_workbook
 from organizations.models import (
@@ -15,6 +21,14 @@ from organizations.models import (
     PlannedInspectionExtension,
     SourceGroup,
 )
+from organizations.source_record_export import (
+    _render_source_group_artifact,
+    _source_group_queryset,
+    _spool_source_group_rows,
+    build_source_record_export_artifacts,
+    build_source_records_export_archive,
+    load_current_source_record_export_generation,
+)
 from rest_framework import status
 from rest_framework.test import APITestCase
 
@@ -25,9 +39,57 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
     """Checks admin-only source-record export contract."""
 
     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()
         self.url = reverse(
             "api_v2:organizations:organization-source-records-export",
         )
+        self.ticket_url = reverse(
+            "api_v2:organizations:organization-source-records-export-ticket",
+        )
+        self.download_url = reverse(
+            "api_v2:organizations:organization-source-records-export-download",
+        )
+
+    def tearDown(self):
+        self.settings_override.disable()
+        self.export_directory.cleanup()
+        super().tearDown()
+
+    @staticmethod
+    def _response_body(response) -> bytes:
+        if response.streaming:
+            return b"".join(response.streaming_content)
+        return response.content
+
+    def test_export_returns_service_unavailable_before_first_nightly_generation(self):
+        self.client.force_authenticate(UserFactory.create_superuser())
+
+        response = self.client.post(
+            self.url,
+            {
+                "sources": [SourceGroup.PLANNED_INSPECTIONS.value],
+                "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_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_exports_selected_sources_to_zip(self):
         self.client.force_authenticate(UserFactory.create_superuser())
@@ -74,27 +136,34 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
             period_start=100,
             period_end=200,
         )
+        generation = build_source_record_export_artifacts()
 
-        response = self.client.post(
-            self.url,
-            {
-                "sources": [
-                    SourceGroup.PLANNED_INSPECTIONS.value,
-                    SourceGroup.FINANCIAL_INDICATORS.value,
-                ],
-                "format": "xlsx",
-            },
-            format="json",
-        )
+        with self.assertNumQueries(0):
+            response = self.client.post(
+                self.url,
+                {
+                    "sources": [
+                        SourceGroup.PLANNED_INSPECTIONS.value,
+                        SourceGroup.FINANCIAL_INDICATORS.value,
+                    ],
+                    "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.assertEqual(generation.artifacts_count, 25)
         self.assertIn(
             'filename="organization_source_records_export_',
             response["Content-Disposition"],
         )
 
-        with zipfile.ZipFile(BytesIO(response.content)) as archive:
+        with zipfile.ZipFile(BytesIO(self._response_body(response))) as archive:
             self.assertEqual(
                 set(archive.namelist()),
                 {"planned-inspections.xlsx", "financial-indicators.json"},
@@ -131,6 +200,53 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
             )
             self.assertEqual(financial_rows[0]["financial_lines"][0]["period_end"], 200)
 
+    def test_admin_uses_one_time_ticket_for_native_zero_sql_download(self):
+        self.client.force_authenticate(UserFactory.create_superuser())
+        build_source_record_export_artifacts()
+
+        with self.assertNumQueries(0):
+            ticket_response = self.client.post(
+                self.ticket_url,
+                {
+                    "sources": [SourceGroup.PLANNED_INSPECTIONS.value],
+                    "format": "json",
+                },
+                format="json",
+            )
+
+        self.assertEqual(ticket_response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(ticket_response.data["expires_in"], 300)
+        self.assertRegex(ticket_response.data["ticket"], r"^[A-Za-z0-9_-]{43}$")
+        self.assertNotIn("download_url", ticket_response.data)
+
+        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)
+        self.assertTrue(download_response.streaming)
+        self.assertEqual(download_response["Content-Type"], "application/zip")
+        self.assertNotIn("Content-Length", download_response)
+        with zipfile.ZipFile(
+            BytesIO(self._response_body(download_response))
+        ) as archive:
+            self.assertEqual(archive.namelist(), ["planned-inspections.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",
+        )
+
     def test_csv_export_uses_bom_and_canonical_columns_before_payload(self):
         self.client.force_authenticate(UserFactory.create_superuser())
         organization = Organization.objects.create(
@@ -152,6 +268,7 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
             title="CSV проверка",
             payload={"nested": {"value": "данные"}},
         )
+        build_source_record_export_artifacts()
 
         response = self.client.post(
             self.url,
@@ -164,7 +281,7 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-        with zipfile.ZipFile(BytesIO(response.content)) as archive:
+        with zipfile.ZipFile(BytesIO(self._response_body(response))) as archive:
             csv_bytes = archive.read("planned-inspections.csv")
             self.assertTrue(csv_bytes.startswith(b"\xef\xbb\xbf"))
             csv_text = csv_bytes.decode("utf-8-sig")
@@ -176,6 +293,178 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
         )
         self.assertIn("payload.nested.value", csv_rows[0])
 
+    def test_xlsx_export_splits_rows_across_bounded_workbook_parts(self):
+        organization = Organization.objects.create(
+            name='ООО "Многолистовая выгрузка"',
+            inn="7707083812",
+        )
+        extension = PlannedInspectionExtension.objects.create(
+            organization=organization,
+            title="Плановые проверки Генпрокуратуры России",
+        )
+        for index in range(3):
+            OrganizationSourceRecord.objects.create(
+                extension=extension,
+                record_type="inspection",
+                source="inspections",
+                external_id=f"INSP-SHEET-{index}",
+                title=f"Проверка {index}",
+                payload={},
+            )
+
+        with override_settings(SOURCE_RECORD_EXPORT_XLSX_ROWS_PER_FILE=2):
+            generation = build_source_record_export_artifacts()
+
+        artifacts = sorted(
+            (
+                item
+                for item in generation.artifacts
+                if item.source_group == SourceGroup.PLANNED_INSPECTIONS.value
+                and item.file_format == "xlsx"
+            ),
+            key=lambda item: item.file_name,
+        )
+
+        self.assertEqual(generation.artifacts_count, 25)
+        self.assertEqual(generation.files_count, 26)
+        self.assertEqual(
+            [item.file_name for item in artifacts],
+            [
+                "planned-inspections-part-001.xlsx",
+                "planned-inspections-part-002.xlsx",
+            ],
+        )
+        first_workbook = load_workbook(artifacts[0].path, read_only=True)
+        second_workbook = load_workbook(artifacts[1].path, read_only=True)
+
+        self.assertEqual(first_workbook.sheetnames, ["data"])
+        self.assertEqual(second_workbook.sheetnames, ["data"])
+        self.assertEqual(
+            len(list(first_workbook["data"].iter_rows(values_only=True))),
+            3,
+        )
+        self.assertEqual(
+            len(list(second_workbook["data"].iter_rows(values_only=True))),
+            2,
+        )
+        self.assertEqual(
+            next(second_workbook["data"].iter_rows(values_only=True))[:4],
+            ("Наименование", "ИНН", "ОГРН", "КПП"),
+        )
+
+        selected_artifacts = [
+            item
+            for item in generation.artifacts
+            if item.source_group == SourceGroup.PLANNED_INSPECTIONS.value
+        ]
+        self.assertEqual(len(selected_artifacts), 4)
+
+        package = build_source_records_export_archive(
+            source_groups=[SourceGroup.PLANNED_INSPECTIONS.value],
+            export_format="xlsx",
+        )
+        archive_bytes = b"".join(package.archive_chunks)
+
+        with zipfile.ZipFile(BytesIO(archive_bytes)) as archive:
+            self.assertEqual(
+                archive.namelist(),
+                [
+                    "planned-inspections-part-001.xlsx",
+                    "planned-inspections-part-002.xlsx",
+                ],
+            )
+        self.assertEqual(package.files_count, 2)
+        self.assertFalse((Path(self.export_directory.name) / "tmp").exists())
+
+    def test_nightly_export_clears_model_ordering_to_avoid_multi_million_row_sort(self):
+        queryset = _source_group_queryset(SourceGroup.GOVERNMENT_PROCUREMENTS.value)
+
+        self.assertFalse(queryset.ordered)
+        self.assertFalse(queryset.query.default_ordering)
+
+    def test_json_artifact_reuses_valid_canonical_spool_without_copying_it(self):
+        organization = Organization.objects.create(
+            name='ООО "JSON без копии"',
+            inn="7707083813",
+        )
+        extension = PlannedInspectionExtension.objects.create(
+            organization=organization,
+            title="Плановые проверки Генпрокуратуры России",
+        )
+        for index in range(2):
+            OrganizationSourceRecord.objects.create(
+                extension=extension,
+                record_type="inspection",
+                source="inspections",
+                external_id=f"INSP-JSON-{index}",
+                title=f"Проверка {index}",
+                payload={"index": index},
+            )
+
+        with TemporaryDirectory() as temporary_directory:
+            spool_path = Path(temporary_directory) / "rows.json"
+            artifact_path = Path(temporary_directory) / "inspections.json"
+            headers, records_count = _spool_source_group_rows(
+                source_group=SourceGroup.PLANNED_INSPECTIONS.value,
+                output_path=spool_path,
+            )
+
+            _render_source_group_artifact(
+                row_spool_path=spool_path,
+                output_path=artifact_path,
+                headers=headers,
+                file_format="json",
+                records_count=records_count,
+            )
+
+            self.assertEqual(records_count, 2)
+            self.assertEqual(len(json.loads(spool_path.read_text())), 2)
+            self.assertTrue(os.path.samefile(spool_path, artifact_path))
+
+    def test_generation_publishes_complete_matrix_and_keeps_previous_on_failure(self):
+        first_generation = build_source_record_export_artifacts()
+        current_generation = load_current_source_record_export_generation()
+
+        self.assertEqual(first_generation.artifacts_count, 25)
+        self.assertEqual(
+            current_generation.generation_id, first_generation.generation_id
+        )
+        self.assertEqual(
+            {
+                artifact.file_format
+                for artifact in current_generation.artifacts
+                if artifact.source_group == SourceGroup.FINANCIAL_INDICATORS.value
+            },
+            {"json"},
+        )
+        self.assertEqual(
+            {
+                artifact.file_format
+                for artifact in current_generation.artifacts
+                if artifact.source_group == SourceGroup.PLANNED_INSPECTIONS.value
+            },
+            {"csv", "xlsx", "json"},
+        )
+
+        original_replace = os.replace
+
+        def fail_generation_publish(source, destination):
+            if Path(source).name.startswith(".building-"):
+                raise OSError("disk full")
+            return original_replace(source, destination)
+
+        with patch(
+            "organizations.source_record_export.os.replace",
+            side_effect=fail_generation_publish,
+        ), self.assertRaises(OSError):
+            build_source_record_export_artifacts()
+
+        current_after_failure = load_current_source_record_export_generation()
+        self.assertEqual(
+            current_after_failure.generation_id,
+            first_generation.generation_id,
+        )
+
     def test_export_rejects_non_admin_user(self):
         self.client.force_authenticate(UserFactory.create_user())
 
@@ -187,8 +476,17 @@ class OrganizationSourceRecordExportApiV2Test(APITestCase):
             },
             format="json",
         )
+        ticket_response = self.client.post(
+            self.ticket_url,
+            {
+                "sources": [SourceGroup.PLANNED_INSPECTIONS.value],
+                "format": "json",
+            },
+            format="json",
+        )
 
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(ticket_response.status_code, status.HTTP_403_FORBIDDEN)
 
     def test_export_rejects_empty_duplicate_and_unknown_values(self):
         self.client.force_authenticate(UserFactory.create_superuser())

tests/apps/organizations/test_tasks.py

@@ -1,11 +1,13 @@
 """Tests for organization source backfill tasks and schedules."""
 
 from importlib import import_module
+from tempfile import TemporaryDirectory
 
 from apps.parsers.models import ParserLoadLog
 from django.apps import apps as django_apps
+from django.conf import settings
 from django.core.cache import cache
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from django.utils import timezone
 from django_celery_beat.models import PeriodicTask
 from organizations.cache import get_organization_api_cache_version
@@ -17,6 +19,7 @@ from organizations.models import (
 from organizations.tasks import (
     backfill_all_organization_sources,
     backfill_organization_sources_for_parser_batch,
+    refresh_source_record_export_artifacts,
 )
 
 from tests.apps.parsers.factories import IndustrialCertificateRecordFactory
@@ -116,3 +119,62 @@ class OrganizationSnapshotScheduleMigrationTest(TestCase):
         self.assertEqual(task.crontab.minute, "30")
         self.assertEqual(task.crontab.hour, "4")
         self.assertEqual(str(task.crontab.timezone), "Europe/Moscow")
+
+
+class SourceRecordExportArtifactsTaskTest(TestCase):
+    """Checks 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:source-record-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_all_artifacts_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_another_generation_holds_lock(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):
+    """Checks the nightly Celery Beat schedule for export artifacts."""
+
+    def test_migration_seeds_nightly_source_record_export_task(self):
+        migration = import_module(
+            "organizations.migrations.0008_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,
+            "organizations.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")