from __future__ import annotations

import csv
from datetime import datetime, timezone
import hashlib
import io
import json
import os
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

from .archive import create_exclusive_bytes, sha256_file
from .models import ContractError, ErrorCode


MANIFEST_COLUMNS = (
    "task_id", "requested_by", "review_owner", "source_url", "source_site", "title",
    "publisher", "report_date", "downloaded_at", "http_status", "content_type", "file_name",
    "relative_path", "bytes", "sha256", "download_status", "error_or_note",
    "source_cache_path", "source_file_name", "android_package", "extension_added",
    "pdf_magic_valid", "remote_sha256", "local_sha256", "openability", "page_count",
    "encryption_status", "schema_version", "row_id", "handoff_id", "run_id", "item_id",
    "slot_id", "query", "candidate_id", "report_identity", "analysts", "selection_reason",
    "remote_bytes", "local_bytes", "quota_reservation_id", "quota_terminal_event_id",
    "quota_artifact_event_id", "status", "stop_code", "reused_without_new_trigger",
    "external_evidence_hash", "manifested_at_utc",
)


def canonical_json_bytes(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=False).encode("utf-8")


def validate_manifest_row(row: Mapping[str, Any]) -> dict[str, str]:
    if tuple(row.keys()) != MANIFEST_COLUMNS:
        raise ContractError(ErrorCode.MANIFEST_INVALID, "keys", "48-key order required")
    result = {k: "" if row[k] is None else str(row[k]) for k in MANIFEST_COLUMNS}
    if result["source_site"] != "慧博 APP（安卓模拟器本地缓存）":
        raise ContractError(ErrorCode.MANIFEST_INVALID, "source_site", "literal mismatch")
    if result["android_package"] != "cn.com.hibor":
        raise ContractError(ErrorCode.MANIFEST_INVALID, "android_package", "literal mismatch")
    if result["download_status"] not in {"SUCCESS", "DUPLICATE", "FAILED", "STOPPED"}:
        raise ContractError(ErrorCode.MANIFEST_INVALID, "download_status", "enum")
    if result["status"] != result["download_status"]:
        raise ContractError(ErrorCode.MANIFEST_INVALID, "status", "must equal download_status")
    if result["content_type"] == "application/pdf" and result["pdf_magic_valid"].lower() != "true":
        raise ContractError(ErrorCode.MANIFEST_INVALID, "content_type", "PDF claim without magic")
    if result["reused_without_new_trigger"].lower() not in {"true", "false"}:
        raise ContractError(ErrorCode.MANIFEST_INVALID, "reused", "BOOL required")
    return result


def _create_or_verify(path: Path, data: bytes,
                      checkpoint: Callable[[], None] | None = None) -> None:
    try:
        create_exclusive_bytes(path, data, checkpoint=checkpoint)
    except FileExistsError as exc:
        if not path.is_file() or path.is_symlink():
            raise ContractError(ErrorCode.PERSIST_LATE, "path", "existing non-file") from exc
        try:
            existing = path.read_bytes()
        except OSError as read_exc:
            raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "path", "existing unreadable") from read_exc
        if existing != data:
            raise ContractError(ErrorCode.PERSIST_LATE, "path", "existing bytes differ") from exc
    if checkpoint:
        checkpoint()


def write_manifest_create_new(path: Path, rows: Sequence[Mapping[str, Any]], *,
                              checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
    buffer = io.StringIO(newline="")
    writer = csv.DictWriter(buffer, fieldnames=MANIFEST_COLUMNS, lineterminator="\r\n")
    writer.writeheader()
    for row in rows:
        writer.writerow(validate_manifest_row(row))
    data = buffer.getvalue().encode("utf-8")
    _create_or_verify(path, data, checkpoint=checkpoint)
    try:
        parsed_rows = list(csv.DictReader(io.StringIO(path.read_text(encoding="utf-8"), newline="")))
    except (OSError, UnicodeError, csv.Error) as exc:
        raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "reopen failed") from exc
    if (parsed_rows and tuple(parsed_rows[0].keys()) != MANIFEST_COLUMNS) or (not parsed_rows and rows):
        raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "reopen schema mismatch")
    if path.read_bytes() != data:
        raise ContractError(ErrorCode.PERSIST_LATE, "manifest", "reopen bytes mismatch")
    return len(data), hashlib.sha256(data).hexdigest()


def write_json_create_new(path: Path, value: Any, *,
                          checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
    data = canonical_json_bytes(value)
    _create_or_verify(path, data, checkpoint=checkpoint)
    parsed = json.loads(path.read_text(encoding="utf-8"))
    if canonical_json_bytes(parsed) != data:
        raise ContractError(ErrorCode.PERSIST_LATE, "json", "canonical roundtrip mismatch")
    return len(data), hashlib.sha256(data).hexdigest()


def write_bytes_create_new(path: Path, data: bytes, *,
                           checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
    """Create a closure artifact or accept only an exact durable replay."""
    _create_or_verify(path, data, checkpoint=checkpoint)
    try:
        reopened = path.read_bytes()
    except OSError as exc:
        raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "bytes", "reopen failed") from exc
    if reopened != data:
        raise ContractError(ErrorCode.PERSIST_LATE, "bytes", "reopen mismatch")
    return len(data), hashlib.sha256(data).hexdigest()


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")