mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-08-02 12:58:30 +00:00
104 lines
3.1 KiB
Python
104 lines
3.1 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 (
|
|
ExportSheet,
|
|
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:
|
|
named_sheets: list[tuple[str, ExportSheet]] = []
|
|
row_count = 0
|
|
for scenario_id in scenario_ids:
|
|
rows = self._rows.rows_for(
|
|
portfolio_id=portfolio_id,
|
|
scenario_id=scenario_id,
|
|
property_ids=property_ids,
|
|
)
|
|
sheet = shape_scenario_sheet(rows)
|
|
named_sheets.append((self._scenarios.name_for(scenario_id), sheet))
|
|
row_count += len(sheet.rows)
|
|
|
|
data = render_workbook(named_sheets)
|
|
key = f"{self._key_prefix}/{export_name}.xlsx"
|
|
self._exports.put_object(key, data)
|
|
url = self._exports.generate_presigned_url(key, self._url_ttl_seconds)
|
|
|
|
return ScenarioExportResult(
|
|
presigned_url=url,
|
|
export_s3_key=key,
|
|
sheet_count=len(named_sheets),
|
|
row_count=row_count,
|
|
)
|