"""Fetch transport: DIRECT MODE first (httpx, conditional requests, per-domain rate limits, robots.txt, backoff), optional
ESCALATED MODE (headless browser → Scrapfly → Firecrawl) that is never required and never automatic unless the connector asks.

    direct → retry → [browser] → [scrapfly] → [firecrawl] → BlockedError (→ review queue by the caller)
"""
from __future__ import annotations

import asyncio
import hashlib
import ipaddress
import logging
import socket
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from urllib.parse import urljoin, urlparse, urlunparse
from urllib.robotparser import RobotFileParser

import httpx

from aiatlas.config import settings

log = logging.getLogger(__name__)

TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504}
BLOCK_STATUS = {401, 403, 451, 999}
REDIRECT_STATUS = {301, 302, 303, 307, 308}
MAX_REDIRECTS = 5

# ---------------------------------------------------------------------------------------------- SSRF guard
# Follow-up targets come from crawled content (links in model cards, feeds, READMEs): every destination — and every redirect
# hop — is validated before a connection is opened. Nothing internal is ever fetched.
_BLOCKED_SUFFIXES = (".local", ".internal", ".localhost", ".localdomain", ".lan", ".home", ".corp", ".intranet")
_BLOCKED_HOSTS = {"localhost", "metadata.google.internal", "metadata", "instance-data", "kubernetes.default", "kubernetes.default.svc"}
_BLOCKED_NETWORKS = [ipaddress.ip_network(n) for n in (
    "0.0.0.0/8",          # unspecified / "this network"
    "10.0.0.0/8",         # RFC1918
    "100.64.0.0/10",      # CGNAT
    "127.0.0.0/8",        # loopback
    "169.254.0.0/16",     # link-local (cloud metadata endpoints 169.254.169.254 …)
    "172.16.0.0/12",      # RFC1918
    "192.0.0.0/24",       # IETF protocol assignments
    "192.168.0.0/16",     # RFC1918
    "198.18.0.0/15",      # benchmarking
    "240.0.0.0/4",        # reserved + broadcast
    "::/128",             # unspecified
    "::1/128",            # loopback
    "fc00::/7",           # ULA
    "fe80::/10",          # link-local
    "::ffff:0:0/96",      # IPv4-mapped (checked again as IPv4 below)
    "64:ff9b::/96",       # NAT64
)]


class BlockedDestination(ValueError):
    """The URL points at a private, local or otherwise non-public destination."""


def _ip_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        ip = ip.ipv4_mapped
    if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_reserved or ip.is_multicast:
        return True
    return any(ip in net for net in _BLOCKED_NETWORKS)


def validate_destination(url: str, *, resolved_ips: list[str] | None = None) -> None:
    """Raise `BlockedDestination` for non-http(s) schemes, local/internal hostnames and private, loopback, link-local, CGNAT,
    IPv6 ULA/link-local or unspecified addresses. `resolved_ips` lets callers (and tests) inject the DNS answer."""
    p = urlparse(url.strip())
    if p.scheme.lower() not in ("http", "https"):
        raise BlockedDestination(f"blocked destination: scheme {p.scheme!r} for {url}")
    host = (p.hostname or "").strip().lower().rstrip(".")
    if not host:
        raise BlockedDestination(f"blocked destination: no host in {url}")
    if p.username or p.password:
        raise BlockedDestination(f"blocked destination: credentials in URL {url}")
    if host in _BLOCKED_HOSTS or host.endswith(_BLOCKED_SUFFIXES) or "." not in host and not _is_ip_literal(host):
        raise BlockedDestination(f"blocked destination: local host {host!r}")
    literal = _parse_ip(host)
    if literal is not None:
        if _ip_blocked(literal):
            raise BlockedDestination(f"blocked destination: non-public address {host}")
        return
    ips = resolved_ips if resolved_ips is not None else _resolve(host, p.port or (443 if p.scheme.lower() == "https" else 80))
    for raw in ips:
        ip = _parse_ip(raw)
        if ip is not None and _ip_blocked(ip):
            raise BlockedDestination(f"blocked destination: {host} resolves to non-public address {raw}")


def _is_ip_literal(host: str) -> bool:
    return _parse_ip(host) is not None


