#!/usr/bin/env python
r"""Rendered webpage crawler with consent verification and Markdown output.

Core flow:
  1. Open the online page in nodriver/Chrome.
  2. Verify and, if necessary, accept cookie consent on the online page.
  3. Wait until the rendered page content is stable.
  4. Capture rendered HTML from the browser.
  5. Download images referenced by the rendered HTML and rewrite image URLs.
  6. Write HTML, TXT, screenshots, crawl metadata, and optionally Markdown.

The saved HTML is a debug/source artifact. The primary AI artifact is Markdown.
When --vision false, Markdown is generated from the post-consent rendered HTML
with local image links. When --vision true, screenshots are sent to the Gemini
vision pipeline and Markdown is rendered from structured webpage/v1 JSON.
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import inspect
import json
import logging
import os
import re
import sys
import time
import traceback
import unicodedata
from collections import deque
from datetime import UTC, datetime
from pathlib import Path
from urllib.parse import parse_qs, urljoin, urlparse, urlunparse
from urllib.robotparser import RobotFileParser

import nodriver as uc
from bs4 import BeautifulSoup

try:
    from markdownify import markdownify as html_to_markdown
except Exception:  # Optional until --markdown true --vision false is used.
    html_to_markdown = None

try:
    from gemini_detect import analyze_cookie_consent
except Exception:  # Optional until consent vision verification is used.
    analyze_cookie_consent = None

try:
    from gemini_detect import detect_cloudflare_challenge
    from gemini_detect import detect_blocking_overlay
except Exception:
    detect_cloudflare_challenge = None
    detect_blocking_overlay = None

try:
    from webpage_vision_markdown import (
        DEFAULT_GEMINI_MODEL,
        DEFAULT_MAX_ATTEMPTS,
        process_webpage_screenshots,
        write_json as write_pipeline_json,
    )
except Exception:  # Optional dependency path for --vision false.
    DEFAULT_GEMINI_MODEL = "google_gemini.gemini-3.1-flash-lite-preview"
    DEFAULT_MAX_ATTEMPTS = 6
    process_webpage_screenshots = None

    def write_pipeline_json(path, payload):
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


try:
    from crawl_db import (
        check_cache,
        save_page_record,
        save_page_links,
        save_page_images,
        get_link_map,
        CacheEntry,
        canonical_hash,
        is_known_target,
        mark_discovered_target,
        product_exists as product_exists_db,
    )
except Exception:
    check_cache = None
    save_page_record = None
    save_page_links = None
    save_page_images = None
    get_link_map = None
    CacheEntry = None
    canonical_hash = None
    is_known_target = None
    mark_discovered_target = None

try:
    from interactions import (
        discover_interactive_targets,
        execute_interaction,
        capture_modal_page,
        restore_page_state,
    )
except Exception:
    discover_interactive_targets = None
    execute_interaction = None
    capture_modal_page = None
    restore_page_state = None

try:
    from page_classifier import classify_page
except Exception:
    classify_page = None

try:
    from gemini_page_analysis import analyze_page
except Exception:
    analyze_page = None


# ---------------------------------------------------------------------------
# Timing log
# ---------------------------------------------------------------------------

class TimingLog:
    """Per-run wall-clock timing .log + .csv inside ``app/timing_logs/``."""

    TIMING_DIR = Path(os.getenv("CRAWLER_TIMING_DIR", str(Path(__file__).resolve().parent / ".." / "data" / "logs" / "timing_logs"))).resolve()

    def __init__(self, log_path: Path, csv_path: Path, description: str) -> None:
        self._log_path = log_path
        self._csv_path = csv_path
        self._start = time.perf_counter()
        self._lock = __import__("threading").Lock()
        self._description = description.strip()

        self.TIMING_DIR.mkdir(parents=True, exist_ok=True)

        if not self._csv_path.exists():
            self._csv_path.write_text("url|task|time_in_seconds\n", encoding="utf-8")

        self._write_log_header()

    def _write_log_header(self) -> None:
        ts = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
        lines = [
            f"# crawler timing log  {ts}",
            f"# input: {self._description}",
            "",
        ]
        self._log_path.write_text("\n".join(lines) + "\n", encoding="utf-8")

    def _now(self) -> float:
        return time.perf_counter() - self._start

    def record(self, url: str, task: str, elapsed: float) -> None:
        row = f"{url}|{task}|{elapsed:.3f}\n"
        with self._lock:
            with self._csv_path.open("a", encoding="utf-8") as f:
                f.write(row)
            log_line = f"{self._now():9.3f}s  TIMING  {url}|{task}|{elapsed:.3f}s\n"
            with self._log_path.open("a", encoding="utf-8") as lf:
                lf.write(log_line)
        log.info("TIMING %s|%s|%.3fs", url, task, elapsed)


# ---------------------------------------------------------------------------
# Paths and logging
# ---------------------------------------------------------------------------
HERE = Path(__file__).resolve().parent
COOKIES_DIR = HERE / "cookies"
COOKIES_DIR.mkdir(parents=True, exist_ok=True)
CONSENT_TEXTS_FILE = HERE / "consent_texts.json"
SITES_ROOT = (HERE / ".." / "data" / "sites_crawled").resolve()
SITES_ROOT.mkdir(parents=True, exist_ok=True)
LOG_PATH = (HERE / ".." / "data" / "logs" / "crawler.log").resolve()
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
_LOG_START = time.perf_counter()


class ElapsedFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        record.elapsed = f"{time.perf_counter() - _LOG_START:9.3f}s"
        return super().format(record)


def configure_logging(*, verbose: bool = False) -> None:
    """Configure console + crawler.log output. The file log is recreated every run."""
    global _LOG_START
    _LOG_START = time.perf_counter()
    level = logging.DEBUG if verbose else logging.INFO
    root = logging.getLogger()
    for handler in list(root.handlers):
        root.removeHandler(handler)
        try:
            handler.close()
        except Exception:
            pass

    formatter = ElapsedFormatter(
        fmt="%(asctime)s [T+%(elapsed)s] [%(levelname)s] %(message)s",
        datefmt="%H:%M:%S",
    )

    console = logging.StreamHandler(sys.stdout)
    console.setLevel(logging.INFO)
    console.setFormatter(formatter)

    file_handler = logging.FileHandler(LOG_PATH, mode="w", encoding="utf-8")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    root.setLevel(logging.DEBUG)
    root.addHandler(console)
    root.addHandler(file_handler)

    # Always suppress websockets binary-frame debug logs (causes UnicodeEncodeError on Windows cp1252 consoles)
    logging.getLogger("websockets").setLevel(logging.WARNING)
    logging.getLogger("nodriver").setLevel(logging.WARNING)
    logging.getLogger("nodriver.core.connection").setLevel(logging.WARNING)
    logging.getLogger("httpcore").setLevel(logging.WARNING)
    logging.getLogger("httpx").setLevel(logging.WARNING)


configure_logging(verbose=False)
log = logging.getLogger("crawler")
_timing_log: TimingLog | None = None


def _timing_record(url: str, task: str, elapsed: float) -> None:
    if _timing_log is not None:
        _timing_log.record(url, task, elapsed)


# ---------------------------------------------------------------------------
# Safe filenames and output layout
# ---------------------------------------------------------------------------
_TOKEN_SAFE = re.compile(r"[^A-Za-z0-9._-]+")
_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".avif", ".bmp"}
_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
_STATIC_RESOURCE_EXTENSIONS = {
    ".7z", ".avi", ".bmp", ".css", ".csv", ".doc", ".docx",
    ".eot", ".gif", ".gz", ".ico", ".jpeg", ".jpg", ".js",
    ".json", ".m4a", ".m4v", ".mov", ".mp3", ".mp4", ".mpeg",
    ".mpg", ".ogg", ".otf", ".pdf", ".png", ".rar", ".rss",
    ".svg", ".tar", ".tgz", ".ttf", ".txt", ".wav", ".webm",
    ".webp", ".woff", ".woff2", ".xls", ".xlsx", ".xml", ".zip",
}


def utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


def _ascii_filename_text(value: str) -> str:
    value = (value or "").strip()
    value = value.replace("ß", "ss").replace("ẞ", "SS")
    value = value.replace("æ", "ae").replace("Æ", "AE").replace("œ", "oe").replace("Œ", "OE")
    value = unicodedata.normalize("NFKD", value)
    return "".join(ch for ch in value if not unicodedata.combining(ch))


def _safe_token(value: str, fallback: str = "page") -> str:
    value = _ascii_filename_text(value)
    value = value.strip().replace("&", " and ")
    value = _TOKEN_SAFE.sub("_", value)
    value = re.sub(r"_+", "_", value).strip("._-")
    return value or fallback


def _png_dimensions(path: Path) -> tuple[int, int] | None:
    try:
        with path.open("rb") as handle:
            header = handle.read(24)
        if len(header) >= 24 and header.startswith(_PNG_SIGNATURE):
            return int.from_bytes(header[16:20], "big"), int.from_bytes(header[20:24], "big")
    except Exception:
        return None
    return None


def _safe_relative(path: Path, root: Path) -> str:
    try:
        return str(path.relative_to(root))
    except ValueError:
        return str(path)


def _real_host(url: str) -> str:
    return (urlparse(url).hostname or "unknown").lower()


def _domain_slug(url: str) -> str:
    host = _real_host(url)
    if host.startswith("www."):
        host = host[4:]
    labels = [_safe_token(part, "domain").lower() for part in host.split(".") if part]
    return "__".join(labels) or "unknown"


def _url_output_stem(url: str) -> str:
    parsed = urlparse(url)
    domain = _domain_slug(url)
    path = parsed.path.strip("/")

    base = domain
    if path and path.lower() not in {"index", "index.html", "index.htm", "index.php"}:
        parts: list[str] = []
        segments = [segment for segment in path.split("/") if segment]
        for i, segment in enumerate(segments):
            raw = segment
            suffix = Path(raw).suffix.lower()
            if suffix:
                raw = Path(raw).stem
            if i == len(segments) - 1 and raw.lower() == "index":
                continue
            token = _safe_token(raw.lower())
            if token:
                parts.append(token)
        if parts:
            base = domain + "__" + "__".join(parts)

    canonical = canonicalize_url(url)
    if canonical:
        cparsed = urlparse(canonical)
        if cparsed.query:
            qhash = hashlib.sha256((_QUERY_HASH_SEED + cparsed.query).encode()).hexdigest()[:8]
            return f"{base}__q_{qhash}"
    return base


def _image_page_stem(url: str) -> str:
    base = _url_output_stem(url)
    # Ensure root pages get __root suffix for image filenames
    parsed = urlparse(url)
    path = parsed.path.strip("/")
    if not path or path.lower() in {"index", "index.html", "index.htm", "index.php"}:
        if "__q_" in base:
            return base.replace(domain_slug := _domain_slug(url), f"{domain_slug}__root", 1)
        return base + "__root"
    return base

# Store the domain_slug call outside the replace for the root case  ... actually let me rewrite cleaner:
def _image_page_stem(url: str) -> str:
    parsed = urlparse(url)
    domain = _domain_slug(url)
    path = parsed.path.strip("/")
    if not path or path.lower() in {"index", "index.html", "index.htm", "index.php"}:
        base = domain + "__root"
    else:
        parts: list[str] = []
        for segment in [segment for segment in path.split("/") if segment]:
            raw = segment
            suffix = Path(raw).suffix.lower()
            if suffix:
                raw = Path(raw).stem
            if raw.lower() == "index":
                continue
            token = _safe_token(raw.lower())
            if token:
                parts.append(token)
        base = domain + ("__" + "__".join(parts) if parts else "__root")
    canonical = canonicalize_url(url)
    if canonical:
        cparsed = urlparse(canonical)
        if cparsed.query:
            qhash = hashlib.sha256((_QUERY_HASH_SEED + cparsed.query).encode()).hexdigest()[:8]
            return f"{base}__q_{qhash}"
    return base


def _page_filename(url: str) -> str:
    return f"{_url_output_stem(url)}.html"


def output_dir_for(start_url: str) -> Path:
    """Return sites_crawled/<real-host>/ for all artifacts of this domain."""
    d = SITES_ROOT / _real_host(start_url)
    d.mkdir(parents=True, exist_ok=True)
    (d / "images").mkdir(parents=True, exist_ok=True)
    (d / "screenshots").mkdir(parents=True, exist_ok=True)
    return d


def _strip_fragment_and_query(url: str) -> str:
    parsed = urlparse(url)
    return urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", ""))


# ---------------------------------------------------------------------------
# URL canonicalization — preserved query params, dropped trackers
# ---------------------------------------------------------------------------
_TRACKING_PARAMS = {
    "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
    "fbclid", "gclid", "msclkid", "dclid", "yclid", "_ga", "_gid",
    "mc_cid", "mc_eid", "ref", "ref_src", "source", "campaign",
    "rnd", "random", "cache", "cachebuster", "cb", "timestamp", "_ts",
}

_MEANINGFUL_PARAMS = {
    "lang", "language", "locale", "tab", "mountingtype", "mounting_type",
    "product", "productid", "product_id", "family", "productfamily",
    "product_family", "category", "filter", "filters", "sort", "page",
    "q", "search", "query", "variant", "configuration", "config",
    "view", "mode", "type", "id",
}

_QUERY_HASH_SEED = "crawler/v2"


def canonicalize_url(url: str, *, base_url: str | None = None, query_mode: str = "semantic") -> str | None:
    if not url or not url.strip():
        return None
    full = url
    if base_url:
        full = urljoin(base_url, url)
    parsed = urlparse(full)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        return None
    scheme = parsed.scheme.lower()
    host = parsed.hostname.lower()
    port = parsed.port
    if port in (80, 443) or port is None:
        netloc = host
    else:
        netloc = f"{host}:{port}"
    path = parsed.path
    # Normalize duplicate slashes
    while "//" in path:
        path = path.replace("//", "/")
    if not path:
        path = "/"
    if query_mode == "path_only":
        return urlunparse((scheme, netloc, path, "", "", ""))
    qs = parsed.query
    if qs:
        params = {}
        for k, v in parse_qs(qs, keep_blank_values=False).items():
            kl = k.lower()
            if kl in _TRACKING_PARAMS:
                continue
            if query_mode == "full":
                params[k] = v
            else:  # semantic
                if kl in _MEANINGFUL_PARAMS:
                    params[k] = v
                elif len(k) >= 3 and not k.startswith("_"):
                    params[k] = v
        if params:
            sorted_pairs = sorted((k, ",".join(v)) for k, v in params.items())
            qs = "&".join(f"{k}={v}" for k, v in sorted_pairs)
            return urlunparse((scheme, netloc, path, "", qs, ""))
    return urlunparse((scheme, netloc, path, "", "", ""))


# ---------------------------------------------------------------------------
# JSON helpers / resource map
# ---------------------------------------------------------------------------

def write_json(path: Path, payload) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    tmp.replace(path)


def write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(text, encoding="utf-8", newline="\n")
    tmp.replace(path)


def append_resource_map(out_dir: Path, entries: list[dict]) -> None:
    if not entries:
        return
    path = out_dir / "_resource_map.json"
    current: dict = {"schema_version": "crawler-resource-map/v1", "updated_at": utc_now_iso(), "resources": []}
    if path.exists():
        try:
            existing = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(existing, dict) and isinstance(existing.get("resources"), list):
                current = existing
        except Exception:
            pass
    known = {str(item.get("source_url")) for item in current.get("resources", []) if isinstance(item, dict)}
    for entry in entries:
        if str(entry.get("source_url")) in known:
            continue
        current.setdefault("resources", []).append(entry)
        known.add(str(entry.get("source_url")))
    current["updated_at"] = utc_now_iso()
    write_json(path, current)


# ---------------------------------------------------------------------------
# robots.txt
# ---------------------------------------------------------------------------
_robots: dict[str, RobotFileParser] = {}


def _get_robots(domain: str) -> RobotFileParser:
    if domain in _robots:
        return _robots[domain]
    rp = RobotFileParser()
    rp.set_url(f"https://{domain}/robots.txt")
    try:
        rp.read()
    except Exception:
        log.warning("robots.txt unavailable for %s - allowing all", domain)
        rp.allow_all = True
    _robots[domain] = rp
    return rp


def can_fetch(url: str, agent: str = "*") -> bool:
    domain = _real_host(url)
    return _get_robots(domain).can_fetch(agent, url) if domain else True


# ---------------------------------------------------------------------------
# nodriver helper wrappers
# ---------------------------------------------------------------------------
async def _maybe_await(value):
    if inspect.isawaitable(value):
        return await value
    return value


async def await_with_timeout(label: str, value, *, timeout: float, default=None, warning: bool = True):
    """Await a nodriver operation with a hard timeout and visible logging."""
    timeout = max(0.5, float(timeout))
    log.debug("BEGIN %s timeout=%.1fs", label, timeout)
    started = time.perf_counter()
    try:
        result = await asyncio.wait_for(_maybe_await(value), timeout=timeout)
        log.debug("END %s elapsed=%.3fs", label, time.perf_counter() - started)
        return result
    except asyncio.TimeoutError:
        message = f"{label} timed out after {timeout:.1f}s; continuing with fallback/default"
        if warning:
            log.warning(message)
        else:
            log.debug(message)
        return default
    except Exception as exc:
        message = f"{label} failed after {time.perf_counter() - started:.3f}s: {exc}; continuing with fallback/default"
        if warning:
            log.warning(message)
        else:
            log.debug(message)
        return default


def _normalize_evaluate_result(result) -> object:
    """Return a plain Python value from any nodriver/CDP evaluate result shape."""
    if result is None:
        return None
    if isinstance(result, (str, int, float, bool, list)):
        return result
    if isinstance(result, dict):
        if "value" in result and "type" in result:
            inner = result["value"]
            if isinstance(inner, dict) and "value" in inner and "type" in inner:
                return inner["value"]
            return inner
        return result
    if hasattr(result, "value"):
        return result.value
    if hasattr(result, "deep_serialized_value") and result.deep_serialized_value is not None:
        return result.deep_serialized_value.value
    if hasattr(result, "text"):
        return result.text
    if isinstance(result, tuple) and len(result) == 2:
        first, second = result
        if hasattr(first, "value") and first.value is not None:
            return first.value
        if first is not None and not isinstance(first, Exception):
            return first
    return result


async def _probe_evaluate(tab, label: str, url: str) -> None:
    """Diagnostic probe: log details of JS evaluation on a tab."""
    scripts = [
        "() => document.readyState",
        "() => ({href: location.href, readyState: document.readyState, title: document.title})",
    ]
    for script in scripts:
        try:
            raw = await asyncio.wait_for(tab.evaluate(script, return_by_value=True), timeout=1.0)
        except asyncio.TimeoutError:
            log.debug("_probe_evaluate %s TIMEOUT script=%s tab_id=%s url=%s", label, script[:60], id(tab), url)
            continue
        except Exception as exc:
            log.debug("_probe_evaluate %s EXCEPTION script=%s tab_id=%s url=%s exc=%s", label, script[:60], id(tab), url, exc)
            continue
        log.debug(
            "_probe_evaluate %s tab_id=%s url=%s raw_type=%s raw_repr=%s normalized_type=%s",
            label, id(tab), url, type(raw).__name__, repr(raw)[:500], type(_normalize_evaluate_result(raw)).__name__,
        )
        normalized = _normalize_evaluate_result(raw)
        if isinstance(normalized, dict) and normalized.get("href"):
            log.debug("_probe_evaluate %s current_url=%s tab_id=%s", label, normalized.get("href"), id(tab))


async def evaluate_value(tab, script: str, *, label: str = "evaluate") -> object:
    """Evaluate JS and return a plain Python value across nodriver/CDP versions.
    Tries three bounded paths and returns None only after all fail."""
    # Path 1: return_by_value (primary nodriver API)
    try:
        result = await asyncio.wait_for(
            tab.evaluate(script, return_by_value=True),
            timeout=5.0,
        )
        normalized = _normalize_evaluate_result(result)
        if normalized is not None and not (isinstance(normalized, dict) and normalized.get("type") == "function"):
            return normalized
    except asyncio.TimeoutError:
        log.debug("evaluate_value path1 timeout for %s", label)
    except Exception as exc:
        log.debug("evaluate_value path1 failed for %s: %s", label, exc)

    # Path 2: bare evaluate (older nodriver return shape)
    try:
        result = await asyncio.wait_for(
            tab.evaluate(script),
            timeout=5.0,
        )
        normalized = _normalize_evaluate_result(result)
        if normalized is not None and not (isinstance(normalized, dict) and normalized.get("type") == "function"):
            return normalized
    except asyncio.TimeoutError:
        log.debug("evaluate_value path2 timeout for %s", label)
    except Exception as exc:
        log.debug("evaluate_value path2 failed for %s: %s", label, exc)

    # Path 3: typed CDP fallback with await_promise
    try:
        from nodriver.cdp.runtime import evaluate as cdp_evaluate
        raw = await asyncio.wait_for(
            tab.send(cdp_evaluate(script, await_promise=True, return_by_value=True, user_gesture=True, allow_unsafe_eval_blocked_by_csp=True)),
            timeout=5.0,
        )
        normalized = _normalize_evaluate_result(raw)
        if normalized is not None:
            return normalized
    except asyncio.TimeoutError:
        log.debug("evaluate_value path3 timeout for %s", label)
    except Exception as exc:
        log.debug("evaluate_value path3 failed for %s: %s", label, exc)

    return None


async def save_viewport_screenshot(
    tab,
    path: Path,
    *,
    expected_width: int | None = None,
    expected_height: int | None = None,
    full_page: bool = False,
) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)

    async def capture_once(label: str) -> None:
        log.debug("Saving screenshot via nodriver Tab.save_screenshot full_page=%s: %s", full_page, path)
        try:
            await await_with_timeout(
                label,
                tab.save_screenshot(str(path), format="png", full_page=full_page),
                timeout=30.0,
                default=None,
                warning=True,
            )
        except TypeError:
            log.warning("Tab.save_screenshot(..., full_page=%s) was not accepted; retrying with older nodriver signature", full_page)
            try:
                await await_with_timeout(label + " old-format", tab.save_screenshot(str(path), format="png"), timeout=30.0)
            except TypeError:
                log.warning("Tab.save_screenshot(..., format='png') was not accepted; retrying with filename only")
                await await_with_timeout(label + " filename-only", tab.save_screenshot(str(path)), timeout=30.0)

    await capture_once("nodriver Tab.save_screenshot()")
    actual = _png_dimensions(path)
    if actual and expected_width and expected_height and not full_page and actual != (int(expected_width), int(expected_height)):
        expected = (int(expected_width), int(expected_height))
        dx = max(0, expected[0] - actual[0])
        dy = max(0, expected[1] - actual[1])
        mismatch_pct_x = (dx / max(expected[0], 1)) * 100
        mismatch_pct_y = (dy / max(expected[1], 1)) * 100
        if mismatch_pct_x < 5 and mismatch_pct_y < 10:
            log.info("Screenshot pixel size mismatch is within tolerance for %s: expected %sx%s, got %sx%s (dx=%s %.1f%%, dy=%s %.1f%%)",
                     _safe_relative(path, SITES_ROOT), expected[0], expected[1], actual[0], actual[1],
                     dx, mismatch_pct_x, dy, mismatch_pct_y)
        else:
            log.warning("Screenshot pixel size mismatch for %s: expected %sx%s, got %sx%s (dx=%s, dy=%s)",
                        _safe_relative(path, SITES_ROOT), expected[0], expected[1], actual[0], actual[1], dx, dy)
    elif actual:
        log.info("Saved viewport screenshot %s (%sx%s)", _safe_relative(path, SITES_ROOT), actual[0], actual[1])
    return path


async def get_tab_content(tab, *, timeout: float = 10.0, required: bool = False) -> str:
    """Get rendered HTML. Primary path uses documented nodriver Tab.get_content()."""
    html = await await_with_timeout(
        "nodriver Tab.get_content()",
        tab.get_content(),
        timeout=timeout,
        default=None,
        warning=True,
    )
    if isinstance(html, str) and html.strip():
        log.debug("Rendered HTML captured through Tab.get_content(): %d characters", len(html))
        return html

    log.warning("Falling back to document.documentElement.outerHTML because Tab.get_content() returned no usable HTML")
    fallback = await await_with_timeout(
        "fallback DOM outerHTML evaluation",
        evaluate_value(tab, "() => document.documentElement ? document.documentElement.outerHTML : ''"),
        timeout=min(max(2.0, float(timeout)), 10.0),
        default="",
        warning=True,
    )
    if isinstance(fallback, str) and fallback.strip():
        log.info("Rendered HTML captured through fallback DOM outerHTML: %d characters", len(fallback))
        return fallback
    if required:
        raise RuntimeError("Could not capture rendered HTML from the browser")
    return ""


def _browser_args_for_viewport(width: int, height: int, *, docker: bool = False) -> list[str]:
    args: list[str] = []
    if docker:
        args.extend([
            "--no-sandbox",
            "--disable-dev-shm-usage",
            "--disable-gpu",
            "--max_old_space_size=2048",
        ])
    if width > 0 and height > 0:
        args.extend([
            f"--window-size={int(width)},{int(height)}",
            f"--screen-info={{0,0 {int(width)}x{int(height)}}}",
            "--window-position=0,0",
            "--force-device-scale-factor=1",
            "--high-dpi-support=1",
            "--hide-scrollbars",
        ])
    return args


async def set_and_verify_viewport(tab, width: int, height: int) -> dict:
    if width > 0 and height > 0:
        await await_with_timeout(
            "nodriver Tab.set_window_state()",
            tab.set_window_state(left=0, top=0, width=int(width), height=int(height), state="normal"),
            timeout=5.0,
            default=None,
            warning=False,
        )
        await await_with_timeout(
            "nodriver Tab.set_window_size()",
            tab.set_window_size(left=0, top=0, width=int(width), height=int(height)),
            timeout=5.0,
            default=None,
            warning=False,
        )
    await await_with_timeout("nodriver tab sleep before viewport check", tab.sleep(0.5), timeout=2.0, default=None, warning=False)
    metrics = await await_with_timeout(
        "viewport metrics evaluation",
        evaluate_value(
            tab,
            """
            () => ({
              width: window.innerWidth || document.documentElement.clientWidth || 0,
              height: window.innerHeight || document.documentElement.clientHeight || 0,
              devicePixelRatio: window.devicePixelRatio || 1,
              url: location.href,
              title: document.title || ''
            })
            """,
        ),
        timeout=5.0,
        default=None,
        warning=True,
    )
    if not isinstance(metrics, dict):
        metrics = {"width": width, "height": height, "devicePixelRatio": 1, "url": "", "title": ""}
    if width and height and (int(metrics.get("width") or 0) != width or int(metrics.get("height") or 0) != height):
        log.warning(
            "Requested viewport %sx%s but browser reports %sx%s",
            width,
            height,
            metrics.get("width"),
            metrics.get("height"),
        )
    else:
        log.info("Viewport verified: %sx%s", metrics.get("width"), metrics.get("height"))
    return metrics


# ---------------------------------------------------------------------------
# Cookie persistence
# ---------------------------------------------------------------------------

def _cookie_file(domain: str) -> Path:
    return COOKIES_DIR / f"{domain}.json"


def _domain_cookie_match(cookie_domain: str, host: str) -> bool:
    cookie_domain = (cookie_domain or "").lstrip(".").lower()
    host = (host or "").lower()
    root = host[4:] if host.startswith("www.") else host
    return cookie_domain in {host, root} or host.endswith("." + cookie_domain) or cookie_domain.endswith("." + root)


def _raw_cdp(method: str, params: dict | None = None):
    result = yield {"method": method, "params": params or {}}
    return result


def _browser_cdp_connection(browser):
    for tab in browser:
        if getattr(tab, "target", None):
            return tab
    return browser


async def _raw_get_domain_cookies(browser, domain: str) -> list[dict]:
    connection = _browser_cdp_connection(browser)
    raw = await connection.send(_raw_cdp("Storage.getCookies"))
    cookies = raw.get("cookies", []) if isinstance(raw, dict) else []
    return [cookie for cookie in cookies if isinstance(cookie, dict) and _domain_cookie_match(str(cookie.get("domain") or ""), domain)]


def _cookie_to_set_param(cookie: dict) -> dict | None:
    if not cookie.get("name"):
        return None
    allowed = {
        "name", "value", "domain", "path", "secure", "httpOnly", "sameSite", "expires",
        "priority", "sourceScheme", "sourcePort", "partitionKey",
    }
    item = {key: cookie[key] for key in allowed if key in cookie and cookie[key] is not None}
    if item.get("expires") == -1:
        item.pop("expires", None)
    item.setdefault("path", "/")
    item.setdefault("value", "")
    return item


async def load_browser_cookies(browser, domain: str) -> bool:
    path = _cookie_file(domain)
    if not path.exists():
        log.info("No saved cookies found for %s at %s", domain, path)
        return False
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        cookies = payload.get("cookies", payload) if isinstance(payload, dict) else payload
        params = [_cookie_to_set_param(cookie) for cookie in cookies if isinstance(cookie, dict)]
        params = [item for item in params if item]
        if not params:
            log.info("No usable saved cookies found for %s at %s", domain, path)
            return False
        result = await await_with_timeout(
            f"raw CDP Storage.setCookies({domain})",
            _browser_cdp_connection(browser).send(_raw_cdp("Storage.setCookies", {"cookies": params})),
            timeout=10.0,
            default="__crawler_timeout_or_failure__",
            warning=True,
        )
        if result == "__crawler_timeout_or_failure__":
            log.warning("Could not load cookies for %s within timeout. Continuing without saved cookies.", domain)
            return False
    except Exception as exc:
        log.warning("Could not load cookies for %s from %s: %s", domain, path, exc)
        return False
    log.info("Loaded %d cookies for %s from %s", len(params), domain, path)
    return True


async def save_browser_cookies(browser, domain: str) -> bool:
    path = _cookie_file(domain)
    try:
        result = await await_with_timeout(
            f"raw CDP Storage.getCookies({domain})",
            _raw_get_domain_cookies(browser, domain),
            timeout=10.0,
            default="__crawler_timeout_or_failure__",
            warning=True,
        )
        if result == "__crawler_timeout_or_failure__":
            log.warning("Could not save cookies for %s within timeout. Continuing crawl without blocking.", domain)
            return False
        payload = {"schema_version": "crawler-cookies/v1", "domain": domain, "saved_at": utc_now_iso(), "cookies": result}
        write_json(path, payload)
        log.info("Saved %d cookies for %s to %s", len(result), domain, path)
        return True
    except Exception as exc:
        log.warning("Could not save cookies for %s: %s", domain, exc)
        return False


async def close_tab_safely(tab, label: str, *, timeout: float = 5.0) -> bool:
    """Close a nodriver tab without allowing tab.close() to block the crawl."""
    if tab is None:
        return True
    task = asyncio.create_task(_maybe_await(tab.close()))
    done, pending = await asyncio.wait({task}, timeout=max(0.5, float(timeout)))
    if pending:
        log.warning("Could not close tab for %s within %.1fs; continuing", label, timeout)

        def _consume_late_close_result(late_task: asyncio.Task) -> None:
            try:
                late_task.result()
            except Exception:
                pass

        task.add_done_callback(_consume_late_close_result)
        return False
    try:
        task.result()
        return True
    except Exception as exc:
        log.debug("Closing tab for %s failed: %s", label, exc)
        return False


# ---------------------------------------------------------------------------
# Learned consent texts
# ---------------------------------------------------------------------------

def load_learned_consent_texts(domain: str) -> list[str]:
    if not CONSENT_TEXTS_FILE.exists():
        return []
    try:
        data = json.loads(CONSENT_TEXTS_FILE.read_text(encoding="utf-8"))
        result: list[str] = []
        if isinstance(data.get("by_domain"), dict) and isinstance(data["by_domain"].get(domain), list):
            result.extend(str(item) for item in data["by_domain"][domain] if str(item).strip())
        if isinstance(data.get("texts"), list):
            result.extend(str(item) for item in data["texts"] if str(item).strip())
        # Prefer explicit accept-all labels, de-prioritize generic OK.
        return sorted(set(result), key=lambda text: (text.strip().lower() in {"ok", "okay"}, len(text)))
    except Exception:
        return []


def save_learned_consent_text(domain: str, text: str | None) -> None:
    text = (text or "").strip()
    if not text:
        return
    data = {"schema_version": "crawler-consent-texts/v1", "updated_at": utc_now_iso(), "texts": [], "by_domain": {}}
    if CONSENT_TEXTS_FILE.exists():
        try:
            loaded = json.loads(CONSENT_TEXTS_FILE.read_text(encoding="utf-8"))
            if isinstance(loaded, dict):
                data.update(loaded)
        except Exception:
            pass
    data.setdefault("texts", [])
    data.setdefault("by_domain", {})
    if text not in data["texts"]:
        data["texts"].append(text)
    data["by_domain"].setdefault(domain, [])
    if text not in data["by_domain"][domain]:
        data["by_domain"][domain].append(text)
    data["updated_at"] = utc_now_iso()
    write_json(CONSENT_TEXTS_FILE, data)
    log.info("Learned consent text for %s: %s", domain, text)


BUILTIN_CONSENT_TEXTS = [
    "Alle akzeptieren", "Alle Cookies akzeptieren", "Allen zustimmen", "Ich stimme zu",
    "Akzeptieren", "Zustimmen", "Einverstanden", "Verstanden", "OK", "Okay",
    "Accept all cookies", "Accept all", "Allow all", "I agree", "Agree", "Accept", "Got it",
]


# ---------------------------------------------------------------------------
# Cookie consent verification
# ---------------------------------------------------------------------------
async def _dom_consent_check(tab) -> bool | None:
    """Cheap DOM check for common cookie consent overlays. Returns True if consent detected, False if clear, None if uncertain."""
    script = """
    () => {
      const selectors = [
        '[id*="cookie"]', '[class*="cookie"]',
        '[id*="consent"]', '[class*="consent"]',
        '[id*="gdpr"]', '[class*="gdpr"]',
        '[id*="cc-banner"]', '[class*="cc-banner"]',
        '[id*="privacy"]', '[class*="privacy"]',
        '[aria-label*="cookie" i]', '[aria-label*="consent" i]',
      ];
      for (const sel of selectors) {
        try {
          const el = document.querySelector(sel);
          if (el && el.offsetParent !== null) {
            const text = (el.textContent || '').toLowerCase();
            const accept_words = ['accept', 'agree', 'allow', 'consent', 'zustimmen', 'akzeptieren', 'einverstanden', 'ok', 'got it'];
            if (accept_words.some(w => text.includes(w))) {
              return true;
            }
          }
        } catch(e) {}
      }
      return false;
    }
    """
    try:
        result = await evaluate_value(tab, script)
        return bool(result) if result is not None else None
    except Exception:
        return None

async def _try_click_consent_by_text(tab, texts: list[str]) -> str | None:
    for text in texts:
        if not text or len(text.strip()) < 2:
            continue
        try:
            el = await tab.find(text, best_match=True, timeout=2)
            if el:
                await el.click()
                log.info("Consent click by visible text: %r", text)
                return text
        except Exception as exc:
            log.debug("Consent text click failed for %r: %s", text, exc)
    return None


def _cloudflare_provider(challenge: dict | None) -> str:
    provider = str((challenge or {}).get("provider") or (challenge or {}).get("cloudflare_provider") or "cloudflare").strip().lower()
    if provider not in {"cloudflare", "akamai", "other", "none"}:
        provider = "other" if provider else "cloudflare"
    return provider if provider != "none" else "cloudflare"


def _cloudflare_confidence(challenge: dict | None) -> float:
    try:
        return max(0.0, min(1.0, float((challenge or {}).get("confidence", (challenge or {}).get("cloudflare_confidence", 0.0)) or 0.0)))
    except Exception:
        return 0.0


def _cloudflare_visible_text(challenge: dict | None) -> str:
    return str((challenge or {}).get("visible_text") or (challenge or {}).get("cloudflare_detected_text") or "").strip()


def _is_cloudflare_blocked(result: dict | None) -> bool:
    return bool(result) and (
        result.get("status") == "blocked_by_cloudflare"
        or bool(result.get("blocked_by_cloudflare"))
        or (bool(result.get("is_challenge")) and _cloudflare_provider(result) != "none")
    )


def _blocked_consent_result(domain: str, challenge: dict, *, enabled: bool = True) -> dict:
    provider = _cloudflare_provider(challenge)
    confidence = _cloudflare_confidence(challenge)
    visible_text = _cloudflare_visible_text(challenge)
    result = {
        "schema_version": "crawler-cookie-consent/v1",
        "checked_at": utc_now_iso(),
        "enabled": bool(enabled),
        "domain": domain,
        "status": "blocked_by_cloudflare",
        "attempts": [],
        "blocked_by_cloudflare": True,
        "blocked_reason": "cloudflare",
        "provider": provider,
        "cloudflare_provider": provider,
        "confidence": confidence,
        "cloudflare_confidence": confidence,
        "visible_text": visible_text,
        "cloudflare_detected_text": visible_text,
        "cloudflare_check": challenge,
    }
    if challenge.get("screenshot"):
        result["screenshot"] = challenge.get("screenshot")
    return result


async def _detect_cloudflare_challenge_on_viewport(
    tab,
    *,
    domain: str,
    out_dir: Path,
    page_stem: str,
    viewport: dict,
    viewport_width: int,
    viewport_height: int,
    source_url: str = "",
    label: str = "challenge_check",
) -> dict:
    if detect_cloudflare_challenge is None:
        return {"checked": False, "is_challenge": False, "provider": "none", "confidence": 0.0, "visible_text": "", "reason": "detector_unavailable"}

    screenshots_dir = out_dir / "screenshots"
    screenshots_dir.mkdir(parents=True, exist_ok=True)
    screenshot_path = screenshots_dir / f"{page_stem}__{label}.png"
    await save_viewport_screenshot(
        tab,
        screenshot_path,
        expected_width=int(viewport.get("width") or viewport_width or 0),
        expected_height=int(viewport.get("height") or viewport_height or 0),
    )
    if not screenshot_path.exists():
        return {"checked": False, "is_challenge": False, "provider": "none", "confidence": 0.0, "visible_text": "", "reason": "screenshot_failed"}

    t_detect = time.perf_counter()
    try:
        challenge = detect_cloudflare_challenge(screenshot_path)
    except Exception as exc:
        log.warning("Gemini bot-protection challenge check failed for %s: %s", page_stem, exc)
        return {"checked": False, "is_challenge": False, "provider": "none", "confidence": 0.0, "visible_text": "", "screenshot": str(screenshot_path), "reason": str(exc)}
    _timing_record(source_url, "cloudflare_detect", time.perf_counter() - t_detect)

    challenge = dict(challenge or {})
    challenge["checked"] = True
    challenge["screenshot"] = str(screenshot_path)
    challenge["domain"] = domain
    log.info(
        "Bot-protection check for %s: challenge=%s provider=%s confidence=%.2f",
        page_stem,
        bool(challenge.get("is_challenge")),
        challenge.get("provider"),
        _cloudflare_confidence(challenge),
    )
    return challenge


async def _quick_cloudflare_challenge_check(
    tab,
    *,
    domain: str,
    out_dir: Path,
    page_stem: str,
    url: str,
    viewport_width: int,
    viewport_height: int,
    wait: float = 0.5,
) -> dict:
    await tab.sleep(wait)
    viewport = await set_and_verify_viewport(tab, viewport_width, viewport_height)
    return await _detect_cloudflare_challenge_on_viewport(
        tab,
        domain=domain,
        out_dir=out_dir,
        page_stem=page_stem,
        viewport=viewport,
        viewport_width=viewport_width,
        viewport_height=viewport_height,
        source_url=url,
        label="quick_challenge",
    )


def _write_cloudflare_blocked_artifacts(out_dir: Path, page_stem: str, url: str, challenge: dict | None) -> Path:
    challenge = dict(challenge or {})
    provider = _cloudflare_provider(challenge)
    confidence = _cloudflare_confidence(challenge)
    visible_text = _cloudflare_visible_text(challenge)
    marker_path = out_dir / f"{page_stem}_blocked.json"
    marker = {
        "schema_version": "crawler-blocked-page/v1",
        "generated_at": utc_now_iso(),
        "url": url,
        "status": "blocked",
        "blocked_reason": "cloudflare",
        "blocked_by": "cloudflare",
        "provider": provider,
        "confidence": confidence,
        "detected_text": visible_text,
        "visible_text": visible_text,
        "screenshot": challenge.get("screenshot"),
    }
    write_json(marker_path, marker)

    crawl_metadata = {
        "schema_version": "crawler-page/v2",
        "generated_at": utc_now_iso(),
        "url": url,
        "status": "blocked",
        "blocked_by_cloudflare": True,
        "cloudflare_provider": provider,
        "cloudflare_confidence": confidence,
        "cloudflare_detected_text": visible_text,
        "blocked_marker_path": str(marker_path.resolve()),
        "cloudflare_screenshot": challenge.get("screenshot"),
        "html_path": None,
        "txt_path": None,
        "images_manifest_path": None,
        "images_count": 0,
        "screenshots_enabled": False,
        "screenshot_metadata": [],
        "markdown_enabled": False,
        "vision_enabled": False,
        "markdown_path": None,
        "structured_json_path": None,
        "render_metrics": {},
    }
    write_json(out_dir / f"{page_stem}.crawl.json", crawl_metadata)
    log.warning("Page blocked by %s challenge: %s", provider, url)
    log.info("Saved %s", _safe_relative(marker_path, SITES_ROOT))
    return marker_path


async def dismiss_blocking_overlays(
    tab,
    *,
    domain: str,
    out_dir: Path,
    page_stem: str,
    viewport_width: int = 1920,
    viewport_height: int = 1080,
    max_attempts: int = 3,
    source_url: str = "",
) -> dict:
    result = {
        "overlays_found": 0,
        "overlays_dismissed": 0,
        "attempts": [],
    }
    if detect_blocking_overlay is None:
        return result

    screenshots_dir = out_dir / "screenshots"
    screenshots_dir.mkdir(parents=True, exist_ok=True)

    for attempt in range(1, max_attempts + 1):
        viewport = await set_and_verify_viewport(tab, viewport_width, viewport_height)
        await tab.sleep(0.5)
        screenshot_path = screenshots_dir / f"{page_stem}__overlay_check_{attempt:02d}.png"
        await save_viewport_screenshot(
            tab, screenshot_path,
            expected_width=int(viewport.get("width") or viewport_width),
            expected_height=int(viewport.get("height") or viewport_height),
        )
        if not screenshot_path.exists():
            break

        screenshot_dims = _png_dimensions(screenshot_path)
        t_detect = time.perf_counter()
        try:
            detection = detect_blocking_overlay(screenshot_path)
        except Exception as exc:
            log.warning("Overlay detection attempt %d failed: %s", attempt, exc)
            break
        _timing_record(source_url, f"overlay_detect_{attempt}", time.perf_counter() - t_detect)

        attempt_record = {
            "attempt": attempt,
            "screenshot": str(screenshot_path),
            "detection": detection,
        }

        if not detection.get("has_blocking_overlay"):
            log.info("No blocking overlay detected (attempt %d)", attempt)
            result["attempts"].append(attempt_record)
            break

        result["overlays_found"] += 1
        log.info(
            "Blocking overlay detected (attempt %d): coverage=%s close_visible=%s confidence=%.2f",
            attempt, detection.get("overlay_coverage"), detection.get("close_button_visible"),
            detection.get("confidence", 0),
        )

        box = detection.get("close_button_box_2d")
        if not detection.get("close_button_visible") or not isinstance(box, list) or len(box) != 4:
            log.warning("Overlay detected but no close button found; trying Escape key")
            try:
                await evaluate_value(tab, "() => { document.dispatchEvent(new KeyboardEvent('keydown', {key: 'Escape', bubbles: true})); return true; }")
            except Exception:
                pass
            await tab.sleep(1.0)
            result["attempts"].append(attempt_record)
            continue

        width = int(screenshot_dims[0]) if screenshot_dims else int(viewport.get("width") or viewport_width)
        height = int(screenshot_dims[1]) if screenshot_dims else int(viewport.get("height") or viewport_height)
        x0, y0, x1, y1 = [float(v) for v in box]
        click_x = int(round(((x0 + x1) / 2.0 / 1000.0) * width))
        click_y = int(round(((y0 + y1) / 2.0 / 1000.0) * height))
        log.info("Clicking overlay close button at pixel x=%d y=%d", click_x, click_y)
        try:
            await tab.mouse_click(click_x, click_y)
        except Exception as exc:
            log.warning("Overlay close click failed: %s", exc)
        await tab.sleep(1.0)
        result["overlays_dismissed"] += 1
        attempt_record["clicked"] = {"x": click_x, "y": click_y}
        result["attempts"].append(attempt_record)

    return result


async def _try_click_consent_by_box(tab, box: list[float] | None, viewport: dict) -> tuple[int, int] | None:
    if not isinstance(box, list) or len(box) != 4:
        return None
    try:
        x0, y0, x1, y1 = [float(v) for v in box]
        width = int(viewport.get("width") or 0) or 1920
        height = int(viewport.get("height") or 0) or 1080
        x = int(round(((x0 + x1) / 2.0 / 1000.0) * width))
        y = int(round(((y0 + y1) / 2.0 / 1000.0) * height))
        await tab.mouse_click(x, y)
        log.info("Consent click by viewport pixel: x=%s y=%s", x, y)
        return x, y
    except Exception as exc:
        log.warning("Consent pixel click failed: %s", exc)
        return None


async def verify_and_accept_cookie_consent(
    browser,
    tab,
    *,
    domain: str,
    out_dir: Path,
    page_stem: str,
    viewport_width: int,
    viewport_height: int,
    enabled: bool = True,
    wait_before_check: float = 3.0,
    max_attempts: int = 2,
    source_url: str = "",
    cloudflare_check_enabled: bool = True,
) -> dict:
    """Verify cookie state via screenshot/Gemini and click accept when needed."""
    result = {
        "schema_version": "crawler-cookie-consent/v1",
        "checked_at": utc_now_iso(),
        "enabled": bool(enabled),
        "domain": domain,
        "status": "not_checked",
        "attempts": [],
    }
    if not enabled:
        result["status"] = "disabled"
        return result

    screenshots_dir = out_dir / "screenshots"
    screenshots_dir.mkdir(parents=True, exist_ok=True)
    await set_and_verify_viewport(tab, viewport_width, viewport_height)
    await tab.sleep(wait_before_check)

    viewport = await set_and_verify_viewport(tab, viewport_width, viewport_height)
    if cloudflare_check_enabled:
        challenge = await _detect_cloudflare_challenge_on_viewport(
            tab,
            domain=domain,
            out_dir=out_dir,
            page_stem=page_stem,
            viewport=viewport,
            viewport_width=viewport_width,
            viewport_height=viewport_height,
            source_url=source_url,
            label="challenge_check",
        )
        result["cloudflare_check"] = challenge
        if challenge.get("is_challenge"):
            result.update(_blocked_consent_result(domain, challenge, enabled=enabled))
            return result

    if analyze_cookie_consent is None:
        log.warning("Gemini consent verifier unavailable; falling back to text selectors only")
        clicked = await _try_click_consent_by_text(tab, load_learned_consent_texts(domain) + BUILTIN_CONSENT_TEXTS)
        if clicked:
            await tab.sleep(1.5)
            await save_browser_cookies(browser, domain)
            result["status"] = "clicked_without_vision_verification"
        else:
            result["status"] = "no_click_without_vision_verification"
        return result

    before_html_saved = False
    for attempt in range(1, max_attempts + 1):
        screenshot_path = screenshots_dir / f"{page_stem}__consent_check_{attempt:02d}_before.png"
        await save_viewport_screenshot(
            tab,
            screenshot_path,
            expected_width=int(viewport.get("width") or viewport_width or 0),
            expected_height=int(viewport.get("height") or viewport_height or 0),
        )
        t_detect = time.perf_counter()
        try:
            check = analyze_cookie_consent(screenshot_path)
        except Exception as exc:
            log.warning("Gemini consent check failed completely: %s; falling back to text selectors", exc)
            clicked = await _try_click_consent_by_text(tab, load_learned_consent_texts(domain) + BUILTIN_CONSENT_TEXTS)
            if clicked:
                await tab.sleep(1.5)
                await save_browser_cookies(browser, domain)
                result["status"] = "clicked_without_vision_verification"
            else:
                result["status"] = "no_click_without_vision_verification"
            return result
        _timing_record(source_url, "consent_detect", time.perf_counter() - t_detect)
        result["attempts"].append({"phase": "before_click", "attempt": attempt, "screenshot": str(screenshot_path), "result": check})
        log.info(
            "Consent check before click: shown=%s action=%s text=%r confidence=%.2f",
            check.get("cookie_consent_shown"),
            check.get("recommended_action"),
            check.get("accept_button_text"),
            float(check.get("confidence") or 0),
        )

        if not check.get("cookie_consent_shown"):
            result["status"] = "not_shown"
            await save_browser_cookies(browser, domain)
            return result

        if not before_html_saved:
            try:
                before_html = await get_tab_content(tab, timeout=5.0, required=False)
                if before_html:
                    write_text(out_dir / f"{page_stem}__before_cookie.html", before_html)
                    before_html_saved = True
                    log.info("Saved before-cookie debug HTML for %s", page_stem)
                else:
                    log.warning("Before-cookie debug HTML was empty; continuing without saving it")
            except Exception as exc:
                log.warning("Skipping before-cookie debug HTML because capture failed: %s", exc)

        texts = []
        if check.get("accept_button_text"):
            texts.append(str(check.get("accept_button_text")))
        texts.extend(load_learned_consent_texts(domain))
        texts.extend(BUILTIN_CONSENT_TEXTS)
        t_click = time.perf_counter()
        clicked_text = await _try_click_consent_by_text(tab, texts)
        clicked_pixel = None
        if not clicked_text:
            clicked_pixel = await _try_click_consent_by_box(tab, check.get("accept_button_box_2d"), viewport)
        _timing_record(source_url, "consent_click", time.perf_counter() - t_click)

        if clicked_text:
            save_learned_consent_text(domain, clicked_text)
        elif check.get("accept_button_text"):
            save_learned_consent_text(domain, str(check.get("accept_button_text")))

        if not clicked_text and clicked_pixel is None:
            result["status"] = "visible_but_click_failed"
            continue

        await tab.sleep(2.0)
        try:
            await await_with_timeout("nodriver tab breathe after consent click", tab, timeout=5.0, default=None, warning=True)
        except Exception as exc:
            log.warning("tab breathe after consent click failed: %s", exc)
        viewport = await set_and_verify_viewport(tab, viewport_width, viewport_height)
        post_path = screenshots_dir / f"{page_stem}__consent_check_{attempt:02d}_after.png"
        await save_viewport_screenshot(
            tab,
            post_path,
            expected_width=int(viewport.get("width") or viewport_width or 0),
            expected_height=int(viewport.get("height") or viewport_height or 0),
        )
        if not post_path.exists():
            log.warning("Post-consent screenshot could not be saved; assuming consent was accepted")
            await save_browser_cookies(browser, domain)
            result["status"] = "accepted_screenshot_failed"
            return result
        t_verify = time.perf_counter()
        try:
            post_check = analyze_cookie_consent(post_path)
        except Exception as exc:
            log.warning("Gemini post-consent check failed: %s; assuming consent accepted", exc)
            await save_browser_cookies(browser, domain)
            result["status"] = "accepted_verification_failed"
            return result
        _timing_record(source_url, "consent_verify", time.perf_counter() - t_verify)
        result["attempts"].append({"phase": "after_click", "attempt": attempt, "screenshot": str(post_path), "result": post_check})
        log.info(
            "Consent check after click: shown=%s confidence=%.2f",
            post_check.get("cookie_consent_shown"),
            float(post_check.get("confidence") or 0),
        )

        if not post_check.get("cookie_consent_shown"):
            await save_browser_cookies(browser, domain)
            result["status"] = "accepted_and_verified"
            try:
                after_html = await get_tab_content(tab, timeout=5.0, required=False)
                if after_html:
                    write_text(out_dir / f"{page_stem}__after_cookie.html", after_html)
                    log.info("Saved after-cookie debug HTML for %s", page_stem)
                else:
                    log.warning("After-cookie debug HTML was empty; continuing to crawl")
            except Exception as exc:
                log.warning("Skipping after-cookie debug HTML because capture failed or timed out: %s", exc)
            log.info("Consent phase complete with status=%s; continuing crawl", result["status"])
            return result

    result["status"] = "still_visible_after_click"
    return result


async def _quick_consent_check(
    tab,
    *,
    domain: str,
    out_dir: Path,
    page_stem: str,
    viewport_width: int,
    viewport_height: int,
) -> bool:
    """Quick check with adaptive wait: starts fast (0.5s), falls back to 5s if consent detected."""
    if analyze_cookie_consent is None:
        log.debug("Gemini consent verifier unavailable; skipping consent check for %s", page_stem)
        return True

    screenshots_dir = out_dir / "screenshots"
    screenshots_dir.mkdir(parents=True, exist_ok=True)
    await set_and_verify_viewport(tab, viewport_width, viewport_height)

    async def _take_and_check(wait: float, label: str) -> bool:
        await tab.sleep(wait)
        check_path = screenshots_dir / f"{page_stem}__quick_consent_{label}.png"
        await save_viewport_screenshot(
            tab,
            check_path,
            expected_width=viewport_width,
            expected_height=viewport_height,
        )
        if not check_path.exists():
            log.warning("Quick consent screenshot could not be captured for %s; proceeding", page_stem)
            return False
        try:
            check = analyze_cookie_consent(check_path)
            shown = check.get("cookie_consent_shown", False)
            log.info("Quick consent check (%s) for %s: shown=%s confidence=%.2f", label, page_stem, shown, float(check.get("confidence") or 0))
            return shown
        except Exception as exc:
            log.warning("Quick consent check failed for %s: %s; proceeding", page_stem, exc)
            return False

    if not await _take_and_check(0.5, "fast"):
        return True

    log.info("Consent detected on %s with fast check; retrying with 5s wait", page_stem)
    if not await _take_and_check(5.0, "slow"):
        return True

    log.warning("Cookie consent still visible on %s after 5s; may need manual handling", page_stem)
    return False


# ---------------------------------------------------------------------------
# Page readiness
# ---------------------------------------------------------------------------
async def wait_for_rendered_content_stable(tab, *, timeout: float = 10.0, stable_cycles: int = 2) -> dict:
    """Wait until rendered content stops changing. Fast-path: one readiness check then immediate return."""
    deadline = time.time() + timeout
    readiness_script = """
    () => {
      const doc = document.documentElement || {};
      const body = document.body || {};
      return {
        readyState: document.readyState,
        textLength: (body.innerText || '').replace(/\\s+/g, ' ').trim().length,
        imageCount: (document.images || []).length,
        scrollHeight: Math.max(doc.scrollHeight || 0, body.scrollHeight || 0, window.innerHeight || 0),
        url: location.href || '',
        title: document.title || ''
      };
    }
    """

    t0 = time.perf_counter()
    metrics = await evaluate_value(tab, readiness_script, label="readiness_check")
    elapsed = time.perf_counter() - t0
    log.debug("Fast readiness check: %.3fs result=%s", elapsed, type(metrics).__name__)

    if isinstance(metrics, dict) and metrics.get("readyState") in ("interactive", "complete"):
        text_len = int(metrics.get("textLength") or 0)
        if text_len > 0:
            log.info("Rendered page stable via fast-path readyState=%s (text=%d chars url=%s)",
                     metrics["readyState"], text_len, metrics.get("url", "")[:80])
            return {
                "readyState": str(metrics["readyState"]),
                "textLength": text_len,
                "imageCount": int(metrics.get("imageCount") or 0),
                "loadedImageCount": int(metrics.get("imageCount") or 0),
                "scrollHeight": int(metrics.get("scrollHeight") or 0),
                "viewportWidth": 0,
                "viewportHeight": 0,
                "title": str(metrics.get("title") or ""),
                "url": str(metrics.get("url") or ""),
            }

    # Only fall into polling if readyState is loading or JS eval failed
    log.debug("Fast-path readiness returned %s; trying html fallback or polling",
              metrics.get("readyState") if isinstance(metrics, dict) else "non-dict")

    # Try get_tab_content as last-resort readiness
    html = await await_with_timeout("readiness HTML fallback", get_tab_content(tab, timeout=1.5, required=False),
                                     timeout=2.0, default="", warning=False)
    if isinstance(html, str) and len(html) > 5000 and "<body" in html.lower():
        log.info("Rendered page ready via get_tab_content fallback (html=%d chars)", len(html))
        return {"readyState": "unknown_html_available", "htmlLength": len(html),
                "waitFallback": "get_content", "textLength": 0, "imageCount": 0,
                "scrollHeight": 0, "viewportWidth": 0, "viewportHeight": 0,
                "title": "", "url": ""}

    # Polling loop for slow pages
    last_signature = None
    stable = 0
    last_metrics = {}
    while time.time() < deadline:
        try:
            metrics = await evaluate_value(
                tab,
                """
                () => {
                  const body = document.body || {};
                  const doc = document.documentElement || {};
                  return {
                    readyState: document.readyState,
                    textLength: (body.innerText || '').replace(/\\s+/g, ' ').trim().length,
                    imageCount: (document.images || []).length,
                    scrollHeight: Math.max(doc.scrollHeight || 0, body.scrollHeight || 0, window.innerHeight || 0),
                  };
                }
                """,
                label="content_poll",
            )
            if isinstance(metrics, dict):
                last_metrics = metrics
                signature = (metrics.get("readyState"), metrics.get("textLength"), metrics.get("scrollHeight"))
                if metrics.get("readyState") == "complete" and signature == last_signature:
                    stable += 1
                else:
                    stable = 0
                last_signature = signature
                if stable >= stable_cycles:
                    log.info(
                        "Rendered page stable: readyState=%s textLength=%s scrollHeight=%s",
                        metrics.get("readyState"), metrics.get("textLength"), metrics.get("scrollHeight"),
                    )
                    return metrics
        except Exception as exc:
            log.debug("Page stability check failed: %s", exc)
        await tab.sleep(0.2)
    log.warning("Timed out waiting for rendered content stability. Last metrics: %s", last_metrics)
    return last_metrics


# ---------------------------------------------------------------------------
# Full-page background image collection
# ---------------------------------------------------------------------------
async def _fullpage_scroll_and_discover(tab) -> list[dict]:
    """Scroll the full page height, let lazy images load, return all visible images from DOM."""
    try:
        metrics = await _page_metrics(tab)
        viewport_h = int(metrics.get("viewport", {}).get("height") or 1080)
        scroll_h = int(metrics.get("scroll_height") or 0)

        # First pass: scroll to bottom to trigger lazy content
        await _scroll_to(tab, 99999)
        await tab.sleep(1.0)
        # Re-measure after lazy content loaded
        metrics = await _page_metrics(tab)
        scroll_h = int(metrics.get("scroll_height") or scroll_h)

        # Second pass: scroll in viewport steps with pauses for lazy images
        step = max(1, viewport_h)
        y = 0
        while y < scroll_h:
            await _scroll_to(tab, y)
            await tab.sleep(0.4)
            y += step
        # Ensure bottom is reached
        await _scroll_to(tab, scroll_h)
        await tab.sleep(0.5)
        await _scroll_to(tab, 0)
        await tab.sleep(0.3)
    except Exception as exc:
        log.debug("Full-page scroll for images failed: %s", exc)

    return await _query_visible_images(tab)


async def _run_fullpage_background(
    tab,
    page_url: str,
    out_dir: Path,
    page_stem: str,
    options: dict,
) -> tuple[Path | None, list[dict]]:
    """Background full-page pass: scroll, discover more images, write __fullpage.md."""
    t_full = time.perf_counter()

    # Scroll full page and discover all images after lazy loading
    visibles = await _fullpage_scroll_and_discover(tab)
    _timing_record(page_url, "fullpage_scroll_background", time.perf_counter() - t_full)

    if not visibles:
        log.info("Full-page pass: no images discovered after scrolling for %s", page_url)
        return None, []

    max_images = int(options.get("max_images") or 0)
    candidates = visibles
    if max_images > 0 and len(candidates) > max_images:
        candidates = candidates[:max_images]
    log.info("Full-page pass: %d images visible, %d selected after scroll", len(visibles), len(candidates))

    # Download images (reuse same pattern)
    images_dir = out_dir / "images"
    images_dir.mkdir(parents=True, exist_ok=True)
    image_stem = _image_page_stem(page_url)
    downloaded: dict[str, str] = {}
    manifest: list[dict] = []
    used_filenames: set[str] = set()
    sequence_counter = [0]
    filename_lock = asyncio.Lock()
    semaphore = asyncio.Semaphore(6)

    async def download_one(img_info: dict) -> tuple[str, dict | None]:
        src = (img_info.get("url") or img_info.get("currentSrc") or img_info.get("src") or "").strip()
        if not src or src.startswith("data:") or src.startswith("blob:"):
            return src, None
        full_url = urljoin(page_url, src)
        parsed = urlparse(full_url)
        if parsed.scheme not in {"http", "https"}:
            return src, None

        async with semaphore:
            resource = await _fetch_resource_bytes_cached(tab, full_url)
            from_cache = resource is not None
            if resource is None:
                resource = await _fetch_resource_bytes(tab, full_url)
            if resource is None:
                return src, None
            content, content_type = resource
            if not (content_type.lower().startswith("image/") or Path(parsed.path).suffix.lower() in _IMAGE_EXTENSIONS):
                return src, None

            async with filename_lock:
                sequence_counter[0] += 1
                idx = sequence_counter[0]
                ext = _image_extension_from_url_or_type(full_url, content_type)
                original_stem = _clean_original_image_filename(full_url, fallback=f"fp_image_{idx:03d}")
                target, filename = _unique_image_target(images_dir, image_stem, original_stem, ext, used_filenames, full_url)
            target.write_bytes(content)
            rel = f"images/{filename}"
            entry = {
                "type": "image",
                "page_url": page_url,
                "source_url": full_url,
                "local_path": str(target.resolve()),
                "relative_path": rel,
                "content_type": content_type,
                "original_filename": Path(parsed.path).name or None,
                "size_bytes": len(content),
                "from_cache": from_cache,
                "alt": img_info.get("alt", ""),
                "title": img_info.get("title", ""),
                "viewport_width": img_info.get("w", 0),
                "viewport_height": img_info.get("h", 0),
                "natural_width": img_info.get("nw", 0),
                "natural_height": img_info.get("nh", 0),
            }
            return full_url, entry

    t_fetch = time.perf_counter()
    tasks = [download_one(img) for img in candidates]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            log.debug("Full-page image download task exception: %s", result)
            continue
        key, entry = result
        if key and entry:
            downloaded[key] = entry["relative_path"]
            manifest.append(entry)
    _timing_record(page_url, "fullpage_image_discovery", time.perf_counter() - t_fetch)

    if not manifest:
        log.info("Full-page pass: no new images downloaded for %s", page_url)
        return None, []

    log.info("Full-page pass: %d images downloaded for %s", len(manifest), page_url)
    cache_hits = sum(1 for e in manifest if e.get("from_cache"))
    if cache_hits > 0:
        log.info("Full-page images: %d/%d from browser cache", cache_hits, len(manifest))

    # Build full-page HTML with all images rewritten
    html = await await_with_timeout(
        "fullpage HTML capture",
        get_tab_content(tab, timeout=float(options.get("html_timeout") or 15.0), required=False),
        timeout=float(options.get("html_timeout") or 15.0),
        default="",
        warning=True,
    )
    if isinstance(html, bytes):
        html = html.decode("utf-8", errors="replace")

    soup = BeautifulSoup(html, "html.parser") if html else BeautifulSoup("", "html.parser")
    for tag in soup.find_all("img"):
        tag_src = str(tag.get("src") or "")
        if not tag_src:
            continue
        full = urljoin(page_url, tag_src)
        if full in downloaded:
            tag["src"] = downloaded[full]
        for attr in ["srcset", "data-srcset"]:
            if not tag.has_attr(attr):
                continue
            rebuilt: list[str] = []
            for src, descriptor in _split_srcset(str(tag.get(attr) or "")):
                src_full = urljoin(page_url, src)
                if src_full in downloaded:
                    rebuilt.append(f"{downloaded[src_full]} {descriptor}".strip())
                else:
                    rebuilt.append(f"{src} {descriptor}".strip())
            if rebuilt:
                tag["srcset"] = ", ".join(rebuilt)

    localized_html = str(soup)

    # Generate full-page Markdown
    t_md = time.perf_counter()
    cleaned_html = clean_html_for_content(localized_html)
    markdown = markdown_from_rendered_html(cleaned_html, page_url)

    output_mode = str(options.get("fullpage_output") or "second")
    if output_mode == "update":
        md_path = out_dir / f"{page_stem}.md"
        images_manifest_path = out_dir / f"{page_stem}.images.json"
    else:
        md_path = out_dir / f"{page_stem}__fullpage.md"
        images_manifest_path = out_dir / f"{page_stem}__fullpage.images.json"

    write_text(md_path, markdown)
    write_json(images_manifest_path, {"schema_version": "crawler-page-images/v1", "page_url": page_url, "images": manifest})
    _timing_record(page_url, "fullpage_markdown_write", time.perf_counter() - t_md)

    log.info("Full-page Markdown saved: %s (%d images)", _safe_relative(md_path, SITES_ROOT), len(manifest))
    return md_path, manifest
# ---------------------------------------------------------------------------
_CONSENT_REMOVE_SELECTORS = [
    ".ch2", "[data-nosnippet='true']", "#onetrust-banner-sdk", ".cookie", ".cookiebanner",
    ".cookie-banner", ".cookie-consent", ".cc-window", ".cc-banner", ".modal-backdrop",
]


def remove_runtime_noise_for_content(soup: BeautifulSoup) -> None:
    for tag in list(soup.find_all(["script", "style", "noscript", "template"])):
        tag.decompose()
    for selector in _CONSENT_REMOVE_SELECTORS:
        try:
            for tag in list(soup.select(selector)):
                tag.decompose()
        except Exception:
            pass


def _image_extension_from_url_or_type(url: str, content_type: str | None = None) -> str:
    content_type = (content_type or "").split(";", 1)[0].strip().lower()
    mapping = {
        "image/jpeg": ".jpg",
        "image/jpg": ".jpg",
        "image/png": ".png",
        "image/gif": ".gif",
        "image/svg+xml": ".svg",
        "image/webp": ".webp",
        "image/avif": ".avif",
        "image/bmp": ".bmp",
    }
    if content_type in mapping:
        return mapping[content_type]
    parsed = urlparse(url)
    query = parse_qs(parsed.query)
    fmt = str((query.get("format") or [""])[0]).lower().strip(".")
    if fmt in {"jpg", "jpeg", "png", "gif", "svg", "webp", "avif", "bmp"}:
        return ".jpg" if fmt == "jpeg" else f".{fmt}"
    suffix = Path(parsed.path).suffix.lower()
    if suffix in _IMAGE_EXTENSIONS:
        return suffix
    return ".img"


def _split_srcset(srcset: str) -> list[tuple[str, str]]:
    result: list[tuple[str, str]] = []
    for part in (srcset or "").split(","):
        item = part.strip()
        if not item:
            continue
        pieces = item.split(None, 1)
        result.append((pieces[0], pieces[1] if len(pieces) > 1 else ""))
    return result


def _safe_url_encode(full_url: str) -> str:
    from urllib.parse import quote, urlparse, urlunparse

    try:
        full_url.encode("ascii")
        return full_url
    except UnicodeEncodeError:
        pass
    parsed = urlparse(full_url)
    safe_path = quote(parsed.path, safe="/:@!$&'()*+,;=")
    safe_query = quote(parsed.query, safe="/:@!$&'()*+,;=")
    return urlunparse((parsed.scheme, parsed.netloc, safe_path, parsed.params, safe_query, parsed.fragment))


# ---------------------------------------------------------------------------
# Browser-cached image fetch via Network.loadNetworkResource (CDP experimental)
# ---------------------------------------------------------------------------
_frame_id_cache: dict[int, str] = {}
_cached_fetch_disabled: bool = False
_RECOVERABLE_NET_ERRORS = {"net::ERR_FAILED", "net::ERR_ABORTED"}


async def _get_main_frame_id(tab) -> str:
    tab_key = id(tab)
    cached = _frame_id_cache.get(tab_key)
    if cached:
        return cached
    from nodriver.cdp.page import get_frame_tree
    frame_tree = await tab.send(get_frame_tree())
    frame_id = frame_tree.frame.id_
    _frame_id_cache[tab_key] = frame_id
    return frame_id


async def _fetch_resource_bytes_cached(
    tab, full_url: str,
    allowed_mime_prefixes: tuple[str, ...] | None = ("image/",),
    allowed_extensions: tuple[str, ...] | None = (".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".avif", ".bmp"),
) -> tuple[bytes, str] | None:
    """Try to get a resource from the browser cache via Network.loadNetworkResource.

    Returns (body, content_type) on success, or None on any failure.
    Callers should fall back to _fetch_resource_bytes() when this returns None.

    If allowed_mime_prefixes is None, all content types are accepted.
    If allowed_extensions is None, all file extensions are accepted.
    """
    import base64 as _base64
    global _cached_fetch_disabled

    if _cached_fetch_disabled:
        return None

    try:
        frame_id = await _get_main_frame_id(tab)
    except Exception as exc:
        log.debug("Failed to get frame tree for cached fetch: %s", exc)
        return None

    from nodriver.cdp.network import load_network_resource, LoadNetworkResourceOptions
    from nodriver.cdp.io import read as io_read, close as io_close

    options = LoadNetworkResourceOptions(disable_cache=False, include_credentials=True)

    try:
        result = await tab.send(load_network_resource(full_url, options, frame_id))
    except Exception as exc:
        msg = str(exc).lower()
        if "not found" in msg or "wasn't found" in msg or "unknown method" in msg:
            _cached_fetch_disabled = True
            log.info("Network.loadNetworkResource not supported by this Chrome; disabling cached image fetch")
        else:
            log.debug("Network.loadNetworkResource call failed for %s: %s", full_url[:120], exc)
        return None

    if not result.success:
        net_err = getattr(result, "net_error_name", "") or ""
        if net_err in _RECOVERABLE_NET_ERRORS:
            log.debug("Network.loadNetworkResource recoverable error for %s: %s", full_url[:120], net_err)
        else:
            log.debug("Network.loadNetworkResource returned success=false for %s (httpStatus=%s, netError=%s)",
                      full_url[:120], result.http_status_code, net_err)
        return None

    if result.stream is None:
        log.debug("Network.loadNetworkResource returned no stream for %s", full_url[:120])
        return None

    try:
        chunks: list[bytes] = []
        while True:
            base64_encoded, data, eof = await tab.send(io_read(result.stream))
            chunk = _base64.b64decode(data) if base64_encoded else data.encode("utf-8")
            chunks.append(chunk)
            if eof:
                break
    finally:
        try:
            await tab.send(io_close(result.stream))
        except Exception:
            pass

    body = b"".join(chunks)
    if not body:
        log.debug("Network.loadNetworkResource returned empty body for %s", full_url[:120])
        return None

    content_type = ""
    if result.headers:
        for key, val in result.headers.items():
            if key.lower() == "content-type":
                content_type = val
                break

    if allowed_mime_prefixes is not None and content_type:
        if not any(content_type.lower().startswith(p) for p in allowed_mime_prefixes):
            if allowed_extensions is None or not any(full_url.lower().endswith(ext) for ext in allowed_extensions):
                return None

    log.debug("Browser-cached fetch: %d bytes from %s", len(body), full_url[:120])
    return body, content_type


async def _fetch_resource_bytes(
    tab, full_url: str,
    allowed_mime_prefixes: tuple[str, ...] | None = ("image/",),
    allowed_extensions: tuple[str, ...] | None = (".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".avif", ".bmp"),
) -> tuple[bytes, str] | None:
    import urllib.request
    from urllib.error import URLError

    safe_url = _safe_url_encode(full_url)
    log.debug("Fetching via urllib: %s", safe_url[:120])
    try:
        resp = await asyncio.to_thread(
            lambda: urllib.request.urlopen(
                urllib.request.Request(safe_url, headers={"User-Agent": "Mozilla/5.0"}),
                timeout=15,
            )
        )
        content_type = resp.headers.get("Content-Type", "")
        if allowed_mime_prefixes is not None and content_type:
            if not any(content_type.lower().startswith(p) for p in allowed_mime_prefixes):
                if allowed_extensions is None or not any(full_url.lower().endswith(ext) for ext in allowed_extensions):
                    return None
        body = resp.read()
        log.debug("Fetched %d bytes from %s", len(body), full_url[:120])
        return body, content_type
    except URLError as exc:
        log.debug("urllib fetch failed for %s: %s", full_url[:120], exc)
    except Exception as exc:
        log.debug("Unexpected fetch error for %s: %s", full_url[:120], exc)
    return None


def _clean_original_image_filename(url: str, fallback: str) -> str:
    parsed = urlparse(url)
    name = Path(parsed.path).name
    stem = Path(name).stem if name else ""
    if not stem:
        query = parse_qs(parsed.query)
        for key in ("file", "filename", "name", "image"):
            values = query.get(key) or []
            if values and str(values[0]).strip():
                stem = Path(str(values[0])).stem
                break
    return _safe_token(stem.lower(), fallback=fallback)


def _unique_image_target(images_dir: Path, page_stem: str, original_stem: str, ext: str, used_filenames: set[str], source_url: str) -> tuple[Path, str]:
    digest = hashlib.sha1(source_url.encode("utf-8")).hexdigest()[:8]
    base_filename = f"{page_stem}__{original_stem}{ext}"
    filename = base_filename
    if filename in used_filenames:
        filename = f"{page_stem}__{original_stem}__{digest}{ext}"
    counter = 2
    while filename in used_filenames:
        filename = f"{page_stem}__{original_stem}__{digest}_{counter}{ext}"
        counter += 1
    used_filenames.add(filename)
    return images_dir / filename, filename


async def download_and_rewrite_images(tab, html: str, *, page_url: str, out_dir: Path, page_stem: str, max_images: int = 0) -> tuple[str, list[dict]]:
    """Discover visible images via browser DOM and download them in parallel; rewrite HTML src attributes."""
    soup = BeautifulSoup(html, "html.parser")
    images_dir = out_dir / "images"
    images_dir.mkdir(parents=True, exist_ok=True)
    image_stem = _image_page_stem(page_url)

    t_scroll = time.perf_counter()
    await _scroll_page_for_images(tab)
    _timing_record(page_url, "visible_image_scroll", time.perf_counter() - t_scroll)

    t_discover = time.perf_counter()
    visibles = await _query_visible_images(tab)
    _timing_record(page_url, "visible_image_discovery", time.perf_counter() - t_discover)
    if not visibles:
        log.info("Images: no visible images discovered for %s", page_url)
        return html, []

    candidates = visibles
    if max_images > 0 and len(candidates) > max_images:
        candidates = candidates[:max_images]
        log.info("Images: %d visible, capped at %d", len(visibles), max_images)

    log.info("Images: %d visible, %d selected for download", len(visibles), len(candidates))

    downloaded: dict[str, str] = {}
    manifest: list[dict] = []
    used_filenames: set[str] = set()
    sequence_counter = [0]
    filename_lock = asyncio.Lock()
    semaphore = asyncio.Semaphore(6)
    cache_time_total = 0.0
    fallback_time_total = 0.0
    cache_time_lock = asyncio.Lock()

    async def download_one(img_info: dict) -> tuple[str, dict | None]:
        nonlocal cache_time_total, fallback_time_total
        src = (img_info.get("url") or img_info.get("currentSrc") or img_info.get("src") or "").strip()
        if not src or src.startswith("data:") or src.startswith("blob:"):
            return src, None
        full_url = urljoin(page_url, src)
        parsed = urlparse(full_url)
        if parsed.scheme not in {"http", "https"}:
            return src, None

        async with semaphore:
            t_cache = time.perf_counter()
            resource = await _fetch_resource_bytes_cached(tab, full_url)
            from_cache = resource is not None
            if from_cache:
                async with cache_time_lock:
                    cache_time_total += time.perf_counter() - t_cache
            else:
                async with cache_time_lock:
                    cache_time_total += time.perf_counter() - t_cache
                t_fallback = time.perf_counter()
                resource = await _fetch_resource_bytes(tab, full_url)
                async with cache_time_lock:
                    fallback_time_total += time.perf_counter() - t_fallback
            if resource is None:
                log.debug("Image download failed: %s", full_url[:120])
                return src, None
            content, content_type = resource
            if not (content_type.lower().startswith("image/") or Path(parsed.path).suffix.lower() in _IMAGE_EXTENSIONS):
                log.debug("Image skipped: non-image response (%s) for %s", content_type, full_url[:120])
                return src, None

            async with filename_lock:
                sequence_counter[0] += 1
                idx = sequence_counter[0]
                ext = _image_extension_from_url_or_type(full_url, content_type)
                original_stem = _clean_original_image_filename(full_url, fallback=f"image_{idx:03d}")
                target, filename = _unique_image_target(images_dir, image_stem, original_stem, ext, used_filenames, full_url)
            target.write_bytes(content)
            rel = f"images/{filename}"
            entry = {
                "type": "image",
                "page_url": page_url,
                "source_url": full_url,
                "local_path": str(target.resolve()),
                "relative_path": rel,
                "content_type": content_type,
                "original_filename": Path(parsed.path).name or None,
                "size_bytes": len(content),
                "from_cache": from_cache,
                "alt": img_info.get("alt", ""),
                "title": img_info.get("title", ""),
                "viewport_width": img_info.get("w", 0),
                "viewport_height": img_info.get("h", 0),
                "natural_width": img_info.get("nw", 0),
                "natural_height": img_info.get("nh", 0),
            }
            if idx % 10 == 0 or idx == len(candidates):
                log.info("Images: %d/%d downloaded", idx, len(candidates))
            log.debug("Image %d/%d: %s -> %s (%d bytes)", idx, len(candidates), full_url, target.resolve(), len(content))
            return full_url, entry

    tasks = [download_one(img) for img in candidates]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            log.debug("Image download task exception: %s", result)
            continue
        key, entry = result
        if key and entry:
            downloaded[key] = entry["relative_path"]
            manifest.append(entry)

    if cache_time_total > 0:
        _timing_record(page_url, "browser_cache_image_fetch", cache_time_total)
    if fallback_time_total > 0:
        _timing_record(page_url, "urllib_image_fallback", fallback_time_total)

    if max_images > 0 and len(candidates) >= max_images:
        log.info("Image cap reached: %d", max_images)
    if manifest:
        log.info("Image download summary for %s: %d/%d saved", page_url, len(manifest), len(candidates))
        cache_hits = sum(1 for e in manifest if e.get("from_cache"))
        if cache_hits > 0:
            log.info("Images: %d/%d served from browser cache", cache_hits, len(manifest))
    else:
        log.info("Image download summary for %s: no images saved from %d candidates", page_url, len(candidates))

    for tag in soup.find_all("img"):
        tag_src = str(tag.get("src") or "")
        if not tag_src:
            continue
        full = urljoin(page_url, tag_src)
        if full in downloaded:
            tag["src"] = downloaded[full]
        for attr in ["srcset", "data-srcset"]:
            if not tag.has_attr(attr):
                continue
            rebuilt: list[str] = []
            for src, descriptor in _split_srcset(str(tag.get(attr) or "")):
                src_full = urljoin(page_url, src)
                if src_full in downloaded:
                    rebuilt.append(f"{downloaded[src_full]} {descriptor}".strip())
                else:
                    rebuilt.append(f"{src} {descriptor}".strip())
            if rebuilt:
                tag["srcset"] = ", ".join(rebuilt)
        for attr in ["data-src", "data-original", "data-lazy", "data-bg", "data-background", "data-image", "poster"]:
            if not tag.has_attr(attr):
                continue
            full = urljoin(page_url, str(tag.get(attr) or ""))
            if full in downloaded:
                if attr == "poster":
                    tag["poster"] = downloaded[full]
                else:
                    if not tag.get("src"):
                        tag["src"] = downloaded[full]

    return str(soup), manifest


async def _scroll_page_for_images(tab) -> None:
    """Scroll through the page to trigger lazy-loaded images. Does not collect results."""
    try:
        metrics = await _page_metrics(tab)
        viewport_h = int(metrics.get("viewport", {}).get("height") or 1080)
        scroll_h = int(metrics.get("scroll_height") or 0)
        step = max(1, viewport_h)
        y = 0
        while y < scroll_h:
            await _scroll_to(tab, y)
            await tab.sleep(0.2)
            y += step
        await _scroll_to(tab, 0)
        await tab.sleep(0.3)
    except Exception as exc:
        log.debug("Scroll-for-images failed: %s", exc)


async def _query_visible_images(tab) -> list[dict]:
    """Query the DOM for loaded visible images. Does not scroll."""
    script = """
    (() => {
      const results = [];
      const seen = new Set();
      for (const img of (document.images || [])) {
        const url = (img.currentSrc || img.src || "").trim();
        if (!url || seen.has(url)) continue;
        seen.add(url);
        const rect = img.getBoundingClientRect ? img.getBoundingClientRect() : {};
        if (img.complete && img.naturalWidth > 0) {
          results.push(JSON.stringify({
            url: url,
            w: rect.width || 0,
            h: rect.height || 0,
            nw: img.naturalWidth || 0,
            nh: img.naturalHeight || 0,
            alt: (img.alt || "").substring(0, 500),
            title: (img.title || "").substring(0, 500),
          }));
        }
      }
      return results;
    })()
    """
    try:
        result = await tab.evaluate(script)
        if isinstance(result, list) and result:
            parsed = []
            for item in result:
                try:
                    if isinstance(item, str):
                        parsed.append(json.loads(item))
                    elif isinstance(item, dict) and item.get("type") == "string" and item.get("value"):
                        parsed.append(json.loads(str(item.get("value"))))
                except Exception:
                    pass
            if parsed:
                return parsed
    except Exception as exc:
        log.warning("Visible image discovery failed: %s", exc)
    return []


async def _discover_visible_images(tab) -> list[dict]:
    """Scroll to trigger lazy loading, then query DOM for visible images (combined, kept for compatibility)."""
    await _scroll_page_for_images(tab)
    return await _query_visible_images(tab)


def clean_html_for_content(html: str) -> str:
    soup = BeautifulSoup(html, "html.parser")
    remove_runtime_noise_for_content(soup)
    return str(soup)


def extract_text_from_html(html: str, page_url: str) -> str:
    soup = BeautifulSoup(html, "html.parser")
    remove_runtime_noise_for_content(soup)
    title = soup.title.get_text(" ", strip=True) if soup.title else ""
    body = soup.body or soup
    lines = []
    for line in body.get_text("\n", strip=True).splitlines():
        line = re.sub(r"\s+", " ", line).strip()
        if line:
            lines.append(line)
    return "\n".join([f"URL: {page_url}", f"Title: {title}", "", *lines, ""])


def markdown_from_rendered_html(html: str, page_url: str) -> str:
    if html_to_markdown is None:
        raise ImportError("markdownify is required for --markdown true when --vision false. Install with: pip install markdownify")
    soup = BeautifulSoup(html, "html.parser")
    remove_runtime_noise_for_content(soup)
    title = soup.title.get_text(" ", strip=True) if soup.title else page_url
    body = soup.body or soup
    markdown = html_to_markdown(
        str(body),
        heading_style="ATX",
        bullets="-",
        strip=["script", "style", "noscript", "template"],
    )
    markdown = re.sub(r"\n{4,}", "\n\n\n", markdown).strip()
    frontmatter = "\n".join([
        "---",
        "schema: rendered-html/v1",
        f'title: "{title.replace(chr(34), chr(39))}"',
        f"source_url: {page_url}",
        f"generated_at: {utc_now_iso()}",
        "---",
    ])
    return f"{frontmatter}\n\n{markdown}\n"


# ---------------------------------------------------------------------------
# Asset download (CSS, JS, fonts) from browser cache for local HTML viewing
# ---------------------------------------------------------------------------

def extract_css_links(html: str, base_url: str) -> list[str]:
    soup = BeautifulSoup(html, "html.parser")
    urls: list[str] = []
    for link in soup.find_all("link"):
        rel = link.get("rel") or []
        if isinstance(rel, str):
            rel = [rel]
        if "stylesheet" not in [r.lower() for r in rel]:
            continue
        href = link.get("href", "")
        if not href:
            continue
        absolute = _normalized_crawl_url(href, base_url, allow_static_resources=True)
        if absolute:
            urls.append(absolute)
    return urls


def extract_js_links(html: str, base_url: str) -> list[str]:
    soup = BeautifulSoup(html, "html.parser")
    urls: list[str] = []
    for script in soup.find_all("script", src=True):
        absolute = _normalized_crawl_url(script["src"], base_url, allow_static_resources=True)
        if absolute:
            urls.append(absolute)
    return urls


def extract_font_urls_from_css(css_content: str, source_url: str) -> list[str]:
    FONT_EXTS = {".woff2", ".woff", ".ttf", ".otf", ".eot"}
    urls: list[str] = []
    for m in re.finditer(r'url\(\s*["\']?([^"\')\s]+)["\']?\s*\)', css_content, re.IGNORECASE):
        ref = m.group(1)
        absolute = _normalized_crawl_url(ref, source_url, allow_static_resources=True)
        if absolute and Path(urlparse(absolute).path).suffix.lower() in FONT_EXTS:
            urls.append(absolute)
    return urls


def _asset_name_from_url_content(url: str, content: bytes, category: str) -> str:
    stem = Path(urlparse(url).path).stem or category
    h = hashlib.sha256(content).hexdigest()[:8]
    ext_map = {"css": ".css", "js": ".js", "font": ".woff2"}
    ext = Path(urlparse(url).path).suffix
    if not ext:
        ext = ext_map.get(category, ".bin")
    return f"{stem}.{h}{ext}"


async def _download_and_rewrite_assets(
    tab, html: str, page_url: str, out_dir: Path, options: dict
) -> str:
    assets_dir = out_dir / "assets"
    assets_dir.mkdir(parents=True, exist_ok=True)
    manifest_path = assets_dir / "_asset_manifest.json"
    manifest: dict[str, dict] = {}
    if manifest_path.exists():
        try:
            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        except Exception:
            manifest = {}

    max_css_per_page = int(options.get("max_css_per_page", 10))
    max_js_per_page = int(options.get("max_js_per_page", 5))
    max_asset_bytes = int(options.get("max_asset_bytes", 5242880))

    # --- CSS ---
    css_urls = extract_css_links(html, page_url)[:max_css_per_page] if max_css_per_page > 0 else extract_css_links(html, page_url)
    log.info("Extracted %d CSS URLs for %s: %s", len(css_urls), page_url, [u.split("?")[0].rsplit("/",1)[-1] for u in css_urls])
    for url in css_urls:
        # Build the original href as it appears in the HTML (relative or absolute with query)
        parsed = urlparse(url)
        rel_href = parsed.path + ("?" + parsed.query if parsed.query else "")
        if url in manifest:
            # Replace both absolute and relative forms
            html = html.replace(url, f"assets/{manifest[url]['file']}")
            html = html.replace(rel_href, f"assets/{manifest[url]['file']}")
            continue
        try:
            resource = await _fetch_resource_bytes_cached(
                tab, url,
                allowed_mime_prefixes=("text/css",),
                allowed_extensions=(".css",),
            )
            if resource is None:
                # Retry without query string (cache-buster params like ?v=...)
                clean_url = url.split("?")[0] if "?" in url else url
                if clean_url != url:
                    resource = await _fetch_resource_bytes_cached(
                        tab, clean_url,
                        allowed_mime_prefixes=("text/css",),
                        allowed_extensions=(".css",),
                    )
            if resource is None:
                try:
                    resource = await _fetch_resource_bytes(
                        tab, url,
                        allowed_mime_prefixes=("text/css",),
                        allowed_extensions=(".css",),
                    )
                except Exception as exc:
                    log.info("CSS network fetch exception for %s: %s", url, exc)
        except Exception as exc:
            log.info("CSS cache fetch exception for %s: %s", url, exc)
            resource = None
        if resource is None:
            log.info("CSS fetch failed for %s (not in browser cache, not via network)", url)
            continue
        content, _ = resource
        if len(content) > max_asset_bytes:
            continue
        fname = _asset_name_from_url_content(url, content, "css")
        (assets_dir / "css").mkdir(parents=True, exist_ok=True)
        (assets_dir / "css" / fname).write_bytes(content)
        # Replace both absolute URL and relative href in HTML
        html = html.replace(url, f"assets/css/{fname}")
        html = html.replace(rel_href, f"assets/css/{fname}")
        manifest[url] = {"file": f"css/{fname}", "size": len(content)}

        # --- Fonts from CSS ---
        css_text = content.decode("utf-8", errors="replace")
        font_urls = extract_font_urls_from_css(css_text, url)
        for font_url in font_urls:
            if font_url in manifest:
                css_text = css_text.replace(font_url, f"../fonts/{manifest[font_url]['file']}")
                continue
            try:
                font_resource = await _fetch_resource_bytes_cached(tab, font_url, allowed_mime_prefixes=None, allowed_extensions=None)
            except Exception:
                font_resource = None
            if font_resource is None:
                continue
            fcontent, _ = font_resource
            if len(fcontent) > max_asset_bytes:
                continue
            ffname = _asset_name_from_url_content(font_url, fcontent, "font")
            (assets_dir / "fonts").mkdir(parents=True, exist_ok=True)
            (assets_dir / "fonts" / ffname).write_bytes(fcontent)
            css_text = css_text.replace(font_url, f"../fonts/{ffname}")
            manifest[font_url] = {"file": f"fonts/{ffname}", "size": len(fcontent)}
        (assets_dir / "css" / fname).write_text(css_text, encoding="utf-8")

    # --- JS ---
    js_urls = extract_js_links(html, page_url)[:max_js_per_page] if max_js_per_page > 0 else extract_js_links(html, page_url)
    for url in js_urls:
        if url in manifest:
            html = html.replace(url, f"assets/{manifest[url]['file']}")
            continue
        try:
            resource = await _fetch_resource_bytes_cached(
                tab, url,
                allowed_mime_prefixes=("text/javascript", "application/javascript"),
                allowed_extensions=(".js",),
            )
            if resource is None:
                resource = await _fetch_resource_bytes(
                    tab, url,
                    allowed_mime_prefixes=("text/javascript", "application/javascript"),
                    allowed_extensions=(".js",),
                )
        except Exception:
            resource = None
        if resource is None:
            continue
        content, _ = resource
        if len(content) > max_asset_bytes:
            continue
        fname = _asset_name_from_url_content(url, content, "js")
        (assets_dir / "js").mkdir(parents=True, exist_ok=True)
        (assets_dir / "js" / fname).write_bytes(content)
        html = html.replace(url, f"assets/js/{fname}")
        manifest[url] = {"file": f"js/{fname}", "size": len(content)}

    write_json(manifest_path, manifest)
    return html


# ---------------------------------------------------------------------------
# Screenshots after consent
# ---------------------------------------------------------------------------
async def _page_metrics(tab) -> dict:
    data = await evaluate_value(
        tab,
        """
        () => {
          const doc = document.documentElement || {};
          const body = document.body || {};
          return {
            title: document.title || '',
            url: location.href || '',
            viewport: {
              width: window.innerWidth || doc.clientWidth || 0,
              height: window.innerHeight || doc.clientHeight || 0
            },
            scroll_width: Math.max(doc.scrollWidth || 0, body.scrollWidth || 0, window.innerWidth || 0),
            scroll_height: Math.max(doc.scrollHeight || 0, body.scrollHeight || 0, window.innerHeight || 0)
          };
        }
        """,
    )
    return data if isinstance(data, dict) else {}


async def _scroll_to(tab, y: int) -> None:
    try:
        await evaluate_value(tab, f"() => {{ window.scrollTo(0, {int(y)}); return window.scrollY || 0; }}")
    except Exception as exc:
        log.debug("Scroll failed: %s", exc)


async def capture_page_screenshots(
    tab,
    *,
    url: str,
    out_dir: Path,
    page_stem: str,
    viewport_width: int,
    viewport_height: int,
    max_tiles: int,
    overlap: int,
    viewport_only: bool = False,
) -> list[dict]:
    await set_and_verify_viewport(tab, viewport_width, viewport_height)
    await tab.sleep(0.5)
    metrics = await _page_metrics(tab)
    viewport = metrics.get("viewport") if isinstance(metrics.get("viewport"), dict) else {}
    actual_width = int(viewport.get("width") or viewport_width or 1920)
    actual_height = int(viewport.get("height") or viewport_height or 1080)

    screenshots_dir = out_dir / "screenshots"
    screenshots_dir.mkdir(parents=True, exist_ok=True)
    captured_at = utc_now_iso()
    metadata: list[dict] = []

    if not viewport_only:
        scroll_height = int(metrics.get("scroll_height") or actual_height)

        # Force lazy content to render by scrolling to bottom and back, then re-measure
        await _scroll_to(tab, 99999)
        await tab.sleep(1.0)
        metrics = await _page_metrics(tab)
        scroll_height = int(metrics.get("scroll_height") or actual_height)
        await _scroll_to(tab, 0)
        await tab.sleep(0.5)

        screenshots_dir = out_dir / "screenshots"
        screenshots_dir.mkdir(parents=True, exist_ok=True)
        captured_at = utc_now_iso()
        metadata: list[dict] = []

    # Always capture viewport screenshot
    tile_path = screenshots_dir / f"{page_stem}__viewport_initial.png"
    full_path = screenshots_dir / f"{page_stem}__fullpage.png"

    await save_viewport_screenshot(tab, tile_path, expected_width=actual_width, expected_height=actual_height)

    tile_pixels = _png_dimensions(tile_path)
    tile_item = {
        "source_url": url,
        "url": str(metrics.get("url") or url),
        "page_title": str(metrics.get("title") or ""),
        "screenshot_number": 1,
        "screenshot_path": str(tile_path.resolve()),
        "local_screenshot_path": str(tile_path.resolve()),
        "viewport": {"width": actual_width, "height": actual_height},
        "screenshot_pixel_size": {
            "width": int(tile_pixels[0]) if tile_pixels else None,
            "height": int(tile_pixels[1]) if tile_pixels else None,
        },
        "capture_timestamp": captured_at,
        "captured_at": captured_at,
        "scroll_y": 0,
        "scroll_position": {"x": 0, "y": 0},
        "tile_index": 1,
        "tile_count": 1,
        "total_available_tiles": 1,
        "truncated_by_screenshot_max_tiles": False,
        "is_last_tile": True,
        "content_continues_after_last_captured_tile": False,
        "screenshot_type": "viewport",
    }
    metadata.append(tile_item)
    log.info("Saved viewport screenshot %s (%dx%d)", _safe_relative(tile_path, SITES_ROOT), tile_pixels[0] if tile_pixels else 0, tile_pixels[1] if tile_pixels else 0)

    if viewport_only:
        return metadata

    await _scroll_to(tab, 0)
    await tab.sleep(0.3)
    try:
        await save_viewport_screenshot(tab, full_path, full_page=True)
        full_pixels = _png_dimensions(full_path)
    except Exception as exc:
        log.debug("Full-page screenshot failed, falling back to tiles: %s", exc)
        full_pixels = None

    if full_pixels and full_pixels[1] and full_pixels[1] > actual_height * 1.1:
        full_item = {
            "source_url": url,
            "url": str(metrics.get("url") or url),
            "page_title": str(metrics.get("title") or ""),
            "screenshot_number": 2,
            "screenshot_path": str(full_path.resolve()),
            "local_screenshot_path": str(full_path.resolve()),
            "viewport": {"width": actual_width, "height": actual_height},
            "screenshot_pixel_size": {
                "width": int(full_pixels[0]) if full_pixels else None,
                "height": int(full_pixels[1]) if full_pixels else None,
            },
            "capture_timestamp": captured_at,
            "captured_at": captured_at,
            "scroll_y": 0,
            "scroll_position": {"x": 0, "y": 0},
            "tile_index": 1,
            "tile_count": 1,
            "total_available_tiles": 1,
            "truncated_by_screenshot_max_tiles": False,
            "is_last_tile": True,
            "content_continues_after_last_captured_tile": False,
            "screenshot_type": "fullpage",
        }
        metadata.append(full_item)
        log.info("Saved full-page screenshot %s (%dx%d)", _safe_relative(full_path, SITES_ROOT), full_pixels[0], full_pixels[1])
    else:
        if full_pixels:
            full_path.unlink(missing_ok=True)
        # Fall back to tile-based for long pages without fullpage support
        scroll_height_val = max(scroll_height, actual_height)
        if scroll_height_val > actual_height * 1.2:
            step = max(1, actual_height - max(0, int(overlap)))
            positions = list(range(0, max(scroll_height_val - actual_height, 0) + 1, step))
            last = max(scroll_height_val - actual_height, 0)
            if last not in positions:
                positions.append(last)
            if not positions:
                positions = [0]
            total_available = len(positions)
            truncated = bool(max_tiles and max_tiles > 0 and total_available > max_tiles)
            if truncated:
                positions = positions[:max_tiles]
            for index, scroll_y in enumerate(positions, start=2):
                await _scroll_to(tab, scroll_y)
                await tab.sleep(0.4)
                sp = screenshots_dir / f"{page_stem}__tile_{index:03d}.png"
                await save_viewport_screenshot(tab, sp, expected_width=actual_width, expected_height=actual_height)
                sp_px = _png_dimensions(sp)
                metadata.append({
                    "source_url": url, "url": str(metrics.get("url") or url),
                    "page_title": str(metrics.get("title") or ""),
                    "screenshot_number": index,
                    "screenshot_path": str(sp.resolve()),
                    "local_screenshot_path": str(sp.resolve()),
                    "viewport": {"width": actual_width, "height": actual_height},
                    "screenshot_pixel_size": {"width": int(sp_px[0]) if sp_px else None, "height": int(sp_px[1]) if sp_px else None},
                    "capture_timestamp": captured_at, "captured_at": captured_at,
                    "scroll_y": int(scroll_y), "scroll_position": {"x": 0, "y": int(scroll_y)},
                    "tile_index": index, "tile_count": len(positions),
                    "total_available_tiles": total_available,
                    "truncated_by_screenshot_max_tiles": truncated,
                    "is_last_tile": index == len(positions),
                    "content_continues_after_last_captured_tile": bool(truncated and index == len(positions)),
                    "screenshot_type": "viewport_tile",
                })
                log.info("Saved %s", _safe_relative(sp, SITES_ROOT))

    await _scroll_to(tab, 0)
    metadata_path = screenshots_dir / f"{page_stem}__screenshots.json"
    write_json(metadata_path, {"schema_version": "webpage-screenshot-capture/v1", "screenshots": metadata})
    log.info("Saved %s", _safe_relative(metadata_path, SITES_ROOT))
    return metadata


# ---------------------------------------------------------------------------
# Link discovery
# ---------------------------------------------------------------------------
def _canonical_or_base_url_from_html(html: str, fallback_url: str) -> str:
    try:
        soup = BeautifulSoup(html, "html.parser")
        canonical = soup.find("link", rel=lambda value: value and "canonical" in value)
        if canonical and canonical.get("href"):
            return _strip_fragment_and_query(urljoin(fallback_url, str(canonical.get("href"))))
        og_url = soup.find("meta", attrs={"property": "og:url"})
        if og_url and og_url.get("content"):
            return _strip_fragment_and_query(urljoin(fallback_url, str(og_url.get("content"))))
        base_tag = soup.find("base", href=True)
        if base_tag and base_tag.get("href"):
            return _strip_fragment_and_query(urljoin(fallback_url, str(base_tag.get("href"))))
    except Exception:
        pass
    return _strip_fragment_and_query(fallback_url)


def _normalized_crawl_url(href: str, base_url: str, *, allow_static_resources: bool = False) -> str | None:
    href = (href or "").strip()
    if not href:
        return None
    lowered = href.lower()
    if lowered.startswith(("#", "javascript:", "mailto:", "tel:", "sms:", "data:", "blob:")):
        return None
    full = _strip_fragment_and_query(urljoin(base_url, href))
    parsed = urlparse(full)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        return None
    if not allow_static_resources:
        suffix = Path(parsed.path).suffix.lower()
        if suffix in _STATIC_RESOURCE_EXTENSIONS:
            return None
    return full


def _is_writerside_or_help_area(base_url: str) -> bool:
    path = urlparse(base_url).path.lower()
    return "/help/" in path or "/devportal/" in path or path.endswith(".html")


def _is_crawlable_child(candidate_url: str, base_url: str, start_area_url: str | None = None, include_subdomains: bool = False) -> bool:
    candidate = urlparse(candidate_url)
    base = urlparse(base_url)
    if include_subdomains:
        cand_host = (candidate.hostname or "").lower().rstrip(".")
        base_host = (base.hostname or "").lower().rstrip(".")
        if cand_host != base_host and not cand_host.endswith("." + base_host):
            return False
    else:
        if candidate.hostname != base.hostname:
            return False
    if _strip_fragment_and_query(candidate_url).rstrip("/") == _strip_fragment_and_query(base_url).rstrip("/"):
        return False
    reference = start_area_url or base_url
    if _is_writerside_or_help_area(reference):
        ref = urlparse(reference)
        base_dir = ref.path.rsplit("/", 1)[0].rstrip("/") + "/"
        return candidate.path.startswith(base_dir) and candidate.path.endswith(".html")
    return True


def discover_children_from_html(html: str, base_url: str, start_area_url: str | None = None, include_subdomains: bool = False) -> list[str]:
    soup = BeautifulSoup(html, "html.parser")
    effective_base = _canonical_or_base_url_from_html(html, base_url)
    selector_groups = [
        "nav a[href], header a[href], .navigation a[href], .navbar a[href], .menu a[href]",
        ".app__nav a[href], .app__toc a[href]",
        "main a[href], article a[href], .main-content a[href], .content a[href]",
        "footer a[href]",
        "a[href]",
    ]
    found: list[str] = []
    seen: set[str] = set()
    for selectors in selector_groups:
        for a in soup.select(selectors):
            candidate = _normalized_crawl_url(a.get("href", ""), effective_base)
            if not candidate or not _is_crawlable_child(candidate, effective_base, start_area_url=start_area_url, include_subdomains=include_subdomains):
                continue
            key = candidate.rstrip("/")
            if key in seen:
                continue
            seen.add(key)
            found.append(candidate)
    log.info("Discovered %d child pages from HTML", len(found))
    return found


def discover_children_with_text(html: str, base_url: str, start_area_url: str | None = None, include_subdomains: bool = False) -> list[dict[str, str]]:
    """Return crawlable child links with visible anchor text.

    This uses the same discovery rules as discover_children_from_html(), but it
    returns dictionaries for API/MCP consumers.
    """
    soup = BeautifulSoup(html, "html.parser")
    effective_base = _canonical_or_base_url_from_html(html, base_url)
    selector_groups = [
        "nav a[href], header a[href], .navigation a[href], .navbar a[href], .menu a[href]",
        ".app__nav a[href], .app__toc a[href]",
        "main a[href], article a[href], .main-content a[href], .content a[href]",
        "footer a[href]",
        "a[href]",
    ]
    found: list[dict[str, str]] = []
    seen: set[str] = set()
    for selectors in selector_groups:
        for a in soup.select(selectors):
            candidate = _normalized_crawl_url(a.get("href", ""), effective_base)
            if not candidate or not _is_crawlable_child(candidate, effective_base, start_area_url=start_area_url, include_subdomains=include_subdomains):
                continue
            key = candidate.rstrip("/")
            if key in seen:
                continue
            seen.add(key)
            text = a.get_text(" ", strip=True).strip() or candidate
            found.append({"url": candidate, "text": text})
    log.info("Discovered %d child links with text from HTML", len(found))
    return found


# ---------------------------------------------------------------------------
# Markdown deduplication
# ---------------------------------------------------------------------------
def _normalize_line(line: str) -> str:
    return " ".join(line.strip().lower().split())


def _is_frontmatter_boundary(line: str) -> bool:
    return line.strip() == "---"


def dedupe_markdown_files(out_dir: Path) -> dict:
    """Remove lines that appear identically across many markdown files (headers, footers, nav).

    Each .md file gets its duplicate lines removed in-place.
    Removed lines are written to a companion ``_removed_parts.md`` file.
    Returns summary dict.
    """
    md_files = sorted(out_dir.glob("*.md"))
    # Exclude companion files from dedup
    md_files = [f for f in md_files if "_removed_parts" not in f.name and "__vision" not in f.name]
    if len(md_files) < 2:
        return {"files": len(md_files), "total_lines_removed": 0}

    file_lines: dict[str, list[tuple[str, bool]]] = {}
    line_counts: dict[str, int] = {}
    total_files = len(md_files)

    for md_path in md_files:
        raw = md_path.read_text(encoding="utf-8")
        lines = raw.splitlines()
        annotated: list[tuple[str, bool]] = []
        in_frontmatter = False
        seen_frontmatter = False
        for line in lines:
            stripped = line.strip()
            if _is_frontmatter_boundary(line):
                if not seen_frontmatter:
                    in_frontmatter = True
                    seen_frontmatter = True
                else:
                    in_frontmatter = False
                annotated.append((line, False))
                continue
            if in_frontmatter:
                annotated.append((line, False))
                continue
            norm = _normalize_line(line)
            if norm:
                line_counts[norm] = line_counts.get(norm, 0) + 1
            annotated.append((line, bool(norm)))
        file_lines[str(md_path)] = annotated

    threshold = max(3, int(total_files * 0.35))
    boilerplate_norms: set[str] = {norm for norm, count in line_counts.items() if count >= threshold}

    total_removed = 0
    for md_path in md_files:
        annotated = file_lines[str(md_path)]
        kept: list[str] = []
        removed: list[str] = []
        in_frontmatter = False
        seen_frontmatter = False
        for line, has_content in annotated:
            if _is_frontmatter_boundary(line):
                if not seen_frontmatter:
                    in_frontmatter = True
                    seen_frontmatter = True
                else:
                    in_frontmatter = False
                kept.append(line)
                continue
            if in_frontmatter or not has_content:
                kept.append(line)
                continue
            norm = _normalize_line(line)
            if norm in boilerplate_norms:
                removed.append(line)
                total_removed += 1
            else:
                kept.append(line)

        md_path.write_text("\n".join(kept) + "\n", encoding="utf-8")
        if removed:
            removed_path = md_path.parent / (md_path.stem + "_removed_parts.md")
            removed_path.write_text("\n".join(removed) + "\n", encoding="utf-8")
            log.info("Deduped %s: %d lines removed -> %s", md_path.name, len(removed), removed_path.name)

    log.info("Deduplication complete: %d boilerplate lines across %d norms in %d files",
             total_removed, len(boilerplate_norms), total_files)
    return {
        "files": total_files,
        "total_lines_removed": total_removed,
        "boilerplate_norms": len(boilerplate_norms),
        "threshold": threshold,
    }


# ---------------------------------------------------------------------------
# Page index: URL <-> stem mapping
# ---------------------------------------------------------------------------
def _save_page_index(out_dir: Path, results: dict[str, bool]) -> None:
    """Write _page_index.json mapping source_url -> stem and stem -> source_url.

    Filename transformations (``/`` -> ``__``, special chars -> ``_``) are lossy,
    so this mapping is required for downstream tools to resolve filenames back to URLs.
    """
    url_to_stem: dict[str, str] = {}
    stem_to_url: dict[str, str] = {}
    for url in sorted(results):
        stem = _url_output_stem(url)
        url_to_stem[url] = stem
        stem_to_url[stem] = url
    write_json(out_dir / "_page_index.json", {
        "schema_version": "crawler-page-index/v1",
        "generated_at": utc_now_iso(),
        "url_to_stem": url_to_stem,
        "stem_to_url": stem_to_url,
    })
    log.info("Saved page index: %d entries", len(url_to_stem))


# ---------------------------------------------------------------------------
# Page state (incremental crawl)
# ---------------------------------------------------------------------------
def _load_page_state(out_dir: Path) -> dict:
    path = out_dir / "_page_state.json"
    if not path.exists():
        return {"schema_version": "crawler-page-state/v1", "pages": {}}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {"schema_version": "crawler-page-state/v1", "pages": {}}


def _save_page_state(out_dir: Path, state: dict) -> None:
    state["updated_at"] = utc_now_iso()
    write_json(out_dir / "_page_state.json", state)


def _content_digest(html: str) -> str:
    return hashlib.sha256(html.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Markdown link rewriting
# ---------------------------------------------------------------------------
def rewrite_markdown_links(out_dir: Path) -> dict:
    """Rewrite cross-page markdown links from live URLs to local .md files.

    Loads ``_page_index.json``, reads each .md file's frontmatter ``source_url``,
    resolves every inline link against it, and rewrites links whose resolved
    absolute URL matches a known crawled page to the corresponding ``stem.md``.
    """
    index_path = out_dir / "_page_index.json"
    if not index_path.exists():
        log.warning("No _page_index.json in %s; skipping link rewriting", out_dir)
        return {"rewritten": 0, "files": 0}

    try:
        raw = index_path.read_text(encoding="utf-8")
        page_index = json.loads(raw)
    except Exception as exc:
        log.warning("Failed to load page index: %s", exc)
        return {"rewritten": 0, "files": 0}

    url_to_stem: dict[str, str] = page_index.get("url_to_stem", {})
    if not url_to_stem:
        return {"rewritten": 0, "files": 0}

    normalized_index: dict[str, str] = {}
    for url, stem in url_to_stem.items():
        clean = _strip_fragment_and_query(url).rstrip("/")
        if clean:
            normalized_index[clean.lower()] = stem

    md_files = sorted(out_dir.glob("*.md"))
    md_files = [f for f in md_files if "_removed_parts" not in f.name and "__vision" not in f.name]
    if not md_files:
        return {"rewritten": 0, "files": 0}

    total_rewritten = 0
    total_unlinked = 0
    files_modified = 0

    for md_path in md_files:
        text = md_path.read_text(encoding="utf-8")
        fm_match = re.search(r'^source_url:\s*(.+)$', text, re.MULTILINE)
        if not fm_match:
            continue
        source_url = fm_match.group(1).strip()
        source_domain = _real_host(source_url)
        new_text, rw, ul = _rewrite_links_in_text(text, source_url, source_domain, normalized_index, md_path.stem)
        if new_text != text:
            md_path.write_text(new_text, encoding="utf-8")
            files_modified += 1
            total_rewritten += rw
            total_unlinked += ul
            log.info("Link rewrite %s: %d rewritten, %d unlinked", md_path.name, rw, ul)

    log.info("Link rewriting complete: %d rewritten, %d unlinked across %d files", total_rewritten, total_unlinked, files_modified)
    return {"rewritten": total_rewritten, "unlinked": total_unlinked, "files": files_modified}


def _rewrite_links_in_text(text: str, source_url: str, source_domain: str, normalized_index: dict[str, str], current_stem: str) -> tuple[str, int, int]:
    """Rewrite markdown links in *text* that match entries in *normalized_index*.

    Returns ``(new_text, rewritten_count, unlinked_count)``.
    Links to internal pages that were NOT crawled are unlinked (label kept, URL dropped).
    """
    links = _find_markdown_links(text)
    if not links:
        return text, 0, 0

    rw = 0
    ul = 0
    result = text
    for start, end, label, href, title in reversed(links):
        if not href or href.startswith("#"):
            continue
        try:
            resolved = urljoin(source_url, href)
        except Exception:
            continue
        clean = _strip_fragment_and_query(resolved).rstrip("/")
        if not clean:
            continue
        stem = normalized_index.get(clean.lower())
        if stem is None:
            stem = normalized_index.get((clean + "/").lower())
        if stem is not None and stem != current_stem:
            target = stem + ".md"
            if title:
                replacement = f"[{label}]({target} \"{title}\")"
            else:
                replacement = f"[{label}]({target})"
            result = result[:start] + replacement + result[end:]
            rw += 1
    return result, rw, 0


def _find_markdown_links(text: str) -> list[tuple[int, int, str, str, str | None]]:
    """Scan *text* for inline markdown links, handling nested brackets in labels.

    Returns a list of ``(start, end, label, url, title)`` tuples.
    """
    links: list[tuple[int, int, str, str, str | None]] = []
    i = 0
    n = len(text)
    while i < n:
        if text[i] != "[":
            i += 1
            continue
        if i > 0 and text[i - 1] == "!":
            i += 1
            continue
        depth = 1
        j = i + 1
        while j < n and depth > 0:
            if text[j] == "\\" and j + 1 < n:
                j += 2
                continue
            if text[j] == "[":
                depth += 1
            elif text[j] == "]":
                depth -= 1
            j += 1
        if depth != 0:
            i += 1
            continue
        label_end = j - 1
        label = text[i + 1 : label_end]

        k = label_end + 1
        if k < n and text[k] == " ":
            k += 1
        if k >= n or text[k] != "(":
            i = label_end + 1
            continue

        url_start = k + 1
        m = url_start
        while m < n and text[m] != ")":
            if text[m] == "\\" and m + 1 < n:
                m += 2
                continue
            m += 1
        if m >= n:
            i = label_end + 1
            continue

        raw_url = text[url_start:m].strip()
        parts = raw_url.split(None, 1)
        url = parts[0] if parts else ""
        title: str | None = None
        if len(parts) > 1:
            t = parts[1].strip()
            if (t.startswith('"') and t.endswith('"')) or (t.startswith("'") and t.endswith("'")):
                title = t[1:-1]

        links.append((i, m + 1, label, url, title))
        i = m + 1
    return links


# ---------------------------------------------------------------------------
def check_markdown_quality(out_dir: Path, page_state: dict | None = None) -> list[dict]:
    """Analyze each .md file for semantic clarity via DeepSeek API.

    Reads DEEPSEEK_API_KEY from environment.
    Writes per-page ``_md_quality.json`` files.
    Returns list of quality reports.
    Skips files whose stem corresponds to an unchanged page in *page_state*.
    """
    import os as _os
    import requests

    api_key = _os.environ.get("DEEPSEEK_API_KEY", "")
    if not api_key:
        log.warning("DEEPSEEK_API_KEY not set; skipping markdown quality check")
        return []

    unchanged_stems: set[str] = set()
    if page_state:
        for url, entry in page_state.get("pages", {}).items():
            if entry.get("status") == "unchanged":
                unchanged_stems.add(entry.get("stem", ""))

    model = "deepseek-v4-flash"
    url = "https://api.deepseek.com/v1/chat/completions"

    md_files = sorted(out_dir.glob("*.md"))
    md_files = [f for f in md_files if "_removed_parts" not in f.name]
    if unchanged_stems:
        skipped = [f for f in md_files if f.stem in unchanged_stems]
        md_files = [f for f in md_files if f.stem not in unchanged_stems]
        if skipped:
            log.info("Quality check: skipping %d unchanged files (%s)", len(skipped), ", ".join(f.name for f in skipped[:5]))
    results: list[dict] = []

    for md_path in md_files:
        text = md_path.read_text(encoding="utf-8")
        if len(text) > 8000:
            text = text[:8000] + "\n\n[... content truncated ...]"

        prompt = f"""Analyze this markdown file from a web crawler converted from HTML.

