"""The fetch port.

One deliberately dumb function: give it a URL, get back a response. Every
collector goes through this, and no collector ever imports a vendor SDK. That
is what makes the supplier underneath a config change rather than a rewrite.

Two rules are enforced here rather than left to each collector:

* Validate on CONTENT, never on status code. Some retail edges answer a blocked
  request with HTTP 202 and an empty body, which a naive client records as a
  success while ingesting nothing.
* A block is a refusal, not an obstacle. When a source turns us away we raise
  SourceBlocked and stop; we do not escalate to heavier tooling.
* A page that is NOT THERE is not a refusal. HTTP 404 and 410 raise PageGone,
  which is a FetchError and never a SourceBlocked, because a product the shop
  no longer sells is its catalogue moving on, not the host turning us away. The
  distinction is load-bearing: a refusal is final for the whole source, so one
  delisted bottle read back with an empty-bodied 404 ended a recheck `blocked`
  and left a public airport's source reading "refused" with Start disabled.

`render()` is the second door, same rules. Some sources draw their prices with
JavaScript, so a text fetch sees a shell; render() asks the browser sidecar
(`main/browser/`, its own container) to load the page as a browser would and
returns what the page drew, plus the JSON bodies of the same-origin API calls
the page itself made. It is roughly fifty times the cost of a text fetch, so it
is a per-URL tool with a per-run budget, never a mode for a whole source.
"""

import gzip
import json
import logging
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from urllib.parse import urlsplit

from app.services.collectors import control
from app.services.collectors.robots import BOT_NAME, Robots, pattern_regex

logger = logging.getLogger(__name__)

# Built from BOT_NAME so the name we honour in robots.txt and the name we send
# cannot drift apart (tests/test_robots.py::TestIdentity pins them equal).
USER_AGENT = f"Mozilla/5.0 (compatible; {BOT_NAME}/0.1; +https://bot.dutyfreeprofessor.com)"
DEFAULT_TIMEOUT = 40

# Renders. A host that publishes no Crawl-delay still gets at least this much
# between renders: a render is a full page load with every script, not one
# request, and ten seconds is what a slow human browsing costs them.
RENDER_FLOOR_DELAY = 10.0
RENDER_TIMEOUT = 90  # the sidecar's own page timeout is shorter; this bounds the HTTP call
# Outside an explicit render_budget() a process may render this many pages.
# A shell session raises it deliberately; a collector declares its own.
RENDER_DEFAULT_CAP = 25


class FetchError(RuntimeError):
    """The request failed for a reason worth retrying."""


class SourceBlocked(RuntimeError):
    """The source declined to serve us. Stop; do not work around it."""


# The two statuses that mean "there is nothing here", as against "not for you".
GONE_STATUSES = (404, 410)


class PageGone(FetchError):
    """The page is not there (HTTP 404 or 410); the host said nothing about us.

    A FetchError on purpose, so every walk that already skips a page it could
    not read skips this one too, and `base.gone()` reads it as the listing
    being gone. A retry will find the same absence, so a caller counts it
    (a recheck: one `existing_missing`) or logs and moves on; it never ends
    the source the way SourceBlocked does.
    """


def is_gone_status(status: int | None) -> bool:
    return status in GONE_STATUSES


@dataclass(slots=True)
class Page:
    url: str
    status: int
    body: bytes

    @property
    def text(self) -> str:
        return self.body.decode("utf-8", errors="replace")

    def json(self) -> object:
        return json.loads(self.body)


def _looks_blocked(status: int, body: bytes) -> bool:
    if status in (401, 403, 406, 429) or status == 202 and len(body) < 512:
        return True
    if not body.strip():
        return True
    lowered = body[:2048].lower()
    return any(
        marker in lowered
        for marker in (b"attention required", b"access denied", b"are you a robot")
    )


