"""Three simultaneous browser attempts per tool, run as separate sequential batches."""
import argparse
import json
import os
import platform
import signal
import subprocess
import sys
import threading
import time
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path

import psutil

HERE = Path(__file__).resolve().parent
OUT_ROOT = Path(os.environ.get("SCRAPINGEVALS_OUTPUT_DIR", HERE)).resolve()
URL = "https://www.amazon.com/s?k=laptop"
ENV_BASE = "/tmp/scrapingevals-{}-amazon-20260929"
PROXY_VARS = ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy")
BLOCK_MARKERS = ("robot check", "enter the characters", "verify you are human", "just a moment", "access denied", "checking your browser")
MIN_AVAILABLE_BYTES = 3 * 1024 ** 3


def now():
    return datetime.now(timezone.utc).isoformat()


def sample_tree(root, samples, stop):
    while not stop.is_set():
        pss = rss = count = 0
        try:
            members = [root, *root.children(recursive=True)]
        except psutil.NoSuchProcess:
            break
        for proc in members:
            try:
                mem = proc.memory_full_info()
                pss += getattr(mem, "pss", mem.rss)
                rss += mem.rss
                count += 1
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass
        samples.append((pss, rss, count))
        stop.wait(0.2)


def peaks(samples):
    return {
        "sampled_peak_process_tree_pss_mb": round(max((x[0] for x in samples), default=0) / 1048576, 1),
        "sampled_peak_process_tree_rss_sum_mb": round(max((x[1] for x in samples), default=0) / 1048576, 1),
        "sampled_max_process_count": max((x[2] for x in samples), default=0),
        "resource_sample_interval_seconds": 0.2,
    }


def worker(tool, attempt, go_path):
    from bs4 import BeautifulSoup

    out = OUT_ROOT / tool / f"attempt-{attempt}"
    out.mkdir(parents=True, exist_ok=True)
    for key in PROXY_VARS:
        os.environ.pop(key, None)
    row = {
        "tool": tool, "tool_version": version(tool), "python_version": platform.python_version(),
        "runtime": platform.platform(), "url": URL, "attempt": attempt, "started_at": now(),
        "command": [sys.executable, *sys.argv],
        "config": {"headless": True, "proxy": None, "os": "linux" if tool == "camoufox" else None,
                   "wait_until": "domcontentloaded", "navigation_timeout_ms": 30000,
                   "content_wait_ms": 8000, "settle_ms": 500},
        "network": "this host, direct outbound; common proxy environment variables removed",
        "resource_metric_caveat": "Sampled Python plus descendant browser process PSS/RSS every 0.2 seconds; may miss short peaks and is not a clean per-browser allocation under concurrent operation.",
    }
    while not go_path.exists():
        time.sleep(0.05)
    samples, stop = [], threading.Event()
    monitor = threading.Thread(target=sample_tree, args=(psutil.Process(), samples, stop), daemon=True)
    monitor.start()
    start = time.perf_counter()
    try:
        if tool == "patchright":
            from patchright.sync_api import sync_playwright
            with sync_playwright() as playwright:
                row["browser_executable"] = playwright.chromium.executable_path
                browser = playwright.chromium.launch(headless=True)
                try:
                    capture(browser, row, out)
                finally:
                    browser.close()
        else:
            from camoufox.pkgman import launch_path
            from camoufox.sync_api import Camoufox
            row["browser_executable"] = launch_path()
            with Camoufox(headless=True, os="linux") as browser:
                capture(browser, row, out)
    except Exception as exc:
        row.update(navigation_outcome=row.get("navigation_outcome", "error"),
                   error=f"{type(exc).__name__}: {str(exc)[:500]}", usable_content=False)
    finally:
        row["wall_seconds"] = round(time.perf_counter() - start, 3)
        row["finished_at"] = now()
        stop.set()
        monitor.join(timeout=1)
        row.update(peaks(samples))
        (out / "result.json").write_text(json.dumps(row, indent=2) + "\n", encoding="utf-8")
        print(json.dumps({"tool": tool, "attempt": attempt, "status": row.get("http_status"),
                          "cards": row.get("product_result_count"), "error": row.get("error")}))


def capture(browser, row, out):
    from bs4 import BeautifulSoup

    row["browser_version"] = browser.version
    page = browser.new_page()
    try:
        row["navigation_start_unix"] = time.time()
        try:
            response = page.goto(URL, wait_until="domcontentloaded", timeout=30000)
        finally:
            row["navigation_end_unix"] = time.time()
        row["http_status"] = response.status if response else None
        row["navigation_outcome"] = "returned"
        if row["http_status"] == 200:
            try:
                page.wait_for_function("() => document.body && document.body.innerText.length >= 500", timeout=8000)
            except Exception:
                pass
        page.wait_for_timeout(500)
        row["final_url"] = page.url
        row["title"] = page.title()
        html = page.content()
        text = page.locator("body").inner_text(timeout=5000)
        (out / "rendered-html.txt").write_text(html, encoding="utf-8")
        (out / "visible-text.txt").write_text(text, encoding="utf-8")
        row.update(rendered_html="rendered-html.txt", visible_text="visible-text.txt",
                   html_bytes=len(html.encode("utf-8")), visible_text_chars=len(text))
        page.screenshot(path=str(out / "screenshot.png"), full_page=False, timeout=10000)
        row["screenshot"] = "screenshot.png"
        cards = [c for c in BeautifulSoup(html, "html.parser").select('[data-component-type="s-search-result"]')
                 if c.get("data-asin") and c.select_one("h2")]
        low = (row["title"] + "\n" + text[:2000]).lower()
        markers = [m for m in BLOCK_MARKERS if m in low]
        row.update(product_result_count=len(cards), required_term_found="laptop" in text.lower(),
                   block_markers=markers, empty_or_short_200=row["http_status"] == 200 and len(text) < 500,
                   error_page_marker="500_503.png" in html or "Sorry! Something went wrong!" in row["title"],
                   usable_content=row["http_status"] == 200 and len(cards) > 0 and "laptop" in text.lower()
                   and len(text) >= 500 and not markers)
    finally:
        page.close()


