"""Programmatic XML upload, processing, validation, and ingestion boundary.""" from __future__ import annotations import hashlib import json import os import tempfile import uuid from dataclasses import dataclass from pathlib import Path from typing import Any, Dict from arr_ingestion.contracts import ( ARTIFACT_ROLES, DELIVERY_SCHEMA_VERSION, RESULT_SCHEMA_VERSION, ArtifactRef, DeliveryEnvelope, IngestionError, ) from arr_ingestion.repository import IngestionRepository, JobRegistration from arr_ingestion.service import IngestionService from arr_processing.local import LocalDailyProcessor from arr_storage.store import ManagedObjectStore, StoredObject from arr_web.contracts import PortalError, validate_upload_filename, validate_xml_payload @dataclass class ProgrammaticUploadCoordinator: """Own the complete upload-to-terminal transaction without Agent or MCP.""" object_store: ManagedObjectStore ingestion_repository: IngestionRepository ingestion_service: IngestionService processor: LocalDailyProcessor processor_version: str rule_set_sha256: str def submit(self, original_filename: str, payload: bytes) -> Dict[str, Any]: uploaded_filename = validate_upload_filename(original_filename) validate_xml_payload(payload) job_id = "arrjob-" + uuid.uuid4().hex attempt_no = 1 registered = False terminal = False try: with tempfile.TemporaryDirectory(prefix="arr-programmatic-") as temporary: root = Path(temporary) upload_path = root / "upload" / "source.xml" self._write_private(upload_path, payload) source = self.object_store.upload_committed( job_id=job_id, attempt_no=attempt_no, role="source_xml", source=upload_path, original_filename="source.xml", ) self.ingestion_repository.register_job( JobRegistration( job_id=job_id, source=source.to_artifact_ref(), processor_version=self.processor_version, rule_set_sha256=self.rule_set_sha256, attempt_no=attempt_no, idempotency_key=self._idempotency_key( job_id, attempt_no, source ), uploaded_filename=uploaded_filename, ) ) registered = True self.ingestion_repository.mark_running(job_id, attempt_no) materialized_source = root / "processor-input" / "source.xml" self.object_store.materialize( source.object_key, materialized_source, source.byte_size, ) processed = self.processor.run( materialized_source, root / "processor-output", ) stored_outputs: Dict[str, StoredObject] = {} for role, path in processed.artifacts.items(): stored_outputs[role] = self.object_store.upload_committed( job_id=job_id, attempt_no=attempt_no, role=role, source=path, original_filename=path.name, ) envelope = self._delivery( job_id=job_id, attempt_no=attempt_no, source=source, outputs=stored_outputs, status=processed.status, business_date=( processed.business_date.isoformat() if processed.business_date is not None else None ), ) raw_envelope = ( json.dumps( envelope.to_dict(), ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), ) + "\n" ).encode("utf-8") outcome = self.ingestion_service.ingest(raw_envelope) terminal = True return { "job_id": job_id, "status": { "recorded_failure": "failed", "recorded_review": "needs_review", }.get(outcome.status, "succeeded"), "ingestion_status": outcome.status, "attempt_no": attempt_no, "business_date": ( outcome.business_date.isoformat() if outcome.business_date is not None else None ), "daily_version_id": outcome.daily_version_id, "version_no": outcome.version_no, "source_sha256": source.sha256, "source_byte_size": source.byte_size, "review_case_id": outcome.review_case_id, "review_revision": outcome.review_revision, "review_completed_items": outcome.review_completed_items, "review_total_items": outcome.review_total_items, } except PortalError: raise except IngestionError as error: if registered and not terminal: self._best_effort_failure(job_id, attempt_no, error.code) retryable = ( error.retryable or error.code.startswith("DATABASE_") or error.code in {"OBJECT_STORE_UNAVAILABLE", "PROCESSOR_UNAVAILABLE"} ) raise PortalError( error.code, error.safe_message, 503 if retryable else 422, ) from None except Exception: if registered and not terminal: self._best_effort_failure( job_id, attempt_no, "PROGRAMMATIC_PROCESSING_FAILED" ) raise PortalError( "PROGRAMMATIC_PROCESSING_FAILED", "文件已安全接收，但程序化处理未能完成", 503, ) from None def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]: try: return self.ingestion_repository.get_price_review(job_id, limit, offset) except IngestionError as error: self._raise_review_portal_error(error) def update_price_review_item( self, job_id: str, item_id: int, case_id: str, revision: int, real_price: str, actor_username: str, ) -> Dict[str, Any]: try: return self.ingestion_repository.update_price_review_item( job_id, item_id, case_id, revision, real_price, actor_username, ) except IngestionError as error: self._raise_review_portal_error(error) def finalize_price_review( self, job_id: str, case_id: str, revision: int, actor_username: str, ) -> Dict[str, Any]: idempotency_key = hashlib.sha256(uuid.uuid4().bytes).hexdigest() try: plan = self.ingestion_repository.begin_price_review_generation( job_id, case_id, revision, actor_username, self.processor_version, self.rule_set_sha256, idempotency_key, ) except IngestionError as error: self._raise_review_portal_error(error) if plan.status == "completed": return { "job_id": job_id, "status": "succeeded", "ingestion_status": "already_committed", "attempt_no": None, "business_date": plan.business_date.isoformat(), "daily_version_id": plan.daily_version_id, "version_no": plan.version_no, "review_case_id": plan.review_case_id, } if plan.status == "processing": return { "job_id": job_id, "status": "needs_review", "ingestion_status": "generation_processing", "attempt_no": plan.attempt_no, "business_date": plan.business_date.isoformat(), "review_case_id": plan.review_case_id, } if ( plan.status != "ready" or plan.attempt_no is None or plan.manual_override_bytes is None or plan.manual_override_sha256 is None or plan.source is None ): raise PortalError("REVIEW_STATE_INVALID", "复核生成状态无效", 409) try: with tempfile.TemporaryDirectory(prefix="arr-review-finalize-") as temporary: root = Path(temporary) source_path = root / "processor-input" / "source.xml" self.object_store.materialize( plan.source.object_key, source_path, plan.source.byte_size, ) manifest_path = root / "review-input" / "manual-override.json" self._write_private(manifest_path, plan.manual_override_bytes) self.ingestion_repository.mark_running(job_id, plan.attempt_no) processed = self.processor.run( source_path, root / "processor-output", manual_overrides=manifest_path, review_job_id=job_id, review_case_id=plan.review_case_id, manual_override_sha256=plan.manual_override_sha256, ) stored_outputs: Dict[str, StoredObject] = {} if processed.status == "success": stored_outputs["manual_override_json"] = self.object_store.upload_committed( job_id=job_id, attempt_no=plan.attempt_no, role="manual_override_json", source=manifest_path, original_filename="manual-override.json", ) for role, path in processed.artifacts.items(): stored_outputs[role] = self.object_store.upload_committed( job_id=job_id, attempt_no=plan.attempt_no, role=role, source=path, original_filename=path.name, ) envelope = self._delivery( job_id=job_id, attempt_no=plan.attempt_no, source=plan.source, outputs=stored_outputs, status=processed.status, business_date=( processed.business_date.isoformat() if processed.business_date is not None else None ), ) raw_envelope = ( json.dumps( envelope.to_dict(), ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), ) + "\n" ).encode("utf-8") outcome = self.ingestion_service.ingest(raw_envelope) return { "job_id": job_id, "status": "failed" if outcome.status == "recorded_failure" else "succeeded", "ingestion_status": outcome.status, "attempt_no": plan.attempt_no, "business_date": ( outcome.business_date.isoformat() if outcome.business_date is not None else None ), "daily_version_id": outcome.daily_version_id, "version_no": outcome.version_no, "review_case_id": plan.review_case_id, } except IngestionError as error: if self._is_retryable_review_generation_error(error): self._best_effort_review_generation_failure( job_id, plan.review_case_id, plan.attempt_no, error.code ) else: self._best_effort_failure(job_id, plan.attempt_no, error.code) self._raise_review_portal_error(error) except Exception: self._best_effort_review_generation_failure( job_id, plan.review_case_id, plan.attempt_no, "REVIEW_GENERATION_FAILED", ) raise PortalError("REVIEW_GENERATION_FAILED", "正式日报生成暂未完成，可使用同一清单重试", 503) from None def cancel_price_review( self, job_id: str, case_id: str, revision: int, actor_username: str, ) -> Dict[str, Any]: try: return self.ingestion_repository.cancel_price_review( job_id, case_id, revision, actor_username ) except IngestionError as error: self._raise_review_portal_error(error) @staticmethod def _raise_review_portal_error(error: IngestionError) -> None: status = 422 if error.code in { "REVIEW_REVISION_CONFLICT", "REVIEW_IMMUTABLE", "REVIEW_NOT_OPEN", "REVIEW_NOT_CANCELLABLE", "REVIEW_RULESET_CHANGED", "REVIEW_STATE_INVALID", }: status = 409 elif error.code in {"REVIEW_NOT_FOUND", "REVIEW_ITEM_NOT_FOUND", "JOB_NOT_FOUND"}: status = 404 elif error.retryable or error.code.startswith("DATABASE_"): status = 503 raise PortalError(error.code, error.safe_message, status) from None def _best_effort_review_generation_failure( self, job_id: str, case_id: str, attempt_no: int, failure_code: str, ) -> None: try: self.ingestion_repository.record_price_review_generation_failure( job_id, case_id, attempt_no, failure_code ) except Exception: pass @staticmethod def _is_retryable_review_generation_error(error: IngestionError) -> bool: """Keep a frozen case open only for failures external to its facts.""" return ( error.retryable or error.code.startswith("DATABASE_") or error.code in { "ARTIFACT_UNREADABLE", "OBJECT_STORE_UNAVAILABLE", "PROCESSOR_TIMEOUT", "PROCESSOR_UNAVAILABLE", "SOURCE_NOT_FOUND", } ) def _best_effort_failure( self, job_id: str, attempt_no: int, failure_code: str, ) -> None: try: self.ingestion_repository.record_failure( job_id, attempt_no, failure_code ) except Exception: pass def _delivery( self, *, job_id: str, attempt_no: int, source: StoredObject | ArtifactRef, outputs: Dict[str, StoredObject], status: str, business_date: object, ) -> DeliveryEnvelope: artifacts: Dict[str, object] = { role: None for role in ARTIFACT_ROLES } source_ref = source.to_artifact_ref() if isinstance(source, StoredObject) else source artifacts["source_xml"] = source_ref.to_dict() for role, stored in outputs.items(): artifacts[role] = stored.to_artifact_ref().to_dict() return DeliveryEnvelope.from_dict( { "delivery_schema_version": DELIVERY_SCHEMA_VERSION, "delivery_id": f"local-{job_id}-a{attempt_no}", "job_id": job_id, "attempt_no": attempt_no, "status": status, "processor_version": self.processor_version, "rule_set_sha256": self.rule_set_sha256, "result_schema_version": RESULT_SCHEMA_VERSION, "business_date": business_date, "artifacts": artifacts, } ) def _idempotency_key( self, job_id: str, attempt_no: int, source: StoredObject, ) -> str: value = "\x00".join( ( "arr-programmatic-v1", job_id, str(attempt_no), source.sha256, self.processor_version, self.rule_set_sha256, ) ).encode("utf-8") return hashlib.sha256(value).hexdigest() @staticmethod def _write_private(path: Path, payload: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(descriptor, "wb") as target: descriptor = -1 target.write(payload) target.flush() os.fsync(target.fileno()) finally: if descriptor >= 0: os.close(descriptor)