import asyncio
import json
import os
import threading
import time
from datetime import datetime, UTC
from pathlib import Path
from typing import Any

import nodriver as uc
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

import crawler
import file_server
import browser_sessions
import crawl_db
import requests


# ---------------------------------------------------------------------------
# Request / response models
# ---------------------------------------------------------------------------
class CrawlSingleRequest(BaseModel):
    """Crawl exactly one URL. The page is opened in Chromium, consent is handled, HTML is captured, images are downloaded and rewritten, and Markdown is generated.

    Set include_page_data=true to get inline markdown, image URLs, and discoverable links in the response.
    Set cache_policy=cache_first to return cached results without opening the browser (sub-50ms for known pages).
    Set interactive_mode=discover_only to find JavaScript buttons, cards, and overlays on the page."""
    url: str = Field(
        description="Full URL of the page to crawl. Query parameters are preserved in 'semantic' mode (e.g. ?MountingType=Anbau). Tracking params (utm_*, fbclid, gclid) are dropped.",
        json_schema_extra={"example": "https://www.prolicht.at/en/product-configurator/product-filter/?MountingType=Anbau"})
    markdown: bool = Field(default=True, description="Generate .md output from rendered HTML. Uses markdownify on the post-consent DOM.")
    images: bool = Field(default=True, description="Download and rewrite images referenced by the page. Images are fetched from browser cache, saved under images/, and rewritten to local paths in HTML before Markdown conversion.")
    screenshots: bool | str = Field(default=False, description="Screenshot mode: false/'none' = no screenshots, true/'full' = viewport + full-page + tiles, 'viewport_initial' = viewport screenshot only (fast, sufficient for consent checks)")
    quality_check: bool = Field(default=False, description="Run DeepSeek semantic quality check on the Markdown output. Requires DEEPSEEK_API_KEY.")
    cloudflare_check: bool = Field(default=True, description="Use Gemini screenshot verification to detect Cloudflare/Akamai bot-protection challenge pages before crawling.")
    max_images: int = Field(default=5, ge=0, description="Maximum images to download per page. 0 = unlimited. Images beyond this cap are not downloaded or rewritten.")
    headless: bool | None = Field(default=None, description="Override CRAWLER_HEADLESS env var. null = use env default (true in production). true = headless Chromium. false = visible browser (requires Xvfb).")
    keep_crawl: bool = Field(default=False, description="Skip reprocessing (image download, Markdown generation) when the page content digest matches a cached version. The browser still navigates to the page to compute the digest.")
    respect_robots: bool = Field(default=True, description="Respect robots.txt disallow rules. Set false to crawl pages that robots.txt blocks.")
    persistent: bool = Field(default=False, description="Save crawl result to DB, run page classification, store link maps. Leave false for fast transient fetches.")
    cache_policy: str = Field(
        default="revalidate",
        description="Controls cache behavior: 'cache_first' = return cached artifacts without opening browser (0.02s). 'revalidate' = open browser but skip reprocessing if unchanged. 'refresh' = always fetch but skip expensive reprocessing if digest matches. 'force' = always fetch and rewrite everything.",
        json_schema_extra={"enum": ["cache_first", "revalidate", "refresh", "force"]})
    crawl_scope: str = Field(
        default="all",
        description="Which pages to crawl: 'all' = crawl everything discovered. 'new_only' = only crawl pages never seen before (checks DB). Use new_only for incremental updates.",
        json_schema_extra={"enum": ["all", "new_only"]})
    query_mode: str = Field(
        default="semantic",
        description="How to handle query parameters in URLs: 'semantic' = keep meaningful params (tab, filter, MountingType), drop tracking params. 'full' = keep all query params. 'path_only' = strip all query params (old behavior).",
        json_schema_extra={"enum": ["semantic", "full", "path_only"]})
    interactive_mode: str = Field(
        default="off",
        description="JavaScript interaction discovery: 'off' = normal link extraction only. 'discover_only' = find and click JS elements (buttons, cards, tabs) and detect navigation/modals. 'full' = also capture modal content as virtual pages.",
        json_schema_extra={"enum": ["off", "discover_only", "full"]})
    max_interactions_per_page: int = Field(default=80, ge=0, description="Hard cap on JS click candidates per page. Prevents runaway interaction discovery.")
    max_virtual_pages_per_page: int = Field(default=50, ge=0, description="Hard cap on modal/interaction virtual pages captured from one real page.")
    include_page_data: bool = Field(default=False, description="Return a rich 'page' object with inline markdown, image URLs, discoverable links, file paths, and errors. Set this when you need the page content immediately in the API response.")
    max_inline_chars: int = Field(default=0, ge=0, description="Maximum characters of inline markdown when include_page_data=true. 0 = no limit (full markdown returned). Use to limit response size for very large pages.")
    download_assets: bool = Field(default=False, description="Download CSS, JS, and font assets from browser cache for offline HTML viewing")
    max_css_per_page: int = Field(default=10, ge=0, description="Max CSS files per page; 0 = unlimited")
    max_js_per_page: int = Field(default=5, ge=0, description="Max JS files per page; 0 = unlimited")
    max_asset_bytes: int = Field(default=5242880, ge=1, description="Max bytes per individual asset download (default 5 MB)")


