"""The rendered-fetch sidecar: one sandboxed Chromium behind a tiny HTTP server.

Two of the promised airports (Singapore, Seoul) publish nothing readable until
their own JavaScript has run, so a text fetch sees a 62 KB shell and no price.
The honest path is a real browser that loads a permitted page and reads what
the page itself drew; this process is that browser. It runs in its own
container (`dfp-browser`) on a network that holds only the app and itself, so
nothing here can reach the database, and the container carries no secrets.

What it never does, by construction rather than by policy text:

* It never clicks, types, scrolls or dismisses anything. A cookie, age or
  shopper-mode dialog that hides the price is a STOP for the collector, never
  a workaround here.
* It never keeps state between renders: every request gets a fresh context
  (no cookies, no storage), and the context is closed before the reply.
* It never talks to anything the caller did not name. `route('**/*')` aborts
  every request to an IP literal, a loopback/private/link-local address, or a
  host outside the source's declared allowlist, which also drops trackers.
* It never fetches a path the host's robots.txt disallows, even when the page
  itself asks for it (scripts, styles, images). The app sends the Disallow
  rules as regexes (`deny_regex`); a same-origin request matching one is
  aborted and counted. A retailer that disallows its own UI assets has said
  no to bots rendering its pages, and the render will show that honestly.
* It never validates by status code. It reports the main response's status,
  a few headers, the rendered HTML and the body text length, and the app
  decides (`fetch.render()`), because that decision is tested logic there.

The policy functions (`egress_allowed`, `host_permitted`) import nothing from
Playwright so the tests in `main/tests/` can exercise them without a browser.
The Playwright import happens in `main()`.

Contract, kept deliberately small (JSON in, JSON out):

    GET  /health            -> {"ok": true, "browser": "<version>"}
    POST /render            {"url", "user_agent", "allow_hosts": [...],
                             "deny_regex": [...], "wait_until", "wait_for",
                             "timeout_ms", "settle_ms", "capture_json",
                             "max_body_bytes"}
                            -> {"url", "final_url", "status", "headers",
                                "html", "text_length", "title",
                                "api_responses": [{"url", "status",
                                                    "content_type", "body"}],
                                "aborted": {"<host>": n},
                                "aborted_paths": {"<path>": n},
                                "console": ["<error lines, capped>"], "elapsed_ms"}
"""

import ipaddress
import json
import logging
import re
import sys
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlsplit

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("dfp-browser")

LISTEN = ("0.0.0.0", 8080)  # the container's own interface; compose publishes no port
DEFAULT_TIMEOUT_MS = 45_000
DEFAULT_SETTLE_MS = 2_000
MAX_SETTLE_MS = 15_000
DEFAULT_MAX_BODY_BYTES = 2_000_000   # per captured JSON body
MAX_TOTAL_CAPTURE_BYTES = 8_000_000  # per render, all captured bodies together
JSON_TYPES = ("application/json", "text/json", "+json")
# How long goto() waits: a server-rendered page is complete at domcontentloaded,
# a client-rendered one needs load (then the settle). Measured 2026-09-05: a
# server-rendered page whose scripts robots.txt disallows never fires load.
WAIT_UNTIL = ("commit", "domcontentloaded", "load", "networkidle")
PlaywrightTimeoutError: type[Exception] = TimeoutError  # rebound to Playwright's in Renderer


def host_permitted(host: str, allow_hosts: list[str]) -> bool:
    """Whether a hostname is inside the source's declared allowlist.

    Exact match or a subdomain of an allowed host. An IP literal of any kind,
    localhost, and the .local/.internal suffixes are never permitted, whatever
    the list says: the allowlist names retailers, and a retailer is a name.
    """
    host = (host or "").strip().lower().rstrip(".")
    if not host:
        return False
    try:
        ipaddress.ip_address(host.strip("[]"))
        return False
    except ValueError:
        pass
    if host == "localhost" or host.endswith((".localhost", ".local", ".internal")):
        return False
    for allowed in allow_hosts:
        allowed = allowed.strip().lower().rstrip(".")
        if allowed and (host == allowed or host.endswith("." + allowed)):
            return True
    return False