def max_overlap(rows):
    events = []
    for row in rows:
        if "navigation_start_unix" in row and "navigation_end_unix" in row:
            events.extend(((row["navigation_start_unix"], 1), (row["navigation_end_unix"], -1)))
    active = maximum = 0
    for _, change in sorted(events):
        active += change
        maximum = max(maximum, active)
    return maximum


def batch(tool):
    base = OUT_ROOT / tool
    base.mkdir(parents=True, exist_ok=True)
    go_path = base / "go.flag"
    if go_path.exists() or (base / "batch.json").exists():
        raise SystemExit("Batch artifacts already exist; set SCRAPINGEVALS_OUTPUT_DIR to a fresh directory")
    available = psutil.virtual_memory().available
    if available < 6 * 1024 ** 3:
        raise SystemExit(f"Preflight stopped: available RAM {available / 1073741824:.2f} GiB < 6 GiB")
    env = os.environ.copy()
    for key in PROXY_VARS:
        env.pop(key, None)
    env_root = Path(ENV_BASE.format(tool))
    if tool == "patchright":
        env["PLAYWRIGHT_BROWSERS_PATH"] = str(env_root / "browsers")
        env["XDG_CACHE_HOME"] = str(env_root / "cache")
    else:
        env["XDG_CACHE_HOME"] = str(env_root / "cache")
    python = str(env_root / "bin/python")
    row = {"tool": tool, "url": URL, "batch_size": 3, "started_at": now(),
           "command": [sys.executable, *sys.argv], "worker_python": python,
           "environment": {key: env[key] for key in ("XDG_CACHE_HOME", "PLAYWRIGHT_BROWSERS_PATH") if key in env},
           "network": "same local host, no proxy", "available_ram_before_gib": round(available / 1073741824, 2),
           "minimum_available_ram_guard_gib": 3,
           "resource_metric_caveat": "Peak sampled PSS/RSS of parent, three workers, and descendants at 0.2-second intervals; may miss short peaks. Sequential batches are not simultaneous cross-tool measurements."}
    processes = []
    start = time.perf_counter()
    for attempt in range(1, 4):
        out = base / f"attempt-{attempt}"
        out.mkdir(parents=True, exist_ok=True)
        with (out / "stdout.log").open("w") as stdout, (out / "stderr.log").open("w") as stderr:
            processes.append(subprocess.Popen([python, __file__, "--worker", tool, str(attempt), str(go_path)],
                                            cwd=Path.cwd(), env=env, stdout=stdout, stderr=stderr,
                                            start_new_session=True))
    samples, stop = [], threading.Event()
    monitor = threading.Thread(target=sample_tree, args=(psutil.Process(), samples, stop), daemon=True)
    monitor.start()
    go_path.write_text(now() + "\n")
    row["released_at"] = now()
    row["available_ram_at_release_gib"] = round(psutil.virtual_memory().available / 1073741824, 2)
    abort_reason = None
    while any(p.poll() is None for p in processes):
        current = psutil.virtual_memory().available
        if current < MIN_AVAILABLE_BYTES:
            abort_reason = f"RAM guard: available {current / 1073741824:.2f} GiB < 3 GiB"
        elif time.perf_counter() - start > 90:
            abort_reason = "90-second batch timeout"
        if abort_reason:
            for p in processes:
                if p.poll() is None:
                    os.killpg(p.pid, signal.SIGTERM)
            break
        time.sleep(0.2)
    exits = [p.wait(timeout=10) for p in processes]
    stop.set()
    monitor.join(timeout=1)
    rows = []
    for attempt in range(1, 4):
        path = base / f"attempt-{attempt}" / "result.json"
        rows.append(json.loads(path.read_text()) if path.exists() else {"attempt": attempt, "error": "worker exited without result.json"})
    row.update(finished_at=now(), batch_wall_seconds=round(time.perf_counter() - start, 3),
               exit_codes=exits, abort_reason=abort_reason, max_navigation_concurrency=max_overlap(rows),
               attempts=[{"attempt": x.get("attempt"), "status": x.get("http_status"),
                          "product_result_count": x.get("product_result_count"),
                          "usable_content": x.get("usable_content"), "wall_seconds": x.get("wall_seconds"),
                          "error": x.get("error")} for x in rows],
               available_ram_after_gib=round(psutil.virtual_memory().available / 1073741824, 2))
    row.update(peaks(samples))
    (base / "batch.json").write_text(json.dumps(row, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(row, indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--worker", action="store_true")
    parser.add_argument("tool", choices=("patchright", "camoufox"))
    parser.add_argument("attempt", nargs="?", type=int)
    parser.add_argument("go_path", nargs="?")
    args = parser.parse_args()
    if args.worker:
        worker(args.tool, args.attempt, Path(args.go_path))
    else:
        batch(args.tool)