class CrawlResponse(BaseModel):
    """Response shape for a single-page crawl including artifact paths and the public markdown URL."""
    url: str = Field(description="Normalized URL that was crawled")
    domain: str = Field(description="Hostname extracted from the URL (e.g. www.prolicht.at)")
    status: str = Field(description='"ok" when the crawl succeeded and artifacts were written, "error" on failure')
    fast_markdown_path: str | None = Field(default=None, description="Absolute container path to the generated .md file")
    markdown_url: str | None = Field(default=None, description="Public short-link URL to download the markdown (e.g. https://host/l/abc123). Redirects to /crawler/markdown/{token} with images rewritten to signed URLs.")
    html_path: str | None = Field(default=None, description="Absolute container path to the saved .html file")
    txt_path: str | None = Field(default=None, description="Absolute container path to the extracted .txt file")
    images_manifest_path: str | None = Field(default=None, description="Absolute container path to the .images.json manifest")
    crawl_metadata_path: str | None = Field(default=None, description="Absolute container path to the .crawl.json metadata file")
    consent_status: str | None = Field(default=None, description="Cookie consent status: 'skipped_cookies_loaded', 'accepted', 'failed', 'cache_hit', or 'blocked_by_cloudflare'")
    blocked_reason: str | None = Field(default=None, description="Reason the crawl was blocked, e.g. 'cloudflare'")
    blocked_provider: str | None = Field(default=None, description="Detected bot-protection provider, e.g. 'cloudflare' or 'akamai'")
    blocked_confidence: float | None = Field(default=None, description="Gemini confidence for the bot-protection detection")
    elapsed_seconds: float = 0


class CrawlUrlsRequest(BaseModel):
    """Crawl a list of URLs in parallel. URLs are grouped by domain and processed concurrently with one browser session per domain."""
    urls: list[str] = Field(description="List of full URLs to crawl. Can span multiple domains. Each URL is processed independently.", min_length=1, json_schema_extra={"example": ["https://example.com/page1", "https://example.com/page2"]})
    markdown: bool = Field(default=True, description="Generate .md output from rendered HTML")
    images: bool = Field(default=True, description="Download and rewrite images referenced by each page")
    screenshots: bool | str = Field(default=False, description="Screenshot mode: false/'none' = no, true/'full' = all, 'viewport_initial' = viewport only")
    quality_check: bool = Field(default=False, description="Run DeepSeek semantic quality check on Markdown output")
    cloudflare_check: bool = Field(default=True, description="Use Gemini screenshot verification to detect Cloudflare/Akamai bot-protection challenge pages before crawling.")
    max_images: int = Field(default=5, ge=0, description="Maximum images per page; 0 = unlimited")
    limit: int = Field(default=0, ge=0, description="Hard cap on total pages across all URLs; 0 = unlimited")
    threads: int = Field(default=4, ge=1, le=16, description="Concurrent crawl threads across domains. Higher values increase parallelism but also resource usage.")
    headless: bool | None = Field(default=None, description="Override CRAWLER_HEADLESS. null = use env default.")
    include_page_data: bool = Field(default=False, description="Return rich page artifact payloads with inline markdown, image URLs, and links for each page.")
    max_inline_chars: int = Field(default=0, ge=0, description="Max inline markdown chars per page when include_page_data=true. 0 = unlimited.")
    respect_robots: bool = Field(default=True, description="Respect robots.txt disallow rules. Set false to crawl pages that robots.txt blocks.")


class CrawlSitemapRequest(BaseModel):
    """Discover and crawl internal pages from a starting URL. Same-host, same-folder HTML pages are discovered recursively. For Writerside/help sites, only .html pages in the same directory are followed. Product detail pages are skipped if the product already exists in the database."""
    url: str = Field(description="Starting page URL for sitemap-style crawl. The crawler opens this page, discovers same-host child links, and crawls them recursively.", json_schema_extra={"example": "https://www.jetbrains.com/help/youtrack/devportal/youtrack-rest-api.html"})
    markdown: bool = Field(default=True, description="Generate .md output from rendered HTML for each page")
    images: bool = Field(default=True, description="Download and rewrite images referenced by each page")
    screenshots: bool | str = Field(default=False, description="Screenshot mode: false/'none' = no, true/'full' = all, 'viewport_initial' = viewport only")
    quality_check: bool = Field(default=False, description="Run DeepSeek semantic quality check on Markdown output")
    cloudflare_check: bool = Field(default=True, description="Use Gemini screenshot verification to detect Cloudflare/Akamai bot-protection challenge pages before crawling.")
    max_images: int = Field(default=5, ge=0, description="Maximum images per page; 0 = unlimited")
    limit: int = Field(default=6, ge=0, description="Hard cap on total pages including the start page. 0 = unlimited. Use to avoid runaway crawls on large sites.")
    max_depth: int = Field(default=0, ge=0, description="Maximum recursion depth from the start page. 0 = unlimited depth.")
    recursive: bool = Field(default=True, description="Follow links discovered on each page to find more pages. Set false to only crawl the start page.")
    threads: int = Field(default=4, ge=1, le=16, description="Concurrent crawl threads. Pages are processed in parallel within the same domain.")
    headless: bool | None = Field(default=None, description="Override CRAWLER_HEADLESS. null = use env default.")
    include_page_data: bool = Field(default=False, description="Return rich page artifact payloads for each crawled page with inline markdown, image URLs, and links.")
    max_inline_chars: int = Field(default=0, ge=0, description="Max inline markdown chars per page when include_page_data=true. 0 = unlimited.")
    include_subdomains: bool = Field(default=False, description="When true, also follow links to subdomains of the start URL hostname (e.g. shop.prolicht.at from www.prolicht.at). Default false = exact hostname match only.")
    keep_crawl: bool = Field(default=False, description="Skip reprocessing (image download, Markdown generation) when the page content digest matches a cached version. Enables incremental updates across runs.")
    download_assets: bool = Field(default=False, description="Download CSS, JS, and font assets from browser cache for offline HTML viewing")
    max_css_per_page: int = Field(default=10, ge=0, description="Max CSS files per page; 0 = unlimited")
    max_js_per_page: int = Field(default=5, ge=0, description="Max JS files per page; 0 = unlimited")
    max_asset_bytes: int = Field(default=5242880, ge=1, description="Max bytes per individual asset download (default 5 MB)")
    respect_robots: bool = Field(default=True, description="Respect robots.txt disallow rules. Set false to crawl pages that robots.txt blocks.")