def egress_allowed(url: str, allow_hosts: list[str]) -> bool:
    """The route decision for one request the page wants to make."""
    parts = urlsplit(url)
    if parts.scheme in ("data", "blob", "about"):
        return True  # no network; inline resources the page carries itself
    if parts.scheme not in ("http", "https"):
        return False
    return host_permitted(parts.hostname or "", allow_hosts)


def path_denied(url: str, origin_host: str, deny: list[re.Pattern]) -> bool:
    """Whether a same-origin request hits one of the host's Disallow rules."""
    parts = urlsplit(url)
    if (parts.hostname or "").lower() != origin_host.lower():
        return False
    path = parts.path or "/"
    if parts.query:
        path = f"{path}?{parts.query}"
    return any(rx.match(path) for rx in deny)


def is_json(content_type: str) -> bool:
    lowered = (content_type or "").lower()
    return any(marker in lowered for marker in JSON_TYPES)


class Renderer:
    """One browser for the life of the process; one fresh context per render."""

    def __init__(self) -> None:
        from playwright.sync_api import TimeoutError as _Timeout  # lazy: see module docstring
        from playwright.sync_api import sync_playwright

        global PlaywrightTimeoutError
        PlaywrightTimeoutError = _Timeout
        self._pw = sync_playwright().start()
        # chromium_sandbox=True keeps Chromium's own renderer sandbox on. The
        # container makes that possible with a seccomp profile that allows
        # user namespaces and chroot without CAP_SYS_ADMIN (browser/seccomp.json).
        self.browser = self._pw.chromium.launch(headless=True, chromium_sandbox=True)
        logger.info("browser_ready version=%s", self.browser.version)

    def render(self, req: dict) -> dict:
        url = str(req.get("url") or "")
        parts = urlsplit(url)
        if parts.scheme not in ("http", "https") or not parts.hostname:
            raise ValueError("url must be absolute http(s)")
        allow_hosts = [parts.hostname, *[str(h) for h in req.get("allow_hosts") or []]]
        deny = [re.compile(str(rx)) for rx in req.get("deny_regex") or []]
        origin_host = parts.hostname
        user_agent = str(req.get("user_agent") or "")
        if not user_agent:
            raise ValueError("user_agent is required: the app decides the identity we present")
        wait_until = str(req.get("wait_until") or "load")
        if wait_until not in WAIT_UNTIL:
            raise ValueError(f"wait_until must be one of {WAIT_UNTIL}")
        timeout_ms = int(req.get("timeout_ms") or DEFAULT_TIMEOUT_MS)
        settle_ms = min(int(req.get("settle_ms") or DEFAULT_SETTLE_MS), MAX_SETTLE_MS)
        # A CSS selector the caller expects once the page has drawn (a price
        # element, say). Waited for after load, up to the page timeout; its
        # absence is reported, never worked around.
        wait_for = str(req.get("wait_for") or "")
        capture_json = bool(req.get("capture_json", True))
        max_body = int(req.get("max_body_bytes") or DEFAULT_MAX_BODY_BYTES)

        aborted: dict[str, int] = {}
        aborted_paths: dict[str, int] = {}
        console: list[str] = []  # errors only; why a page did not draw is diagnostic gold
        captured: list[dict] = []
        captured_bytes = 0
        started = time.monotonic()

        context = self.browser.new_context(
            user_agent=user_agent,
            locale="en-US",
            viewport={"width": 1280, "height": 900},
            java_script_enabled=True,
            accept_downloads=False,
        )
        try:
            def route_handler(route, request):
                if not egress_allowed(request.url, allow_hosts):
                    host = urlsplit(request.url).hostname or request.url[:40]
                    aborted[host] = aborted.get(host, 0) + 1
                    route.abort("blockedbyclient")
                elif deny and path_denied(request.url, origin_host, deny):
                    path = urlsplit(request.url).path[:80]
                    aborted_paths[path] = aborted_paths.get(path, 0) + 1
                    route.abort("blockedbyclient")
                else:
                    route.continue_()

            context.route("**/*", route_handler)

            def on_response(response):
                nonlocal captured_bytes
                if not capture_json:
                    return
                ctype = response.headers.get("content-type", "")
                if not is_json(ctype):
                    return
                try:
                    body = response.body()
                except Exception:  # a body the browser no longer holds; not worth failing for
                    return
                if len(body) > max_body or captured_bytes + len(body) > MAX_TOTAL_CAPTURE_BYTES:
                    return
                captured_bytes += len(body)
                captured.append(
                    {
                        "url": response.url,
                        "status": response.status,
                        "content_type": ctype,
                        "body": body.decode("utf-8", "replace"),
                    }
                )

            page = context.new_page()
            page.on("response", on_response)
            page.on("pageerror", lambda err: len(console) < 40 and console.append(f"pageerror: {str(err)[:300]}"))
            page.on("console", lambda msg: msg.type in ("error", "warning") and len(console) < 40 and console.append(f"{msg.type}: {msg.text[:300]}"))
            main = None
            timed_out = False

            def on_document(response):
                # The main frame's document response, kept even when goto()
                # later times out: a challenge page that loops never fires
                # "load", and its status and headers are the whole answer.
                nonlocal main
                if main is None and response.request.is_navigation_request() and response.frame == page.main_frame:
                    main = response

            page.on("response", on_document)
            try:
                page.goto(url, wait_until=wait_until, timeout=timeout_ms)
            except PlaywrightTimeoutError:
                timed_out = True
            waited_for = None
            if wait_for and not timed_out:
                try:
                    page.wait_for_selector(wait_for, state="attached", timeout=timeout_ms)
                    waited_for = True
                except PlaywrightTimeoutError:
                    waited_for = False
            try:
                page.wait_for_load_state("networkidle", timeout=settle_ms)
            except Exception:
                pass  # a page that never goes idle still has a DOM worth reading
            if settle_ms:
                page.wait_for_timeout(min(settle_ms, MAX_SETTLE_MS))
            html = page.content()
            try:
                text = page.inner_text("body")
            except Exception:
                text = ""
            headers = {}
            status = None
            final_url = page.url
            if main is not None:
                status = main.status
                final_url = main.url
                wanted = ("cf-mitigated", "server", "content-type", "x-robots-tag", "retry-after")
                headers = {k: v for k, v in main.headers.items() if k.lower() in wanted}
            return {
                "url": url,
                "final_url": final_url,
                "status": status,
                "headers": headers,
                "html": html,
                "text_length": len(text.strip()),
                "title": page.title(),
                "api_responses": captured,
                "aborted": aborted,
                "aborted_paths": aborted_paths,
                "timed_out": timed_out,
                "waited_for": waited_for,
                "console": console,
                "elapsed_ms": int((time.monotonic() - started) * 1000),
            }
        finally:
            context.close()