def fetch(url: str, *, accept: str = "application/json", delay: float = 0.0) -> Page:
    """Fetch one URL politely.

    Raises SourceBlocked if the source refuses, PageGone if the URL is not there.
    """
    # The control hook (Stream AW4): unbound it sleeps `delay` as before; bound by a run it
    # commits the batch, heartbeats, reads the page's control and sleeps the live pace. Called
    # at 0 too, so the deliberate first fetch still reads the control once.
    control.wait(float(delay or 0.0))
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": accept,
            "Accept-Language": "en",
            # ProductVariant pages run to 650KB uncompressed and compress about
            # sevenfold. Asking for gzip is politeness as much as economy: it
            # is their bandwidth we are spending.
            "Accept-Encoding": "gzip",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response:
            status, body = response.status, response.read()
            if response.headers.get("Content-Encoding", "").lower() == "gzip":
                body = gzip.decompress(body)
    except urllib.error.HTTPError as exc:
        body = exc.read() if exc.fp else b""
        # Before the block test, which an empty body alone satisfies: a 404 is usually
        # empty-bodied, and reading that as a refusal is how one delisted product stopped a
        # whole airport's source.
        if is_gone_status(exc.code):
            raise PageGone(f"{url} is gone (HTTP {exc.code})") from exc
        if _looks_blocked(exc.code, body):
            raise SourceBlocked(f"{url} refused with HTTP {exc.code}") from exc
        raise FetchError(f"{url} returned HTTP {exc.code}") from exc
    except (urllib.error.URLError, TimeoutError) as exc:
        raise FetchError(f"{url} failed: {exc}") from exc

    if _looks_blocked(status, body):
        raise SourceBlocked(f"{url} returned an empty or challenge body (HTTP {status})")
    return Page(url=url, status=status, body=body)


def fetch_json(url: str, *, delay: float = 0.0) -> object:
    return fetch(url, delay=delay).json()


# --- rendered fetch -----------------------------------------------------------


class RenderBudgetExhausted(FetchError):
    """The run has rendered as many pages as it declared it would."""


class RenderRefused(SourceBlocked):
    """A render that is a refusal on content, carrying what did arrive.

    A collector never reads a refused page as a catalogue, but a probe wants
    to see the headers and the API bodies that came back before the stop.
    """

    def __init__(self, message: str, rendered: "Rendered") -> None:
        super().__init__(message)
        self.rendered = rendered


@dataclass(slots=True)
class ApiResponse:
    """One JSON response the page fetched for itself while rendering."""

    url: str
    status: int
    body: str

    def json(self) -> object:
        return json.loads(self.body)


@dataclass(slots=True)
class Rendered:
    url: str
    final_url: str
    status: int | None
    html: str
    text_length: int
    title: str
    headers: dict[str, str]
    api_responses: list[ApiResponse] = field(default_factory=list)
    aborted: dict[str, int] = field(default_factory=dict)
    # Same-origin requests the page made to paths its robots.txt disallows.
    aborted_paths: dict[str, int] = field(default_factory=dict)
    # The page never fired "load" inside the sidecar's timeout; what is here
    # is the DOM as it stood. A looping challenge page looks like this.
    timed_out: bool = False
    # None when no selector was asked for; False when it never appeared.
    waited_for: bool | None = None
    # Console errors and uncaught exceptions the page raised (capped): the
    # reason a client-rendered page drew nothing is usually written here.
    console: list[str] = field(default_factory=list)
    elapsed_ms: int = 0

    @property
    def text(self) -> str:
        return self.html

    def api_json(self, path_fragment: str) -> Iterator[object]:
        """Parsed bodies of the captured responses whose URL contains the fragment."""
        for response in self.api_responses:
            if path_fragment in response.url:
                try:
                    yield response.json()
                except ValueError:
                    continue


@dataclass(slots=True)
class RenderBudget:
    cap: int
    count: int = 0
    label: str = ""


_budget: ContextVar[RenderBudget | None] = ContextVar("render_budget", default=None)
_last_render_at: dict[str, float] = {}


@contextmanager
def render_budget(cap: int, label: str = "") -> Iterator[RenderBudget]:
    """Declare how many renders a run may make; the count is logged at exit.

    Cost discipline is the point: a collector states its cap up front, and a
    loop that would exceed it stops with RenderBudgetExhausted instead of
    quietly spending fifty text fetches' worth per page it did not need.
    """
    budget = RenderBudget(cap=cap, label=label)
    token = _budget.set(budget)
    try:
        yield budget
    finally:
        _budget.reset(token)
        logger.info("render_budget label=%s cap=%d used=%d", label or "-", cap, budget.count)