class CrawlMultiResponse(BaseModel):
    """Response for multi-page crawl operations (urls, sitemap)."""
    status: str = Field(description='"ok" if all pages succeeded, "partial" if some failed, "error" if all failed')
    total: int = 0
    succeeded: int = 0
    failed: int = 0
    failed_urls: list[str] = Field(default_factory=list, description="URLs that failed to crawl")
    results: dict[str, bool] = Field(default_factory=dict, description="Map of URL -> success boolean")
    markdown_urls: dict[str, str | None] = Field(default_factory=dict, description="Map of URL -> public markdown download URL (short link)")
    blocked_urls: dict[str, dict[str, Any]] = Field(default_factory=dict, description="Map of URL -> bot-protection block details when detected")
    pages: list[dict] = Field(default_factory=list, description="Rich page objects when include_page_data=true. Each contains url, domain, status, markdown, images, links, paths, errors.")
    elapsed_seconds: float = 0


class ArtifactPageRequest(BaseModel):
    """Read already-crawled page artifacts from disk without opening the browser. Use this to retrieve pages that were previously crawled."""
    url: str = Field(description="Full URL of the already-crawled page. The page must have been crawled previously (files must exist on disk).")
    include_markdown: bool = Field(default=True, description="Return inline markdown text from the saved .md file")
    include_links: bool = Field(default=True, description="Return discoverable child links extracted from the saved HTML")
    include_images: bool = Field(default=True, description="Return public image URLs (signed short links) from the images manifest")
    max_images: int = Field(default=5, ge=0, description="Maximum image URLs to return; 0 = unlimited")
    max_inline_chars: int = Field(default=0, ge=0, description="Maximum characters of inline markdown; 0 = unlimited (full file)")


class ArtifactPagesRequest(BaseModel):
    """Batch-read already-crawled page artifacts. No browser activity."""
    urls: list[str] = Field(description="Full URLs of already-crawled pages", min_length=1, json_schema_extra={"example": ["https://www.prolicht.at", "https://www.heise.de"]})
    include_markdown: bool = Field(default=True, description="Return inline markdown for each page")
    include_links: bool = Field(default=True, description="Return discoverable links for each page")
    include_images: bool = Field(default=True, description="Return image URLs for each page")
    max_images: int = Field(default=5, ge=0, description="Maximum image URLs per page; 0 = unlimited")
    max_inline_chars: int = Field(default=0, ge=0, description="Max markdown chars per page; 0 = unlimited")


# ---------------------------------------------------------------------------
# Option builders
# ---------------------------------------------------------------------------
def request_options(req: CrawlSingleRequest) -> dict[str, Any]:
    return {
        "markdown": req.markdown,
        "vision": False,
        "screenshots": req.screenshots,
        "images": req.images,
        "quality_check": req.quality_check,
        "quality_mode": "fast",
        "vision_on_bad_quality": False,
        "fullpage_background": False,
        "fullpage_output": "second",
        "max_images": req.max_images,
        "viewport_width": 1920,
        "viewport_height": 1080,
        "screenshot_max_tiles": 0,
        "screenshot_overlap": 120,
        "force_vision": False,
        "consent_check": True,
        "cloudflare_check": getattr(req, "cloudflare_check", True),
        "consent_wait": 3.0,
        "consent_attempts": 2,
        "content_timeout": 10.0,
        "navigation_timeout": 30.0,
        "html_timeout": 15.0,
        "threads": 1,
        "keep_crawl": req.keep_crawl,
        "persistent": req.persistent,
        "cache_policy": req.cache_policy,
        "crawl_scope": req.crawl_scope,
        "query_mode": req.query_mode,
        "interactive_mode": req.interactive_mode,
        "max_interactions_per_page": req.max_interactions_per_page,
        "max_virtual_pages_per_page": req.max_virtual_pages_per_page,
        "docker": True,
        "download_assets": getattr(req, "download_assets", False),
        "max_css_per_page": getattr(req, "max_css_per_page", 10),
        "max_js_per_page": getattr(req, "max_js_per_page", 5),
        "max_asset_bytes": getattr(req, "max_asset_bytes", 5242880),
        "respect_robots": getattr(req, "respect_robots", True),
    }


def multi_request_options(req: CrawlUrlsRequest | CrawlSitemapRequest) -> dict[str, Any]:
    threads = getattr(req, "threads", 4)
    return {
        "markdown": req.markdown,
        "vision": False,
        "screenshots": req.screenshots,
        "images": req.images,
        "quality_check": req.quality_check,
        "quality_mode": "fast",
        "vision_on_bad_quality": False,
        "fullpage_background": False,
        "fullpage_output": "second",
        "max_images": req.max_images,
        "viewport_width": 1920,
        "viewport_height": 1080,
        "screenshot_max_tiles": 0,
        "screenshot_overlap": 120,
        "force_vision": False,
        "consent_check": True,
        "cloudflare_check": getattr(req, "cloudflare_check", True),
        "consent_wait": 3.0,
        "consent_attempts": 2,
        "content_timeout": 10.0,
        "navigation_timeout": 30.0,
        "html_timeout": 15.0,
        "threads": threads,
        "keep_crawl": getattr(req, "keep_crawl", False),
        "docker": True,
        "include_subdomains": getattr(req, "include_subdomains", False),
        "download_assets": getattr(req, "download_assets", False),
        "max_css_per_page": getattr(req, "max_css_per_page", 10),
        "max_js_per_page": getattr(req, "max_js_per_page", 5),
        "max_asset_bytes": getattr(req, "max_asset_bytes", 5242880),
        "respect_robots": getattr(req, "respect_robots", True),
    }


