"""Controlled PixelRAG platform probe worker.

This worker executes the machine actions emitted by platform-probe readiness:
public empty-context visual capture for recoverable marketplace interstitials,
or structured-source/backoff receipts for hard blocked platforms. It never
reads sessions/cookies, logs in, writes DB rows, or promotes price truth.
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
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,
    DEFAULT_ARTIFACT_ROOT,
)
from services.pixelrag_platform_probe_service import (
    build_pixelrag_platform_probe_readiness,
)


POLICY = "controlled_pixelrag_platform_probe_worker_v1"
DEFAULT_LIMIT = 25
DEFAULT_TIMEOUT_SECONDS = 30
DEFAULT_SETTLE_MS = 800
DEFAULT_MAX_TILES = 12
DEFAULT_WORKER_RECEIPT_ROOT = os.getenv(
    "PIXELRAG_PLATFORM_PROBE_WORKER_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_platform_probe_worker_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_platform_probe_worker_receipts",
)
DEFAULT_CAPTURE_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_PLATFORM_PROBE_CAPTURE_OUTPUT_ROOT",
    str(DEFAULT_ARTIFACT_ROOT),
)
CAPTURE_READY_STATUSES = {
    "ready_for_public_context_probe",
    "ready_for_platform_probe",
}


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 _capture_script_path() -> Path:
    return Path(__file__).resolve().parents[1] / "scripts" / "ops" / "capture_pixelrag_visual_evidence.py"


def _base_item(item: Mapping[str, Any]) -> dict[str, Any]:
    fallback = item.get("structured_source_fallback")
    if not isinstance(fallback, Mapping):
        fallback = {}
    return {
        "platform": item.get("platform"),
        "manifest_id": item.get("manifest_id"),
        "source_type": item.get("source_type"),
        "source_receipt_path": item.get("source_receipt_path"),
        "source_capture_receipt_path": item.get("source_capture_receipt_path"),
        "probe_status": item.get("probe_status"),
        "barrier_type": item.get("barrier_type"),
        "url": item.get("url"),
        "title": item.get("title"),
        "structured_source_fallback_available": bool(fallback.get("available")),
        "structured_source_count": len(list(fallback.get("sources") or [])),
        "writes_database": False,
        "model_call_performed": False,
        "primary_human_gate_count": 0,
    }


def _write_worker_receipt(
    *,
    output_root: Path,
    item: Mapping[str, Any],
    worker_item: Mapping[str, Any],
) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "platform_probe_worker_receipt.json"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    receipt = dict(worker_item)
    receipt["artifact_write_performed"] = True
    receipt["receipt_path"] = str(target)
    target.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return str(target)


def _dry_run_item(item: Mapping[str, Any]) -> dict[str, Any]:
    base = _base_item(item)
    probe_status = str(item.get("probe_status") or "")
    manifest = item.get("capture_manifest_preview")
    capture_ready = (
        probe_status in CAPTURE_READY_STATUSES
        and isinstance(manifest, Mapping)
        and bool(manifest.get("success"))
    )
    if capture_ready:
        base.update({
            "worker_status": "dry_run_ready_for_public_context_capture",
            "ready_for_execution": True,
            "capture_execution_ready": True,
            "capture_manifest_id": manifest.get("manifest_id"),
            "public_browser_context_policy": (
                (manifest.get("public_browser_context") or {}).get("context_policy")
            ),
            "network_call_performed": False,
            "artifact_write_performed": False,
            "next_machine_action": "run_pixelrag_platform_probe_worker_execute",
        })
        return base
    base.update({
        "worker_status": "dry_run_structured_source_fallback_ready",
        "ready_for_execution": True,
        "capture_execution_ready": False,
        "network_call_performed": False,
        "artifact_write_performed": False,
        "structured_source_fallback": item.get("structured_source_fallback") or {},
        "next_machine_action": "run_structured_source_or_platform_backoff_policy",
    })
    return base


def _skipped_item(item: Mapping[str, Any]) -> dict[str, Any]:
    base = _base_item(item)
    base.update({
        "worker_status": "skipped_not_probe_ready",
        "ready_for_execution": False,
        "capture_execution_ready": False,
        "network_call_performed": False,
        "artifact_write_performed": False,
        "next_machine_action": item.get("next_machine_action")
        or "refresh_pixelrag_platform_probe_readiness",
    })
    return base


def _structured_source_item(
    item: Mapping[str, Any],
    *,
    output_root: Path,
    execute: bool,
    write_receipt: bool,
) -> dict[str, Any]:
    base = _base_item(item)
    fallback = item.get("structured_source_fallback") or {}
    base.update({
        "worker_status": (
            "executed_structured_source_fallback_package"
            if execute
            else "dry_run_structured_source_fallback_ready"
        ),
        "ready_for_execution": True,
        "capture_execution_ready": False,
        "network_call_performed": False,
        "artifact_write_performed": False,
        "structured_source_fallback": fallback,
        "structured_source_package": {
            "adapter_code": fallback.get("adapter_code"),
            "available": bool(fallback.get("available")),
            "source_count": len(list(fallback.get("sources") or [])),
            "network_request_allowed": bool(fallback.get("network_request_allowed")),
            "database_write_allowed": False,
            "dry_run_only": True,
            "blocked_page_not_product_data": True,
        },
        "next_machine_action": "run_structured_source_or_platform_backoff_policy",
    })
    if execute and write_receipt:
        base["receipt_path"] = _write_worker_receipt(
            output_root=output_root,
            item=item,
            worker_item=base,
        )
        base["artifact_write_performed"] = True
    return base


def _run_capture_subprocess(
    *,
    manifest: Mapping[str, Any],
    capture_output_root: Path,
    timeout_seconds: int,
    settle_ms: int,
    max_tiles: int,
    capture_script: Path | None = None,
) -> dict[str, Any]:
    script = capture_script or _capture_script_path()
    completed = subprocess.run(
        [
            sys.executable,
            str(script),
            "--manifest-json",
            json.dumps(dict(manifest), ensure_ascii=False),
            "--output-dir",
            str(capture_output_root),
            "--timeout",
            str(max(1, int(timeout_seconds or DEFAULT_TIMEOUT_SECONDS))),
            "--settle-ms",
            str(max(0, int(settle_ms or DEFAULT_SETTLE_MS))),
            "--max-tiles",
            str(max(1, int(max_tiles or DEFAULT_MAX_TILES))),
        ],
        capture_output=True,
        check=False,
        text=True,
    )
    try:
        payload = json.loads(completed.stdout)
    except json.JSONDecodeError:
        payload = {}
    stderr_excerpt = completed.stderr[:500]
    if len(completed.stderr) > 500:
        stderr_excerpt = f"{completed.stderr[:250]}...{completed.stderr[-250:]}"
    return {
        "returncode": completed.returncode,
        "stdout_json": payload,
        "stdout_excerpt": completed.stdout[:500],
        "stderr_excerpt": stderr_excerpt,
    }


def _capture_runtime_unavailable(result: Mapping[str, Any]) -> bool:
    haystack = " ".join(
        [
            str(result.get("stdout_excerpt") or ""),
            str(result.get("stderr_excerpt") or ""),
        ]
    ).lower()
    return (
        "modulenotfounderror" in haystack
        and "playwright" in haystack
    ) or "no module named 'playwright'" in haystack or (
        "from playwright.sync_api" in haystack
        and "traceback" in haystack
    ) or (
        "playwright" in haystack
        and (
            "executable doesn't exist" in haystack
            or "please run" in haystack
            or "browsertype.launch" in haystack
        )
    )


def _capture_item(
    item: Mapping[str, Any],
    *,
    output_root: Path,
    capture_output_root: Path,
    timeout_seconds: int,
    settle_ms: int,
    max_tiles: int,
    write_receipt: bool,
    capture_script: Path | None,
) -> dict[str, Any]:
    base = _base_item(item)
    manifest = item.get("capture_manifest_preview")
    if not isinstance(manifest, Mapping) or not manifest.get("success"):
        base.update({
            "worker_status": "skipped_missing_capture_manifest",
            "ready_for_execution": False,
            "capture_execution_ready": False,
            "network_call_performed": False,
            "artifact_write_performed": False,
            "next_machine_action": "refresh_pixelrag_platform_probe_manifest_or_use_structured_source",
        })
        if write_receipt:
            base["receipt_path"] = _write_worker_receipt(
                output_root=output_root,
                item=item,
                worker_item=base,
            )
            base["artifact_write_performed"] = True
        return base

    result = _run_capture_subprocess(
        manifest=manifest,
        capture_output_root=capture_output_root,
        timeout_seconds=timeout_seconds,
        settle_ms=settle_ms,
        max_tiles=max_tiles,
        capture_script=capture_script,
    )
    capture_payload = result["stdout_json"]
    files = list(capture_payload.get("files") or []) if isinstance(capture_payload, Mapping) else []
    tile_file_count = sum(1 for file_item in files if file_item.get("kind") == "tile")
    receipt_path = ""
    planned_output = capture_payload.get("planned_output") if isinstance(capture_payload, Mapping) else {}
    if isinstance(planned_output, Mapping):
        receipt_path = str(planned_output.get("receipt") or "")
    for file_item in files:
        if file_item.get("kind") == "receipt":
            receipt_path = str(file_item.get("path") or receipt_path)

    captured_ok = (
        result["returncode"] == 0
        and isinstance(capture_payload, Mapping)
        and capture_payload.get("success") is True
        and capture_payload.get("status") == "captured"
    )
    runtime_missing = not captured_ok and _capture_runtime_unavailable(result)
    structured_fallback_available = bool(
        (item.get("structured_source_fallback") or {}).get("available")
    )
    capture_fallback = not captured_ok and structured_fallback_available
    base.update({
        "worker_status": (
            "executed_capture_ok"
            if captured_ok
            else (
                (
                    "capture_runtime_unavailable_structured_fallback_package"
                    if runtime_missing
                    else "capture_failed_structured_fallback_package"
                )
                if capture_fallback
                else "capture_worker_error"
            )
        ),
        "ready_for_execution": True,
        "capture_execution_ready": True,
        "capture_manifest_id": manifest.get("manifest_id"),
        "capture_returncode": result["returncode"],
        "capture_status": capture_payload.get("status") if isinstance(capture_payload, Mapping) else "",
        "capture_http_status": int(capture_payload.get("http_status") or 0)
        if isinstance(capture_payload, Mapping)
        else 0,
        "capture_tile_file_count": tile_file_count,
        "capture_receipt_path": receipt_path,
        "network_call_performed": bool(not runtime_missing),
        "artifact_write_performed": bool(captured_ok),
        "structured_source_fallback": item.get("structured_source_fallback") or {},
        "structured_source_package": (
            {
                "adapter_code": (item.get("structured_source_fallback") or {}).get("adapter_code"),
                "available": True,
                "source_count": len(list((item.get("structured_source_fallback") or {}).get("sources") or [])),
                "network_request_allowed": bool(
                    (item.get("structured_source_fallback") or {}).get("network_request_allowed")
                ),
                "database_write_allowed": False,
                "dry_run_only": True,
                "blocked_page_not_product_data": True,
                "capture_runtime_unavailable": runtime_missing,
            }
            if capture_fallback
            else None
        ),
        "capture_stdout_excerpt": result["stdout_excerpt"] if not captured_ok else "",
        "capture_stderr_excerpt": result["stderr_excerpt"] if not captured_ok else "",
        "next_machine_action": (
            "run_pixelrag_rag_candidate_replay_after_probe_capture"
            if captured_ok
            else (
                (
                    "run_structured_source_or_install_pixelrag_capture_runtime"
                    if runtime_missing
                    else "run_structured_source_or_retry_platform_capture"
                )
                if capture_fallback
                else "repair_pixelrag_platform_probe_worker_capture_runtime_or_use_structured_source"
            )
        ),
    })
    if write_receipt:
        base["receipt_path"] = _write_worker_receipt(
            output_root=output_root,
            item=item,
            worker_item=base,
        )
        base["artifact_write_performed"] = True
    return base


def run_pixelrag_platform_probe_worker(
    *,
    artifact_root: str | Path | None = None,
    vlm_receipt_root: str | Path | None = None,
    output_root: str | Path | None = None,
    capture_output_root: str | Path | None = None,
    platform: str | tuple[str, ...] | list[str] | None = None,
    max_age_hours: int | None = None,
    limit: int | None = None,
    timeout: int | None = None,
    settle_ms: int | None = None,
    max_tiles: int | None = None,
    execute: bool = False,
    write_receipt: bool = False,
    capture_script: str | Path | None = None,
) -> dict[str, Any]:
    """Run or dry-run public platform probe machine actions."""
    output = Path(output_root or DEFAULT_WORKER_RECEIPT_ROOT)
    capture_output = Path(capture_output_root or DEFAULT_CAPTURE_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))
    timeout_seconds = max(1, min(int(timeout or DEFAULT_TIMEOUT_SECONDS), 120))
    settle = max(0, min(int(settle_ms or DEFAULT_SETTLE_MS), 5000))
    tiles = max(1, min(int(max_tiles or DEFAULT_MAX_TILES), 80))
    generated_at = datetime.now(timezone.utc).isoformat()
    capture_script_path = Path(capture_script) if capture_script else None
    readiness_artifact_root = artifact_root or DEFAULT_ARTIFACT_ROOT

    readiness = build_pixelrag_platform_probe_readiness(
        artifact_root=readiness_artifact_root,
        vlm_receipt_root=vlm_receipt_root,
        platform=platforms,
        max_age_hours=max_age,
        limit=item_limit,
    )
    probe_items = list(readiness.get("probe_items") or [])
    worker_items: list[dict[str, Any]] = []
    for item in probe_items:
        if not item.get("probe_ready"):
            worker_items.append(_skipped_item(item))
            continue
        probe_status = str(item.get("probe_status") or "")
        manifest = item.get("capture_manifest_preview")
        can_capture = (
            probe_status in CAPTURE_READY_STATUSES
            and isinstance(manifest, Mapping)
            and bool(manifest.get("success"))
        )
        if not execute:
            worker_items.append(_dry_run_item(item))
            continue
        if can_capture:
            worker_items.append(_capture_item(
                item,
                output_root=output,
                capture_output_root=capture_output,
                timeout_seconds=timeout_seconds,
                settle_ms=settle,
                max_tiles=tiles,
                write_receipt=write_receipt,
                capture_script=capture_script_path,
            ))
        else:
            worker_items.append(_structured_source_item(
                item,
                output_root=output,
                execute=True,
                write_receipt=write_receipt,
            ))

    ready_count = sum(1 for item in probe_items if item.get("probe_ready"))
    dry_run_count = sum(1 for item in worker_items if str(item.get("worker_status") or "").startswith("dry_run_"))
    structured_count = sum(
        1 for item in worker_items if "structured" in str(item.get("worker_status") or "")
    )
    capture_ready_count = sum(1 for item in worker_items if item.get("capture_execution_ready"))
    capture_execute_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_capture")
    )
    capture_ok_count = sum(1 for item in worker_items if item.get("worker_status") == "executed_capture_ok")
    capture_error_count = sum(1 for item in worker_items if item.get("worker_status") == "capture_worker_error")
    executed_structured_count = sum(
        1
        for item in worker_items
        if "structured" in str(item.get("worker_status") or "")
        and not str(item.get("worker_status") or "").startswith("dry_run_")
    )
    executed_count = capture_execute_count + executed_structured_count
    skipped_count = sum(1 for item in worker_items if str(item.get("worker_status") or "").startswith("skipped_"))
    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
    network_call_performed = any(bool(item.get("network_call_performed")) for item in worker_items)
    artifact_write_performed = any(bool(item.get("artifact_write_performed")) for item in worker_items)

    if capture_error_count:
        status = "critical" if execute and capture_ok_count == 0 and capture_ready_count else "warning"
    elif probe_items and ready_count:
        status = "ok"
    elif probe_items:
        status = "warning"
    else:
        status = "warning"

    if not probe_items:
        next_action = "run_pixelrag_visual_capture_worker"
    elif capture_error_count:
        next_action = "repair_pixelrag_platform_probe_worker_capture_runtime_or_use_structured_source"
    elif not execute and capture_ready_count:
        next_action = "run_pixelrag_platform_probe_worker_execute"
    elif execute and capture_ok_count:
        next_action = "run_pixelrag_rag_candidate_replay_after_probe_capture"
    elif structured_count:
        next_action = "run_structured_source_or_platform_backoff_policy"
    else:
        next_action = "refresh_pixelrag_platform_probe_readiness"

    summary = {
        "probe_candidate_count": len(probe_items),
        "ready_count": ready_count,
        "skipped_count": skipped_count,
        "dry_run_count": dry_run_count,
        "capture_ready_count": capture_ready_count,
        "capture_execute_count": capture_execute_count,
        "capture_ok_count": capture_ok_count,
        "capture_error_count": capture_error_count,
        "structured_fallback_count": structured_count,
        "executed_structured_count": executed_structured_count,
        "executed_count": executed_count,
        "receipt_written_count": receipt_written_count,
        "network_call_performed": network_call_performed,
        "artifact_write_performed": artifact_write_performed,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in probe_items}),
    }
    return {
        "success": status != "critical",
        "policy": POLICY,
        "status": status,
        "generated_at": generated_at,
        "artifact_root": str(readiness_artifact_root),
        "vlm_receipt_root": str(
            vlm_receipt_root
            or readiness.get("vlm_receipt_root")
            or ""
        ),
        "output_root": str(output),
        "capture_output_root": str(capture_output),
        "platform_filter": list(platforms),
        "max_age_hours": max_age,
        "limit": item_limit,
        "timeout_seconds": timeout_seconds,
        "settle_ms": settle,
        "max_tiles": tiles,
        "execute": bool(execute),
        "write_receipt": bool(write_receipt),
        "summary": summary,
        "worker_items": worker_items,
        "source_readiness": {
            "policy": readiness.get("policy"),
            "status": readiness.get("status"),
            "summary": readiness.get("summary"),
            "next_machine_action": readiness.get("next_machine_action"),
        },
        "controlled_apply": {
            "network_call": bool(execute and network_call_performed),
            "model_call": False,
            "artifact_write": artifact_write_performed,
            "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,
            "blocked_pages_are_not_product_data": True,
            "visual_fields_are_candidate_evidence_only": True,
            "requires_rag_candidate_replay_after_capture": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "DEFAULT_CAPTURE_OUTPUT_ROOT",
    "DEFAULT_WORKER_RECEIPT_ROOT",
    "POLICY",
    "run_pixelrag_platform_probe_worker",
]