def make_handler(renderer: Renderer):
    class Handler(BaseHTTPRequestHandler):
        def _send(self, code: int, payload: dict) -> None:
            body = json.dumps(payload).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):  # noqa: N802 (http.server's contract)
            if self.path == "/health":
                self._send(200, {"ok": True, "browser": renderer.browser.version})
            else:
                self._send(404, {"error": "not found"})

        def do_POST(self):  # noqa: N802
            if self.path != "/render":
                self._send(404, {"error": "not found"})
                return
            length = int(self.headers.get("Content-Length") or 0)
            try:
                req = json.loads(self.rfile.read(length) or b"{}")
                result = renderer.render(req)
            except ValueError as exc:
                self._send(400, {"error": str(exc)})
                return
            except Exception as exc:  # the browser failed; the app treats this as retryable
                logger.warning("render_failed url=%s error=%s", req.get("url") if isinstance(req, dict) else "?", exc)
                self._send(502, {"error": f"{type(exc).__name__}: {exc}"[:2000]})
                return
            logger.info(
                "rendered url=%s status=%s text=%d api=%d aborted=%s denied_paths=%d timed_out=%s ms=%d",
                result["url"], result["status"], result["text_length"],
                len(result["api_responses"]), sorted(result["aborted"]),
                sum(result["aborted_paths"].values()), result["timed_out"], result["elapsed_ms"],
            )
            self._send(200, result)

        def log_message(self, fmt, *args):  # quiet the per-request access line
            return

    return Handler


def main() -> int:
    renderer = Renderer()
    server = HTTPServer(LISTEN, make_handler(renderer))
    logger.info("listening on %s:%d", *LISTEN)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    sys.exit(main())