def _api_timing_init() -> None:
    if crawler._timing_log is not None:
        return
    ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    stem = f"{ts}_api"
    crawler._timing_log = crawler.TimingLog(
        crawler.TimingLog.TIMING_DIR / f"{stem}.log",
        crawler.TimingLog.TIMING_DIR / f"{stem}.csv",
        "API request",
    )


# ---------------------------------------------------------------------------
# Worker + FastAPI app
# ---------------------------------------------------------------------------
class CrawlerWorker:
    def __init__(self) -> None:
        self.loop = None
        self.thread = None

    def start(self) -> None:
        if self.thread is not None:
            return
        self.thread = threading.Thread(target=self._run_loop, name="crawler-nodriver", daemon=True)
        self.thread.start()
        while self.loop is None:
            time.sleep(0.05)

    def _run_loop(self) -> None:
        self.loop = uc.loop()
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    async def submit(self, coro):
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        return await asyncio.wrap_future(future)

    def stop(self) -> None:
        if self.loop is not None:
            self.loop.call_soon_threadsafe(self.loop.stop)


app = FastAPI(title="crawler", version="1.0")
worker = CrawlerWorker()
file_server.register_routes(app)


@app.on_event("startup")
async def startup() -> None:
    crawler.SITES_ROOT = Path(os.getenv("CRAWLER_OUT", "/crawler/sites_crawled")).resolve()
    crawler.SITES_ROOT.mkdir(parents=True, exist_ok=True)
    file_server.init_config(crawler.SITES_ROOT)
    worker.start()


@app.on_event("shutdown")
async def shutdown() -> None:
    worker.stop()


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.get("/health")
async def health() -> dict[str, Any]:
    """Worker and browser readiness check.

    Returns the crawler worker status. The worker is a daemon thread that owns the nodriver event loop.
    When worker_started=true, the browser is ready to accept crawl requests."""
    return {"status": "ok", "worker_started": worker.loop is not None}


@app.post("/artifacts/page")
async def artifact_page(req: ArtifactPageRequest) -> dict[str, Any]:
    """Read an already-crawled page from disk without opening the browser.

    This endpoint reads saved .md, .html, and .images.json files for a previously crawled URL.
    No browser activity occurs. Returns inline markdown, image URLs, discoverable links,
    and file paths. Use this when you know the page was already crawled and want the content
    without the latency of re-crawling."""
    return _build_page_artifact(
        req.url,
        include_markdown=req.include_markdown,
        include_links=req.include_links,
        include_images=req.include_images,
        max_images=req.max_images,
        max_inline_chars=req.max_inline_chars,
    )


@app.post("/artifacts/pages")
async def artifact_pages(req: ArtifactPagesRequest) -> dict[str, Any]:
    """Batch-read already-crawled pages from disk.

    Same as /artifacts/page but accepts a list of URLs. Returns an array of page objects.
    No browser activity. Use this to retrieve multiple previously crawled pages in one call."""
    pages = [
        _build_page_artifact(
            url,
            include_markdown=req.include_markdown,
            include_links=req.include_links,
            include_images=req.include_images,
            max_images=req.max_images,
            max_inline_chars=req.max_inline_chars,
        )
        for url in req.urls
    ]
    return {"status": "ok", "total": len(pages), "pages": pages}


def _resolve_headless(req_headless: bool | None) -> bool:
    if req_headless is not None:
        return req_headless
    return os.getenv("CRAWLER_HEADLESS", "true").lower() in {"1", "true", "yes", "on"}


def _build_markdown_url(url: str) -> str | None:
    """Build a public markdown URL for a crawled page if its .md file exists."""
    domain = crawler._real_host(url)
    stem = crawler._url_output_stem(url)
    md_path = crawler.SITES_ROOT / domain / f"{stem}.md"
    if not md_path.exists():
        return None
    token, _ = file_server.build_markdown_token(domain, md_path.name)
    md_url = file_server.build_public_markdown_url(token)
    if not md_url:
        return None
    short, _, _ = file_server.shorten_public_url(md_url)
    return short or md_url




def _read_blocked_details(urls: list[str]) -> dict[str, dict[str, Any]]:
    blocked: dict[str, dict[str, Any]] = {}
    for url in urls:
        domain = crawler._real_host(url)
        stem = crawler._url_output_stem(url)
        metadata_path = crawler.SITES_ROOT / domain / f"{stem}.crawl.json"
        if not metadata_path.exists():
            continue
        try:
            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
        except Exception:
            continue
        if metadata.get("blocked_by_cloudflare"):
            blocked[url] = {
                "blocked_reason": "cloudflare",
                "blocked_provider": metadata.get("cloudflare_provider"),
                "blocked_confidence": metadata.get("cloudflare_confidence"),
                "blocked_marker_path": metadata.get("blocked_marker_path"),
            }
    return blocked

def _build_markdown_urls(results: dict[str, bool]) -> dict[str, str | None]:
    return {url: _build_markdown_url(url) for url in results}


# ---------------------------------------------------------------------------
# Artifact helpers for endpoint responses
# ---------------------------------------------------------------------------
def _page_output_stem(url: str) -> str:
    return crawler._url_output_stem(url)


def _safe_read_text(path: Path, max_chars: int) -> tuple[str, str | None]:
    if not path.exists() or not path.is_file():
        return "", f"missing file: {path}"
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
        if max_chars <= 0:
            return text, None
        return text[:max_chars], None
    except Exception as exc:
        return "", f"failed to read {path}: {exc}"


def _safe_read_json(path: Path) -> tuple[dict[str, Any], str | None]:
    if not path.exists() or not path.is_file():
        return {}, f"missing file: {path}"
    try:
        return json.loads(path.read_text(encoding="utf-8", errors="replace")), None
    except Exception as exc:
        return {}, f"failed to read json {path}: {exc}"


