156 lines
5.7 KiB
Python
156 lines
5.7 KiB
Python
"""ARR-owned execution of the frozen Opera daily processor."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Dict, Mapping, Optional
|
|
|
|
from arr_ingestion.contracts import IngestionError
|
|
from arr_ingestion.validation import ProcessorPolicy, strict_json_file
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocalProcessingOutput:
|
|
status: str
|
|
business_date: Optional[date]
|
|
artifacts: Mapping[str, Path]
|
|
exit_code: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocalDailyProcessor:
|
|
"""Run one isolated deterministic process and expose only its artifact paths."""
|
|
|
|
policy: ProcessorPolicy
|
|
timeout_seconds: int = 240
|
|
|
|
def __post_init__(self) -> None:
|
|
if (
|
|
not isinstance(self.timeout_seconds, int)
|
|
or isinstance(self.timeout_seconds, bool)
|
|
or not 10 <= self.timeout_seconds <= 900
|
|
):
|
|
raise ValueError("local processor timeout is invalid")
|
|
if not (self.policy.skill_root / "scripts" / "process_daily.py").is_file():
|
|
raise ValueError("ARR daily processor is unavailable")
|
|
|
|
def run(self, source_xml: Path, output_dir: Path) -> LocalProcessingOutput:
|
|
if not source_xml.is_file():
|
|
raise IngestionError("SOURCE_NOT_FOUND", "registered source XML is unavailable")
|
|
output_dir.mkdir(parents=True, exist_ok=False, mode=0o700)
|
|
result_path = output_dir / "result.json"
|
|
structured_path = output_dir / "structured-result.json"
|
|
command = [
|
|
self.policy.python_binary,
|
|
str((self.policy.skill_root / "scripts" / "process_daily.py").resolve()),
|
|
"--xml",
|
|
str(source_xml.resolve()),
|
|
"--output-dir",
|
|
str(output_dir.resolve()),
|
|
"--result-json",
|
|
str(result_path.resolve()),
|
|
"--structured-result-json",
|
|
str(structured_path.resolve()),
|
|
]
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
timeout=self.timeout_seconds,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
raise IngestionError(
|
|
"PROCESSOR_TIMEOUT",
|
|
"deterministic XML processing exceeded its time limit",
|
|
) from None
|
|
except OSError:
|
|
raise IngestionError(
|
|
"PROCESSOR_UNAVAILABLE",
|
|
"deterministic XML processing could not start",
|
|
) from None
|
|
|
|
result = strict_json_file(result_path, "processor result")
|
|
structured = strict_json_file(structured_path, "structured processor result")
|
|
status = result.get("status")
|
|
if (
|
|
status not in {"success", "failed"}
|
|
or structured.get("status") != status
|
|
or (completed.returncode == 0) != (status == "success")
|
|
):
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID",
|
|
"processor exit status and result artifacts do not agree",
|
|
)
|
|
|
|
raw_date = result.get("business_date")
|
|
business_date: Optional[date]
|
|
if raw_date is None:
|
|
business_date = None
|
|
elif isinstance(raw_date, str):
|
|
try:
|
|
business_date = date.fromisoformat(raw_date)
|
|
except ValueError:
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "processor business date is invalid"
|
|
) from None
|
|
else:
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "processor business date is invalid"
|
|
)
|
|
|
|
outputs = result.get("outputs")
|
|
if not isinstance(outputs, dict):
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "processor output manifest is invalid"
|
|
)
|
|
artifacts: Dict[str, Path] = {
|
|
"result_json": result_path,
|
|
"structured_result_json": structured_path,
|
|
}
|
|
if status == "success":
|
|
artifacts["daily_report"] = self._output_path(
|
|
output_dir, outputs.get("daily_report"), ".xlsx"
|
|
)
|
|
if outputs.get("exception_report") is not None or business_date is None:
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "successful processor output is incomplete"
|
|
)
|
|
else:
|
|
artifacts["exception_report"] = self._output_path(
|
|
output_dir, outputs.get("exception_report"), ".xlsx"
|
|
)
|
|
if outputs.get("daily_report") is not None:
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "failed processor output is inconsistent"
|
|
)
|
|
return LocalProcessingOutput(
|
|
status=str(status),
|
|
business_date=business_date,
|
|
artifacts=artifacts,
|
|
exit_code=completed.returncode,
|
|
)
|
|
|
|
@staticmethod
|
|
def _output_path(output_dir: Path, value: object, suffix: str) -> Path:
|
|
if (
|
|
not isinstance(value, str)
|
|
or not value
|
|
or Path(value).name != value
|
|
or not value.lower().endswith(suffix)
|
|
):
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "processor output filename is invalid"
|
|
)
|
|
candidate = (output_dir / value).resolve()
|
|
if candidate.parent != output_dir.resolve() or not candidate.is_file():
|
|
raise IngestionError(
|
|
"PROCESSOR_RESULT_INVALID", "processor output artifact is unavailable"
|
|
)
|
|
return candidate
|