mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
"""The Scenario Export orchestrator (ADR-0065).
|
|
|
|
Ties one export together: for each requested Scenario, read its Properties'
|
|
persisted rows, shape them into a sheet, render the sheet-per-scenario branded
|
|
workbook, upload it to the exports bucket, mint a presigned URL, and email the
|
|
requester. Best-effort email (the link is also recorded on ``sub_task.outputs``);
|
|
a selection that yields no rows for any Scenario is a recorded failure — never an
|
|
empty workbook emailed to the user.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Protocol, Sequence
|
|
|
|
from domain.scenario_export.scenario_sheet import (
|
|
PropertyScenarioData,
|
|
shape_scenario_sheet,
|
|
)
|
|
from domain.tasks.subtasks import SubTaskFailure
|
|
from infrastructure.s3.s3_client import S3Client
|
|
from infrastructure.xlsx.scenario_export_workbook import render_workbook
|
|
from repositories.email.email_sender import EmailSender
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ExportRowReader(Protocol):
|
|
def rows_for(
|
|
self,
|
|
*,
|
|
portfolio_id: int,
|
|
scenario_id: int,
|
|
property_ids: Sequence[int],
|
|
) -> list[PropertyScenarioData]: ...
|
|
|
|
|
|
class ScenarioNames(Protocol):
|
|
def name_for(self, scenario_id: int) -> str: ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScenarioExportResult:
|
|
"""What a completed export reports (lands on ``sub_task.outputs``)."""
|
|
|
|
presigned_url: str
|
|
export_s3_key: str
|
|
sheet_count: int
|
|
row_count: int
|
|
|
|
|
|
class ScenarioExportOrchestrator:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
rows: ExportRowReader,
|
|
scenarios: ScenarioNames,
|
|
exports: S3Client,
|
|
email: EmailSender,
|
|
url_ttl_seconds: int,
|
|
key_prefix: str = "scenario-exports",
|
|
) -> None:
|
|
self._rows = rows
|
|
self._scenarios = scenarios
|
|
self._exports = exports
|
|
self._email = email
|
|
self._url_ttl_seconds = url_ttl_seconds
|
|
self._key_prefix = key_prefix
|
|
|
|
def run(
|
|
self,
|
|
*,
|
|
portfolio_id: int,
|
|
scenario_ids: Sequence[int],
|
|
property_ids: Sequence[int],
|
|
recipient_email: str,
|
|
export_name: str,
|
|
) -> ScenarioExportResult:
|
|
raise NotImplementedError
|