def _build_public_image_urls_for_manifest(domain: str, manifest_path: str | None, max_images: int) -> tuple[list[dict[str, str]], list[str]]:
    images = []
    errors = []
    if not manifest_path:
        return images, errors
    mp = Path(manifest_path)
    data, err = _safe_read_json(mp)
    if err:
        errors.append(err)
        return images, errors
    for img in data.get("images", []):
        if max_images > 0 and len(images) >= max_images:
            break
        rel = img.get("relative_path") or img.get("local_path") or ""
        if not rel:
            continue
        name = Path(rel).name
        token, _ = file_server.build_image_token(domain, rel)
        img_url = file_server.build_public_image_url(token)
        if img_url:
            short, _, _ = file_server.shorten_public_url(img_url)
            images.append({"name": name, "url": short or img_url, "relative_path": rel})
        else:
            images.append({"name": name, "url": "", "relative_path": rel})
    return images, errors


def _discover_links_for_html(html_path: str | None, url: str) -> tuple[list[dict[str, str]], list[str]]:
    errors = []
    if not html_path:
        return [], errors
    hp = Path(html_path)
    if not hp.exists():
        return [], [f"missing html: {html_path}"]
    try:
        html = hp.read_text(encoding="utf-8", errors="replace")
        links = crawler.discover_children_with_text(html, url)
        return links, errors
    except Exception as exc:
        return [], [f"failed to read links from {html_path}: {exc}"]


def _build_page_artifact(
    url: str,
    *,
    ok: bool | None = None,
    crawl_result: dict[str, Any] | None = None,
    include_markdown: bool = True,
    include_links: bool = True,
    include_images: bool = True,
    max_images: int = 5,
    max_inline_chars: int = 0,
) -> dict[str, Any]:
    domain = crawler._real_host(url)
    stem = _page_output_stem(url)
    out_dir = crawler.SITES_ROOT / domain

    errors: list[str] = []

    md_path = out_dir / f"{stem}.md"
    html_path = out_dir / f"{stem}.html"
    txt_path = out_dir / f"{stem}.txt"
    images_manifest_path = out_dir / f"{stem}.images.json"
    crawl_metadata_path = out_dir / f"{stem}.crawl.json"

    if crawl_result:
        if crawl_result.get("fast_markdown_path"):
            md_path = Path(crawl_result["fast_markdown_path"])
        if crawl_result.get("html_path"):
            html_path = Path(crawl_result["html_path"])
        if crawl_result.get("txt_path"):
            txt_path = Path(crawl_result["txt_path"])
        if crawl_result.get("images_manifest_path"):
            images_manifest_path = Path(crawl_result["images_manifest_path"])

    crawl_metadata: dict[str, Any] = {}
    if crawl_metadata_path.exists():
        try:
            crawl_metadata = json.loads(crawl_metadata_path.read_text(encoding="utf-8"))
        except Exception:
            crawl_metadata = {}

    # Determine status
    if crawl_metadata.get("blocked_by_cloudflare"):
        status = "blocked"
    else:
        if ok is None:
            ok = md_path.exists()
        status = "ok" if ok and md_path.exists() else "error"

    # Inline markdown
    markdown = ""
    if include_markdown:
        markdown, md_err = _safe_read_text(md_path, max_inline_chars)
        if md_err:
            errors.append(md_err)

    # Markdown URL
    markdown_url = _build_markdown_url(url) or ""

    # Images
    images: list[dict[str, str]] = []
    if include_images and max_images > 0:
        mp_str = str(images_manifest_path.resolve()) if images_manifest_path.exists() else None
        images, img_errs = _build_public_image_urls_for_manifest(domain, mp_str, max_images)
        errors.extend(img_errs)

    # Links
    links: list[dict[str, str]] = []
    if include_links:
        links, link_errs = _discover_links_for_html(str(html_path.resolve()) if html_path.exists() else None, url)
        errors.extend(link_errs)

    return {
        "url": url,
        "domain": domain,
        "status": status,
        "blocked_reason": "cloudflare" if crawl_metadata.get("blocked_by_cloudflare") else None,
        "blocked_provider": crawl_metadata.get("cloudflare_provider"),
        "blocked_confidence": crawl_metadata.get("cloudflare_confidence"),
        "title": "",
        "markdown": markdown,
        "markdown_url": markdown_url,
        "images": images,
        "links": links,
        "paths": {
            "markdown": str(md_path.resolve()),
            "html": str(html_path.resolve()),
            "txt": str(txt_path.resolve()),
            "images_manifest": str(images_manifest_path.resolve()),
            "crawl_metadata": str(crawl_metadata_path.resolve()),
        },
        "errors": errors[:20],
    }


@app.post("/crawl/single")
async def crawl_single(req: CrawlSingleRequest) -> dict[str, Any]:
    """Crawl exactly one URL and return paths to generated artifacts.

    The fast path produces rendered post-consent Markdown with browser-cached
    visible images. When saved cookies are valid and no consent overlay is
    visible, Gemini consent detection is skipped entirely.
    """
    started = time.perf_counter()
    _api_timing_init()
    headless = _resolve_headless(req.headless)
    options = request_options(req)
    try:
        result = await worker.submit(crawler.fetch_single_result(req.url, headless=headless, options=options))
        if isinstance(result, dict):
            result = {k: v for k, v in result.items()}
        else:
            result = {"url": req.url, "domain": crawler._real_host(req.url), "status": "error"}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    result["elapsed_seconds"] = round(time.perf_counter() - started, 3)

    domain = str(result.get("domain") or crawler._real_host(req.url))
    md_path = result.get("fast_markdown_path")
    if md_path and Path(md_path).exists():
        file_rel = Path(md_path).name
        token, _ = file_server.build_markdown_token(domain, file_rel)
        md_url = file_server.build_public_markdown_url(token)
        if md_url:
            short_url, _, _ = file_server.shorten_public_url(md_url)
            result["markdown_url"] = short_url or md_url

    if req.include_page_data:
        result["page"] = _build_page_artifact(
            req.url,
            ok=result.get("status") == "ok",
            crawl_result=result,
            include_markdown=True,
            include_links=True,
            include_images=req.images,
            max_images=req.max_images,
            max_inline_chars=req.max_inline_chars,
        )

    return result


