feat: limit source exports to current year

src/apps/external_data/management/commands/build_source_record_exports.py

@@ -20,6 +20,7 @@ class Command(BaseAppCommand):
             {
                 "generation_id": generation.generation_id,
                 "generated_at": generation.generated_at,
+                "export_year": generation.export_year,
                 "artifacts_count": generation.artifacts_count,
                 "files_count": generation.files_count,
                 "records_count": generation.records_count,

src/apps/external_data/migrations/0008_export_date_indexes.py

@@ -0,0 +1,34 @@
+"""Index the remaining date fields used by current-year exports."""
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    """Add indexes for nullable certificate and security-registry dates."""
+
+    dependencies = [
+        ("external_data", "0007_seed_nightly_source_record_exports"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="industrialcertificate",
+            name="issue_date",
+            field=models.DateField(
+                blank=True,
+                db_index=True,
+                null=True,
+                verbose_name="дата выдачи",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="informationsecurityregistryentry",
+            name="issued_at",
+            field=models.DateField(
+                blank=True,
+                db_index=True,
+                null=True,
+                verbose_name="дата выдачи",
+            ),
+        ),
+    ]

src/apps/external_data/models.py

@@ -44,7 +44,12 @@ class IndustrialCertificate(UUIDPrimaryKeyMixin, TimestampMixin, models.Model):
     certificate_number = models.CharField(
         _("номер сертификата"), max_length=100, db_index=True
     )
-    issue_date = models.DateField(_("дата выдачи"), null=True, blank=True)
+    issue_date = models.DateField(
+        _("дата выдачи"),
+        null=True,
+        blank=True,
+        db_index=True,
+    )
     expiry_date = models.DateField(_("дата окончания"), null=True, blank=True)
     certificate_file_url = models.TextField(
         _("ссылка на файл сертификата"), blank=True, default=""
@@ -254,7 +259,12 @@ class InformationSecurityRegistryEntry(
         blank=True,
         default="",
     )
-    issued_at = models.DateField(_("дата выдачи"), null=True, blank=True)
+    issued_at = models.DateField(
+        _("дата выдачи"),
+        null=True,
+        blank=True,
+        db_index=True,
+    )
     expires_at = models.DateField(_("дата окончания"), null=True, blank=True)
 
     class Meta:

src/apps/external_data/source_record_export.py

@@ -24,6 +24,7 @@ from apps.external_data.models import (
     BankruptcyProcedure,
     DefenseUnreliableSupplier,
     FinancialReport,
+    FinancialReportLine,
     IndustrialCertificate,
     IndustrialProduct,
     InformationSecurityRegistryEntry,
@@ -34,7 +35,7 @@ from apps.external_data.models import (
 )
 from django.conf import settings
 from django.core.cache import cache
-from django.db.models import QuerySet, prefetch_related_objects
+from django.db.models import Prefetch, Q, QuerySet, prefetch_related_objects
 from django.utils import timezone
 from openpyxl import Workbook
 
@@ -43,7 +44,7 @@ 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
+EXPORT_MANIFEST_VERSION = 2
 CURRENT_EXPORT_MANIFEST_FILE_NAME = "current.json"
 GENERATION_MANIFEST_FILE_NAME = "manifest.json"
 GENERATION_DIRECTORY_NAME = "generations"
@@ -117,6 +118,7 @@ class SourceModelExportSpec:
     payload_aliases: tuple[tuple[str, str], ...] = ()
     payload_export_fields: tuple[str, ...] | None = None
     prefetch_related: tuple[str, ...] = ()
+    export_year_lookup: str | None = None
 
 
 @dataclass(frozen=True)
@@ -153,6 +155,7 @@ SOURCE_GROUP_EXPORT_SPECS: dict[str, SourceGroupExportSpec] = {
                 status_field="status",
                 load_batch_field="load_batch",
                 prefetch_related=("lines",),
+                export_year_lookup="lines__year",
             ),
         ),
     ),
@@ -436,6 +439,7 @@ class SourceRecordExportGeneration:
 
     generation_id: str
     generated_at: str
+    export_year: int
     artifacts: tuple[SourceRecordExportArtifact, ...]
     records_count: int
 
@@ -514,6 +518,7 @@ def build_source_record_export_artifacts(
     generations_directory.mkdir(parents=True, exist_ok=True)
 
     generated_at_datetime = _normalize_generation_datetime(now or timezone.now())
+    export_year = _export_year(generated_at_datetime)
     generation_id = (
         f"{generated_at_datetime.strftime('%Y%m%dT%H%M%SZ')}-{uuid4().hex[:8]}"
     )
@@ -530,6 +535,7 @@ def build_source_record_export_artifacts(
             headers, records_count = _spool_source_group_rows(
                 source_spec=source_spec,
                 output_path=row_spool_path,
+                export_year=export_year,
             )
             source_record_counts[source_group] = records_count
 
@@ -569,6 +575,7 @@ def build_source_record_export_artifacts(
         generation = SourceRecordExportGeneration(
             generation_id=generation_id,
             generated_at=generated_at_datetime.isoformat(),
+            export_year=export_year,
             artifacts=tuple(artifacts),
             records_count=sum(source_record_counts.values()),
         )
@@ -622,6 +629,7 @@ def build_source_records_export_archive(
     source_groups: Sequence[str],
     export_format: str,
     export_directory: str | Path | None = None,
+    requested_at: datetime | None = None,
 ) -> SourceRecordExportArchive:
     """Package selected prepared files without querying external-data tables."""
 
@@ -629,6 +637,13 @@ def build_source_records_export_archive(
     generation = load_current_source_record_export_generation(
         export_directory=root_directory,
     )
+    requested_at_datetime = _normalize_generation_datetime(
+        requested_at or timezone.now()
+    )
+    if generation.export_year != _export_year(requested_at_datetime):
+        raise SourceRecordExportArtifactsUnavailable(
+            "Prepared source-record export belongs to a different calendar year."
+        )
     artifacts_by_key: dict[
         tuple[str, str],
         list[SourceRecordExportArtifact],
@@ -654,10 +669,11 @@ def build_source_records_export_archive(
             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_name=_build_source_records_archive_name(
+            source_groups=source_groups,
+            requested_at=requested_at_datetime,
+        ),
         archive_chunks=_stream_zip_archive(selected_artifacts),
         files_count=len(selected_artifacts),
         generated_at=generation.generated_at,
@@ -671,9 +687,11 @@ def create_source_record_export_download_ticket(
 ) -> SourceRecordExportDownloadTicket:
     """Validate prepared files and cache a short-lived download capability."""
 
+    requested_at = _normalize_generation_datetime(timezone.now())
     package = build_source_records_export_archive(
         source_groups=source_groups,
         export_format=export_format,
+        requested_at=requested_at,
     )
     expires_in = max(
         1,
@@ -685,7 +703,11 @@ def create_source_record_export_download_ticket(
             )
         ),
     )
-    payload = {"sources": list(source_groups), "format": export_format}
+    payload = {
+        "sources": list(source_groups),
+        "format": export_format,
+        "requested_at": requested_at.isoformat(),
+    }
     for _attempt in range(3):
         ticket = secrets.token_urlsafe(32)
         if cache.add(
@@ -718,6 +740,7 @@ def consume_source_record_export_download_ticket(
     try:
         source_groups = payload["sources"]
         export_format = payload["format"]
+        requested_at_value = payload["requested_at"]
         if (
             not isinstance(source_groups, list)
             or not source_groups
@@ -728,14 +751,19 @@ def consume_source_record_export_download_ticket(
             )
             or len(source_groups) != len(set(source_groups))
             or export_format not in EXPORT_FORMATS
+            or not isinstance(requested_at_value, str)
         ):
             raise ValueError
+        requested_at = datetime.fromisoformat(requested_at_value)
+        if timezone.is_naive(requested_at):
+            raise ValueError
     except (KeyError, TypeError, ValueError):
         raise SourceRecordExportTicketInvalid from None
 
     return build_source_records_export_archive(
         source_groups=source_groups,
         export_format=export_format,
+        requested_at=requested_at,
     )
 
 
@@ -788,6 +816,24 @@ def _normalize_generation_datetime(value: datetime) -> datetime:
     return value.astimezone(UTC)
 
 
+def _export_year(value: datetime) -> int:
+    """Return the calendar year in the configured service timezone."""
+    return timezone.localtime(value).year
+
+
+def _build_source_records_archive_name(
+    *,
+    source_groups: Sequence[str],
+    requested_at: datetime,
+) -> str:
+    source_name = "__".join(
+        SOURCE_GROUP_EXPORT_SPECS[source_group].file_stem
+        for source_group in source_groups
+    )
+    timestamp = timezone.localtime(requested_at).strftime("%Y%m%d_%H%M%S")
+    return f"{source_name}_{timestamp}.zip"
+
+
 def _source_group_export_formats(source_group: str) -> tuple[str, ...]:
     if source_group == FINANCIAL_SOURCE_GROUP:
         return (EXPORT_FORMAT_JSON,)
@@ -808,25 +854,59 @@ 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(
+def _source_model_queryset(
+    model_spec: SourceModelExportSpec,
+    *,
+    export_year: int,
+) -> QuerySet:
+    queryset = cast(
         QuerySet,
         model_spec.model.objects.select_related("organization").order_by(),
     )
+    if model_spec.export_year_lookup:
+        return queryset.filter(
+            **{model_spec.export_year_lookup: export_year}
+        ).distinct()
+    if model_spec.record_date_field:
+        return queryset.filter(
+            Q(**{f"{model_spec.record_date_field}__year": export_year})
+            | Q(
+                **{
+                    f"{model_spec.record_date_field}__isnull": True,
+                    "created_at__year": export_year,
+                }
+            )
+        )
+    return queryset.filter(created_at__year=export_year)
 
 
 def _iter_source_model_records(
     model_spec: SourceModelExportSpec,
+    *,
+    export_year: int,
 ) -> Iterator[Any]:
-    iterator = _source_model_queryset(model_spec).iterator(
-        chunk_size=SOURCE_RECORD_EXPORT_ITERATOR_CHUNK_SIZE
-    )
+    iterator = _source_model_queryset(
+        model_spec,
+        export_year=export_year,
+    ).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)
+            prefetches = [
+                Prefetch(
+                    related_name,
+                    queryset=FinancialReportLine.objects.filter(
+                        year=export_year
+                    ).order_by(),
+                )
+                if related_name == "lines"
+                and model_spec.export_year_lookup == "lines__year"
+                else related_name
+                for related_name in model_spec.prefetch_related
+            ]
+            prefetch_related_objects(batch, *prefetches)
         yield from batch
 
 
@@ -851,13 +931,17 @@ def _spool_source_group_rows(
     *,
     source_spec: SourceGroupExportSpec,
     output_path: Path,
+    export_year: int,
 ) -> 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):
+            for record in _iter_source_model_records(
+                model_spec,
+                export_year=export_year,
+            ):
                 row = _build_record_row(
                     record,
                     source_spec=source_spec,
@@ -1149,6 +1233,7 @@ def _generation_manifest_payload(
         "version": EXPORT_MANIFEST_VERSION,
         "generation_id": generation.generation_id,
         "generated_at": generation.generated_at,
+        "export_year": generation.export_year,
         "records_count": generation.records_count,
         "artifacts_count": generation.artifacts_count,
         "files_count": generation.files_count,
@@ -1180,6 +1265,9 @@ def _generation_from_manifest(
         generation_id = str(payload["generation_id"])
         generated_at = str(payload["generated_at"])
         datetime.fromisoformat(generated_at)
+        export_year = int(payload["export_year"])
+        if not 1 <= export_year <= 9999:
+            raise ValueError("Source-record export year is invalid.")
         records_count = int(payload["records_count"])
         artifact_payloads = payload["artifacts"]
         if not isinstance(artifact_payloads, list):
@@ -1234,6 +1322,7 @@ def _generation_from_manifest(
     return SourceRecordExportGeneration(
         generation_id=generation_id,
         generated_at=generated_at,
+        export_year=export_year,
         artifacts=tuple(artifacts),
         records_count=records_count,
     )

src/apps/external_data/tasks.py

@@ -35,6 +35,7 @@ def refresh_source_record_export_artifacts(self) -> dict:  # noqa: ARG001
             "status": "success",
             "generation_id": generation.generation_id,
             "generated_at": generation.generated_at,
+            "export_year": generation.export_year,
             "artifacts_count": generation.artifacts_count,
             "files_count": generation.files_count,
             "records_count": generation.records_count,