"""Controlled PixelRAG source-contract replay worker.

This worker turns platform-probe structured fallback receipts into replayable
source-contract receipts. It does not fetch marketplace pages, call models,
write databases, or promote 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_rag_candidate_replay_service import (
    DEFAULT_PLATFORM_PROBE_WORKER_RECEIPT_ROOT,
    build_pixelrag_rag_candidate_replay_readback,
)


POLICY = "controlled_pixelrag_source_contract_replay_worker_v1"
SOURCE_CONTRACT_VERSION = "pixelrag_marketplace_source_contract_v1"
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_SOURCE_CONTRACT_REPLAY_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_source_contract_replay_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_source_contract_replay_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 _source_contract_candidates(readback: Mapping[str, Any]) -> list[dict[str, Any]]:
    candidates: list[dict[str, Any]] = []
    for item in list(readback.get("candidates") or []):
        if not isinstance(item, Mapping):
            continue
        if item.get("candidate_status") != "source_contract_fallback":
            continue
        if not item.get("eligible_for_source_contract_replay"):
            continue
        candidates.append(dict(item))
    return candidates


def _contract_checks(candidate: Mapping[str, Any]) -> dict[str, bool]:
    worker = candidate.get("platform_probe_worker")
    if not isinstance(worker, Mapping):
        worker = {}
    package = worker.get("structured_source_package")
    if not isinstance(package, Mapping):
        package = {}
    return {
        "structured_source_package_available": bool(package.get("available")),
        "source_count_positive": int(package.get("source_count") or 0) > 0,
        "database_write_disallowed": not bool(package.get("database_write_allowed")),
        "network_request_deferred": not bool(package.get("network_request_allowed")),
        "dry_run_only_contract": bool(package.get("dry_run_only")),
        "blocked_page_not_product_data": bool(package.get("blocked_page_not_product_data")),
        "source_worker_receipt_path_present": bool(candidate.get("receipt_path")),
        "source_receipt_path_present": bool(worker.get("source_receipt_path")),
    }


def _source_contract_preview(candidate: Mapping[str, Any]) -> dict[str, Any]:
    worker = candidate.get("platform_probe_worker")
    if not isinstance(worker, Mapping):
        worker = {}
    package = worker.get("structured_source_package")
    if not isinstance(package, Mapping):
        package = {}
    platform = str(candidate.get("platform") or "unknown")
    manifest_id = str(candidate.get("manifest_id") or "unknown")
    adapter_code = str(package.get("adapter_code") or platform)
    return {
        "source_contract_version": SOURCE_CONTRACT_VERSION,
        "contract_id": f"{platform}:{manifest_id}:{adapter_code}",
        "adapter_code": adapter_code,
        "source_count": int(package.get("source_count") or 0),
        "capture_runtime_unavailable": bool(package.get("capture_runtime_unavailable")),
        "barrier_type": worker.get("barrier_type"),
        "expected_offer_fields": [
            "platform",
            "platform_product_id",
            "title",
            "url",
            "image_url",
            "price",
            "currency",
            "availability",
            "seller",
            "promotion_badges",
            "shipping_badges",
            "evidence_time",
            "source_method",
            "confidence",
        ],
        "required_before_data_promotion": [
            "public_source_boundary",
            "rate_limit_contract",
            "provenance_contract",
            "identity_matcher_replay",
            "promotion_gate",
            "embedding_signature_guard",
        ],
    }


def _worker_item(
    candidate: Mapping[str, Any],
    *,
    execute: bool,
) -> dict[str, Any]:
    checks = _contract_checks(candidate)
    check_count = len(checks)
    pass_count = sum(1 for passed in checks.values() if passed)
    ready = pass_count == check_count
    worker = candidate.get("platform_probe_worker")
    if not isinstance(worker, Mapping):
        worker = {}
    package = worker.get("structured_source_package")
    if not isinstance(package, Mapping):
        package = {}
    status = (
        "executed_source_contract_replay_ready"
        if execute and ready
        else (
            "dry_run_ready_for_source_contract_replay"
            if ready
            else "skipped_source_contract_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": candidate.get("platform"),
        "manifest_id": candidate.get("manifest_id"),
        "source_type": "platform_probe_worker_receipt",
        "source_worker_receipt_path": candidate.get("receipt_path"),
        "source_receipt_path": worker.get("source_receipt_path"),
        "source_contract_status": "ready" if ready else "blocked",
        "ready_for_execution": ready,
        "execute": bool(execute),
        "adapter_code": package.get("adapter_code"),
        "structured_source_count": int(package.get("source_count") or 0),
        "capture_runtime_unavailable": bool(package.get("capture_runtime_unavailable")),
        "barrier_type": worker.get("barrier_type"),
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "source_contract_checks": checks,
        "source_contract_check_count": check_count,
        "source_contract_check_pass_count": pass_count,
        "source_contract_preview": _source_contract_preview(candidate),
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "source_contract_is_candidate_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_replay_preflight"
            if execute and ready
            else (
                "run_pixelrag_source_contract_replay_worker_execute"
                if ready
                else "repair_source_contract_fallback_package"
            )
        ),
    }


def _write_replay_receipt(*, output_root: Path, item: Mapping[str, Any]) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "source_contract_replay_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_source_contract_replay_worker(
    *,
    artifact_root: str | Path | None = None,
    platform_probe_worker_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 source-contract replay for platform probe worker receipts."""
    output = Path(output_root or DEFAULT_OUTPUT_ROOT)
    worker_root = Path(
        platform_probe_worker_receipt_root
        or DEFAULT_PLATFORM_PROBE_WORKER_RECEIPT_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))
    readback = build_pixelrag_rag_candidate_replay_readback(
        artifact_root=artifact_root,
        platform_probe_worker_receipt_root=worker_root,
        platform=platforms,
        max_age_hours=max_age,
        limit=item_limit,
    )
    candidates = _source_contract_candidates(readback)
    worker_items: list[dict[str, Any]] = []
    for candidate in candidates:
        item = _worker_item(candidate, execute=execute)
        if execute and write_receipt and item.get("ready_for_execution"):
            item["receipt_path"] = _write_replay_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"))
    capture_runtime_gap_count = sum(
        1 for item in worker_items if item.get("capture_runtime_unavailable")
    )
    guard_failed_count = sum(
        1 for item in worker_items if item.get("source_contract_status") != "ready"
    )

    if not worker_items:
        status = "warning"
    elif guard_failed_count:
        status = "warning"
    else:
        status = "ok"

    if not worker_items:
        next_action = "run_pixelrag_platform_probe_worker_or_rag_candidate_replay"
    elif not execute and ready_count:
        next_action = "run_pixelrag_source_contract_replay_worker_execute"
    elif execute and receipt_written_count:
        next_action = "run_marketplace_source_contract_adapter_replay_preflight"
    elif guard_failed_count:
        next_action = "repair_source_contract_fallback_package"
    else:
        next_action = "refresh_pixelrag_source_contract_replay_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,
        "capture_runtime_gap_count": capture_runtime_gap_count,
        "guard_failed_count": guard_failed_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": datetime.now(timezone.utc).isoformat(),
        "source_contract_version": SOURCE_CONTRACT_VERSION,
        "artifact_root": str(artifact_root or readback.get("artifact_root") or ""),
        "platform_probe_worker_receipt_root": str(worker_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,
        "source_readback": {
            "policy": readback.get("policy"),
            "status": readback.get("status"),
            "summary": readback.get("summary"),
            "next_machine_action": readback.get("next_machine_action"),
        },
        "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,
            "source_contract_is_candidate_only": True,
            "blocked_pages_are_not_product_data": 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__ = [
    "DEFAULT_OUTPUT_ROOT",
    "POLICY",
    "SOURCE_CONTRACT_VERSION",
    "run_pixelrag_source_contract_replay_worker",
]
