"""Fetch transport (spec §13–14, §109–118): httpx direct mode with conditional requests, per-domain rate limiting and concurrency
caps, robots.txt awareness, bounded redirects (each hop SSRF-checked), size limits, failure classification and an *optional*
headless-browser mode that is only used when a sensor's connector asks for it.

Crawler inputs are untrusted: every destination and every redirect hop is validated before a socket is opened.
"""
from __future__ import annotations

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

import httpx

from companyatlas.config import settings
from companyatlas.taxonomy import FailureClass
from companyatlas.urls import registrable_domain

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}

# ---------------------------------------------------------------------------------------------- SSRF guard (spec §115–116)
_BLOCKED_SUFFIXES = (".local", ".internal", ".localhost", ".localdomain", ".lan", ".home", ".corp", ".intranet", ".maclustr.io", ".ts.net")
_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", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8", "169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.168.0.0/16",
    "198.18.0.0/15", "240.0.0.0/4", "::/128", "::1/128", "fc00::/7", "fe80::/10", "::ffff:0:0/96", "64:ff9b::/96",
)]


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 isinstance(ip, ipaddress.IPv6Address) and ip in ipaddress.ip_network("64:ff9b::/96"):
        ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)          # NAT64: judge the embedded IPv4
    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 _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 []
    return sorted({info[4][0] for info in infos})


