"""Controlled PixelRAG marketplace source-contract adapter preflight.

This worker verifies source-contract replay receipts before any marketplace
adapter can run. It does not fetch external pages, call models, write databases,
or promote source-contract candidates into price truth.
"""

from __future__ import annotations

import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from services.pixelrag_crawler_integration_service import DEFAULT_ARTIFACT_MAX_AGE_HOURS
from services.pixelrag_source_contract_replay_worker_service import (
    DEFAULT_OUTPUT_ROOT as DEFAULT_SOURCE_CONTRACT_REPLAY_RECEIPT_ROOT,
    SOURCE_CONTRACT_VERSION,
)


POLICY = "controlled_pixelrag_marketplace_source_contract_adapter_preflight_v1"
ADAPTER_PREFLIGHT_VERSION = "pixelrag_marketplace_adapter_preflight_v1"
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_MARKETPLACE_ADAPTER_PREFLIGHT_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_marketplace_adapter_preflight_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_marketplace_adapter_preflight_receipts",
)


def _normalise_platforms(
    platform: str | tuple[str, ...] | list[str] | None,
) -> tuple[str, ...]:
    if isinstance(platform, str):
        value = platform.strip().lower()
        return (value,) if value else ()
    return tuple(
        str(item or "").strip().lower()
        for item in (platform or ())
        if str(item or "").strip()
    )


def _safe_segment(value: Any) -> str:
    text = str(value or "unknown").strip().lower()
    text = re.sub(r"[^a-z0-9._-]+", "-", text)
    return text.strip("-") or "unknown"