@app.post("/crawl/urls")
async def crawl_urls(req: CrawlUrlsRequest) -> dict[str, Any]:
    """Crawl a list of URLs spanning one or more domains.

    URLs are grouped by domain internally and processed with one browser session
    per domain. This is equivalent to CLI ``--fetch-urls``.
    """
    started = time.perf_counter()
    _api_timing_init()
    headless = _resolve_headless(req.headless)
    options = multi_request_options(req)
    try:
        results: dict[str, bool] = await worker.submit(
            crawler.fetch_urls(req.urls, headless=headless, delay=0.5, limit=req.limit, options=options)
        )
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc

    succeeded = [u for u, ok in results.items() if ok]
    failed = [u for u, ok in results.items() if not ok]
    if not results:
        overall = "error"
    elif len(failed) == 0:
        overall = "ok"
    elif len(succeeded) > 0:
        overall = "partial"
    else:
        overall = "error"

    response: dict[str, Any] = {
        "status": overall,
        "total": len(results),
        "succeeded": len(succeeded),
        "failed": len(failed),
        "failed_urls": failed,
        "results": results,
        "markdown_urls": _build_markdown_urls(results),
        "blocked_urls": _read_blocked_details(list(results.keys())),
        "elapsed_seconds": round(time.perf_counter() - started, 3),
    }
    if req.include_page_data:
        response["pages"] = [
            _build_page_artifact(
                url,
                ok=ok,
                include_markdown=True,
                include_links=True,
                include_images=req.images,
                max_images=req.max_images,
                max_inline_chars=req.max_inline_chars,
            )
            for url, ok in results.items()
        ]
    return response


@app.post("/crawl/sitemap")
async def crawl_sitemap(req: CrawlSitemapRequest) -> dict[str, Any]:
    """Discover and crawl internal pages from a starting URL.

    Opens the given URL, discovers same-host links, and crawls pages up to
    ``limit``. Equivalent to CLI ``--fetch-sitemap``.
    """
    started = time.perf_counter()
    _api_timing_init()
    headless = _resolve_headless(req.headless)
    options = multi_request_options(req)
    try:
        results: dict[str, bool] = await worker.submit(
            crawler.fetch_sitemap(
                req.url,
                limit=req.limit,
                recursive=req.recursive,
                max_depth=req.max_depth,
                headless=headless,
                delay=0.5,
                options=options,
            )
        )
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc

    succeeded = [u for u, ok in results.items() if ok]
    failed = [u for u, ok in results.items() if not ok]
    if not results:
        overall = "error"
    elif len(failed) == 0:
        overall = "ok"
    elif len(succeeded) > 0:
        overall = "partial"
    else:
        overall = "error"

    response: dict[str, Any] = {
        "status": overall,
        "total": len(results),
        "succeeded": len(succeeded),
        "failed": len(failed),
        "failed_urls": failed,
        "results": results,
        "markdown_urls": _build_markdown_urls(results),
        "blocked_urls": _read_blocked_details(list(results.keys())),
        "elapsed_seconds": round(time.perf_counter() - started, 3),
    }
    if req.include_page_data:
        response["pages"] = [
            _build_page_artifact(
                url,
                ok=ok,
                include_markdown=True,
                include_links=True,
                include_images=req.images,
                max_images=req.max_images,
                max_inline_chars=req.max_inline_chars,
            )
            for url, ok in results.items()
        ]
    return response


@app.post("/browser/restart")
async def browser_restart() -> dict[str, Any]:
    return {"status": "not_implemented_yet"}


# ---------------------------------------------------------------------------
# Browser session endpoints (Phase 1)
# ---------------------------------------------------------------------------
class CreateSessionRequest(BaseModel):
    url: str = Field(description="URL to open in the browser session")
    viewport_width: int = Field(default=1920)
    viewport_height: int = Field(default=1080)


@app.post("/browser/sessions")
async def create_browser_session(req: CreateSessionRequest) -> dict[str, Any]:
    """Open a URL in a new browser tab and keep it alive for iterative inspection.

    Returns a session_id that can be used with /browser/sessions/{id}/actions and
    /browser/sessions/{id}/actions/{action_id}/click to explore the page in real time.
    Sessions auto-expire after 10 minutes of inactivity."""
    headless = _resolve_headless(None)

    async def _create():
        browser, consent_tab, domain, out_dir, consent = await crawler._prepare_browser(
            req.url, headless=headless, options={"docker": True},
        )
        tab = consent_tab
        current = await crawler._current_location(tab)
        if current and not crawler._urls_match(current, req.url):
            tab = await browser.get(req.url, new_tab=True)
            await crawler.set_and_verify_viewport(tab, req.viewport_width, req.viewport_height)
        sid = await browser_sessions.create_session(tab, req.url)
        return {"session_id": sid, "url": req.url, "status": "ok"}

    try:
        return await worker.submit(_create())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.get("/browser/sessions/{session_id}/actions")
async def get_session_actions(session_id: str) -> dict[str, Any]:
    """List all clickable elements in the current browser session.

    Returns action_id, text, tag, role, selector_candidates, and bounding box for each element.
    Safety filters skip cookie banners, login, cart, language, and social sharing buttons."""
    try:
        actions = await worker.submit(browser_sessions.get_actions(session_id))
        return {"session_id": session_id, "actions": actions}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.post("/browser/sessions/{session_id}/actions/{action_id}/click")
