"""The Bulk Document Download orchestrator (ADR-0060). Ties the pieces together for one run: gather a property set's uploaded files (by `landlord_property_id`), enrich each with its display address, lay them out with the Download Package plan, stream the chosen files into a single ZIP, upload it, mint a presigned URL, and email the requester. Best-effort — skipped rows are reported, not fatal; a wholly-empty selection is a recorded failure (ADR-0060). """ from __future__ import annotations import os import tempfile import zipfile from dataclasses import dataclass from datetime import datetime from typing import Optional, Protocol, Sequence, cast from domain.bulk_document_download.package_plan import ( ResolvedDocument, SkippedDocument, plan_download_package, ) from domain.tasks.subtasks import SubTaskFailure from infrastructure.postgres.uploaded_file_table import UploadedFile from infrastructure.s3.s3_client import S3Client from repositories.email.email_sender import EmailSender class UploadedFileReader(Protocol): def by_landlord_property_ids( self, landlord_property_ids: Sequence[str] ) -> list[UploadedFile]: ... class AddressResolver(Protocol): def address_for(self, landlord_property_id: str) -> str: ... class DocumentDownloader(Protocol): def download(self, bucket: str, key: str, dest_path: str) -> None: ... @dataclass(frozen=True) class DownloadPackageResult: """What a completed run reports (lands in `sub_task.outputs`, ADR-0060).""" presigned_url: str package_s3_key: str included: int skipped: tuple[SkippedDocument, ...] class BulkDocumentDownloadOrchestrator: def __init__( self, *, uploaded_files: UploadedFileReader, addresses: AddressResolver, documents: DocumentDownloader, packages: S3Client, email: EmailSender, url_ttl_seconds: int, package_key_prefix: str = "bulk-downloads", ) -> None: self._uploaded_files = uploaded_files self._addresses = addresses self._documents = documents self._packages = packages self._email = email self._url_ttl_seconds = url_ttl_seconds self._package_key_prefix = package_key_prefix def run( self, *, landlord_property_ids: Sequence[str], recipient_email: str, package_name: str, ) -> DownloadPackageResult: files = self._uploaded_files.by_landlord_property_ids(landlord_property_ids) resolved = [self._resolve(file) for file in files] plan = plan_download_package([r for r in resolved if r is not None]) skipped: list[SkippedDocument] = list(plan.skipped) if not plan.entries: raise self._empty_failure(skipped) # Stream to a /tmp file and multipart-upload it (ADR-0060): a Download # Package can be several GB, so it is never held whole in memory. Each # source file is streamed to its own temp file, added to the on-disk ZIP, # then deleted; the finished archive is streamed up from disk. Best-effort # (ADR-0060): a file that can't be read — deleted, or in a bucket the role # can't reach — becomes a SkippedDocument, it does not fail the package. key = f"{self._package_key_prefix}/{package_name}.zip" included = 0 with tempfile.TemporaryDirectory() as tmp_dir: zip_path = os.path.join(tmp_dir, "package.zip") with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive: for index, entry in enumerate(plan.entries): member_path = os.path.join(tmp_dir, f"member-{index}") try: self._documents.download( entry.s3_bucket, entry.s3_key, member_path ) except Exception: skipped.append( SkippedDocument( landlord_property_id=entry.landlord_property_id, s3_key=entry.s3_key, reason="unreadable", ) ) continue archive.write(member_path, arcname=entry.zip_path) os.remove(member_path) included += 1 if included == 0: # Every file that resolved failed to read — no package to send. raise self._empty_failure(skipped) self._packages.upload_file(zip_path, key) url = self._packages.generate_presigned_url(key, self._url_ttl_seconds) self._email.send( to=recipient_email, subject="Your document download is ready", body=( f"Your document download is ready. It contains " f"{included} document(s).\n\n" f"Download it here (link valid for " f"{self._url_ttl_seconds // 60} minutes):\n{url}" ), ) return DownloadPackageResult( presigned_url=url, package_s3_key=key, included=included, skipped=tuple(skipped), ) def _empty_failure(self, skipped: list[SkippedDocument]) -> SubTaskFailure: """A recorded failure when the selection yields no readable documents (ADR-0060) — never an empty ZIP emailed to the user. The skip detail rides on the failure.""" return SubTaskFailure( "no documents could be packaged for the selection", details={ "skipped": [ { "landlord_property_id": s.landlord_property_id, "s3_key": s.s3_key, "reason": s.reason, } for s in skipped ] }, ) def _resolve(self, file: UploadedFile) -> Optional[ResolvedDocument]: if file.landlord_property_id is None: return None return ResolvedDocument( landlord_property_id=file.landlord_property_id, address=self._addresses.address_for(file.landlord_property_id), document_type=file.file_type, s3_bucket=file.s3_file_bucket, s3_key=file.s3_file_key, # TODO: the cast is only needed because UploadedFile types # s3_upload_timestamp as `object` (though it is NOT NULL datetime in # the DB). Type that column `datetime` on the table so callers don't # each cast — see PR #1498 review. uploaded_at=cast(datetime, file.s3_upload_timestamp), )