def render_wait(requested: float, crawl_delay: float | None, floor: float | None = None) -> float:
    """Seconds between renders on one host: the slowest of theirs, ours, and the floor.

    "Ours" is the run's delay and, when the collector declares one, the source's
    own floor (`render_floor_seconds`: a host that answered a steady run of
    renders with a 403 gets a slower pace by declaration, not by someone
    remembering a flag). The general floor applies only where the host publishes
    no Crawl-delay; a host that names a rate gets the slower of its rate and
    ours, as fetch() does. Without a source floor nothing here changed.
    """
    ours = max(float(requested), float(floor or 0.0))
    if crawl_delay:
        return max(ours, float(crawl_delay))
    return max(ours, RENDER_FLOOR_DELAY)


_CHALLENGE_MARKERS = (
    "just a moment",
    "attention required",
    "access denied",
    "are you a robot",
    "verify you are human",
    "checking your browser",
    "enable javascript and cookies to continue",
)


def render_refusal(
    status: int | None, headers: dict[str, str], title: str, html: str, text_length: int
) -> str | None:
    """Why a rendered page is a refusal, or None when it is a page.

    The Seoul retailer answers product pages with HTTP 403 and
    `cf-mitigated: challenge`, and a browser that cannot pass the challenge
    sees a "Just a moment..." interstitial; a shell that never drew is an
    empty render. Both are refusals, decided on what arrived, never on the
    status code alone. A 404 or a 410 never reaches here: render() raises
    PageGone first, because a page that is not there draws nothing and would
    otherwise read as an empty render, which is a refusal.
    """
    lowered_headers = {k.lower(): v.lower() for k, v in (headers or {}).items()}
    if "challenge" in lowered_headers.get("cf-mitigated", ""):
        return "challenge page (cf-mitigated)"
    probe = f"{title or ''}\n{(html or '')[:4096]}".lower()
    for marker in _CHALLENGE_MARKERS:
        if marker in probe:
            return f"challenge page ({marker!r})"
    if status is not None and status in (401, 403, 406, 429):
        return f"HTTP {status}"
    if text_length < 40:
        return f"empty render ({text_length} characters of text)"
    return None


def disallow_regexes(robots: Robots) -> list[str]:
    """The host's Disallow rules as anchored regexes for the sidecar's route filter.

    A page rendering in a browser fetches its own scripts, styles and images,
    and a Disallow covers those requests as much as a crawler's own. The
    translation is the robots matcher's own (`robots.pattern_regex`), not a
    copy of it: two copies of one matcher drift, and the page's requests must
    be judged exactly as the crawler's are. tests/test_render.py and
    tests/test_robots.py pin the sidecar filter to the matcher.
    """
    return [pattern_regex(pattern) for pattern in robots.disallows]


def _sidecar_url() -> str:
    from app.config import settings  # lazy: fetch is imported by the robots module at test time

    return settings.browser_url.rstrip("/")


def _sidecar_render(payload: dict, *, base_url: str) -> dict:
    request = urllib.request.Request(
        f"{base_url}/render",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=RENDER_TIMEOUT) as response:
            return json.loads(response.read())
    except urllib.error.HTTPError as exc:
        body = exc.read() if exc.fp else b""
        try:
            detail = json.loads(body).get("error", "")
        except ValueError:
            detail = body[:200].decode("utf-8", "replace")
        raise FetchError(f"browser sidecar answered HTTP {exc.code}: {detail}") from exc
    except (urllib.error.URLError, TimeoutError) as exc:
        raise FetchError(f"browser sidecar unreachable: {exc}") from exc


