337 lines
11 KiB
Python
337 lines
11 KiB
Python
"""Company-isolated orchestration and privacy-minimized batch results."""
|
|
|
|
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, Sequence, Tuple
|
|
|
|
from company_reports.contracts import (
|
|
COMPANY_NAMES,
|
|
RESULT_SCHEMA_VERSION,
|
|
CompanyReport,
|
|
ErrorCode,
|
|
ReportProblem,
|
|
)
|
|
from company_reports.core import build_all_company_reports, validate_as_of_date
|
|
from company_reports.publishing import (
|
|
BuildError,
|
|
BuiltWorkbook,
|
|
PublicationError,
|
|
PublicationOutcome,
|
|
private_staging_directory,
|
|
)
|
|
from company_reports.repository import ReportRepository, RepositoryError, ReservedReport
|
|
|
|
|
|
BUSINESS_ERROR_CODES = {
|
|
ErrorCode.SOURCE_VERSION_MISSING,
|
|
ErrorCode.GROUP_CODE_MISSING,
|
|
ErrorCode.GROUP_CODE_NOT_FOUND,
|
|
ErrorCode.BOOKING_PARSE_FAILED,
|
|
ErrorCode.ROOM_ITEMS_MISSING,
|
|
ErrorCode.STAY_DATE_INVALID,
|
|
ErrorCode.NIGHTS_CONFLICT,
|
|
ErrorCode.TOTAL_PRICE_INVALID,
|
|
}
|
|
|
|
|
|
class WorkbookBuilder(Protocol):
|
|
def build(self, report: CompanyReport, work_dir: Path) -> BuiltWorkbook:
|
|
...
|
|
|
|
|
|
class ReportPublisher(Protocol):
|
|
def publish(
|
|
self,
|
|
report: CompanyReport,
|
|
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
|
|
companies: Tuple[str, ...] = COMPANY_NAMES
|
|
|
|
def validate(self) -> None:
|
|
validate_as_of_date(self.report_year, self.report_month, self.as_of_date)
|
|
if not self.companies or len(set(self.companies)) != len(self.companies):
|
|
raise ValueError("company selection is empty or contains duplicates")
|
|
unsupported = [company for company in self.companies if company not in COMPANY_NAMES]
|
|
if unsupported:
|
|
raise ValueError("company selection is unsupported")
|
|
|
|
|
|
def _safe_problem(problem: ReportProblem) -> Dict[str, Any]:
|
|
return {
|
|
"code": problem.code,
|
|
"stage": problem.stage,
|
|
"period": problem.period,
|
|
"record_ids": list(problem.record_ids),
|
|
}
|
|
|
|
|
|
def _runtime_problem(code: str, period: str, stage: str) -> Dict[str, Any]:
|
|
return {
|
|
"code": code,
|
|
"stage": stage,
|
|
"period": period,
|
|
"record_ids": [],
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CompanyRunResult:
|
|
company: str
|
|
status: str
|
|
row_count: int
|
|
period_row_counts: Mapping[str, int]
|
|
report_version_id: Optional[int] = None
|
|
version_no: Optional[int] = None
|
|
artifact: Optional[Mapping[str, Any]] = None
|
|
warnings: Tuple[Mapping[str, Any], ...] = tuple()
|
|
errors: Tuple[Mapping[str, Any], ...] = tuple()
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {
|
|
"company": self.company,
|
|
"status": self.status,
|
|
"row_count": self.row_count,
|
|
"period_row_counts": dict(self.period_row_counts),
|
|
"warnings": [dict(warning) for warning in self.warnings],
|
|
"errors": [dict(error) for error in self.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)
|
|
return payload
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BatchRunResult:
|
|
request: RunRequest
|
|
status: str
|
|
companies: Tuple[CompanyRunResult, ...]
|
|
|
|
@property
|
|
def exit_code(self) -> int:
|
|
if self.status == "success":
|
|
return 0
|
|
if self.status == "partial_failure":
|
|
return 2
|
|
codes = {
|
|
str(error.get("code", ""))
|
|
for company in self.companies
|
|
for error in company.errors
|
|
}
|
|
return 2 if codes and codes.issubset(BUSINESS_ERROR_CODES) else 4
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"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(),
|
|
"requested_companies": list(self.request.companies),
|
|
"companies": [company.to_dict() for company in self.companies],
|
|
}
|
|
|
|
|
|
class CompanyReportService:
|
|
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 _status(results: Sequence[CompanyRunResult]) -> str:
|
|
successful = sum(result.status == "success" for result in results)
|
|
if successful == len(results):
|
|
return "success"
|
|
if successful == 0:
|
|
return "failed"
|
|
return "partial_failure"
|
|
|
|
@staticmethod
|
|
def _failed(
|
|
company: str,
|
|
code: str,
|
|
period: str,
|
|
stage: str,
|
|
row_count: int = 0,
|
|
period_counts: Optional[Mapping[str, int]] = None,
|
|
) -> CompanyRunResult:
|
|
return CompanyRunResult(
|
|
company=company,
|
|
status="failed",
|
|
row_count=row_count,
|
|
period_row_counts=period_counts or {},
|
|
errors=(_runtime_problem(code, period, stage),),
|
|
)
|
|
|
|
def run(self, request: RunRequest) -> BatchRunResult:
|
|
request.validate()
|
|
period = f"{request.report_year:04d}-{request.report_month:02d}"
|
|
try:
|
|
snapshot = self._repository.load_snapshot(
|
|
request.report_year,
|
|
request.report_month,
|
|
request.as_of_date,
|
|
)
|
|
except RepositoryError as error:
|
|
failed = tuple(
|
|
self._failed(company, error.code, period, "source")
|
|
for company in request.companies
|
|
)
|
|
return BatchRunResult(request, "failed", failed)
|
|
|
|
reports = build_all_company_reports(
|
|
request.report_year,
|
|
request.report_month,
|
|
request.as_of_date,
|
|
snapshot,
|
|
request.companies,
|
|
)
|
|
results = []
|
|
for report in reports:
|
|
period_counts = {period.key: len(period.rows) for period in report.periods}
|
|
if not report.valid:
|
|
results.append(
|
|
CompanyRunResult(
|
|
company=report.company,
|
|
status="failed",
|
|
row_count=report.row_count,
|
|
period_row_counts=period_counts,
|
|
warnings=tuple(_safe_problem(item) for item in report.warnings),
|
|
errors=tuple(_safe_problem(item) for item in report.errors),
|
|
)
|
|
)
|
|
continue
|
|
|
|
reservation: Optional[ReservedReport] = None
|
|
try:
|
|
reservation = self._repository.reserve_report(report)
|
|
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,
|
|
)
|
|
results.append(
|
|
CompanyRunResult(
|
|
company=report.company,
|
|
status="success",
|
|
row_count=report.row_count,
|
|
period_row_counts=period_counts,
|
|
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,
|
|
"byte_size": outcome.artifact.byte_size,
|
|
"mime_type": outcome.artifact.mime_type,
|
|
"storage_provider": outcome.artifact.storage_provider,
|
|
"bucket_alias": outcome.artifact.bucket_alias,
|
|
"semantic_sha256": str(
|
|
built.summary.get("semantic_sha256", "")
|
|
),
|
|
},
|
|
warnings=tuple(_safe_problem(item) for item in report.warnings),
|
|
)
|
|
)
|
|
except BuildError as error:
|
|
if reservation is not None:
|
|
try:
|
|
self._repository.mark_failed(
|
|
reservation, error.code, error.safe_message
|
|
)
|
|
except Exception:
|
|
pass
|
|
results.append(
|
|
self._failed(
|
|
report.company,
|
|
error.code,
|
|
period,
|
|
"xlsx",
|
|
report.row_count,
|
|
period_counts,
|
|
)
|
|
)
|
|
except (PublicationError, RepositoryError) as error:
|
|
results.append(
|
|
self._failed(
|
|
report.company,
|
|
error.code,
|
|
period,
|
|
"publish",
|
|
report.row_count,
|
|
period_counts,
|
|
)
|
|
)
|
|
except Exception:
|
|
if reservation is not None:
|
|
try:
|
|
self._repository.mark_failed(
|
|
reservation,
|
|
ErrorCode.INTERNAL_ERROR,
|
|
"company report processing failed",
|
|
)
|
|
except Exception:
|
|
pass
|
|
results.append(
|
|
self._failed(
|
|
report.company,
|
|
ErrorCode.INTERNAL_ERROR,
|
|
period,
|
|
"internal",
|
|
report.row_count,
|
|
period_counts,
|
|
)
|
|
)
|
|
|
|
company_results = tuple(results)
|
|
return BatchRunResult(request, self._status(company_results), company_results)
|
|
|
|
|
|
def write_batch_result(path: Path, result: BatchRunResult) -> 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)
|