def validate_destination(url: str, *, resolved_ips: list[str] | None = None) -> None:
    p = urlparse(url.strip())
    if p.scheme.lower() not in ("http", "https"):
        raise BlockedDestination(f"scheme {p.scheme!r}")
    host = (p.hostname or "").strip().lower().rstrip(".")
    if not host:
        raise BlockedDestination("no host")
    if p.username or p.password:
        raise BlockedDestination("credentials in URL")
    literal = _parse_ip(host)
    if literal is None and (host in _BLOCKED_HOSTS or host.endswith(_BLOCKED_SUFFIXES) or "." not in host):
        raise BlockedDestination(f"local host {host!r}")
    if literal is not None:
        if _ip_blocked(literal):
            raise BlockedDestination(f"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"{host} resolves to non-public address {raw}")


async def validate_destination_async(url: str) -> None:
    p = urlparse(url.strip())
    host = (p.hostname or "").strip().lower()
    validate_destination(url, resolved_ips=[])
    if host and _parse_ip(host) is None:
        ips = await asyncio.to_thread(_resolve, host, p.port or (443 if p.scheme.lower() == "https" else 80))
        validate_destination(url, resolved_ips=ips)


# ---------------------------------------------------------------------------------------------- text decoding / sanity
_META_CHARSET_RE = re.compile(rb"""<meta[^>]+charset\s*=\s*["']?\s*([a-zA-Z0-9_.:-]+)""", re.IGNORECASE)
_XML_ENC_RE = re.compile(rb"""^\s*<\?xml[^>]*encoding\s*=\s*["']([a-zA-Z0-9_.:-]+)["']""", re.IGNORECASE)
_CONTROL_RE = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")


def text_quality(text: str) -> float:
    """Share of the text that is NOT replacement characters / C0 control bytes (1.0 = clean). Empty text is 1.0."""
    if not text:
        return 1.0
    bad = text.count("\ufffd") + len(_CONTROL_RE.findall(text))
    return max(0.0, 1.0 - bad / len(text))


def declared_encodings(content: bytes, content_type: str) -> list[str]:
    out: list[str] = []
    if content.startswith(b"\xef\xbb\xbf"):
        out.append("utf-8-sig")
    elif content.startswith((b"\xff\xfe", b"\xfe\xff")):
        out.append("utf-16")
    ct = (content_type or "").lower()
    if "charset=" in ct:
        out.append(ct.split("charset=", 1)[1].split(";")[0].strip().strip('"'))
    head = content[:4096]
    m = _XML_ENC_RE.search(head) or _META_CHARSET_RE.search(head)
    if m:
        out.append(m.group(1).decode("ascii", "ignore"))
    seen: set[str] = set()
    result = []
    for e in out + ["utf-8", "cp1252"]:
        e = e.lower().replace("_", "-")
        if e in ("iso-8859-1", "latin-1", "latin1", "ascii", "us-ascii"):
            e = "cp1252"          # superset; what browsers actually do
        if e not in seen:
            seen.add(e)
            result.append(e)
    return result


def decode_text(content: bytes, content_type: str = "") -> str:
    best: tuple[float, str] | None = None
    for enc in declared_encodings(content, content_type):
        try:
            candidate = content.decode(enc, errors="replace")
        except LookupError:
            continue
        q = text_quality(candidate)
        if q >= 0.99:
            return candidate
        if best is None or q > best[0]:
            best = (q, candidate)
    return best[1] if best else content.decode("utf-8", errors="replace")


_TEXTUAL_TYPES = ("text/", "html", "xml", "json", "javascript", "rss", "atom", "csv")


def looks_binary(content: bytes, content_type: str) -> bool:
    """A body announced as text/HTML/XML/JSON whose bytes are not text (undecoded compression, wrong Content-Type)."""
    ct = (content_type or "").lower()
    if not any(t in ct for t in _TEXTUAL_TYPES) and ct:
        return False                      # PDFs, images… are judged by their connectors, not here
    sample = content[:8192]
    if not sample:
        return False
    if sample[:2] == b"\x1f\x8b" or sample[:4] == b"\x28\xb5\x2f\xfd":       # gzip / zstd magic left undecoded
        return True
    text_bytes = sum(1 for b in sample if 32 <= b < 127 or b in (9, 10, 13) or b >= 128)
    high = sum(1 for b in sample if b >= 128)
    controls = len(sample) - text_bytes
    if controls / len(sample) > 0.02:
        return True
    if high / len(sample) > 0.30:        # legitimate UTF-8 (CJK) has structure; verify it decodes cleanly
        return text_quality(sample.decode("utf-8", errors="replace")) < 0.90
    return False


# ---------------------------------------------------------------------------------------------- results / errors


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


class BlockedError(FetchError):
    """Access denied by the origin or by robots — never bypassed."""


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

    def __init__(self, duration_ms: int = 0):
        super().__init__("not modified")
        self.duration_ms = duration_ms


@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 = "http"
    redirects: int = 0
    sha256: str = field(init=False)

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

    @property
    def etag(self) -> str | None:
        return self.headers.get("etag")

    @property
    def last_modified(self) -> str | None:
        return self.headers.get("last-modified")

    @property
    def text(self) -> str:
        """Decoded body: BOM → HTTP charset → in-document declaration (<meta charset>, XML prolog) → UTF-8; if the chosen codec leaves
        more than 1 % replacement characters, the alternatives are tried and the cleanest decode wins (never garbage in, never silently)."""
        return decode_text(self.content, self.content_type)

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

    @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" or self.content[:200].lstrip().startswith((b"<rss", b"<feed", b"<urlset", b"<sitemapindex"))

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

        return orjson.loads(self.content)


def classify_exception(exc: BaseException) -> FailureClass:
    if isinstance(exc, FetchError):
        return exc.failure
    if isinstance(exc, httpx.ConnectTimeout | httpx.ReadTimeout | httpx.WriteTimeout | httpx.PoolTimeout | asyncio.TimeoutError | TimeoutError):
        return FailureClass.TIMEOUT
    if isinstance(exc, httpx.RemoteProtocolError | httpx.ProtocolError) or exc.__class__.__name__ in ("ProtocolError", "StreamReset", "ConnectionTerminated"):
        return FailureClass.TIMEOUT            # transient transport-level protocol errors (h2 connection reuse, resets): retry policy, not a mystery
    if isinstance(exc, httpx.ConnectError):
        msg = str(exc).lower()
        if "nodename" in msg or "name or service" in msg or "getaddrinfo" in msg or "temporary failure in name" in msg or "no address" in msg:
            return FailureClass.DNS
        return FailureClass.TIMEOUT
    return FailureClass.UNKNOWN


def _status_failure(status: int) -> FailureClass:
    if status == 429:
        return FailureClass.RATE_LIMIT
    if status in (404, 410):
        return FailureClass.PAGE_REMOVED
    if status in BLOCK_STATUS:
        return FailureClass.BOT_CHALLENGE if status in (403, 999) else FailureClass.HTTP_4XX
    if 400 <= status < 500:
        return FailureClass.HTTP_4XX
    return FailureClass.HTTP_5XX


# ---------------------------------------------------------------------------------------------- politeness primitives


class DomainGovernor:
    """Per-domain minimum spacing + concurrency cap (in-process). Cluster-wide fairness comes from Postgres domain budgets."""

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

    def set_crawl_delay(self, domain: str, seconds: float) -> None:
        self._delay[domain] = min(120.0, max(0.0, seconds))

    async def acquire(self, domain: str, per_min: int) -> asyncio.Semaphore:
        sem = self._sem.get(domain)
        if sem is None:
            sem = self._sem[domain] = asyncio.Semaphore(settings.domain_max_concurrency)
        await sem.acquire()
        async with self._lock:
            spacing = max(60.0 / max(1, per_min), self._delay.get(domain, 0.0))
            now = time.monotonic()
            ready = self._next.get(domain, 0.0)
            wait = max(0.0, ready - now)
            self._next[domain] = max(now, ready) + spacing
        if wait > 0:
            await asyncio.sleep(wait)
        return sem


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

    async def policy(self, client: httpx.AsyncClient, url: str) -> tuple[bool, float | None]:
        """(allowed, crawl_delay_seconds)."""
        if not settings.respect_robots:
            return True, None
        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()
            delay: float | None = None
            try:
                r = await client.get(f"{key}/robots.txt", timeout=15, follow_redirects=False)
                if r.status_code == 200 and len(r.content) < 512 * 1024 and b"<html" not in r.content[:512].lower():
                    rp.parse(r.text.splitlines())  # type: ignore[union-attr]
                    agent = settings.user_agent.split("/")[0]
                    try:
                        d = rp.crawl_delay(agent) or rp.crawl_delay("*")  # type: ignore[union-attr]
                        delay = float(d) if d else None
                    except Exception:  # noqa: BLE001
                        delay = None
                else:
                    rp = None
            except Exception:  # noqa: BLE001
                rp = None
            self._cache[key] = (time.monotonic() + 12 * 3600, rp, delay)
            cached = self._cache[key]
        rp, delay = cached[1], cached[2]
        if rp is None:
            return True, None
        agent = settings.user_agent.split("/")[0]
        try:
            return (rp.can_fetch(agent, url) or rp.can_fetch("*", url)), delay
        except Exception:  # noqa: BLE001
            return True, delay


governor = DomainGovernor()
robots = RobotsCache()


# ---------------------------------------------------------------------------------------------- fetcher


class Fetcher:
    """Shared httpx client (one per worker process). `get()` raises NotModified / FetchError / BlockedError.
    HTTP/1.1 by default: h2 connection reuse against thousands of heterogeneous origins produced `ConnectionInputs.SEND_SETTINGS in state CLOSED`
    protocol errors (2026-09-13); HTTP/2 stays available per Fetcher (`http2=True`)."""

    def __init__(self, *, timeout_s: float | None = None, headers: dict[str, str] | None = None, http2: bool = False,
                 max_connections: int | None = None):
        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,application/rss+xml,text/*;q=0.9,*/*;q=0.7",
                        "Accept-Language": "en-US,en;q=0.9,fr;q=0.6,de;q=0.4,*;q=0.2",
                        **(headers or {})}
        self._client: httpx.AsyncClient | None = None
        self._http2 = http2
        self._max_connections = max_connections or max(16, settings.fetch_concurrency * 2)

    async def __aenter__(self) -> Self:
        await self.open()
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.close()

    async def open(self) -> None:
        if self._client is None:
            self._client = httpx.AsyncClient(headers=self.headers, timeout=httpx.Timeout(self.timeout_s, connect=15), follow_redirects=False,
                                             http2=self._http2, limits=httpx.Limits(max_connections=self._max_connections,
                                                                                    max_keepalive_connections=self._max_connections // 2))

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

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

    async def get(self, url: str, *, etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1,
                  retries: int | None = None, accept: str | None = None, rate_per_min: int | None = None, respect_robots: bool = True,
                  max_bytes: int | None = None) -> FetchResult:
        return await self.request("GET", url, etag=etag, last_modified=last_modified, min_bytes=min_bytes, retries=retries, accept=accept,
                                  rate_per_min=rate_per_min, respect_robots=respect_robots, max_bytes=max_bytes)

    async def post_json(self, url: str, payload: object, *, accept: str | None = "application/json", **kw: object) -> FetchResult:
        """POST a JSON body to a *public* endpoint that a public page itself calls (spec §13 mode C, e.g. Workday job search)."""
        return await self.request("POST", url, json=payload, accept=accept, **kw)  # type: ignore[arg-type]

    async def request(self, method: str, url: str, *, json: object | None = None, headers: dict[str, str] | None = None,
                      etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1, retries: int | None = None,
                      accept: str | None = None, rate_per_min: int | None = None, respect_robots: bool = True,
                      max_bytes: int | None = None) -> FetchResult:
        """Same guarantees as `get()` (SSRF guard on every hop, robots, per-domain governor, size caps, failure classes) for any
        method. Non-GET requests are never retried on transport errors beyond the configured retries and never follow redirects
        across registrable domains with the body."""
        client = self.client
        method = method.upper()
        retries = settings.fetch_retries if retries is None else retries
        try:
            await validate_destination_async(url)
        except BlockedDestination as exc:
            raise FetchError(f"blocked destination: {exc}", url=url, failure=FailureClass.BLOCKED_DESTINATION) from exc
        domain = registrable_domain(url)
        if respect_robots:
            allowed, delay = await robots.policy(client, url)
            if delay:
                governor.set_crawl_delay(domain, delay)
            if not allowed:
                raise BlockedError(f"robots.txt disallows {url}", url=url, failure=FailureClass.ROBOTS)
        req_headers: dict[str, str] = dict(headers or {})
        if etag:
            req_headers["If-None-Match"] = etag
        if last_modified:
            req_headers["If-Modified-Since"] = last_modified
        if accept:
            req_headers["Accept"] = accept
        headers = req_headers
        current = url
        hops = 0
        attempt = 0
        while True:
            sem = await governor.acquire(registrable_domain(current), rate_per_min or settings.default_rate_per_min)
            t0 = time.perf_counter()
            try:
                async with client.stream(method, current, headers=headers, json=json) as r:
                    if r.status_code == 304:
                        raise NotModified(int((time.perf_counter() - t0) * 1000))
                    if r.status_code in REDIRECT_STATUS:
                        location = r.headers.get("location")
                        if not location:
                            raise FetchError(f"http {r.status_code} without Location", status=r.status_code, url=url, failure=FailureClass.REDIRECT)
                        hops += 1
                        if hops > settings.max_redirects:
                            raise FetchError(f"too many redirects (> {settings.max_redirects})", status=r.status_code, url=url, failure=FailureClass.REDIRECT)
                        nxt = urljoin(current, location)
                        try:
                            await validate_destination_async(nxt)
                        except BlockedDestination as exc:
                            raise FetchError(f"blocked redirect {current} → {nxt}: {exc}", status=r.status_code, url=url,
                                             failure=FailureClass.BLOCKED_DESTINATION) from exc
                        if registrable_domain(nxt) != registrable_domain(current):
                            headers.pop("If-None-Match", None)
                            headers.pop("If-Modified-Since", None)
                            if method != "GET":
                                raise FetchError(f"{method} redirected off-domain {current} → {nxt}", status=r.status_code, url=url,
                                                 failure=FailureClass.REDIRECT)
                        if r.status_code in (301, 302, 303) and method != "GET":
                            method, json = "GET", None      # per RFC 9110 user agents switch to GET
                        current = nxt
                        continue
                    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)
                        attempt += 1
                        await asyncio.sleep(delay)
                        continue
                    if r.status_code in BLOCK_STATUS:
                        raise BlockedError(f"http {r.status_code} for {current}", status=r.status_code, url=url, failure=_status_failure(r.status_code))
                    if r.status_code >= 400:
                        raise FetchError(f"http {r.status_code} for {current}", status=r.status_code, url=url, failure=_status_failure(r.status_code))
                    return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, hops=hops, max_bytes=max_bytes)
            except (NotModified, FetchError):
                raise
            except (httpx.TimeoutException, httpx.TransportError) as exc:
                if attempt < retries:
                    attempt += 1
                    await asyncio.sleep(1.5 * (2 ** attempt))
                    continue
                raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url, failure=classify_exception(exc)) from exc
            finally:
                sem.release()

    async def _read(self, r: httpx.Response, *, url: str, final_url: str, t0: float, min_bytes: int, hops: int, max_bytes: int | None) -> FetchResult:
        limit = max_bytes or settings.max_body_bytes
        declared = r.headers.get("content-length")
        if declared and declared.isdigit() and int(declared) > limit:
            raise FetchError(f"declared size {declared} exceeds {limit}", status=r.status_code, url=url, failure=FailureClass.TOO_LARGE)
        chunks: list[bytes] = []
        size = 0
        async for chunk in r.aiter_bytes():
            size += len(chunk)
            if size > limit:
                raise FetchError(f"body exceeds {limit} bytes", status=r.status_code, url=url, failure=FailureClass.TOO_LARGE)
            chunks.append(chunk)
        content = b"".join(chunks)
        enc = (r.headers.get("content-encoding") or "").lower()
        if enc and enc not in ("identity",) and enc not in ("gzip", "deflate", "br", "zstd"):
            raise FetchError(f"unsupported content-encoding {enc!r}", status=r.status_code, url=url, failure=FailureClass.PARSING)
        if content and looks_binary(content, r.headers.get("content-type", "")):
            raise FetchError(f"body is not text (content-encoding={enc or 'none'}, type={r.headers.get('content-type', '?')[:40]})",
                             status=r.status_code, url=url, failure=FailureClass.PARSING)
        if len(content) < min_bytes:
            raise FetchError(f"short body ({len(content)} bytes)", status=r.status_code, url=url, failure=FailureClass.PARSING)
        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="http", redirects=hops)
        if res.is_html and looks_like_challenge(content):
            raise BlockedError(f"anti-bot challenge page at {final_url}", status=r.status_code, url=url, failure=FailureClass.BOT_CHALLENGE)
        return res

    # ------------------------------------------------------------------------------------------ optional browser mode (spec §13 B)
    async def get_rendered(self, url: str, *, wait_ms: int = 1500) -> FetchResult:
        """Headless Chromium render — pooled by `settings.browser_concurrency`; only when a connector declares fetch_mode=browser."""
        if not settings.browser_enabled:
            raise FetchError("browser mode disabled", url=url, failure=FailureClass.UNKNOWN)
        await validate_destination_async(url)
        async with _browser_slots():
            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:
                    ctx = await browser.new_context(user_agent=settings.user_agent, java_script_enabled=True)
                    await ctx.route("**/*", lambda route: route.abort() if route.request.resource_type in ("image", "media", "font") else route.continue_())
                    page = await ctx.new_page()
                    resp = await page.goto(url, wait_until="domcontentloaded", timeout=int(self.timeout_s * 1000))
                    await page.wait_for_timeout(wait_ms)
                    html = await page.content()
                    status = resp.status if resp else 200
                    final = page.url
                finally:
                    await browser.close()
            return FetchResult(url=url, final_url=final, 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")


_browser_sem: asyncio.Semaphore | None = None


def _browser_slots() -> asyncio.Semaphore:
    global _browser_sem
    if _browser_sem is None:
        _browser_sem = asyncio.Semaphore(settings.browser_concurrency)
    return _browser_sem


def looks_like_challenge(content: bytes) -> bool:
    if len(content) > 80_000:
        return False
    head = content[:20000].lower()
    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"_pxappid", b"datadome", b"enable javascript and cookies to continue", b"request unsuccessful. incapsula",
               b"bot detection", b"are you a robot")
    return sum(m in head for m in markers) >= 2


def file_result(path: str, *, url: str, content_type: str = "text/html") -> FetchResult:
    """Wrap a fixture file as a FetchResult (tests, `catlas run-sensor --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")


__all__ = [
    "BlockedDestination",
    "BlockedError",
    "FetchError",
    "FetchResult",
    "Fetcher",
    "NotModified",
    "classify_exception",
    "decode_text",
    "file_result",
    "governor",
    "looks_binary",
    "looks_like_challenge",
    "robots",
    "text_quality",
    "validate_destination",
    "validate_destination_async",
]