def render(
    url: str,
    *,
    delay: float = RENDER_FLOOR_DELAY,
    floor: float | None = None,
    robots: Robots | None = None,
    allow_hosts: tuple[str, ...] | list[str] = (),
    capture_json: bool = True,
    wait_until: str = "load",
    wait_for: str = "",
    settle_ms: int | None = None,
    sidecar_url: str | None = None,
) -> Rendered:
    """Load one permitted URL in the browser sidecar and return what it drew.

    Same contract as fetch(): our honest identity, robots re-checked (a fresh
    read of the host's rules unless the run passes the Robots it already
    read), the host's crawl-delay honoured with a ten-second floor (and the
    source's own `floor`, when its collector declares one), content
    validated never status, SourceBlocked on a challenge or an empty render
    and PageGone when the page is not there.
    `allow_hosts` names the asset hosts the page may also reach; everything
    else the page asks for is aborted in the sidecar and reported back.
    `wait_until` is "load" for a page that draws with JavaScript and
    "domcontentloaded" for one the server rendered (whose scripts we may not
    even be permitted to fetch). `wait_for` is a CSS selector the caller
    expects once the page has drawn; `settle_ms` is the quiet time after that.
    """
    from app.services.collectors.robots import check_allowed  # lazy: circular at import time

    parts = urlsplit(url)
    if parts.scheme not in ("http", "https") or not parts.hostname:
        raise ValueError(f"render needs an absolute http(s) URL, got {url!r}")
    origin = f"{parts.scheme}://{parts.netloc}"

    budget = _budget.get()
    if budget is None:
        budget = RenderBudget(cap=RENDER_DEFAULT_CAP, label="default")
        _budget.set(budget)
    if budget.count >= budget.cap:
        raise RenderBudgetExhausted(
            f"render cap {budget.cap} reached ({budget.label or 'default'}); not rendering {url}"
        )

    if robots is None:
        robots = check_allowed(origin, [url], fresh=True)
    elif not robots.allows(url):
        raise SourceBlocked(f"{robots.host} robots.txt disallows {url}")

    wait = render_wait(delay, robots.crawl_delay, floor)
    last = _last_render_at.get(parts.hostname)
    remaining = 0.0 if last is None else max(0.0, wait - (time.monotonic() - last))
    # The control hook, after render_wait: bound, it also takes the live pace into account, so a
    # render is only ever slower than the sidecar's own spacing, never faster.
    control.wait(remaining)
    _last_render_at[parts.hostname] = time.monotonic()
    budget.count += 1

    base = (sidecar_url or _sidecar_url()).rstrip("/")
    if not base:
        raise FetchError("no browser sidecar configured (BROWSER_URL); render() cannot run")
    result = _sidecar_render(
        {
            "url": url,
            "user_agent": USER_AGENT,
            "allow_hosts": list(allow_hosts),
            "deny_regex": disallow_regexes(robots),
            "wait_until": wait_until,
            "wait_for": wait_for,
            "settle_ms": settle_ms,
            "capture_json": capture_json,
        },
        base_url=base,
    )
    rendered = Rendered(
        url=url,
        final_url=result.get("final_url") or url,
        status=result.get("status"),
        html=result.get("html") or "",
        text_length=int(result.get("text_length") or 0),
        title=result.get("title") or "",
        headers=dict(result.get("headers") or {}),
        api_responses=[
            ApiResponse(url=r["url"], status=int(r["status"]), body=r.get("body") or "")
            for r in result.get("api_responses") or []
        ],
        aborted={str(k): int(v) for k, v in (result.get("aborted") or {}).items()},
        aborted_paths={str(k): int(v) for k, v in (result.get("aborted_paths") or {}).items()},
        timed_out=bool(result.get("timed_out")),
        waited_for=result.get("waited_for"),
        console=[str(line) for line in result.get("console") or []],
        elapsed_ms=int(result.get("elapsed_ms") or 0),
    )
    if rendered.aborted:
        logger.info("render_aborted_hosts url=%s hosts=%s", url, sorted(rendered.aborted))
    if rendered.aborted_paths:
        logger.info(
            "render_disallowed_paths url=%s paths=%s", url, sorted(rendered.aborted_paths)
        )
    # `wait` is in the run log so a pace change and what followed it can be read
    # together later (the first Singapore run: 18 clean pages, then a 403).
    logger.info(
        "rendered url=%s status=%s text=%d api=%d timed_out=%s ms=%d wait=%.0f budget=%d/%d",
        url, rendered.status, rendered.text_length, len(rendered.api_responses),
        rendered.timed_out, rendered.elapsed_ms, wait, budget.count, budget.cap,
    )
    if is_gone_status(rendered.status):
        raise PageGone(f"{url} is gone (HTTP {rendered.status})")
    refusal = render_refusal(
        rendered.status, rendered.headers, rendered.title, rendered.html, rendered.text_length
    )
    if refusal:
        raise RenderRefused(f"{url} refused: {refusal} (HTTP {rendered.status})", rendered)
    return rendered