def _parse_iso_datetime(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None


def _receipt_candidates(
    root: Path,
    *,
    platforms: tuple[str, ...],
    limit: int,
) -> list[Path]:
    if not root.exists():
        return []
    candidates: list[Path] = []
    if platforms:
        for platform in platforms:
            candidates.extend((root / platform).glob("*/source_contract_replay_receipt.json"))
    else:
        candidates.extend(root.glob("*/*/source_contract_replay_receipt.json"))
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def _load_receipt(path: Path) -> tuple[dict[str, Any], list[str]]:
    try:
        return json.loads(path.read_text(encoding="utf-8")), []
    except (OSError, json.JSONDecodeError) as exc:
        return {}, [str(exc)[:300]]


def _adapter_preflight_checks(
    receipt: Mapping[str, Any],
    *,
    stale: bool,
    errors: list[str],
) -> dict[str, bool]:
    preview = receipt.get("source_contract_preview")
    if not isinstance(preview, Mapping):
        preview = {}
    promotion = receipt.get("promotion_boundary")
    if not isinstance(promotion, Mapping):
        promotion = {}
    source_checks = receipt.get("source_contract_checks")
    if not isinstance(source_checks, Mapping):
        source_checks = {}
    required_before = set(str(item) for item in list(preview.get("required_before_data_promotion") or []))
    expected_fields = set(str(item) for item in list(preview.get("expected_offer_fields") or []))
    return {
        "receipt_parse_ok": not errors,
        "receipt_fresh": not stale,
        "source_contract_replay_ready": (
            receipt.get("worker_status") == "executed_source_contract_replay_ready"
        ),
        "source_contract_version_supported": (
            preview.get("source_contract_version") == SOURCE_CONTRACT_VERSION
        ),
        "adapter_code_present": bool(preview.get("adapter_code") or receipt.get("adapter_code")),
        "source_count_positive": int(preview.get("source_count") or 0) > 0,
        "source_contract_checks_all_passed": (
            int(receipt.get("source_contract_check_pass_count") or 0)
            == int(receipt.get("source_contract_check_count") or -1)
            and int(receipt.get("source_contract_check_count") or 0) > 0
        ),
        "blocked_page_not_product_data": bool(
            source_checks.get("blocked_page_not_product_data")
        ),
        "database_write_disallowed": bool(source_checks.get("database_write_disallowed")),
        "network_request_deferred": bool(source_checks.get("network_request_deferred")),
        "public_source_boundary_required": "public_source_boundary" in required_before,
        "rate_limit_contract_required": "rate_limit_contract" in required_before,
        "provenance_contract_required": "provenance_contract" in required_before,
        "identity_and_promotion_gates_required": (
            "identity_matcher_replay" in required_before
            and "promotion_gate" in required_before
        ),
        "canonical_offer_fields_present": {
            "platform",
            "title",
            "url",
            "price",
            "currency",
            "evidence_time",
            "source_method",
            "confidence",
        }.issubset(expected_fields),
        "direct_price_write_disallowed": not bool(
            promotion.get("direct_price_write_allowed")
        ),
    }


def _preflight_payload(receipt: Mapping[str, Any]) -> dict[str, Any]:
    preview = receipt.get("source_contract_preview")
    if not isinstance(preview, Mapping):
        preview = {}
    return {
        "adapter_preflight_version": ADAPTER_PREFLIGHT_VERSION,
        "source_contract_version": preview.get("source_contract_version"),
        "adapter_code": preview.get("adapter_code") or receipt.get("adapter_code"),
        "contract_id": preview.get("contract_id"),
        "barrier_type": preview.get("barrier_type") or receipt.get("barrier_type"),
        "capture_runtime_unavailable": bool(
            preview.get("capture_runtime_unavailable")
            or receipt.get("capture_runtime_unavailable")
        ),
        "source_count": int(preview.get("source_count") or 0),
        "expected_offer_fields": list(preview.get("expected_offer_fields") or []),
        "required_before_data_promotion": list(
            preview.get("required_before_data_promotion") or []
        ),
        "adapter_execution_mode": "preflight_only_no_network_no_db",
        "allowed_next_step": "run_marketplace_source_contract_adapter_dry_run",
    }


def _worker_item(
    receipt_path: Path,
    *,
    now: datetime,
    max_age_hours: int,
    execute: bool,
) -> dict[str, Any]:
    receipt, errors = _load_receipt(receipt_path)
    generated_at = _parse_iso_datetime(receipt.get("generated_at"))
    if generated_at is None:
        try:
            generated_at = datetime.fromtimestamp(receipt_path.stat().st_mtime, tz=timezone.utc)
        except OSError:
            generated_at = None
    age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None
    stale = age_hours is None or age_hours > max_age_hours
    checks = _adapter_preflight_checks(receipt, stale=stale, errors=errors)
    check_count = len(checks)
    pass_count = sum(1 for passed in checks.values() if passed)
    ready = pass_count == check_count
    platform = str(receipt.get("platform") or receipt_path.parent.parent.name).strip().lower()
    manifest_id = str(receipt.get("manifest_id") or receipt_path.parent.name).strip()
    status = (
        "executed_adapter_preflight_ready"
        if execute and ready
        else (
            "dry_run_ready_for_adapter_preflight"
            if ready
            else "skipped_adapter_preflight_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": platform,
        "manifest_id": manifest_id,
        "source_type": "source_contract_replay_receipt",
        "source_contract_replay_receipt_path": str(receipt_path),
        "source_worker_receipt_path": receipt.get("source_worker_receipt_path"),
        "source_receipt_path": receipt.get("source_receipt_path"),
        "adapter_code": (receipt.get("source_contract_preview") or {}).get("adapter_code")
        if isinstance(receipt.get("source_contract_preview"), Mapping)
        else receipt.get("adapter_code"),
        "adapter_preflight_status": "ready" if ready else "blocked",
        "ready_for_execution": ready,
        "execute": bool(execute),
        "generated_at": generated_at.isoformat() if generated_at else None,
        "age_hours": round(age_hours, 3) if age_hours is not None else None,
        "stale": stale,
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "adapter_preflight_checks": checks,
        "adapter_preflight_check_count": check_count,
        "adapter_preflight_check_pass_count": pass_count,
        "adapter_preflight": _preflight_payload(receipt),
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "adapter_preflight_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": (
            "run_marketplace_source_contract_adapter_dry_run"
            if execute and ready
            else (
                "run_marketplace_source_contract_adapter_preflight_execute"
                if ready
                else "repair_marketplace_source_contract_adapter_preflight_inputs"
            )
        ),
    }


def _write_preflight_receipt(*, output_root: Path, item: Mapping[str, Any]) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "marketplace_adapter_preflight_receipt.json"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    receipt = dict(item)
    receipt["artifact_write_performed"] = True
    receipt["receipt_path"] = str(target)
    receipt["generated_at"] = datetime.now(timezone.utc).isoformat()
    receipt["policy"] = POLICY
    target.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return str(target)


def run_pixelrag_marketplace_source_contract_adapter_preflight(
    *,
    source_contract_replay_receipt_root: str | Path | None = None,
    output_root: str | Path | None = None,
    platform: str | tuple[str, ...] | list[str] | None = None,
    max_age_hours: int | None = None,
    limit: int | None = None,
    execute: bool = False,
    write_receipt: bool = False,
) -> dict[str, Any]:
    """Run or dry-run marketplace source-contract adapter preflight."""
    source_root = Path(
        source_contract_replay_receipt_root
        or DEFAULT_SOURCE_CONTRACT_REPLAY_RECEIPT_ROOT
    )
    output = Path(output_root or DEFAULT_OUTPUT_ROOT)
    platforms = _normalise_platforms(platform)
    max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
    item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
    now = datetime.now(timezone.utc)
    receipt_paths = _receipt_candidates(source_root, platforms=platforms, limit=item_limit)
    worker_items: list[dict[str, Any]] = []
    for receipt_path in receipt_paths:
        item = _worker_item(
            receipt_path,
            now=now,
            max_age_hours=max_age,
            execute=execute,
        )
        if execute and write_receipt and item.get("ready_for_execution"):
            item["receipt_path"] = _write_preflight_receipt(output_root=output, item=item)
            item["artifact_write_performed"] = True
        worker_items.append(item)

    ready_count = sum(1 for item in worker_items if item.get("ready_for_execution"))
    blocked_count = len(worker_items) - ready_count
    dry_run_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("dry_run_")
    )
    executed_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_")
    )
    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
    guard_failed_count = sum(
        1 for item in worker_items if item.get("adapter_preflight_status") != "ready"
    )
    capture_runtime_gap_count = sum(
        1
        for item in worker_items
        if (item.get("adapter_preflight") or {}).get("capture_runtime_unavailable")
    )
    if not worker_items:
        status = "warning"
    elif guard_failed_count:
        status = "warning"
    else:
        status = "ok"

    if not worker_items:
        next_action = "run_pixelrag_source_contract_replay_worker"
    elif not execute and ready_count:
        next_action = "run_marketplace_source_contract_adapter_preflight_execute"
    elif execute and receipt_written_count:
        next_action = "run_marketplace_source_contract_adapter_dry_run"
    elif guard_failed_count:
        next_action = "repair_marketplace_source_contract_adapter_preflight_inputs"
    else:
        next_action = "refresh_marketplace_source_contract_adapter_preflight_candidates"

    summary = {
        "candidate_count": len(worker_items),
        "ready_count": ready_count,
        "blocked_count": blocked_count,
        "dry_run_count": dry_run_count,
        "executed_count": executed_count,
        "receipt_written_count": receipt_written_count,
        "guard_failed_count": guard_failed_count,
        "capture_runtime_gap_count": capture_runtime_gap_count,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in worker_items}),
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": bool(receipt_written_count),
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
    }
    return {
        "success": status != "critical",
        "policy": POLICY,
        "status": status,
        "generated_at": now.isoformat(),
        "adapter_preflight_version": ADAPTER_PREFLIGHT_VERSION,
        "source_contract_replay_receipt_root": str(source_root),
        "output_root": str(output),
        "platform_filter": list(platforms),
        "max_age_hours": max_age,
        "limit": item_limit,
        "execute": bool(execute),
        "write_receipt": bool(write_receipt and execute),
        "summary": summary,
        "worker_items": worker_items,
        "controlled_apply": {
            "network_call": False,
            "model_call": False,
            "artifact_write": bool(receipt_written_count),
            "db_write": False,
            "writes_database": False,
            "writes_database_count": 0,
            "secret_read": False,
            "raw_cookie_or_session_read": False,
            "credentialed_session_allowed": False,
            "login_allowed": False,
            "production_price_write": False,
            "primary_human_gate_count": 0,
        },
        "promotion_boundary": {
            "writes_ai_insights": False,
            "writes_price_tables": False,
            "adapter_preflight_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "ADAPTER_PREFLIGHT_VERSION",
    "DEFAULT_OUTPUT_ROOT",
    "POLICY",
    "run_pixelrag_marketplace_source_contract_adapter_preflight",
]
