"""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)