def _parse_ip(host: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
    try:
        return ipaddress.ip_address(host.strip("[]").split("%")[0])
    except ValueError:
        return None


def _resolve(host: str, port: int) -> list[str]:
    try:
        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror:
        return []   # unresolvable hosts fail later with a transport error, not as SSRF
    return sorted({info[4][0] for info in infos})


async def validate_destination_async(url: str) -> None:
    """Non-blocking wrapper (DNS resolution runs in a thread)."""
    p = urlparse(url.strip())
    host = (p.hostname or "").strip().lower()
    if not host or _parse_ip(host) is not None:
        validate_destination(url, resolved_ips=[])
        return
    validate_destination(url, resolved_ips=[])          # scheme / hostname rules first (no DNS)
    ips = await asyncio.to_thread(_resolve, host, p.port or (443 if p.scheme.lower() == "https" else 80))
    validate_destination(url, resolved_ips=ips)


class FetchError(Exception):
    def __init__(self, message: str, *, status: int | None = None, url: str = ""):
        super().__init__(message)
        self.status = status
        self.url = url


class BlockedError(FetchError):
    """Access denied by the origin after every allowed transport (never bypass: park in the review queue)."""


class NotModified(Exception):
    """HTTP 304 — content unchanged since our stored validators."""


@dataclass
class FetchResult:
    url: str
    final_url: str
    status: int
    headers: dict[str, str]
    content: bytes
    content_type: str
    fetched_at: datetime
    duration_ms: int
    transport: str = "direct"
    from_cache: bool = False
    sha256: str = field(init=False)

    def __post_init__(self) -> None:
        self.sha256 = hashlib.sha256(self.content).hexdigest()

    @property
    def text(self) -> str:
        enc = "utf-8"
        ct = self.content_type.lower()
        if "charset=" in ct:
            enc = ct.split("charset=", 1)[1].split(";")[0].strip().strip('"') or "utf-8"
        try:
            return self.content.decode(enc, errors="replace")
        except LookupError:
            return self.content.decode("utf-8", errors="replace")

    @property
    def is_html(self) -> bool:
        return "html" in self.content_type or self.content[:256].lstrip().lower().startswith((b"<!doctype html", b"<html"))

    @property
    def is_json(self) -> bool:
        return "json" in self.content_type or self.content[:1] in (b"{", b"[")

    @property
    def is_xml(self) -> bool:
        return "xml" in self.content_type or self.content[:5] == b"<?xml"

    @property
    def is_pdf(self) -> bool:
        return "pdf" in self.content_type or self.content[:4] == b"%PDF"

    def json(self):  # type: ignore[no-untyped-def]
        import orjson

        return orjson.loads(self.content)


def canonicalize_url(url: str) -> str:
    """Stable URL key: lowercase scheme/host, no fragment, no tracking params, no trailing slash duplication."""
    p = urlparse(url.strip())
    query = "&".join(sorted(q for q in p.query.split("&") if q and not q.lower().startswith(
        ("utm_", "ref=", "ref_", "fbclid", "gclid", "mc_cid", "mc_eid", "_hs", "igshid", "source="))))
    path = p.path or "/"
    if len(path) > 1 and path.endswith("/") and not path.endswith("//"):
        path = path.rstrip("/") or "/"
    return urlunparse((p.scheme.lower() or "https", p.netloc.lower(), path, "", query, ""))


def domain_of(url: str) -> str:
    host = urlparse(url).netloc.lower()
    return host.removeprefix("www.")


class _RateLimiter:
    """Per-domain minimum spacing between requests (token bucket, in-process). Cluster-wide fairness comes from Redis locks
    at the connector level (one connector run at a time)."""

    def __init__(self) -> None:
        self._next: dict[str, float] = {}
        self._locks: dict[str, asyncio.Lock] = {}

    async def wait(self, domain: str, per_min: int) -> None:
        lock = self._locks.setdefault(domain, asyncio.Lock())
        async with lock:
            spacing = 60.0 / max(1, per_min)
            now = time.monotonic()
            ready = self._next.get(domain, 0.0)
            if ready > now:
                await asyncio.sleep(ready - now)
                now = time.monotonic()
            self._next[domain] = now + spacing


class _Robots:
    def __init__(self) -> None:
        self._cache: dict[str, tuple[float, RobotFileParser | None]] = {}

    async def allowed(self, client: httpx.AsyncClient, url: str) -> bool:
        if not settings.respect_robots:
            return True
        p = urlparse(url)
        key = f"{p.scheme}://{p.netloc}"
        cached = self._cache.get(key)
        if cached is None or cached[0] < time.monotonic():
            rp: RobotFileParser | None = RobotFileParser()
            try:
                r = await client.get(f"{key}/robots.txt", timeout=15, follow_redirects=False)   # host already validated by the caller
                if r.status_code == 200 and len(r.content) < 512 * 1024:
                    rp.parse(r.text.splitlines())  # type: ignore[union-attr]
                else:
                    rp = None  # no robots or error → allowed
            except Exception:  # noqa: BLE001
                rp = None
            self._cache[key] = (time.monotonic() + 6 * 3600, rp)
            cached = self._cache[key]
        rp = cached[1]
        if rp is None:
            return True
        try:
            return rp.can_fetch(settings.user_agent.split("/")[0], url) or rp.can_fetch("*", url)
        except Exception:  # noqa: BLE001
            return True


class Fetcher:
    """One Fetcher per connector run. Direct mode by default; `escalate=True` enables the optional chain."""

    def __init__(self, *, rate_per_min: int | None = None, robots: bool = True, timeout_s: float | None = None,
                 headers: dict[str, str] | None = None, http2: bool = True):
        self.rate_per_min = rate_per_min or settings.default_rate_per_min
        self.robots = robots
        self.timeout_s = timeout_s or settings.http_timeout_s
        self.headers = {"User-Agent": settings.user_agent, "Accept": "text/html,application/xhtml+xml,application/xml,application/json,text/*;q=0.9,*/*;q=0.8",
                        "Accept-Language": "en-US,en;q=0.8", **(headers or {})}
        self._client: httpx.AsyncClient | None = None
        self._http2 = http2

    async def __aenter__(self) -> Fetcher:
        # redirects are followed manually in `get()` so that every hop goes through the SSRF guard
        self._client = httpx.AsyncClient(headers=self.headers, timeout=httpx.Timeout(self.timeout_s, connect=20), follow_redirects=False,
                                         http2=self._http2, limits=httpx.Limits(max_connections=16, max_keepalive_connections=8))
        return self

    async def __aexit__(self, *exc: object) -> None:
        if self._client:
            await self._client.aclose()
            self._client = None

    @property
    def client(self) -> httpx.AsyncClient:
        if self._client is None:
            raise RuntimeError("use `async with Fetcher() as f`")
        return self._client

    async def get(self, url: str, *, etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1,
                  escalate: bool = False, retries: int = 2, accept: str | None = None, rate_per_min: int | None = None) -> FetchResult:
        """Direct fetch with conditional headers. Raises NotModified (304), FetchError, BlockedError.
        Every destination and every redirect hop (max 5) is validated against the SSRF guard before a connection is opened."""
        client = self.client
        try:
            await validate_destination_async(url)
        except BlockedDestination as exc:
            raise FetchError(f"blocked destination: {exc}", url=url) from exc
        if self.robots and not await _robots.allowed(client, url):
            raise BlockedError(f"robots.txt disallows {url}", status=None, url=url)
        headers: dict[str, str] = {}
        if etag:
            headers["If-None-Match"] = etag
        if last_modified:
            headers["If-Modified-Since"] = last_modified
        if accept:
            headers["Accept"] = accept
        last_exc: Exception | None = None
        current = url
        hops = 0
        for attempt in range(retries + 1):
            await _limiter.wait(domain_of(current), rate_per_min or self.rate_per_min)
            t0 = time.perf_counter()
            try:
                async with client.stream("GET", current, headers=headers) as r:
                    if r.status_code == 304:
                        raise NotModified()
                    if r.status_code in REDIRECT_STATUS:
                        location = r.headers.get("location")
                        if not location:
                            raise FetchError(f"http {r.status_code} without Location for {current}", status=r.status_code, url=url)
                        hops += 1
                        if hops > MAX_REDIRECTS:
                            raise FetchError(f"too many redirects (> {MAX_REDIRECTS}) for {url}", status=r.status_code, url=url)
                        nxt = urljoin(current, location)
                        try:
                            await validate_destination_async(nxt)
                        except BlockedDestination as exc:
                            raise FetchError(f"blocked destination: redirect {current} → {nxt}: {exc}", status=r.status_code, url=url) from exc
                        if domain_of(nxt) != domain_of(current):
                            headers.pop("If-None-Match", None)      # validators belong to the original resource only
                            headers.pop("If-Modified-Since", None)
                        current = nxt
                        # a redirect is not a retry: loop again without consuming the attempt budget
                        return await self._follow(client, current, headers, url=url, hops=hops, min_bytes=min_bytes, escalate=escalate,
                                                  retries=retries, rate_per_min=rate_per_min)
                    if r.status_code in TRANSIENT_STATUS and attempt < retries:
                        retry_after = r.headers.get("retry-after")
                        delay = min(60.0, float(retry_after)) if retry_after and retry_after.isdigit() else 2.0 * (2 ** attempt)
                        log.info("transient status, backing off", extra={"url": url, "status": r.status_code, "delay": delay})
                        await asyncio.sleep(delay)
                        continue
                    if r.status_code in BLOCK_STATUS:
                        if escalate:
                            return await self._escalate(url, reason=f"http {r.status_code}")
                        raise BlockedError(f"http {r.status_code} for {url}", status=r.status_code, url=url)
                    if r.status_code >= 400:
                        raise FetchError(f"http {r.status_code} for {url}", status=r.status_code, url=url)
                    return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, escalate=escalate)
            except NotModified:
                raise
            except (BlockedError, FetchError):
                raise
            except (httpx.TimeoutException, httpx.TransportError) as exc:
                last_exc = exc
                if attempt < retries:
                    await asyncio.sleep(1.5 * (2 ** attempt))
                    continue
                raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url) from exc
        raise FetchError(f"fetch failed: {last_exc}", url=url)

    async def _follow(self, client: httpx.AsyncClient, current: str, headers: dict[str, str], *, url: str, hops: int, min_bytes: int,
                      escalate: bool, retries: int, rate_per_min: int | None) -> FetchResult:
        """Continue a redirect chain: each hop is validated, capped at MAX_REDIRECTS, rate-limited per domain."""
        while True:
            await _limiter.wait(domain_of(current), rate_per_min or self.rate_per_min)
            t0 = time.perf_counter()
            try:
                async with client.stream("GET", current, headers=headers) as r:
                    if r.status_code in REDIRECT_STATUS:
                        location = r.headers.get("location")
                        if not location:
                            raise FetchError(f"http {r.status_code} without Location for {current}", status=r.status_code, url=url)
                        hops += 1
                        if hops > MAX_REDIRECTS:
                            raise FetchError(f"too many redirects (> {MAX_REDIRECTS}) for {url}", status=r.status_code, url=url)
                        nxt = urljoin(current, location)
                        try:
                            await validate_destination_async(nxt)
                        except BlockedDestination as exc:
                            raise FetchError(f"blocked destination: redirect {current} → {nxt}: {exc}", status=r.status_code, url=url) from exc
                        if domain_of(nxt) != domain_of(current):
                            headers.pop("If-None-Match", None)
                            headers.pop("If-Modified-Since", None)
                        current = nxt
                        continue
                    if r.status_code == 304:
                        raise NotModified()
                    if r.status_code in BLOCK_STATUS:
                        if escalate:
                            return await self._escalate(url, reason=f"http {r.status_code}")
                        raise BlockedError(f"http {r.status_code} for {current}", status=r.status_code, url=url)
                    if r.status_code >= 400:
                        raise FetchError(f"http {r.status_code} for {current}", status=r.status_code, url=url)
                    return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, escalate=escalate)
            except (httpx.TimeoutException, httpx.TransportError) as exc:
                raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url) from exc

    async def _read(self, r: httpx.Response, *, url: str, final_url: str, t0: float, min_bytes: int, escalate: bool) -> FetchResult:
        chunks: list[bytes] = []
        size = 0
        async for chunk in r.aiter_bytes():
            size += len(chunk)
            if size > settings.max_body_bytes:
                raise FetchError(f"body exceeds {settings.max_body_bytes} bytes", status=r.status_code, url=url)
            chunks.append(chunk)
        content = b"".join(chunks)
        if len(content) < min_bytes:
            raise FetchError(f"suspiciously short body ({len(content)} bytes) for {url}", status=r.status_code, url=url)
        res = FetchResult(url=url, final_url=final_url, status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()},
                          content=content, content_type=r.headers.get("content-type", "").lower(), fetched_at=datetime.now(UTC),
                          duration_ms=int((time.perf_counter() - t0) * 1000), transport="direct")
        if escalate and res.is_html and _looks_like_challenge(content):
            return await self._escalate(url, reason="anti-bot challenge page")
        return res

    # ---------------------------------------------------------------------------------------------- escalation (optional)
    async def _escalate(self, url: str, *, reason: str) -> FetchResult:
        log.warning("escalating fetch", extra={"url": url, "reason": reason})
        if settings.browser_enabled:
            try:
                return await self._browser(url)
            except Exception as exc:  # noqa: BLE001
                log.info("browser transport failed", extra={"url": url, "error": str(exc)})
        if settings.scrapfly_api_key:
            try:
                return await self._scrapfly(url)
            except Exception as exc:  # noqa: BLE001
                log.info("scrapfly transport failed", extra={"url": url, "error": str(exc)})
        if settings.firecrawl_api_key:
            try:
                return await self._firecrawl(url)
            except Exception as exc:  # noqa: BLE001
                log.info("firecrawl transport failed", extra={"url": url, "error": str(exc)})
        raise BlockedError(f"blocked after escalation ({reason}): {url}", url=url)

    async def _browser(self, url: str) -> FetchResult:
        from playwright.async_api import async_playwright  # optional dependency

        t0 = time.perf_counter()
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            try:
                page = await browser.new_page(user_agent=settings.user_agent)
                resp = await page.goto(url, wait_until="networkidle", timeout=int(self.timeout_s * 1000))
                html = await page.content()
                status = resp.status if resp else 200
            finally:
                await browser.close()
        return FetchResult(url=url, final_url=url, status=status, headers={}, content=html.encode(), content_type="text/html; charset=utf-8",
                           fetched_at=datetime.now(UTC), duration_ms=int((time.perf_counter() - t0) * 1000), transport="browser")

    async def _scrapfly(self, url: str) -> FetchResult:
        t0 = time.perf_counter()
        r = await self.client.get("https://api.scrapfly.io/scrape", params={"key": settings.scrapfly_api_key, "url": url, "asp": "true",
                                                                          "render_js": "true", "country": "us"}, timeout=120)
        r.raise_for_status()
        data = r.json()["result"]
        content = (data.get("content") or "").encode()
        return FetchResult(url=url, final_url=data.get("url") or url, status=int(data.get("status_code") or 200), headers={},
                           content=content, content_type=(data.get("content_type") or "text/html").lower(), fetched_at=datetime.now(UTC),
                           duration_ms=int((time.perf_counter() - t0) * 1000), transport="scrapfly")

    async def _firecrawl(self, url: str) -> FetchResult:
        t0 = time.perf_counter()
        r = await self.client.post("https://api.firecrawl.dev/v1/scrape", json={"url": url, "formats": ["rawHtml"]},
                                   headers={"Authorization": f"Bearer {settings.firecrawl_api_key}"}, timeout=120)
        r.raise_for_status()
        data = r.json().get("data", {})
        content = (data.get("rawHtml") or data.get("html") or "").encode()
        return FetchResult(url=url, final_url=(data.get("metadata") or {}).get("sourceURL") or url, status=200, headers={}, content=content,
                           content_type="text/html", fetched_at=datetime.now(UTC), duration_ms=int((time.perf_counter() - t0) * 1000),
                           transport="firecrawl")


def _looks_like_challenge(content: bytes) -> bool:
    head = content[:20000].lower()
    if len(content) > 60_000:
        return False
    markers = (b"just a moment", b"cf-chl-", b"challenge-platform", b"attention required", b"verify you are human", b"access denied",
               b"captcha", b"perimeterx", b"_px", b"datadome", b"enable javascript and cookies to continue")
    return sum(m in head for m in markers) >= 2


def file_result(path: str, *, url: str, content_type: str = "application/octet-stream") -> FetchResult:
    """Wrap a local file as a FetchResult (fixtures, seed snapshots, `aia run x --file`)."""
    with open(path, "rb") as fh:
        content = fh.read()
    return FetchResult(url=url, final_url=url, status=200, headers={}, content=content, content_type=content_type,
                       fetched_at=datetime.now(UTC), duration_ms=0, transport="file")


_limiter = _RateLimiter()
_robots = _Robots()

__all__ = ["MAX_REDIRECTS", "BlockedDestination", "BlockedError", "FetchError", "FetchResult", "Fetcher", "NotModified", "canonicalize_url", "domain_of",
           "file_result", "validate_destination", "validate_destination_async"]