247 lines
8.3 KiB
Python
247 lines
8.3 KiB
Python
"""Orchestration and privacy-minimized results for monthly channel reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Mapping, Optional, Protocol, Tuple
|
|
|
|
from monthly_reports.contracts import (
|
|
RESULT_SCHEMA_VERSION,
|
|
ErrorCode,
|
|
MonthlyReport,
|
|
MonthlyReportError,
|
|
)
|
|
from monthly_reports.core import build_monthly_report, validate_as_of
|
|
from monthly_reports.publishing import (
|
|
BuildError,
|
|
BuiltWorkbook,
|
|
PublicationError,
|
|
PublicationOutcome,
|
|
private_staging_directory,
|
|
)
|
|
from monthly_reports.repository import ReportRepository, RepositoryError, ReservedReport
|
|
|
|
|
|
class WorkbookBuilder(Protocol):
|
|
def build(self, report: MonthlyReport, work_dir: Path) -> BuiltWorkbook:
|
|
...
|
|
|
|
|
|
class ReportPublisher(Protocol):
|
|
def publish(
|
|
self,
|
|
report: MonthlyReport,
|
|
reservation: ReservedReport,
|
|
built: BuiltWorkbook,
|
|
repository: ReportRepository,
|
|
work_dir: Path,
|
|
) -> PublicationOutcome:
|
|
...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunRequest:
|
|
report_year: int
|
|
report_month: int
|
|
as_of_date: date
|
|
|
|
def validate(self) -> None:
|
|
validate_as_of(self.report_year, self.report_month, self.as_of_date)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunResult:
|
|
request: RunRequest
|
|
status: str
|
|
row_count: int = 0
|
|
channel_manifest: Tuple[Tuple[str, int, int], ...] = tuple()
|
|
report_version_id: Optional[int] = None
|
|
version_no: Optional[int] = None
|
|
artifact: Optional[Mapping[str, Any]] = None
|
|
error_code: Optional[str] = None
|
|
error_stage: Optional[str] = None
|
|
|
|
@property
|
|
def exit_code(self) -> int:
|
|
if self.status == "success":
|
|
return 0
|
|
if self.error_code in {
|
|
ErrorCode.SOURCE_INVALID,
|
|
ErrorCode.SOURCE_SNAPSHOT_STALE,
|
|
}:
|
|
return 2
|
|
return 4
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {
|
|
"schema_version": RESULT_SCHEMA_VERSION,
|
|
"status": self.status,
|
|
"report_year": self.request.report_year,
|
|
"report_month": self.request.report_month,
|
|
"as_of_date": self.request.as_of_date.isoformat(),
|
|
"row_count": self.row_count,
|
|
"channel_manifest": [
|
|
{
|
|
"worksheet": worksheet,
|
|
"worksheet_order": order,
|
|
"row_count": row_count,
|
|
}
|
|
for worksheet, order, row_count in self.channel_manifest
|
|
],
|
|
"warnings": [],
|
|
"errors": [],
|
|
}
|
|
if self.report_version_id is not None:
|
|
payload["report_version_id"] = self.report_version_id
|
|
if self.version_no is not None:
|
|
payload["version_no"] = self.version_no
|
|
if self.artifact is not None:
|
|
payload["artifact"] = dict(self.artifact)
|
|
if self.error_code is not None:
|
|
payload["errors"] = [
|
|
{
|
|
"code": self.error_code,
|
|
"stage": self.error_stage or "internal",
|
|
}
|
|
]
|
|
return payload
|
|
|
|
|
|
class MonthlyReportService:
|
|
def __init__(
|
|
self,
|
|
repository: ReportRepository,
|
|
builder: WorkbookBuilder,
|
|
publisher: ReportPublisher,
|
|
staging_root: Path,
|
|
) -> None:
|
|
self._repository = repository
|
|
self._builder = builder
|
|
self._publisher = publisher
|
|
self._staging_root = staging_root
|
|
|
|
@staticmethod
|
|
def _failed(
|
|
request: RunRequest,
|
|
code: str,
|
|
stage: str,
|
|
report: Optional[MonthlyReport] = None,
|
|
) -> RunResult:
|
|
return RunResult(
|
|
request=request,
|
|
status="failed",
|
|
row_count=report.row_count if report else 0,
|
|
channel_manifest=report.channel_manifest if report else tuple(),
|
|
error_code=code,
|
|
error_stage=stage,
|
|
)
|
|
|
|
def run(self, request: RunRequest) -> RunResult:
|
|
request.validate()
|
|
try:
|
|
snapshot = self._repository.load_snapshot(
|
|
request.report_year,
|
|
request.report_month,
|
|
request.as_of_date,
|
|
)
|
|
except RepositoryError as error:
|
|
return self._failed(request, error.code, "source")
|
|
try:
|
|
report = build_monthly_report(
|
|
request.report_year,
|
|
request.report_month,
|
|
request.as_of_date,
|
|
snapshot,
|
|
)
|
|
except MonthlyReportError as error:
|
|
return self._failed(request, error.code, "source")
|
|
|
|
reservation: Optional[ReservedReport] = None
|
|
try:
|
|
reservation = self._repository.reserve_report(report)
|
|
if reservation.already_published:
|
|
artifact = reservation.published_artifact
|
|
if artifact is None:
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"published monthly artifact identity is missing",
|
|
)
|
|
return RunResult(
|
|
request=request,
|
|
status="success",
|
|
row_count=report.row_count,
|
|
channel_manifest=report.channel_manifest,
|
|
report_version_id=reservation.report_version_id,
|
|
version_no=reservation.version_no,
|
|
artifact={
|
|
"filename": artifact.original_filename,
|
|
"storage_key": artifact.storage_key,
|
|
"sha256": artifact.sha256,
|
|
},
|
|
)
|
|
with private_staging_directory(self._staging_root) as temp_dir:
|
|
work_dir = Path(temp_dir)
|
|
built = self._builder.build(report, work_dir)
|
|
outcome = self._publisher.publish(
|
|
report,
|
|
reservation,
|
|
built,
|
|
self._repository,
|
|
work_dir,
|
|
)
|
|
return RunResult(
|
|
request=request,
|
|
status="success",
|
|
row_count=report.row_count,
|
|
channel_manifest=report.channel_manifest,
|
|
report_version_id=reservation.report_version_id,
|
|
version_no=reservation.version_no,
|
|
artifact={
|
|
"filename": report.filename,
|
|
"storage_key": outcome.artifact.storage_key,
|
|
"sha256": outcome.artifact.sha256,
|
|
"semantic_sha256": str(built.summary.get("semantic_sha256", "")),
|
|
},
|
|
)
|
|
except BuildError as error:
|
|
if reservation is not None:
|
|
try:
|
|
self._repository.mark_failed(reservation, error.code, error.safe_message)
|
|
except Exception:
|
|
pass
|
|
return self._failed(request, error.code, "xlsx", report)
|
|
except (PublicationError, RepositoryError) as error:
|
|
return self._failed(request, error.code, "publish", report)
|
|
except Exception:
|
|
if reservation is not None:
|
|
try:
|
|
self._repository.mark_failed(
|
|
reservation,
|
|
ErrorCode.INTERNAL_ERROR,
|
|
"monthly report processing failed",
|
|
)
|
|
except Exception:
|
|
pass
|
|
return self._failed(request, ErrorCode.INTERNAL_ERROR, "internal", report)
|
|
|
|
|
|
def write_run_result(path: Path, result: RunResult) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temporary = path.parent / f".{path.name}.tmp-{uuid.uuid4().hex}"
|
|
descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
|
json.dump(result.to_dict(), handle, ensure_ascii=False, sort_keys=True, indent=2)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
os.chmod(path, 0o600)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|