Rate how well an LLM could understand and extract factual information.

Check: Can values/numbers be assigned to their labels? Are there orphaned values? Are tables preserved?

You MUST output EXACTLY this JSON structure (no other fields, no markdown):

{{"understandable": true, "confidence": "sure", "value_alignment_score": 5, "issues": ["example issue"], "summary": "one sentence"}}

- understandable: true if facts can be extracted, false if too scrambled
- confidence: "sure", "high", "maybe", or "cannot_understand"
- value_alignment_score: integer 1 (scrambled) to 5 (perfectly clear)
- issues: list of problem strings (empty list if none)
- summary: one sentence describing overall quality

Markdown:
{text[:6000]}"""

        headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
        payload = {
            "model": model,
            "max_tokens": 4096,
            "temperature": 0.1,
            "response_format": {"type": "json_object"},
            "messages": [{"role": "user", "content": prompt}],
        }

        result: dict = {}
        for attempt in range(3):
            try:
                r = requests.post(url, headers=headers, json=payload, timeout=60)
                if r.status_code == 200:
                    data = r.json()
                    choice = data["choices"][0]["message"]
                    content = choice.get("content", "") or ""
                    if not content and not choice.get("reasoning_content"):
                        if attempt < 2:
                            time.sleep(1)
                            continue
                    raw = content.strip()
                    if raw.startswith("```"):
                        raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
                        if raw.endswith("```"):
                            raw = raw[:-3]
                        raw = raw.strip()
                    if raw:
                        result = json.loads(raw)
                        break
                elif r.status_code in (429, 500, 502, 503, 504):
                    delay = 2 * (2 ** attempt)
                    time.sleep(delay)
            except json.JSONDecodeError:
                if attempt < 2:
                    time.sleep(1)
            except Exception as exc:
                if attempt < 2:
                    time.sleep(1)
        else:
            result = {
                "understandable": False,
                "confidence": "cannot_understand",
                "value_alignment_score": 1,
                "issues": ["DeepSeek API failed after 3 attempts"],
                "summary": "API failure",
            }

        # Normalize field names that DeepSeek might use instead
        if "overall_rating" in result and "value_alignment_score" not in result:
            result["value_alignment_score"] = int(result.pop("overall_rating"))
        if "understand" in result and "understandable" not in result:
            result["understandable"] = bool(result.pop("understand"))
        if isinstance(result.get("value_assignment"), dict):
            va = result["value_assignment"]
            if "rating" in va and "value_alignment_score" not in result:
                result["value_alignment_score"] = int(va["rating"])
            if "issues" in va and not result.get("issues"):
                issues_val = va["issues"]
                if isinstance(issues_val, str):
                    result["issues"] = [issues_val]
                elif isinstance(issues_val, list):
                    result["issues"] = issues_val
        result.setdefault("understandable", result.get("confidence") not in ("cannot_understand", None, ""))
        result.setdefault("confidence", "maybe" if result.get("understandable") else "cannot_understand")
        result.setdefault("value_alignment_score", 2)
        result.setdefault("issues", [])
        result.setdefault("summary", "")
        result["_page_stem"] = md_path.stem

        results.append(result)
        log.info("Markdown quality check for %s: confidence=%s alignment=%s",
                 md_path.name, result.get("confidence"), result.get("value_alignment_score"))
        quality_path = md_path.with_suffix("").with_suffix(md_path.suffix + "_quality.json")
        write_json(quality_path, result)

    return results
def _urls_match(a: str, b: str) -> bool:
    a_norm = a.rstrip("/")
    b_norm = b.rstrip("/")
    if a_norm == b_norm:
        return True
    pa, pb = urlparse(a_norm), urlparse(b_norm)
    return pa.scheme == pb.scheme and pa.netloc == pb.netloc and pa.path.rstrip("/") == pb.path.rstrip("/")


async def _current_location(tab) -> str:
    try:
        loc = await await_with_timeout(
            "Runtime.evaluate location.href",
            evaluate_value(tab, "() => location.href || ''"),
            timeout=5.0,
            default="",
            warning=True,
        )
        return str(loc).strip() if loc else ""
    except Exception:
        return ""


async def _fetch_live_page(tab, url: str, options: dict) -> tuple[str | None, dict]:
    if options.get("respect_robots", True) and not can_fetch(url):
        log.warning("Blocked by robots.txt: %s", url)
        return None, {}
    log.info("Fetching live page: %s", url)

    nav_timeout = float(options.get("navigation_timeout") or 30.0)
    current = await _current_location(tab)
    if current and _urls_match(current, url):
        log.info("Tab already on target URL %s; skipping navigation", url)
        await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
        t_wait = time.perf_counter()
        metrics = await wait_for_rendered_content_stable(tab, timeout=float(options.get("content_timeout") or 20.0))
        _timing_record(url, "content_wait", time.perf_counter() - t_wait)
        t_html = time.perf_counter()
        html = await get_tab_content(tab, timeout=float(options.get("html_timeout") or 15.0), required=True)
        _timing_record(url, "html_capture", time.perf_counter() - t_html)
        log.info("Fetched rendered HTML for %s: %d characters", url, len(html))
        return html, metrics

    result = await await_with_timeout("nodriver Tab.get(url)", tab.get(url), timeout=nav_timeout, default=None, warning=True)
    if result is None:
        log.warning("Navigation to %s timed out; attempting fallback content capture from current tab", url)

    await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
    t_wait = time.perf_counter()
    metrics = await wait_for_rendered_content_stable(tab, timeout=float(options.get("content_timeout") or 20.0))
    _timing_record(url, "content_wait", time.perf_counter() - t_wait)

    # Scroll to trigger lazy-loaded images, carousels, and font downloads
    await _scroll_to(tab, 99999)
    await tab.sleep(2.0)
    await _scroll_to(tab, 0)
    await tab.sleep(0.5)

    t_html = time.perf_counter()
    html = await get_tab_content(tab, timeout=float(options.get("html_timeout") or 15.0), required=True)
    _timing_record(url, "html_capture", time.perf_counter() - t_html)
    log.info("Fetched rendered HTML for %s: %d characters", url, len(html))
    return html, metrics


async def _fetch_and_save(tab, url: str, out_dir: Path, options: dict) -> tuple[bool, str | None]:
    log.info("Begin fetch/save for %s", url)
    page_stem = _url_output_stem(url)
    page_path = out_dir / f"{page_stem}.html"

    page_state = options.get("_page_state")
    keep_crawl = bool(options.get("keep_crawl", False))
    consent_result = options.get("_consent_result")
    if _is_cloudflare_blocked(consent_result):
        marker_path = _write_cloudflare_blocked_artifacts(out_dir, page_stem, url, consent_result)
        if keep_crawl and page_state is not None:
            page_state.setdefault("pages", {})[url] = {
                "stem": page_stem,
                "content_digest": None,
                "html_bytes": 0,
                "status": "blocked_by_cloudflare",
                "blocked_marker_path": str(marker_path.resolve()),
                "crawled_at": utc_now_iso(),
            }
        return False, None

    raw_html, render_metrics = await _fetch_live_page(tab, url, options)
    if raw_html is None:
        return False, None

    if keep_crawl and page_state is not None:
        check_html = clean_html_for_content(raw_html)
        new_digest = _content_digest(check_html)
        pages = page_state.setdefault("pages", {})
        prev = pages.get(url, {})
        if prev.get("content_digest") == new_digest:
            prev["crawled_at"] = utc_now_iso()
            prev["status"] = "unchanged"
            log.info("Content unchanged for %s; skipping reprocess", url)
            _timing_record(url, "html_save", 0)
            _timing_record(url, "markdown_generate", 0)
            _timing_record(url, "markdown_save", 0)
            return True, raw_html

    # DB cache check: if page already crawled and content unchanged, skip
    if keep_crawl and check_cache is not None:
        try:
            entry = check_cache(url)
            if entry and entry.found and entry.is_fresh and entry.content_digest:
                check_html_local = clean_html_for_content(raw_html)
                new_digest_local = _content_digest(check_html_local)
                # Try digest match first, then time-based fallback if files exist
                md_exists = (out_dir / f"{entry.stem}.md").exists()
                digest_match = entry.content_digest == new_digest_local
                if digest_match or (md_exists and entry.crawled_at):
                    log.info("DB cache hit: content %s for %s (crawled %.0fs ago)",
                             "matches" if digest_match else "recent + files on disk",
                             url, time.time() - entry.crawled_at)
                    _timing_record(url, "html_save", 0)
                    _timing_record(url, "markdown_generate", 0)
                    _timing_record(url, "markdown_save", 0)
                    return True, raw_html
        except Exception as exc:
            log.debug("DB cache check error (non-fatal): %s", exc)

    t_save = time.perf_counter()
    if options.get("images", False):
        localized_html, image_manifest = await download_and_rewrite_images(tab, raw_html, page_url=url, out_dir=out_dir, page_stem=page_stem, max_images=int(options.get("max_images") or 0))
    else:
        localized_html = raw_html
        image_manifest = []

    if options.get("download_assets", False):
        log.info("Downloading assets (CSS/JS/fonts) from browser cache for %s", url)
        try:
            localized_html = await _download_and_rewrite_assets(tab, localized_html, url, out_dir, options)
        except Exception as exc:
            log.warning("Asset download failed for %s: %s; continuing without assets", url, exc)

    cleaned_html = clean_html_for_content(localized_html)

    write_text(page_path, cleaned_html)
    text_path = page_path.with_suffix(".txt")
    vector_text = extract_text_from_html(cleaned_html, url)
    write_text(text_path, vector_text)
    images_manifest_path = out_dir / f"{page_stem}.images.json"
    write_json(images_manifest_path, {"schema_version": "crawler-page-images/v1", "page_url": url, "images": image_manifest})
    append_resource_map(out_dir, image_manifest)

    log.info("Saved %s", _safe_relative(page_path, SITES_ROOT))
    log.info("Saved %s", _safe_relative(text_path, SITES_ROOT))
    log.info("Saved %s", _safe_relative(images_manifest_path, SITES_ROOT))
    _timing_record(url, "html_save", time.perf_counter() - t_save)

    screenshot_metadata: list[dict] = []
    screenshots_mode = options.get("screenshots", False)
    screenshots_active = screenshots_mode is True or str(screenshots_mode).lower() in ("true", "full", "viewport_initial")
    if screenshots_active:
        screenshot_metadata = await capture_page_screenshots(
            tab,
            url=url,
            out_dir=out_dir,
            page_stem=page_stem,
            viewport_width=int(options.get("viewport_width") or 1920),
            viewport_height=int(options.get("viewport_height") or 1080),
            max_tiles=int(options.get("screenshot_max_tiles") or 0),
            overlap=int(options.get("screenshot_overlap") or 120),
            viewport_only=(screenshots_mode == "viewport_initial"),
        )

    md_path = page_path.with_suffix(".md")
    structured_json_path = page_path.with_suffix(".json")
    if options.get("markdown", False):
        t_md_gen = time.perf_counter()
        if options.get("vision", False):
            if process_webpage_screenshots is None:
                raise ImportError("webpage_vision_markdown.py or google-genai dependencies are unavailable")
            if not screenshot_metadata:
                raise RuntimeError("--vision true requires screenshots to be enabled")
            process_webpage_screenshots(
                screenshot_metadata=screenshot_metadata,
                output_md_path=md_path,
                output_json_path=structured_json_path,
                model=str(options.get("gemini_model") or DEFAULT_GEMINI_MODEL),
                max_attempts=int(options.get("max_attempts") or DEFAULT_MAX_ATTEMPTS),
                text_hint=None,
                force=bool(options.get("force_vision", False)),
            )
        else:
            markdown = markdown_from_rendered_html(cleaned_html, url)
            write_text(md_path, markdown)
        _timing_record(url, "markdown_generate", time.perf_counter() - t_md_gen)
        t_md_save = time.perf_counter()
        log.info("Saved %s", _safe_relative(md_path, SITES_ROOT))
        if not options.get("vision", False) and structured_json_path.exists() is False:
            pass  # vision-only
        else:
            log.info("Saved %s", _safe_relative(structured_json_path, SITES_ROOT))
        _timing_record(url, "markdown_save", time.perf_counter() - t_md_save)
    crawl_metadata = {
        "schema_version": "crawler-page/v2",
        "generated_at": utc_now_iso(),
        "url": url,
        "html_path": str(page_path.resolve()),
        "txt_path": str(text_path.resolve()),
        "images_manifest_path": str(images_manifest_path.resolve()),
        "images_count": len(image_manifest),
        "screenshots_enabled": bool(options.get("screenshots", False)),
        "screenshot_metadata": screenshot_metadata,
        "markdown_enabled": bool(options.get("markdown", False)),
        "vision_enabled": bool(options.get("vision", False)),
        "markdown_path": str(md_path.resolve()) if md_path.exists() else None,
        "structured_json_path": str(structured_json_path.resolve()) if structured_json_path.exists() else None,
        "render_metrics": render_metrics,
    }
    crawl_metadata_path = page_path.with_suffix(".crawl.json")
    write_json(crawl_metadata_path, crawl_metadata)
    log.info("Saved %s", _safe_relative(crawl_metadata_path, SITES_ROOT))

    if keep_crawl and page_state is not None:
        pages = page_state.setdefault("pages", {})
        pages[url] = {
            "stem": page_stem,
            "content_digest": _content_digest(clean_html_for_content(raw_html)),
            "html_bytes": len(raw_html.encode("utf-8")),
            "status": "ok",
            "crawled_at": utc_now_iso(),
        }

    # Save to MariaDB cache (only when persistent=true)
    if options.get("persistent", False) and save_page_record is not None:
        try:
            domain = _real_host(url)
            md_path_check = page_path.with_suffix(".md")
            md_bytes = md_path_check.stat().st_size if md_path_check.exists() else 0
            img_bytes = sum(e.get("size_bytes", 0) for e in image_manifest)
            save_page_record(
                url=url, domain=domain, stem=page_stem, status="ok",
                content_digest=_content_digest(clean_html_for_content(raw_html)),
                markdown_bytes=md_bytes, html_bytes=len(raw_html.encode("utf-8")),
                image_count=len(image_manifest), images_bytes=img_bytes,
                crawled_at=time.time(),
                http_status=render_metrics.get("http_status") if render_metrics else None,
                elapsed_seconds=None, consent_status=None,
            )
            if image_manifest:
                save_page_images(url, image_manifest)
        except Exception as exc:
            log.debug("DB save error (non-fatal): %s", exc)

    # Page classification (only when persistent=true)
    if options.get("persistent", False) and save_page_record is not None and classify_page is not None:
        try:
            visible_text = ""
            text_path_check = page_path.with_suffix(".txt")
            if text_path_check.exists():
                visible_text = text_path_check.read_text(encoding="utf-8", errors="replace")
            if not visible_text:
                visible_text = cleaned_html if 'cleaned_html' in dir() else ""
            title = ""
            soup_check = BeautifulSoup(raw_html, "html.parser")
            if soup_check.title:
                title = soup_check.title.get_text(" ", strip=True)
            headings = [h.get_text(" ", strip=True) for h in soup_check.find_all(["h1", "h2", "h3"])][:30]
            link_list = discover_children_with_text(raw_html, url, include_subdomains=options.get("include_subdomains", False))
            classification = classify_page(
                url=url, title=title, visible_text=visible_text,
                links=link_list, headings=headings,
            )
            if classification:
                from crawl_db import _connect
                conn = _connect()
                with conn.cursor() as cur:
                    cur.execute(
                        "UPDATE pages SET page_type = %s WHERE url_hash = %s",
                        (classification.get("page_class", "normal_page"), _url_hash(url)),
                    )
                conn.commit()

                # Gemini analysis for low-confidence classifications
                gemini_enabled = options.get("gemini_analysis", True)
                if gemini_enabled and analyze_page is not None and classification.get("confidence", 1) < 0.85:
                    log.info("Low classifier confidence (%.2f), running Gemini analysis for %s",
                             classification["confidence"], url[:80])
                    gemini_result = analyze_page(
                        url=url, title=title, visible_text=visible_text,
                        links=link_list, headings=headings,
                    )
                    if gemini_result:
                        classification["gemini"] = gemini_result
                        classification["classified_by"] = "gemini"
                        classification["confidence"] = gemini_result.get("confidence", classification["confidence"])
                        classification["page_class"] = gemini_result.get("page_class", classification["page_class"])

                classification.pop("signals", None)  # keep response compact
        except Exception as exc:
            log.debug("Page classification failed (non-fatal): %s", exc)

    return True, raw_html


def _page_worker_timeout(options: dict) -> float:
    """Upper bound for one page fetch/save cycle.

    This protects the crawl from nodriver/CDP calls that never return after
    consent or during screenshot/content capture. The value intentionally keeps
    a generous floor because screenshots and markdown generation can be slow on
    media-heavy pages.
    """
    return max(
        180.0,
        float(options.get("navigation_timeout") or 30.0)
        + float(options.get("content_timeout") or 20.0)
        + float(options.get("html_timeout") or 15.0)
        + 120.0,
    )


# ---------------------------------------------------------------------------
# Browser lifecycle / public API
# ---------------------------------------------------------------------------
async def _prepare_browser(first_url: str, *, headless: bool, options: dict):
    domain = _real_host(first_url)
    out_dir = output_dir_for(first_url)
    docker_mode = bool(options.get("docker")) or os.getenv("CRAWLER_DOCKER", "").lower() == "true"
    chrome_binary = os.getenv("CRAWLER_CHROME_BINARY", "")
    browser_args = _browser_args_for_viewport(
        int(options.get("viewport_width") or 1920),
        int(options.get("viewport_height") or 1080),
        docker=docker_mode,
    )
    nav_timeout = float(options.get("navigation_timeout") or 30.0)
    log.info("Starting browser through nodriver uc.start(headless=%s, browser_args=%s)", headless, browser_args)
    start_kwargs = {"headless": headless, "browser_args": browser_args}
    if chrome_binary:
        start_kwargs["browser_executable_path"] = chrome_binary
    browser = await await_with_timeout(
        "nodriver uc.start()",
        uc.start(**start_kwargs),
        timeout=max(30.0, nav_timeout),
        default=None,
        warning=True,
    )
    if browser is None:
        raise RuntimeError("Could not start browser through nodriver within timeout")
    cookies_loaded = await load_browser_cookies(browser, domain)
    tab = await await_with_timeout(
        f"nodriver browser.get() consent page {first_url}",
        browser.get(first_url),
        timeout=nav_timeout,
        default=None,
        warning=True,
    )
    if tab is None:
        browser.stop()
        raise RuntimeError(f"Could not open consent page within timeout: {first_url}")
    await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))

    # Verify consent on the online page before any final capture/output.
    page_stem = _url_output_stem(first_url)
    consent_enabled = bool(options.get("consent_check", True))

    t_consent = time.perf_counter()
    if cookies_loaded and consent_enabled:
        challenge = None
        if bool(options.get("cloudflare_check", True)):
            await tab.sleep(float(options.get("consent_wait") or 3.0))
            viewport = await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
            challenge = await _detect_cloudflare_challenge_on_viewport(
                tab,
                domain=domain,
                out_dir=out_dir,
                page_stem=page_stem,
                viewport=viewport,
                viewport_width=int(options.get("viewport_width") or 1920),
                viewport_height=int(options.get("viewport_height") or 1080),
                source_url=first_url,
                label="challenge_check",
            )
        if challenge and challenge.get("is_challenge"):
            consent_result = _blocked_consent_result(domain, challenge, enabled=consent_enabled)
        else:
            dom_has_consent = await _dom_consent_check(tab)
            if dom_has_consent is False:
                log.info("Consent: skipped (cookies loaded, no consent overlay detected via DOM)")
                consent_result = {
                    "schema_version": "crawler-cookie-consent/v1",
                    "checked_at": utc_now_iso(),
                    "enabled": True,
                    "domain": domain,
                    "status": "skipped_cookies_loaded",
                    "attempts": [],
                }
                if challenge is not None:
                    consent_result["cloudflare_check"] = challenge
            else:
                consent_result = await verify_and_accept_cookie_consent(
                    browser,
                    tab,
                    domain=domain,
                    out_dir=out_dir,
                    page_stem=page_stem,
                    viewport_width=int(options.get("viewport_width") or 1920),
                    viewport_height=int(options.get("viewport_height") or 1080),
                    enabled=consent_enabled,
                    wait_before_check=float(options.get("consent_wait") or 3.0),
                    max_attempts=int(options.get("consent_attempts") or 2),
                    source_url=first_url,
                    cloudflare_check_enabled=challenge is None and bool(options.get("cloudflare_check", True)),
                )
    else:
        consent_result = await verify_and_accept_cookie_consent(
            browser,
            tab,
            domain=domain,
            out_dir=out_dir,
            page_stem=page_stem,
            viewport_width=int(options.get("viewport_width") or 1920),
            viewport_height=int(options.get("viewport_height") or 1080),
            enabled=consent_enabled,
            wait_before_check=float(options.get("consent_wait") or 3.0),
            max_attempts=int(options.get("consent_attempts") or 2),
            source_url=first_url,
            cloudflare_check_enabled=bool(options.get("cloudflare_check", True)),
        )
    _timing_record(first_url, "consent_prepare", time.perf_counter() - t_consent)
    write_json(out_dir / "_consent_snapshots.json", consent_result)
    log.info("Consent verification returned status=%s", consent_result.get("status"))
    if consent_result.get("status") == "still_visible_after_click":
        log.warning("Cookie consent may still be visible. Final artifacts may be contaminated by the overlay.")
    if not _is_cloudflare_blocked(consent_result):
        t_overlay = time.perf_counter()
        overlay_result = await dismiss_blocking_overlays(
            tab,
            domain=domain,
            out_dir=out_dir,
            page_stem=page_stem,
            viewport_width=int(options.get("viewport_width") or 1920),
            viewport_height=int(options.get("viewport_height") or 1080),
            max_attempts=3,
            source_url=first_url,
        )
        _timing_record(first_url, "overlay_dismiss", time.perf_counter() - t_overlay)
        if overlay_result.get("overlays_found", 0) > 0:
            write_json(out_dir / "_overlay_dismiss.json", overlay_result)
            log.info("Overlay dismissal: found=%d dismissed=%d", overlay_result["overlays_found"], overlay_result["overlays_dismissed"])
    log.info("Browser prepared; crawl may proceed")
    return browser, tab, domain, out_dir, consent_result


async def fetch_single_result(url: str, *, headless: bool = False, options: dict | None = None) -> dict[str, object]:
    options = options or {}
    t0 = time.perf_counter()

    cache_policy = options.get("cache_policy", "revalidate")
    query_mode = options.get("query_mode", "semantic")
    canonical = canonicalize_url(url, query_mode=query_mode)
    target_url = canonical or _strip_fragment_and_query(url)
    page_stem = _url_output_stem(target_url)

    # Cache-first: check DB before browser startup
    if cache_policy == "cache_first" and check_cache is not None and canonical:
        try:
            entry = check_cache(url, canonical=canonical)
            domain = _real_host(target_url)
            out_dir = output_dir_for(target_url)
            if entry and entry.found and entry.is_fresh:
                md_path = out_dir / f"{page_stem}.md"
                if md_path.exists():
                    log.info("Cache-first hit for %s (crawled %.0fs ago)", target_url, time.time() - entry.crawled_at)
                    _timing_record(target_url, "cache_hit", time.perf_counter() - t0)
                    return {
                        "url": target_url, "domain": domain, "status": "ok",
                        "cache_hit": True,
                        "fast_markdown_path": str(md_path.resolve()),
                        "html_path": str((out_dir / f"{page_stem}.html").resolve()),
                        "txt_path": str((out_dir / f"{page_stem}.txt").resolve()),
                        "images_manifest_path": str((out_dir / f"{page_stem}.images.json").resolve()),
                        "crawl_metadata_path": str((out_dir / f"{page_stem}.crawl.json").resolve()),
                        "consent_status": "cache_hit",
                    }
        except Exception as exc:
            log.debug("Cache-first check failed (non-fatal): %s", exc)

    if options.get("respect_robots", True) and not can_fetch(target_url):
        log.warning("Blocked by robots.txt (pre-browser check): %s", target_url)
        return {
            "url": target_url,
            "domain": _real_host(target_url),
            "status": "blocked",
            "blocked_reason": "robots_txt",
            "consent_status": None,
            "fast_markdown_path": None,
            "elapsed_seconds": round(time.perf_counter() - t0, 3),
        }

    api_result: dict[str, object] = {
        "url": target_url,
        "domain": _real_host(target_url),
        "status": "error",
    }
    browser, consent_tab, domain, out_dir, consent_result = await _prepare_browser(url, headless=headless, options=options)
    _timing_record(url, "consent", time.perf_counter() - t0)
    options["_consent_result"] = consent_result
    if _is_cloudflare_blocked(consent_result):
        marker_path = _write_cloudflare_blocked_artifacts(out_dir, page_stem, target_url, consent_result)
        api_result.update({
            "status": "blocked",
            "blocked_reason": "cloudflare",
            "blocked_provider": _cloudflare_provider(consent_result),
            "blocked_confidence": _cloudflare_confidence(consent_result),
            "cloudflare_visible_text": _cloudflare_visible_text(consent_result),
            "blocked_marker_path": str(marker_path.resolve()),
            "consent_status": consent_result.get("status"),
            "fast_markdown_path": None,
            "html_path": None,
            "txt_path": None,
            "images_manifest_path": None,
            "crawl_metadata_path": str((out_dir / f"{page_stem}.crawl.json").resolve()),
        })
        write_json(out_dir / f"{page_stem}.api.json", api_result)
        await close_tab_safely(consent_tab, f"blocked page tab {target_url}")
        browser.stop()
        return api_result
    if options.get("keep_crawl", False):
        options["_page_state"] = _load_page_state(out_dir)

    current = await _current_location(consent_tab)
    if current and _urls_match(current, target_url):
        log.info("Reusing prepared tab for single page crawl: %s", target_url)
        tab = consent_tab
        _timing_record(target_url, "navigate", 0.0)
    else:
        log.info("Opening fresh tab for single page crawl: %s (current=%s)", target_url, current)
        t_nav = time.perf_counter()
        tab = await await_with_timeout(
            f"nodriver browser.get(new_tab=True) for {target_url}",
            browser.get(target_url, new_tab=True),
            timeout=float(options.get("navigation_timeout") or 30.0),
            default=None,
            warning=True,
        )
        if tab is None:
            browser.stop()
            return api_result
        await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
        _timing_record(target_url, "navigate", time.perf_counter() - t_nav)

    fullpage_md_path = None
    fullpage_manifest = []
    quality_results = []
    api_result: dict = {"url": target_url, "domain": _real_host(target_url), "status": "error"}

    try:
        try:
            ok, _ = await asyncio.wait_for(_fetch_and_save(tab, target_url, out_dir, options), timeout=_page_worker_timeout(options))
        except asyncio.TimeoutError:
            log.warning("Single page crawl timed out for %s after %.1fs", target_url, _page_worker_timeout(options))
            ok = False
        await save_browser_cookies(browser, domain)
        md_results = {_strip_fragment_and_query(url): ok}

        # Fast Markdown is already written. Record state.
        fast_md_path = out_dir / f"{page_stem}.md"
        api_result["fast_markdown_path"] = str(fast_md_path.resolve()) if fast_md_path.exists() else None
        api_result["html_path"] = str((out_dir / f"{page_stem}.html").resolve())
        api_result["txt_path"] = str((out_dir / f"{page_stem}.txt").resolve())
        api_result["images_manifest_path"] = str((out_dir / f"{page_stem}.images.json").resolve())
        api_result["crawl_metadata_path"] = str((out_dir / f"{page_stem}.crawl.json").resolve())
        api_result["consent_status"] = consent_result.get("status")
        api_result["status"] = "ok" if ok else "error"

        # Step 5-6: Full-page background pass (while tab is still open)
        if ok and options.get("fullpage_background", False):
            log.info("Starting full-page background pass for %s", target_url)
            try:
                fp_md_path, fp_manifest = await _run_fullpage_background(
                    tab, target_url, out_dir, page_stem, options,
                )
                if fp_md_path:
                    fullpage_md_path = str(fp_md_path.resolve())
                    fullpage_manifest = fp_manifest
                    api_result["fullpage_markdown_path"] = fullpage_md_path
            except Exception as exc:
                log.warning("Full-page background pass failed for %s: %s", target_url, exc)

        t_post = time.perf_counter()
        # Post-processing
        if options.get("markdown", False):
            try:
                dedupe_markdown_files(out_dir)
            except Exception as exc:
                log.warning("Markdown deduplication failed: %s; continuing", exc)
            try:
                _save_page_index(out_dir, md_results)
                rewrite_markdown_links(out_dir)
            except Exception as exc:
                log.warning("Markdown link rewriting failed: %s; continuing", exc)

            # Step 7: Quality modes
            quality_mode = str(options.get("quality_mode") or "fast")
            do_quality = options.get("quality_check", False)
            if do_quality and quality_mode == "blocking":
                t_q = time.perf_counter()
                try:
                    quality_results = check_markdown_quality(out_dir, options.get("_page_state"))
                    _timing_record(target_url, "quality_classify", time.perf_counter() - t_q)
                except Exception as exc:
                    log.warning("Markdown quality check failed: %s; continuing", exc)
            elif do_quality and quality_mode == "background":
                # Launch quality check but don't await it
                async def _bg_quality():
                    t_q = time.perf_counter()
                    try:
                        results = await asyncio.to_thread(check_markdown_quality, out_dir, options.get("_page_state"))
                        _timing_record(target_url, "quality_classify", time.perf_counter() - t_q)
                        return results
                    except Exception as exc:
                        log.warning("Background quality check failed: %s; continuing", exc)
                        return []
                asyncio.create_task(_bg_quality())  # fire-and-forget for now
            else:
                log.debug("No Markdown quality check for %s (mode=%s)", _safe_relative(out_dir, SITES_ROOT), quality_mode)

        _save_page_index(out_dir, md_results)
        if options.get("_page_state"):
            _save_page_state(out_dir, options["_page_state"])

        # Step 8: Vision markdown only if quality is bad
        if options.get("vision_on_bad_quality", False) and quality_results:
            bad_quality = any(
                r.get("confidence") in ("cannot_understand", "maybe") or r.get("value_alignment_score", 5) <= 2
                for r in quality_results
            )
            if bad_quality and process_webpage_screenshots is not None:
                log.info("Quality is bad for %s; running vision markdown", target_url)
                t_vision = time.perf_counter()
                try:
                    # Capture screenshots if not already present
                    screenshot_metadata = await capture_page_screenshots(
                        tab, url=target_url, out_dir=out_dir, page_stem=page_stem,
                        viewport_width=int(options.get("viewport_width") or 1920),
                        viewport_height=int(options.get("viewport_height") or 1080),
                        max_tiles=int(options.get("screenshot_max_tiles") or 0),
                        overlap=int(options.get("screenshot_overlap") or 120),
                    )
                    vision_md_path = out_dir / f"{page_stem}__vision.md"
                    vision_json_path = out_dir / f"{page_stem}__vision.json"
                    process_webpage_screenshots(
                        screenshot_metadata=screenshot_metadata,
                        output_md_path=vision_md_path,
                        output_json_path=vision_json_path,
                        model=str(options.get("gemini_model") or DEFAULT_GEMINI_MODEL),
                        max_attempts=int(options.get("max_attempts") or DEFAULT_MAX_ATTEMPTS),
                        text_hint=None,
                        force=bool(options.get("force_vision", False)),
                    )
                    api_result["vision_markdown_path"] = str(vision_md_path.resolve())
                    _timing_record(target_url, "vision_markdown", time.perf_counter() - t_vision)
                except Exception as exc:
                    log.warning("Vision markdown fallback failed for %s: %s", target_url, exc)
            else:
                log.debug("Quality is acceptable for %s; skipping vision markdown", target_url)

        _timing_record(target_url, "postprocess", time.perf_counter() - t_post)

        # Step 9: Write API result contract
        api_result.update({
            "quality": quality_results[0] if quality_results else None,
            "timing_csv_path": str(_timing_log._csv_path.resolve()) if _timing_log else None,
            "timing_log_path": str(_timing_log._log_path.resolve()) if _timing_log else None,
        })
        write_json(out_dir / f"{page_stem}.api.json", api_result)

        # Interactive discovery: click JS elements, capture modals (only when persistent + interactive mode enabled)
        interactive_mode = options.get("interactive_mode", "off")
        if ok and options.get("persistent", False) and interactive_mode in {"discover_only", "full"} and discover_interactive_targets is not None:
            try:
                t_int = time.perf_counter()
                candidates = await discover_interactive_targets(tab, target_url)
                max_int = int(options.get("max_interactions_per_page", 80))
                max_virtual = int(options.get("max_virtual_pages_per_page", 50))
                interactive_results = []
                virtual_count = 0
                for idx, candidate in enumerate(candidates[:max_int]):
                    result = await execute_interaction(tab, target_url, candidate)
                    if not result:
                        continue
                    int_entry = {
                        "candidate": {
                            "text": candidate.get("text", "")[:100],
                            "tag": candidate.get("tag", ""),
                            "action_key": candidate.get("action_key", ""),
                        },
                        "result_type": result.get("result_type", "unknown"),
                    }
                    if result.get("result_type") == "url_navigation":
                        new_url = result.get("after_url", "")
                        if new_url:
                            int_entry["discovered_url"] = new_url
                        await restore_page_state(tab, target_url)
                    elif result.get("result_type") == "modal":
                        if virtual_count < max_virtual:
                            modal = await capture_modal_page(
                                tab, target_url, out_dir,
                                candidate.get("text", "modal"),
                                candidate.get("action_key", ""),
                            )
                            if modal:
                                int_entry["modal_page"] = modal
                                virtual_count += 1
                            await restore_page_state(tab, target_url)
                    elif result.get("result_type") in ("no_change", "failed"):
                        await restore_page_state(tab, target_url)
                    else:
                        int_entry["dom_changed"] = True
                    interactive_results.append(int_entry)
                    if idx % 10 == 0 and idx > 0:
                        log.info("Interactive: %d/%d processed", idx, len(candidates[:max_int]))
                api_result["interactive_results"] = interactive_results
                _timing_record(target_url, "interactive_discovery", time.perf_counter() - t_int)
                log.info("Interactive done for %s: %d actions, %d virtual", target_url, len(interactive_results), virtual_count)
            except Exception as exc:
                log.warning("Interactive discovery failed for %s: %s", target_url, exc)
                api_result["interactive_error"] = str(exc)

        _timing_record(target_url, "total_page", time.perf_counter() - t0)
        return api_result
    finally:
        await close_tab_safely(tab, f"single page tab {target_url}")
        browser.stop()


async def fetch_single(url: str, *, headless: bool = False, options: dict | None = None) -> bool:
    result = await fetch_single_result(url, headless=headless, options=options)
    return result.get("status") == "ok"

async def fetch_page_urls(file_path: str, *, headless: bool = False, delay: float = 0.5, limit: int = 0, options: dict | None = None) -> dict[str, bool]:
    p = Path(file_path)
    if not p.exists():
        log.error("File not found: %s", file_path)
        return {}
    urls = [line.strip() for line in p.read_text(encoding="utf-8").splitlines() if line.strip() and not line.strip().startswith("#")]
    if not urls:
        return {}
    return await fetch_urls(urls, headless=headless, delay=delay, limit=limit, options=options)


async def fetch_urls(urls: list[str], *, headless: bool = False, delay: float = 0.5, limit: int = 0, options: dict | None = None) -> dict[str, bool]:
    """Crawl a list of URLs spanning one or more domains.

    URLs are grouped by domain internally. Domain groups are processed in
    parallel, with one browser session per domain. This keeps cookie handling
    isolated per site and avoids a single site's consent/tab-close issue from
    blocking all remaining URLs.
    """
    options = options or {}
    if not urls:
        return {}

    normalized_urls = [_strip_fragment_and_query(url) for url in urls if str(url).strip()]
    if limit and limit > 0:
        normalized_urls = normalized_urls[:limit]
    if not normalized_urls:
        return {}

    domain_groups: dict[str, list[str]] = {}
    for url in normalized_urls:
        domain = _real_host(url)
        domain_groups.setdefault(domain, []).append(url)

    domain_list = sorted(domain_groups.keys())
    total_threads = max(1, int(options.get("threads") or 4))
    domain_concurrency = max(1, min(len(domain_list), total_threads))
    domain_sem = asyncio.Semaphore(domain_concurrency)
    all_results: dict[str, bool] = {}
    all_results_lock = asyncio.Lock()

    async def _process_domain_group(di: int, domain_name: str, domain_urls: list[str]) -> dict[str, bool]:
        async with domain_sem:
            first_url = domain_urls[0]
            log.info("Domain group %d/%d (%s): %d URLs", di + 1, len(domain_list), domain_name, len(domain_urls))

            browser = None
            domain_options = dict(options)
            domain_options.pop("_page_state", None)
            results: dict[str, bool] = {}
            failed: list[str] = []

            try:
                t_consent = time.perf_counter()
                browser, consent_tab, prepared_domain, out_dir, consent_result = await _prepare_browser(first_url, headless=headless, options=domain_options)
                _timing_record(first_url, "consent", time.perf_counter() - t_consent)
                domain_name = prepared_domain
                domain_options["_consent_result"] = consent_result
                if _is_cloudflare_blocked(consent_result):
                    marker_path = _write_cloudflare_blocked_artifacts(out_dir, _url_output_stem(first_url), first_url, consent_result)
                    log.warning("Domain group %s is blocked by %s; skipping %d URL(s)", domain_name, _cloudflare_provider(consent_result), len(domain_urls))
                    for blocked_url in domain_urls:
                        if blocked_url != first_url:
                            _write_cloudflare_blocked_artifacts(out_dir, _url_output_stem(blocked_url), blocked_url, consent_result)
                        results[blocked_url] = False
                    return results
                if domain_options.get("keep_crawl", False):
                    domain_options["_page_state"] = _load_page_state(out_dir)

                if len(domain_list) > 1:
                    per_domain_threads = 1
                else:
                    per_domain_threads = max(1, min(len(domain_urls), total_threads))
                sem = asyncio.Semaphore(per_domain_threads)
                lock = asyncio.Lock()

                async def _fetch_url_worker(url: str, existing_tab=None):
                    async with sem:
                        t_page = time.perf_counter()
                        tab = existing_tab
                        close_when_done = existing_tab is None
                        ok = False
                        try:
                            if tab is None:
                                t_nav = time.perf_counter()
                                tab = await await_with_timeout(
                                    f"nodriver browser.get(new_tab=True) for {url}",
                                    browser.get(url, new_tab=True),
                                    timeout=float(domain_options.get("navigation_timeout") or 30.0),
                                    default=None,
                                    warning=True,
                                )
                                if tab is None:
                                    raise TimeoutError(f"opening new tab timed out for {url}")
                                await set_and_verify_viewport(tab, int(domain_options.get("viewport_width") or 1920), int(domain_options.get("viewport_height") or 1080))
                                _timing_record(url, "navigate", time.perf_counter() - t_nav)
                            page_stem = _url_output_stem(url)
                            blocked_challenge = None
                            if domain_options.get("cloudflare_check", True) and existing_tab is None:
                                blocked_challenge = await _quick_cloudflare_challenge_check(
                                    tab,
                                    domain=domain_name,
                                    out_dir=out_dir,
                                    page_stem=page_stem,
                                    url=url,
                                    viewport_width=int(domain_options.get("viewport_width") or 1920),
                                    viewport_height=int(domain_options.get("viewport_height") or 1080),
                                )
                                if blocked_challenge.get("is_challenge"):
                                    _write_cloudflare_blocked_artifacts(out_dir, page_stem, url, blocked_challenge)
                                    ok = False
                            if blocked_challenge is None or not blocked_challenge.get("is_challenge"):
                                if domain_options.get("consent_check", True) and existing_tab is None:
                                    consent_ok = await _quick_consent_check(
                                        tab,
                                        domain=domain_name,
                                        out_dir=out_dir,
                                        page_stem=page_stem,
                                        viewport_width=int(domain_options.get("viewport_width") or 1920),
                                        viewport_height=int(domain_options.get("viewport_height") or 1080),
                                    )
                                    if not consent_ok:
                                        log.warning("Cookie consent detected on %s; crawling anyway", url)
                                worker_timeout = _page_worker_timeout(domain_options)
                                try:
                                    ok, _ = await asyncio.wait_for(_fetch_and_save(tab, url, out_dir, domain_options), timeout=worker_timeout)
                                except asyncio.TimeoutError:
                                    log.warning("Worker timed out for %s after %.1fs; marking as failed", url, worker_timeout)
                                    ok = False
                        except Exception as exc:
                            log.warning("Worker for %s failed: %s", url, exc)
                            ok = False
                        finally:
                            if close_when_done and tab is not None:
                                await close_tab_safely(tab, f"worker tab {url}")
                        async with lock:
                            results[url] = ok
                            if not ok:
                                failed.append(url)
                        _timing_record(url, "total_page", time.perf_counter() - t_page)

                tasks: list[asyncio.Task] = []
                for url in domain_urls:
                    tasks.append(asyncio.create_task(_fetch_url_worker(url)))
                await asyncio.gather(*tasks, return_exceptions=True)

                await save_browser_cookies(browser, domain_name)

                if options.get("markdown", False):
                    try:
                        dedupe_markdown_files(out_dir)
                    except Exception as exc:
                        log.warning("Markdown deduplication failed for %s: %s; continuing", domain_name, exc)
                    try:
                        _save_page_index(out_dir, results)
                        rewrite_markdown_links(out_dir)
                    except Exception as exc:
                        log.warning("Markdown link rewriting failed for %s: %s; continuing", domain_name, exc)
                    if domain_options.get("quality_check", False):
                        try:
                            check_markdown_quality(out_dir, domain_options.get("_page_state"))
                        except Exception as exc:
                            log.warning("Markdown quality check failed for %s: %s; continuing", domain_name, exc)
                    else:
                        log.debug("No Markdown quality check for %s", _safe_relative(out_dir, SITES_ROOT))
                _save_page_index(out_dir, results)
                if domain_options.get("_page_state"):
                    _save_page_state(out_dir, domain_options["_page_state"])

                log.info("Domain %s complete: %d/%d pages", domain_name, sum(1 for v in results.values() if v), len(domain_urls))
                return results
            except Exception as exc:
                log.warning("Domain group %s failed: %s", domain_name, exc)
                return {url: False for url in domain_urls}
            finally:
                if browser is not None:
                    try:
                        browser.stop()
                    except Exception as exc:
                        log.debug("browser.stop() failed for %s: %s", domain_name, exc)

    async def _process_and_merge(di: int, domain_name: str, domain_urls: list[str]) -> None:
        group_results = await _process_domain_group(di, domain_name, domain_urls)
        async with all_results_lock:
            all_results.update(group_results)

    tasks = [
        asyncio.create_task(_process_and_merge(di, domain, domain_groups[domain]))
        for di, domain in enumerate(domain_list)
    ]
    await asyncio.gather(*tasks, return_exceptions=True)

    log.info("All domains complete: %d/%d pages across %d domains", sum(1 for v in all_results.values() if v), len(all_results), len(domain_list))
    return all_results

async def _crawl_worker(
    browser,
    url: str,
    depth: int,
    out_dir: Path,
    domain: str,
    options: dict,
    results: dict[str, bool],
    failed: list[str],
    discovered: list[dict],
    queue: deque,
    seen: set[str],
    lock: asyncio.Lock,
    sem: asyncio.Semaphore,
    start_area_url: str,
    limit: int,
    recursive: bool,
    max_depth: int,
    delay: float,
):
    async with sem:
        async with lock:
            if limit and len(results) >= limit:
                return
        try:
            worker_timeout = _page_worker_timeout(options)
            ok, html = await asyncio.wait_for(
                _crawl_worker_single(
                    browser, url, depth, out_dir, domain, options,
                ),
                timeout=worker_timeout,
            )
        except asyncio.TimeoutError:
            log.warning("Worker timed out for %s; marking as failed", url)
            ok = False
            html = ""
        except Exception as exc:
            log.warning("Worker failed for %s: %s", url, exc)
            ok = False
            html = ""

        async with lock:
            results[url] = ok
            if not ok:
                failed.append(url)

        children: list[str] = []
        if ok and html and (recursive or depth == 0):
            if max_depth <= 0 or depth < max_depth:
                children = discover_children_from_html(html, url, start_area_url=start_area_url, include_subdomains=options.get("include_subdomains", False))

        async with lock:
            discovered.append({"url": url, "depth": depth, "children": children})
            if children:
                log.info("Link discovery for %s: %d crawlable children", url, len(children))
                for child in children:
                    key = child.rstrip("/")
                    if key in seen:
                        continue
                    if limit and len(results) + len(queue) >= limit:
                        break

                    # Product dedupe: skip known product detail pages
                    if product_exists_db is not None and ("/products/p/" in child or "/product-configurator/product-families/" in child):
                        try:
                            canonical = canonicalize_url(child, query_mode=options.get("query_mode", "semantic"))
                            if canonical and product_exists_db(canonical):
                                log.info("Product already known, skipping: %s", child[:120])
                                continue
                        except Exception:
                            pass

                    seen.add(key)
                    queue.append((child, depth + 1))


async def _crawl_worker_single(
    browser,
    url: str,
    depth: int,
    out_dir: Path,
    domain: str,
    options: dict,
) -> tuple[bool, str]:
    tab = None
    try:
        tab = await await_with_timeout(
            f"nodriver browser.get(new_tab=True) for {url}",
            browser.get(url, new_tab=True),
            timeout=float(options.get("navigation_timeout") or 30.0),
            default=None,
            warning=True,
        )
        if tab is None:
            log.warning("Failed to open tab for %s: browser.get timed out", url)
            return False, ""
        await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
    except Exception as exc:
        log.warning("Failed to open tab for %s: %s", url, exc)
        return False, ""

    try:
        page_stem = _url_output_stem(url)
        if options.get("cloudflare_check", True):
            challenge = await _quick_cloudflare_challenge_check(
                tab,
                domain=domain,
                out_dir=out_dir,
                page_stem=page_stem,
                url=url,
                viewport_width=int(options.get("viewport_width") or 1920),
                viewport_height=int(options.get("viewport_height") or 1080),
            )
            if challenge.get("is_challenge"):
                _write_cloudflare_blocked_artifacts(out_dir, page_stem, url, challenge)
                return False, ""

        if options.get("consent_check", True):
            consent_ok = await _quick_consent_check(
                tab,
                domain=domain,
                out_dir=out_dir,
                page_stem=page_stem,
                viewport_width=int(options.get("viewport_width") or 1920),
                viewport_height=int(options.get("viewport_height") or 1080),
            )
            if not consent_ok:
                log.warning("Cookie consent detected on %s; crawling anyway", url)

        ok, html = await _fetch_and_save(tab, url, out_dir, options)
        return ok, html or ""
    finally:
        if tab is not None:
            await close_tab_safely(tab, f"crawl worker tab {url}")
            log.debug("Closed tab for %s", url)


async def fetch_sitemap(
    start_url: str,
    *,
    limit: int = 0,
    recursive: bool = True,
    max_depth: int = 0,
    headless: bool = False,
    delay: float = 0.5,
    include_subdomains: bool = False,
    options: dict | None = None,
) -> dict[str, bool]:
    options = options or {}
    start_url = _strip_fragment_and_query(start_url)
    browser, consent_tab, domain, out_dir, consent_result = await _prepare_browser(start_url, headless=headless, options=options)
    options["_consent_result"] = consent_result
    if _is_cloudflare_blocked(consent_result):
        page_stem = _url_output_stem(start_url)
        marker_path = _write_cloudflare_blocked_artifacts(out_dir, page_stem, start_url, consent_result)
        write_json(out_dir / "_crawl_run.json", {
            "schema_version": "crawler-run/v1",
            "generated_at": utc_now_iso(),
            "start_url": start_url,
            "domain": domain,
            "status": "blocked",
            "blocked_reason": "cloudflare",
            "blocked_provider": _cloudflare_provider(consent_result),
            "blocked_confidence": _cloudflare_confidence(consent_result),
            "blocked_marker_path": str(marker_path.resolve()),
            "results": {start_url: False},
            "consent": consent_result,
        })
        await close_tab_safely(consent_tab, f"blocked sitemap tab {start_url}")
        browser.stop()
        return {start_url: False}
    if options.get("keep_crawl", False):
        options["_page_state"] = _load_page_state(out_dir)

    log.info("Opening fresh tab for crawl start page: %s", start_url)
    tab = await await_with_timeout(
        f"nodriver browser.get(new_tab=True) for crawl start {start_url}",
        browser.get(start_url, new_tab=True),
        timeout=float(options.get("navigation_timeout") or 30.0),
        default=None,
        warning=True,
    )
    if tab is None:
        log.warning("Could not open fresh crawl tab for %s; marking start page failed", start_url)
        browser.stop()
        write_json(out_dir / "_crawl_run.json", {
            "schema_version": "crawler-run/v1",
            "generated_at": utc_now_iso(),
            "start_url": start_url,
            "limit": limit,
            "threads": 0,
            "results": {start_url: False},
            "consent": consent_result,
        })
        return {start_url: False}
    await set_and_verify_viewport(tab, int(options.get("viewport_width") or 1920), int(options.get("viewport_height") or 1080))
    threads = max(1, int(options.get("threads") or 8))

    results: dict[str, bool] = {}
    discovered_log: list[dict] = []
    failed: list[str] = []
    queue: deque[tuple[str, int]] = deque()
    seen: set[str] = {start_url.rstrip("/")}
    lock = asyncio.Lock()
    sem = asyncio.Semaphore(threads)

    try:
        # Phase 1: crawl start page serially on a fresh tab after consent handling.
        log.info("Crawling start page: %s", start_url)
        try:
            ok, html = await asyncio.wait_for(_fetch_and_save(tab, start_url, out_dir, options), timeout=_page_worker_timeout(options))
        except asyncio.TimeoutError:
            log.warning("Start page crawl timed out for %s after %.1fs", start_url, _page_worker_timeout(options))
            ok = False
            html = ""
        results[start_url] = ok
        if not ok:
            failed.append(start_url)
        if limit and len(results) >= limit:
            log.info("Download limit reached after start page: %d", limit)

        # Discover links from start page
        if not limit or len(results) < limit:
            should_discover = bool(html) and (recursive or True)
            if should_discover and (max_depth <= 0 or 0 < max_depth):
                children = discover_children_from_html(html or "", start_url, start_area_url=start_url, include_subdomains=include_subdomains)
                discovered_log.append({"url": start_url, "depth": 0, "children": children})
                log.info("Link discovery for %s: %d crawlable children", start_url, len(children))
                for child in children:
                    key = child.rstrip("/")
                    if key in seen:
                        continue
                    if limit and len(results) + len(queue) >= limit:
                        log.info("Discovery stopped because --limit=%d is already fully queued", limit)
                        break
                    seen.add(key)
                    queue.append((child, 1))

        # Phase 2: concurrent crawling of discovered pages
        if queue and (not limit or len(results) < limit):
            log.info("Starting concurrent crawl with %d threads for up to %d queued pages", threads, len(queue))
            active_tasks: list[asyncio.Task] = []

            while queue and (not limit or len(results) < limit):
                # Launch workers for queued URLs (up to threads limit)
                while queue and len(active_tasks) < threads:
                    if limit and len(results) + len(active_tasks) >= limit:
                        break
                    if not queue:
                        break
                    url, depth = queue.popleft()
                    task = asyncio.create_task(
                        _crawl_worker(
                            browser, url, depth, out_dir, domain, options,
                            results, failed, discovered_log, queue, seen,
                            lock, sem, start_url, limit, recursive, max_depth, delay,
                        )
                    )
                    active_tasks.append(task)

                if not active_tasks:
                    break

                # Wait for at least one task to complete
                done, pending = await asyncio.wait(active_tasks, return_when=asyncio.FIRST_COMPLETED)
                active_tasks = list(pending)

                for task in done:
                    try:
                        await task
                    except Exception as exc:
                        log.warning("Worker task failed: %s", exc)

            # Drain remaining active tasks
            if active_tasks:
                await asyncio.gather(*active_tasks, return_exceptions=True)

            log.info("Concurrent crawl phase complete: %d results so far", len(results))

        if options.get("markdown", False):
            try:
                dedupe_markdown_files(out_dir)
            except Exception as exc:
                log.warning("Markdown deduplication failed: %s; continuing", exc)
            try:
                _save_page_index(out_dir, results)
                rewrite_markdown_links(out_dir)
            except Exception as exc:
                log.warning("Markdown link rewriting failed: %s; continuing", exc)
        if options.get("quality_check", False):
            try:
                check_markdown_quality(out_dir, options.get("_page_state"))
            except Exception as exc:
                log.warning("Markdown quality check failed: %s; continuing", exc)
        else:
            log.debug("No Markdown quality check for %s", _safe_relative(out_dir, SITES_ROOT))

        await save_browser_cookies(browser, domain)
    finally:
        await close_tab_safely(tab, f"sitemap start tab {start_url}")
        browser.stop()

    write_json(out_dir / "_crawl_run.json", {
        "schema_version": "crawler-run/v1",
        "generated_at": utc_now_iso(),
        "start_url": start_url,
        "limit": limit,
        "threads": threads,
        "results": results,
        "consent": consent_result,
    })
    write_json(out_dir / "_discovered_urls.json", {"schema_version": "crawler-discovered-urls/v1", "pages": discovered_log})
    write_json(out_dir / "_failed_urls.json", {"schema_version": "crawler-failed-urls/v1", "failed": failed})
    _save_page_index(out_dir, results)
    if options.get("_page_state"):
        _save_page_state(out_dir, options["_page_state"])
    log.info("Done: %d/%d pages saved.", sum(1 for v in results.values() if v), len(results))
    return results


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_bool(value: str | bool) -> bool:
    if isinstance(value, bool):
        return value
    normalized = str(value).strip().lower()
    if normalized in {"1", "true", "yes", "y", "on"}:
        return True
    if normalized in {"0", "false", "no", "n", "off"}:
        return False
    raise argparse.ArgumentTypeError("expected true or false")


def build_options(args: argparse.Namespace) -> dict:
    return {
        "markdown": bool(args.markdown),
        "vision": bool(args.vision),
        "screenshots": bool(args.screenshots),
        "images": bool(args.images),
        "quality_check": bool(args.quality_check),
        "quality_mode": str(args.quality_mode),
        "vision_on_bad_quality": bool(args.vision_on_bad_quality),
        "fullpage_background": bool(args.fullpage_background),
        "fullpage_output": str(args.fullpage_output),
        "max_images": args.max_images,
        "gemini_model": args.gemini_model,
        "max_attempts": args.max_attempts,
        "viewport_width": args.viewport_width,
        "viewport_height": args.viewport_height,
        "screenshot_max_tiles": args.screenshot_max_tiles,
        "screenshot_overlap": args.screenshot_overlap,
        "force_vision": args.force_vision,
        "consent_check": bool(args.consent_check),
        "cloudflare_check": bool(args.cloudflare_check),
        "respect_robots": bool(args.respect_robots),
        "consent_wait": args.consent_wait,
        "consent_attempts": args.consent_attempts,
        "content_timeout": args.content_timeout,
        "navigation_timeout": args.navigation_timeout,
        "html_timeout": args.html_timeout,
        "threads": args.threads,
        "keep_crawl": False if args.fresh else bool(args.keep_crawl),
        "download_assets": bool(args.download_assets),
        "max_css_per_page": args.max_css_per_page,
        "max_js_per_page": args.max_js_per_page,
    }


async def _log_pending_tasks_periodically(label: str, interval: float = 30.0) -> None:
    """Write task stacks to crawler.log while a crawl is still running."""
    while True:
        await asyncio.sleep(max(5.0, float(interval)))
        current = asyncio.current_task()
        tasks = [task for task in asyncio.all_tasks() if task is not current and not task.done()]
        log.debug("Watchdog %s: %d pending asyncio tasks", label, len(tasks))
        for idx, task in enumerate(tasks[:20], start=1):
            log.debug("Watchdog %s task %d: %r", label, idx, task)
            frames = task.get_stack(limit=4)
            if not frames:
                continue
            formatted = "".join(traceback.format_stack(frames[-1], limit=6)).rstrip()
            log.debug("Watchdog %s task %d stack:\n%s", label, idx, formatted)


async def _run_with_watchdog(coro, label: str):
    watcher = asyncio.create_task(_log_pending_tasks_periodically(label))
    try:
        return await coro
    finally:
        watcher.cancel()
        try:
            await watcher
        except asyncio.CancelledError:
            pass


def run_crawler_coroutine(coro, label: str):
    return uc.loop().run_until_complete(_run_with_watchdog(coro, label))


def main() -> None:
    # Force UTF-8 on Windows consoles to prevent UnicodeEncodeError from websockets binary ping logs
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

    p = argparse.ArgumentParser(description="Rendered webpage crawler with consent verification and Markdown output")
    g = p.add_mutually_exclusive_group(required=True)
    g.add_argument("--fetch-single", metavar="URL")
    g.add_argument("--fetch-page-urls", metavar="FILE")
    g.add_argument("--fetch-urls", nargs="+", metavar="URL", help="Crawl a list of URLs (can span multiple domains)")
    g.add_argument("--fetch-sitemap", metavar="URL")
    g.add_argument("--url", metavar="URL", help="Alias for --fetch-single")

    p.add_argument("--limit", type=int, default=0, help="Max pages to download; applies to --fetch-sitemap, --fetch-urls, and --fetch-page-urls; 0 = unlimited")
    p.add_argument("--max-depth", type=int, default=0, help="Max recursive sitemap depth; 0 = unlimited")
    p.add_argument("--no-recursive", type=parse_bool, default=False, metavar="true|false", help="Only fetch start page, don't follow links; default false")
    p.add_argument("--headless", type=parse_bool, default=True, metavar="true|false", help="Run Chrome headless; default true")
    p.add_argument("--delay", type=float, default=0.5, help="Delay between page fetches")
    p.add_argument("--out", default=None, help="Output root. Defaults to ../sites_crawled relative to crawler.py")

    p.add_argument("--screenshots", type=parse_bool, default=False, metavar="true|false", help="Capture post-consent screenshots; default false")
    p.add_argument("--markdown", type=parse_bool, default=False, metavar="true|false", help="Generate .md output")
    p.add_argument("--vision", type=parse_bool, default=False, metavar="true|false", help="Use screenshot vision pipeline for Markdown instead of rendered HTML")
    p.add_argument("--images", type=parse_bool, default=False, metavar="true|false", help="Download and rewrite images referenced by page HTML; default false")
    p.add_argument("--quality-check", type=parse_bool, default=False, metavar="true|false", help="Run DeepSeek semantic quality check on Markdown output; default false")
    p.add_argument("--max-images", type=int, default=250, help="Maximum images per page; 0 = unlimited; default 250")
    p.add_argument("--keep-crawl", type=parse_bool, default=False, metavar="true|false", help="Incremental crawl: skip pages whose cleaned HTML digest matches previous run; default false")
    p.add_argument("--fresh", type=parse_bool, default=False, metavar="true|false", help="Force full re-crawl: overrides --keep-crawl, reprocesses every page; default false")
    p.add_argument("--download-assets", type=parse_bool, default=False, metavar="true|false", help="Download CSS/JS/fonts from browser cache, rewrite HTML paths; default false")
    p.add_argument("--max-css-per-page", type=int, default=10, help="Max CSS files per page when --download-assets; default 10")
    p.add_argument("--max-js-per-page", type=int, default=5, help="Max JS files per page when --download-assets; default 5")
    p.add_argument("--gemini-model", default=DEFAULT_GEMINI_MODEL, help="Gemini model for consent detection and --vision true")
    p.add_argument("--max-attempts", type=int, default=DEFAULT_MAX_ATTEMPTS, help="Maximum Gemini retry attempts per screenshot")
    p.add_argument("--viewport-width", type=int, default=1920, help="Requested viewport width")
    p.add_argument("--viewport-height", type=int, default=1080, help="Requested viewport height")
    p.add_argument("--screenshot-max-tiles", type=int, default=0, help="Maximum screenshot tiles per page; 0 = all tiles")
    p.add_argument("--screenshot-overlap", type=int, default=120, help="Pixel overlap between vertical screenshot tiles")
    p.add_argument("--force-vision", type=parse_bool, default=False, metavar="true|false", help="Regenerate vision .md/.json when --vision true; default false")

    p.add_argument("--consent-check", type=parse_bool, default=True, metavar="true|false", help="Use Gemini screenshot verification for cookie consent; default true")
    p.add_argument("--cloudflare-check", type=parse_bool, default=True, metavar="true|false", help="Use Gemini screenshot verification for Cloudflare/Akamai bot protection; default true")
    p.add_argument("--no-cloudflare-check", dest="cloudflare_check", action="store_false", help="Disable Gemini bot-protection challenge detection")
    p.add_argument("--respect-robots", type=parse_bool, default=True, metavar="true|false", help="Respect robots.txt disallow rules; default true")
    p.add_argument("--no-respect-robots", dest="respect_robots", action="store_false", help="Ignore robots.txt and crawl anyway")
    p.add_argument("--consent-wait", type=float, default=3.0, help="Seconds to wait before first consent screenshot")
    p.add_argument("--consent-attempts", type=int, default=2, help="Maximum consent click/verify attempts")
    p.add_argument("--content-timeout", type=float, default=10.0, help="Seconds to wait for rendered content stability")
    p.add_argument("--navigation-timeout", type=float, default=30.0, help="Seconds to wait for nodriver page navigation")
    p.add_argument("--html-timeout", type=float, default=15.0, help="Seconds to wait for rendered HTML capture")
    p.add_argument("--verbose", "-v", type=parse_bool, default=False, metavar="true|false", help="Enable debug-level logging; default false")
    p.add_argument("--threads", type=int, default=4, help="Concurrent crawl threads (1 = sequential); default 4")

    p.add_argument("--fullpage-background", type=parse_bool, default=False, metavar="true|false", help="After fast .md is written, scroll full page to discover more images; default false")
    p.add_argument("--fullpage-output", default="second", choices=["second", "update"], help="second = write _fullpage.md; update = overwrite page.md; default second")
    p.add_argument("--quality-mode", default="fast", choices=["fast", "background", "blocking"], help="fast = skip quality; background = after fast .md; blocking = wait for result; default fast")
    p.add_argument("--vision-on-bad-quality", type=parse_bool, default=False, metavar="true|false", help="Run Gemini vision Markdown when quality is bad; default false")
    args = p.parse_args()

    configure_logging(verbose=bool(args.verbose))
    log.info("crawler.py started")
    log.info("Log file: %s", LOG_PATH)

    global SITES_ROOT
    if args.out:
        SITES_ROOT = Path(args.out).resolve()
        SITES_ROOT.mkdir(parents=True, exist_ok=True)
    log.info("Output root: %s", SITES_ROOT)

    single_url = args.fetch_single or args.url

    run_stem = "crawl"
    run_desc = "crawler run"
    url_list: list[str] = []
    if single_url:
        run_stem = _safe_token(single_url, "page")[:80]
        run_desc = single_url
        url_list = [single_url]
    elif args.fetch_urls:
        run_stem = "urls_" + _safe_token(args.fetch_urls[0], "page")[:60]
        run_desc = "; ".join(args.fetch_urls[:10])
        url_list = list(args.fetch_urls)
    elif args.fetch_page_urls:
        run_stem = "file_" + Path(args.fetch_page_urls).stem[:60]
        run_desc = f"url file: {args.fetch_page_urls}"
    elif args.fetch_sitemap:
        run_stem = "sitemap_" + _safe_token(args.fetch_sitemap, "site")[:60]
        run_desc = f"sitemap: {args.fetch_sitemap}"
        url_list = [args.fetch_sitemap]

    global _timing_log
    ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    stem = f"{ts}_{run_stem}"
    _timing_log = TimingLog(
        TimingLog.TIMING_DIR / f"{stem}.log",
        TimingLog.TIMING_DIR / f"{stem}.csv",
        url_list[0] if len(url_list) == 1 else (run_desc),
    )

    options = build_options(args)
    if options.get("vision", False) and not options.get("screenshots", False):
        log.info("--vision true requires screenshots; forcing --screenshots true")
        options["screenshots"] = True
    if single_url:
        ok = run_crawler_coroutine(fetch_single(single_url, headless=args.headless, options=options), "fetch-single")
        sys.exit(0 if ok else 1)

    if args.fetch_page_urls:
        res = run_crawler_coroutine(fetch_page_urls(args.fetch_page_urls, headless=args.headless, delay=args.delay, limit=args.limit, options=options), "fetch-page-urls")
        failed = [u for u, ok in res.items() if not ok]
        if failed:
            log.warning("%d failed:", len(failed))
            for u in failed:
                log.warning("  %s", u)
        sys.exit(1 if failed else 0)

    if args.fetch_urls:
        res = run_crawler_coroutine(fetch_urls(args.fetch_urls, headless=args.headless, delay=args.delay, limit=args.limit, options=options), "fetch-urls")
        failed = [u for u, ok in res.items() if not ok]
        if failed:
            log.warning("%d failed:", len(failed))
            for u in failed:
                log.warning("  %s", u)
        sys.exit(1 if failed else 0)

    if args.fetch_sitemap:
        res = run_crawler_coroutine(fetch_sitemap(
            args.fetch_sitemap,
            limit=args.limit,
            recursive=not args.no_recursive,
            max_depth=args.max_depth,
            headless=args.headless,
            delay=args.delay,
            options=options,
        ), "fetch-sitemap")
        failed = [u for u, ok in res.items() if not ok]
        if failed:
            log.warning("%d failed:", len(failed))
            for u in failed:
                log.warning("  %s", u)
        sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()