async def click_session_action(session_id: str, action_id: str) -> dict[str, Any]:
    """Click an element in the live browser session and detect the outcome.

    Returns the detected outcome: url_changed, modal_opened, content_changed, no_change, or failed.
    Includes before/after URL and content length for diff analysis."""
    try:
        result = await worker.submit(browser_sessions.click_action(session_id, action_id, timeout=5.0))
        if result is None:
            raise HTTPException(status_code=404, detail="Session not found or expired")
        return {"session_id": session_id, "status": "ok", **result}
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.delete("/browser/sessions/{session_id}")
async def close_browser_session(session_id: str) -> dict[str, Any]:
    """Close a browser session and release its tab resources."""
    try:
        closed = await worker.submit(browser_sessions.close_session(session_id))
        return {"session_id": session_id, "closed": closed}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.get("/browser/sessions/{session_id}/dom")
async def get_session_dom(session_id: str, max_nodes: int = 300) -> dict[str, Any]:
    """Return a simplified DOM tree optimized for LLM consumption.

    Only visible, meaningful elements are included (h1-h6, a, button, input, table, etc.).
    Section headings are tracked and attached to child nodes. CSS selectors included for
    each node. Max 500 nodes, truncated to max_nodes."""
    try:
        dom = await worker.submit(browser_sessions.get_dom(session_id, max_nodes=max_nodes))
        if dom is None:
            raise HTTPException(status_code=404, detail="Session not found or expired")
        return {"session_id": session_id, "status": "ok", **dom}
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


class GeminiAnalysisRequest(BaseModel):
    session_id: str = Field(description="Browser session ID from /browser/sessions")
    analysis_type: str = Field(default="page_parser_discovery", description="Type of analysis: page_parser_discovery, classify_page, extract_configurator")
    include_screenshot: bool = Field(default=True)
    include_dom: bool = Field(default=True)
    include_actions: bool = Field(default=True)


@app.post("/analysis/gemini/page")
async def gemini_page_analysis(req: GeminiAnalysisRequest) -> dict[str, Any]:
    """Send page screenshot + DOM + actions to Gemini Flash Lite for structured analysis.

    Returns page_class, recommended_actions, candidate_extractors, and page regions.
    Uses the gemini_page_analysis module (input-digest caching prevents duplicate API calls)."""
    try:
        async def _analyze():
            from gemini_page_analysis import analyze_page, _call_gemini
            sess = await browser_sessions.get_session(req.session_id)
            if not sess:
                return {"status": "error", "error": "Session not found"}

            # Gather visible text
            visible_text = ""
            try:
                vt = await sess.tab.evaluate(
                    "() => (document.body.textContent || '').trim().slice(0, 30000)",
                    return_by_value=True,
                )
                visible_text = str(vt) if vt else ""
            except Exception:
                pass

            # Gather DOM summary
            dom = await browser_sessions.get_dom(req.session_id, max_nodes=200)
            dom_summary = json.dumps({
                "title": dom.get("title", ""),
                "nodes": dom.get("nodes", [])[:100],
            }, ensure_ascii=False)[:8000] if dom else ""

            # Gather actions
            actions = await browser_sessions.get_actions(req.session_id)
            action_summary = json.dumps(actions[:50] if actions else [], ensure_ascii=False)[:4000]

            title = dom.get("title", "") if dom else ""

            # Build prompt
            prompt = (
                f"You analyze a webpage to suggest how to build a parser for it.\n"
                f"URL: {sess.url}\nTitle: {title}\n\n"
                f"Visible text:\n{visible_text[:20000]}\n\n"
                f"DOM summary:\n{dom_summary}\n\n"
                f"Clickable actions:\n{action_summary}\n\n"
                f"Return JSON with: page_class, confidence, site, main_regions (list of region_name + contains), "
                f"product_identity (product_name, product_family if any), "
                f"candidate_extractors (list of field + method + selector_hint), "
                f"recommended_actions (list of text + reason)"
            )

            result = _call_gemini(prompt)
            if result:
                return {"status": "ok", "session_id": req.session_id, "analysis": result}
            return {"status": "error", "error": "Gemini call returned no result"}

        return await worker.submit(_analyze())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    return JSONResponse(status_code=500, content={"detail": str(exc)})


# ---------------------------------------------------------------------------
# Parser testing & project endpoints (Phase 3)
# ---------------------------------------------------------------------------
class TestSelectorRequest(BaseModel):
    session_id: str = Field(description="Browser session ID")
    selector: str = Field(description="CSS selector to test against the live DOM")
    return_text: bool = Field(default=True, description="Return matched element text")
    limit: int = Field(default=20, ge=1, le=100)


class TestExtractorRequest(BaseModel):
    session_id: str = Field(description="Browser session ID")
    extractor: dict = Field(description="Extractor definition: name, type, section_heading, row_selector, columns")
    limit: int = Field(default=50, ge=1, le=500)


class CreateParserProjectRequest(BaseModel):
    domain: str = Field(description="Target domain, e.g. xal.com")
    parser_name: str = Field(description="Unique parser name")
    parser_type: str = Field(default="product_discovery", description="product_discovery|catalogue_discovery")


class SaveManifestRequest(BaseModel):
    manifest: dict = Field(description="Parser manifest: page_classifiers, actions, extractors, validations, test_urls")


class ValidateParserRequest(BaseModel):
    test_urls: list[str] = Field(description="URLs to validate the parser against")
    checks: dict = Field(default_factory=dict, description="Validation checks: min_products, min_variants, required_fields")


class RunParserRequest(BaseModel):
    urls: list[str] = Field(description="URLs to run the parser against")
    persist: bool = Field(default=True, description="Save extracted data to DB")
    cache_policy: str = Field(default="cache_first")


@app.post("/parser/test-selector")
async def test_selector(req: TestSelectorRequest) -> dict[str, Any]:
    """Test a CSS selector against a live browser session. Returns matched elements with text."""
    try:
        async def _test():
            sess = await browser_sessions.get_session(req.session_id)
            if not sess:
                return {"status": "error", "error": "Session not found"}
            js = f"""
            (function() {{
                var els = document.querySelectorAll('{req.selector.replace("'", "\\'")}');
                var results = [];
                var limit = Math.min(els.length, {req.limit});
                for (var i = 0; i < limit; i++) {{
                    results.push({{
                        index: i,
                        text: '{req.return_text}' === 'True' ? (els[i].textContent || '').trim().slice(0, 500) : '',
                        tag: els[i].tagName.toLowerCase(),
                        id: els[i].id || '',
                        classes: (els[i].className || '').toString().slice(0, 150),
                        href: els[i].getAttribute('href') || '',
                        visible: els[i].getBoundingClientRect().width > 0
                    }});
                }}
                return {{match_count: els.length, matches: results}};
            }})()
            """
            from nodriver.cdp.runtime import evaluate as cdp_evaluate
            raw = await sess.tab.send(cdp_evaluate(js, await_promise=True, return_by_value=True))
            obj = raw[0] if isinstance(raw, tuple) else raw
            return {"status": "ok", "selector": req.selector, "session_id": req.session_id, **obj.value}
        return await worker.submit(_test())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.post("/parser/test-extractor")
async def test_extractor(req: TestExtractorRequest) -> dict[str, Any]:
    """Test an extractor definition against a live browser session."""
    try:
        async def _test():
            sess = await browser_sessions.get_session(req.session_id)
            if not sess:
                return {"status": "error", "error": "Session not found"}
            ext = req.extractor
            rows = ext.get("row_selector", "tr")
            cols = ext.get("columns", {})
            section = ext.get("section_heading", "")
            sec_sel = f"section:has-text('{section}'), div:has-text('{section}'), table:has-text('{section}')" if section else "document"

            js = f"""
            (function() {{
                var container = {sec_sel};
                if (typeof container === 'string') container = document.querySelector(container);
                if (!container) return {{status:'error', error:'Container not found: {section}'}};
                var rows = container.querySelectorAll('{rows.replace("'", "\\'")}');
                var results = [];
                var limit = Math.min(rows.length, {req.limit});
                for (var i = 0; i < limit; i++) {{
                    var row = {{}};
                    var colsJson = {json.dumps(cols)};
                    for (var key in colsJson) {{
                        var colName = colsJson[key];
                        var cell = rows[i].querySelector('td, th');
                        if (cell) row[key] = (cell.textContent || '').trim().slice(0, 200);
                    }}
                    row['_row_text'] = (rows[i].textContent || '').trim().slice(0, 500);
                    results.push(row);
                }}
                return {{status:'ok', row_count: rows.length, extracted: results.length, records: results}};
            }})()
            """
            from nodriver.cdp.runtime import evaluate as cdp_evaluate
            raw = await sess.tab.send(cdp_evaluate(js, await_promise=True, return_by_value=True))
            obj = raw[0] if isinstance(raw, tuple) else raw
            return obj.value
        return await worker.submit(_test())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@app.post("/parser/projects")
async def create_project(req: CreateParserProjectRequest) -> dict[str, Any]:
    project = crawl_db.create_parser_project(req.domain, req.parser_name, req.parser_type)
    return {"status": "ok", "project": project}


@app.get("/parser/projects/{project_id}")
async def get_project(project_id: str) -> dict[str, Any]:
    project = crawl_db.get_parser_project(project_id)
    if not project:
        raise HTTPException(status_code=404, detail="Project not found")
    return {"status": "ok", "project": project}


@app.put("/parser/projects/{project_id}/manifest")
async def save_manifest(project_id: str, req: SaveManifestRequest) -> dict[str, Any]:
    ok = crawl_db.save_parser_manifest(project_id, req.manifest)
    return {"status": "ok" if ok else "error", "project_id": project_id}


@app.post("/parser/projects/{project_id}/validate")
async def validate_parser(project_id: str, req: ValidateParserRequest) -> dict[str, Any]:
    project = crawl_db.get_parser_project(project_id)
    if not project:
        raise HTTPException(status_code=404, detail="Project not found")
    errors = []
    for url in req.test_urls:
        try:
            r = requests.post(f"http://localhost:8114/crawl/single", json={
                "url": url, "markdown": True, "images": False, "include_page_data": True, "max_inline_chars": 5000,
            }, timeout=120)
            page = r.json().get("page", {})
            checks = req.checks
            if checks.get("min_products", 0) > 0:
                if not page.get("links"):
                    errors.append({"check": "min_products", "expected": checks["min_products"], "actual": 0, "url": url})
            for field in checks.get("required_fields", []):
                if field not in str(page.get("markdown", "")):
                    errors.append({"check": f"required_field:{field}", "url": url, "hint": f"Field '{field}' not found in markdown"})
        except Exception as exc:
            errors.append({"url": url, "error": str(exc)})
    status = "passed" if not errors else "failed"
    crawl_db.update_parser_validation(project_id, status, errors)
    return {"status": "ok", "validation_status": status, "errors": errors}


@app.post("/parser/projects/{project_id}/run")
async def run_parser(project_id: str, req: RunParserRequest) -> dict[str, Any]:
    results = []
    for url in req.urls:
        try:
            r = requests.post(f"http://localhost:8114/crawl/single", json={
                "url": url, "markdown": True, "images": False, "include_page_data": True, "cache_policy": req.cache_policy,
            }, timeout=120)
            results.append({"url": url, "status": r.json().get("status", "error")})
        except Exception as exc:
            results.append({"url": url, "error": str(exc)})
    return {"status": "ok", "project_id": project_id, "results": results}
