spb/company-atlas

1"""Fetch transport (spec §13–14, §109–118): httpx direct mode with conditional requests, per-domain rate limiting and concurrency2caps, robots.txt awareness, bounded redirects (each hop SSRF-checked), size limits, failure classification and an *optional*3headless-browser mode that is only used when a sensor's connector asks for it.45Crawler inputs are untrusted: every destination and every redirect hop is validated before a socket is opened.6"""7from __future__ import annotations89import asyncio10import hashlib11import ipaddress12import logging13import re14import socket15import time16from dataclasses import dataclass, field17from datetime import UTC, datetime18from typing import Self19from urllib.parse import urljoin, urlparse20from urllib.robotparser import RobotFileParser2122import httpx2324from companyatlas.config import settings25from companyatlas.taxonomy import FailureClass26from companyatlas.urls import registrable_domain2728log = logging.getLogger(__name__)2930TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504}31BLOCK_STATUS = {401, 403, 451, 999}32REDIRECT_STATUS = {301, 302, 303, 307, 308}3334# ---------------------------------------------------------------------------------------------- SSRF guard (spec §115–116)35_BLOCKED_SUFFIXES = (".local", ".internal", ".localhost", ".localdomain", ".lan", ".home", ".corp", ".intranet", ".maclustr.io", ".ts.net")36_BLOCKED_HOSTS = {"localhost", "metadata.google.internal", "metadata", "instance-data", "kubernetes.default", "kubernetes.default.svc"}37_BLOCKED_NETWORKS = [ipaddress.ip_network(n) for n in (38    "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",39    "198.18.0.0/15", "240.0.0.0/4", "::/128", "::1/128", "fc00::/7", "fe80::/10", "::ffff:0:0/96", "64:ff9b::/96",40)]414243class BlockedDestination(ValueError):44    """The URL points at a private, local or otherwise non-public destination."""454647def _ip_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:48    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:49        ip = ip.ipv4_mapped50    if isinstance(ip, ipaddress.IPv6Address) and ip in ipaddress.ip_network("64:ff9b::/96"):51        ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)          # NAT64: judge the embedded IPv452    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:53        return True54    return any(ip in net for net in _BLOCKED_NETWORKS)555657def _parse_ip(host: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:58    try:59        return ipaddress.ip_address(host.strip("[]").split("%")[0])60    except ValueError:61        return None626364def _resolve(host: str, port: int) -> list[str]:65    try:66        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)67    except socket.gaierror:68        return []69    return sorted({info[4][0] for info in infos})707172def validate_destination(url: str, *, resolved_ips: list[str] | None = None) -> None:73    p = urlparse(url.strip())74    if p.scheme.lower() not in ("http", "https"):75        raise BlockedDestination(f"scheme {p.scheme!r}")76    host = (p.hostname or "").strip().lower().rstrip(".")77    if not host:78        raise BlockedDestination("no host")79    if p.username or p.password:80        raise BlockedDestination("credentials in URL")81    literal = _parse_ip(host)82    if literal is None and (host in _BLOCKED_HOSTS or host.endswith(_BLOCKED_SUFFIXES) or "." not in host):83        raise BlockedDestination(f"local host {host!r}")84    if literal is not None:85        if _ip_blocked(literal):86            raise BlockedDestination(f"non-public address {host}")87        return88    ips = resolved_ips if resolved_ips is not None else _resolve(host, p.port or (443 if p.scheme.lower() == "https" else 80))89    for raw in ips:90        ip = _parse_ip(raw)91        if ip is not None and _ip_blocked(ip):92            raise BlockedDestination(f"{host} resolves to non-public address {raw}")939495async def validate_destination_async(url: str) -> None:96    p = urlparse(url.strip())97    host = (p.hostname or "").strip().lower()98    validate_destination(url, resolved_ips=[])99    if host and _parse_ip(host) is None:100        ips = await asyncio.to_thread(_resolve, host, p.port or (443 if p.scheme.lower() == "https" else 80))101        validate_destination(url, resolved_ips=ips)102103104# ---------------------------------------------------------------------------------------------- text decoding / sanity105_META_CHARSET_RE = re.compile(rb"""<meta[^>]+charset\s*=\s*["']?\s*([a-zA-Z0-9_.:-]+)""", re.IGNORECASE)106_XML_ENC_RE = re.compile(rb"""^\s*<\?xml[^>]*encoding\s*=\s*["']([a-zA-Z0-9_.:-]+)["']""", re.IGNORECASE)107_CONTROL_RE = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")108109110def text_quality(text: str) -> float:111    """Share of the text that is NOT replacement characters / C0 control bytes (1.0 = clean). Empty text is 1.0."""112    if not text:113        return 1.0114    bad = text.count("\ufffd") + len(_CONTROL_RE.findall(text))115    return max(0.0, 1.0 - bad / len(text))116117118def declared_encodings(content: bytes, content_type: str) -> list[str]:119    out: list[str] = []120    if content.startswith(b"\xef\xbb\xbf"):121        out.append("utf-8-sig")122    elif content.startswith((b"\xff\xfe", b"\xfe\xff")):123        out.append("utf-16")124    ct = (content_type or "").lower()125    if "charset=" in ct:126        out.append(ct.split("charset=", 1)[1].split(";")[0].strip().strip('"'))127    head = content[:4096]128    m = _XML_ENC_RE.search(head) or _META_CHARSET_RE.search(head)129    if m:130        out.append(m.group(1).decode("ascii", "ignore"))131    seen: set[str] = set()132    result = []133    for e in out + ["utf-8", "cp1252"]:134        e = e.lower().replace("_", "-")135        if e in ("iso-8859-1", "latin-1", "latin1", "ascii", "us-ascii"):136            e = "cp1252"          # superset; what browsers actually do137        if e not in seen:138            seen.add(e)139            result.append(e)140    return result141142143def decode_text(content: bytes, content_type: str = "") -> str:144    best: tuple[float, str] | None = None145    for enc in declared_encodings(content, content_type):146        try:147            candidate = content.decode(enc, errors="replace")148        except LookupError:149            continue150        q = text_quality(candidate)151        if q >= 0.99:152            return candidate153        if best is None or q > best[0]:154            best = (q, candidate)155    return best[1] if best else content.decode("utf-8", errors="replace")156157158_TEXTUAL_TYPES = ("text/", "html", "xml", "json", "javascript", "rss", "atom", "csv")159160161def looks_binary(content: bytes, content_type: str) -> bool:162    """A body announced as text/HTML/XML/JSON whose bytes are not text (undecoded compression, wrong Content-Type)."""163    ct = (content_type or "").lower()164    if not any(t in ct for t in _TEXTUAL_TYPES) and ct:165        return False                      # PDFs, images… are judged by their connectors, not here166    sample = content[:8192]167    if not sample:168        return False169    if sample[:2] == b"\x1f\x8b" or sample[:4] == b"\x28\xb5\x2f\xfd":       # gzip / zstd magic left undecoded170        return True171    text_bytes = sum(1 for b in sample if 32 <= b < 127 or b in (9, 10, 13) or b >= 128)172    high = sum(1 for b in sample if b >= 128)173    controls = len(sample) - text_bytes174    if controls / len(sample) > 0.02:175        return True176    if high / len(sample) > 0.30:        # legitimate UTF-8 (CJK) has structure; verify it decodes cleanly177        return text_quality(sample.decode("utf-8", errors="replace")) < 0.90178    return False179180181# ---------------------------------------------------------------------------------------------- results / errors182183184class FetchError(Exception):185    def __init__(self, message: str, *, status: int | None = None, url: str = "", failure: FailureClass = FailureClass.UNKNOWN):186        super().__init__(message)187        self.status = status188        self.url = url189        self.failure = failure190191192class BlockedError(FetchError):193    """Access denied by the origin or by robots — never bypassed."""194195196class NotModified(Exception):197    """HTTP 304 — content unchanged since our stored validators."""198199    def __init__(self, duration_ms: int = 0):200        super().__init__("not modified")201        self.duration_ms = duration_ms202203204@dataclass205class FetchResult:206    url: str207    final_url: str208    status: int209    headers: dict[str, str]210    content: bytes211    content_type: str212    fetched_at: datetime213    duration_ms: int214    transport: str = "http"215    redirects: int = 0216    sha256: str = field(init=False)217218    def __post_init__(self) -> None:219        self.sha256 = hashlib.sha256(self.content).hexdigest()220221    @property222    def etag(self) -> str | None:223        return self.headers.get("etag")224225    @property226    def last_modified(self) -> str | None:227        return self.headers.get("last-modified")228229    @property230    def text(self) -> str:231        """Decoded body: BOM → HTTP charset → in-document declaration (<meta charset>, XML prolog) → UTF-8; if the chosen codec leaves232        more than 1 % replacement characters, the alternatives are tried and the cleanest decode wins (never garbage in, never silently)."""233        return decode_text(self.content, self.content_type)234235    @property236    def is_html(self) -> bool:237        head = self.content[:512].lstrip().lower()238        return "html" in self.content_type or head.startswith((b"<!doctype html", b"<html")) or b"<html" in head239240    @property241    def is_json(self) -> bool:242        return "json" in self.content_type or self.content[:1] in (b"{", b"[")243244    @property245    def is_xml(self) -> bool:246        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"))247248    def json(self):  # type: ignore[no-untyped-def]249        import orjson250251        return orjson.loads(self.content)252253254def classify_exception(exc: BaseException) -> FailureClass:255    if isinstance(exc, FetchError):256        return exc.failure257    if isinstance(exc, httpx.ConnectTimeout | httpx.ReadTimeout | httpx.WriteTimeout | httpx.PoolTimeout | asyncio.TimeoutError | TimeoutError):258        return FailureClass.TIMEOUT259    if isinstance(exc, httpx.RemoteProtocolError | httpx.ProtocolError) or exc.__class__.__name__ in ("ProtocolError", "StreamReset", "ConnectionTerminated"):260        return FailureClass.TIMEOUT            # transient transport-level protocol errors (h2 connection reuse, resets): retry policy, not a mystery261    if isinstance(exc, httpx.ConnectError):262        msg = str(exc).lower()263        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:264            return FailureClass.DNS265        return FailureClass.TIMEOUT266    return FailureClass.UNKNOWN267268269def _status_failure(status: int) -> FailureClass:270    if status == 429:271        return FailureClass.RATE_LIMIT272    if status in (404, 410):273        return FailureClass.PAGE_REMOVED274    if status in BLOCK_STATUS:275        return FailureClass.BOT_CHALLENGE if status in (403, 999) else FailureClass.HTTP_4XX276    if 400 <= status < 500:277        return FailureClass.HTTP_4XX278    return FailureClass.HTTP_5XX279280281# ---------------------------------------------------------------------------------------------- politeness primitives282283284class DomainGovernor:285    """Per-domain minimum spacing + concurrency cap (in-process). Cluster-wide fairness comes from Postgres domain budgets."""286287    def __init__(self) -> None:288        self._next: dict[str, float] = {}289        self._sem: dict[str, asyncio.Semaphore] = {}290        self._delay: dict[str, float] = {}291        self._lock = asyncio.Lock()292293    def set_crawl_delay(self, domain: str, seconds: float) -> None:294        self._delay[domain] = min(120.0, max(0.0, seconds))295296    async def acquire(self, domain: str, per_min: int) -> asyncio.Semaphore:297        sem = self._sem.get(domain)298        if sem is None:299            sem = self._sem[domain] = asyncio.Semaphore(settings.domain_max_concurrency)300        await sem.acquire()301        async with self._lock:302            spacing = max(60.0 / max(1, per_min), self._delay.get(domain, 0.0))303            now = time.monotonic()304            ready = self._next.get(domain, 0.0)305            wait = max(0.0, ready - now)306            self._next[domain] = max(now, ready) + spacing307        if wait > 0:308            await asyncio.sleep(wait)309        return sem310311312class RobotsCache:313    def __init__(self) -> None:314        self._cache: dict[str, tuple[float, RobotFileParser | None, float | None]] = {}315316    async def policy(self, client: httpx.AsyncClient, url: str) -> tuple[bool, float | None]:317        """(allowed, crawl_delay_seconds)."""318        if not settings.respect_robots:319            return True, None320        p = urlparse(url)321        key = f"{p.scheme}://{p.netloc}"322        cached = self._cache.get(key)323        if cached is None or cached[0] < time.monotonic():324            rp: RobotFileParser | None = RobotFileParser()325            delay: float | None = None326            try:327                r = await client.get(f"{key}/robots.txt", timeout=15, follow_redirects=False)328                if r.status_code == 200 and len(r.content) < 512 * 1024 and b"<html" not in r.content[:512].lower():329                    rp.parse(r.text.splitlines())  # type: ignore[union-attr]330                    agent = settings.user_agent.split("/")[0]331                    try:332                        d = rp.crawl_delay(agent) or rp.crawl_delay("*")  # type: ignore[union-attr]333                        delay = float(d) if d else None334                    except Exception:  # noqa: BLE001335                        delay = None336                else:337                    rp = None338            except Exception:  # noqa: BLE001339                rp = None340            self._cache[key] = (time.monotonic() + 12 * 3600, rp, delay)341            cached = self._cache[key]342        rp, delay = cached[1], cached[2]343        if rp is None:344            return True, None345        agent = settings.user_agent.split("/")[0]346        try:347            return (rp.can_fetch(agent, url) or rp.can_fetch("*", url)), delay348        except Exception:  # noqa: BLE001349            return True, delay350351352governor = DomainGovernor()353robots = RobotsCache()354355356# ---------------------------------------------------------------------------------------------- fetcher357358359class Fetcher:360    """Shared httpx client (one per worker process). `get()` raises NotModified / FetchError / BlockedError.361    HTTP/1.1 by default: h2 connection reuse against thousands of heterogeneous origins produced `ConnectionInputs.SEND_SETTINGS in state CLOSED`362    protocol errors (2026-09-13); HTTP/2 stays available per Fetcher (`http2=True`)."""363364    def __init__(self, *, timeout_s: float | None = None, headers: dict[str, str] | None = None, http2: bool = False,365                 max_connections: int | None = None):366        self.timeout_s = timeout_s or settings.http_timeout_s367        self.headers = {"User-Agent": settings.user_agent,368                        "Accept": "text/html,application/xhtml+xml,application/xml,application/json,application/rss+xml,text/*;q=0.9,*/*;q=0.7",369                        "Accept-Language": "en-US,en;q=0.9,fr;q=0.6,de;q=0.4,*;q=0.2",370                        **(headers or {})}371        self._client: httpx.AsyncClient | None = None372        self._http2 = http2373        self._max_connections = max_connections or max(16, settings.fetch_concurrency * 2)374375    async def __aenter__(self) -> Self:376        await self.open()377        return self378379    async def __aexit__(self, *exc: object) -> None:380        await self.close()381382    async def open(self) -> None:383        if self._client is None:384            self._client = httpx.AsyncClient(headers=self.headers, timeout=httpx.Timeout(self.timeout_s, connect=15), follow_redirects=False,385                                             http2=self._http2, limits=httpx.Limits(max_connections=self._max_connections,386                                                                                    max_keepalive_connections=self._max_connections // 2))387388    async def close(self) -> None:389        if self._client is not None:390            await self._client.aclose()391            self._client = None392393    @property394    def client(self) -> httpx.AsyncClient:395        if self._client is None:396            raise RuntimeError("Fetcher not opened — use `async with Fetcher() as f` or `await f.open()`")397        return self._client398399    async def get(self, url: str, *, etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1,400                  retries: int | None = None, accept: str | None = None, rate_per_min: int | None = None, respect_robots: bool = True,401                  max_bytes: int | None = None) -> FetchResult:402        return await self.request("GET", url, etag=etag, last_modified=last_modified, min_bytes=min_bytes, retries=retries, accept=accept,403                                  rate_per_min=rate_per_min, respect_robots=respect_robots, max_bytes=max_bytes)404405    async def post_json(self, url: str, payload: object, *, accept: str | None = "application/json", **kw: object) -> FetchResult:406        """POST a JSON body to a *public* endpoint that a public page itself calls (spec §13 mode C, e.g. Workday job search)."""407        return await self.request("POST", url, json=payload, accept=accept, **kw)  # type: ignore[arg-type]408409    async def request(self, method: str, url: str, *, json: object | None = None, headers: dict[str, str] | None = None,410                      etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1, retries: int | None = None,411                      accept: str | None = None, rate_per_min: int | None = None, respect_robots: bool = True,412                      max_bytes: int | None = None) -> FetchResult:413        """Same guarantees as `get()` (SSRF guard on every hop, robots, per-domain governor, size caps, failure classes) for any414        method. Non-GET requests are never retried on transport errors beyond the configured retries and never follow redirects415        across registrable domains with the body."""416        client = self.client417        method = method.upper()418        retries = settings.fetch_retries if retries is None else retries419        try:420            await validate_destination_async(url)421        except BlockedDestination as exc:422            raise FetchError(f"blocked destination: {exc}", url=url, failure=FailureClass.BLOCKED_DESTINATION) from exc423        domain = registrable_domain(url)424        if respect_robots:425            allowed, delay = await robots.policy(client, url)426            if delay:427                governor.set_crawl_delay(domain, delay)428            if not allowed:429                raise BlockedError(f"robots.txt disallows {url}", url=url, failure=FailureClass.ROBOTS)430        req_headers: dict[str, str] = dict(headers or {})431        if etag:432            req_headers["If-None-Match"] = etag433        if last_modified:434            req_headers["If-Modified-Since"] = last_modified435        if accept:436            req_headers["Accept"] = accept437        headers = req_headers438        current = url439        hops = 0440        attempt = 0441        while True:442            sem = await governor.acquire(registrable_domain(current), rate_per_min or settings.default_rate_per_min)443            t0 = time.perf_counter()444            try:445                async with client.stream(method, current, headers=headers, json=json) as r:446                    if r.status_code == 304:447                        raise NotModified(int((time.perf_counter() - t0) * 1000))448                    if r.status_code in REDIRECT_STATUS:449                        location = r.headers.get("location")450                        if not location:451                            raise FetchError(f"http {r.status_code} without Location", status=r.status_code, url=url, failure=FailureClass.REDIRECT)452                        hops += 1453                        if hops > settings.max_redirects:454                            raise FetchError(f"too many redirects (> {settings.max_redirects})", status=r.status_code, url=url, failure=FailureClass.REDIRECT)455                        nxt = urljoin(current, location)456                        try:457                            await validate_destination_async(nxt)458                        except BlockedDestination as exc:459                            raise FetchError(f"blocked redirect {current} → {nxt}: {exc}", status=r.status_code, url=url,460                                             failure=FailureClass.BLOCKED_DESTINATION) from exc461                        if registrable_domain(nxt) != registrable_domain(current):462                            headers.pop("If-None-Match", None)463                            headers.pop("If-Modified-Since", None)464                            if method != "GET":465                                raise FetchError(f"{method} redirected off-domain {current} → {nxt}", status=r.status_code, url=url,466                                                 failure=FailureClass.REDIRECT)467                        if r.status_code in (301, 302, 303) and method != "GET":468                            method, json = "GET", None      # per RFC 9110 user agents switch to GET469                        current = nxt470                        continue471                    if r.status_code in TRANSIENT_STATUS and attempt < retries:472                        retry_after = r.headers.get("retry-after")473                        delay = min(60.0, float(retry_after)) if retry_after and retry_after.isdigit() else 2.0 * (2 ** attempt)474                        attempt += 1475                        await asyncio.sleep(delay)476                        continue477                    if r.status_code in BLOCK_STATUS:478                        raise BlockedError(f"http {r.status_code} for {current}", status=r.status_code, url=url, failure=_status_failure(r.status_code))479                    if r.status_code >= 400:480                        raise FetchError(f"http {r.status_code} for {current}", status=r.status_code, url=url, failure=_status_failure(r.status_code))481                    return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, hops=hops, max_bytes=max_bytes)482            except (NotModified, FetchError):483                raise484            except (httpx.TimeoutException, httpx.TransportError) as exc:485                if attempt < retries:486                    attempt += 1487                    await asyncio.sleep(1.5 * (2 ** attempt))488                    continue489                raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url, failure=classify_exception(exc)) from exc490            finally:491                sem.release()492493    async def _read(self, r: httpx.Response, *, url: str, final_url: str, t0: float, min_bytes: int, hops: int, max_bytes: int | None) -> FetchResult:494        limit = max_bytes or settings.max_body_bytes495        declared = r.headers.get("content-length")496        if declared and declared.isdigit() and int(declared) > limit:497            raise FetchError(f"declared size {declared} exceeds {limit}", status=r.status_code, url=url, failure=FailureClass.TOO_LARGE)498        chunks: list[bytes] = []499        size = 0500        async for chunk in r.aiter_bytes():501            size += len(chunk)502            if size > limit:503                raise FetchError(f"body exceeds {limit} bytes", status=r.status_code, url=url, failure=FailureClass.TOO_LARGE)504            chunks.append(chunk)505        content = b"".join(chunks)506        enc = (r.headers.get("content-encoding") or "").lower()507        if enc and enc not in ("identity",) and enc not in ("gzip", "deflate", "br", "zstd"):508            raise FetchError(f"unsupported content-encoding {enc!r}", status=r.status_code, url=url, failure=FailureClass.PARSING)509        if content and looks_binary(content, r.headers.get("content-type", "")):510            raise FetchError(f"body is not text (content-encoding={enc or 'none'}, type={r.headers.get('content-type', '?')[:40]})",511                             status=r.status_code, url=url, failure=FailureClass.PARSING)512        if len(content) < min_bytes:513            raise FetchError(f"short body ({len(content)} bytes)", status=r.status_code, url=url, failure=FailureClass.PARSING)514        res = FetchResult(url=url, final_url=final_url, status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()},515                          content=content, content_type=r.headers.get("content-type", "").lower(), fetched_at=datetime.now(UTC),516                          duration_ms=int((time.perf_counter() - t0) * 1000), transport="http", redirects=hops)517        if res.is_html and looks_like_challenge(content):518            raise BlockedError(f"anti-bot challenge page at {final_url}", status=r.status_code, url=url, failure=FailureClass.BOT_CHALLENGE)519        return res520521    # ------------------------------------------------------------------------------------------ optional browser mode (spec §13 B)522    async def get_rendered(self, url: str, *, wait_ms: int = 1500) -> FetchResult:523        """Headless Chromium render — pooled by `settings.browser_concurrency`; only when a connector declares fetch_mode=browser."""524        if not settings.browser_enabled:525            raise FetchError("browser mode disabled", url=url, failure=FailureClass.UNKNOWN)526        await validate_destination_async(url)527        async with _browser_slots():528            from playwright.async_api import async_playwright  # optional dependency529530            t0 = time.perf_counter()531            async with async_playwright() as p:532                browser = await p.chromium.launch(headless=True)533                try:534                    ctx = await browser.new_context(user_agent=settings.user_agent, java_script_enabled=True)535                    await ctx.route("**/*", lambda route: route.abort() if route.request.resource_type in ("image", "media", "font") else route.continue_())536                    page = await ctx.new_page()537                    resp = await page.goto(url, wait_until="domcontentloaded", timeout=int(self.timeout_s * 1000))538                    await page.wait_for_timeout(wait_ms)539                    html = await page.content()540                    status = resp.status if resp else 200541                    final = page.url542                finally:543                    await browser.close()544            return FetchResult(url=url, final_url=final, status=status, headers={}, content=html.encode(), content_type="text/html; charset=utf-8",545                               fetched_at=datetime.now(UTC), duration_ms=int((time.perf_counter() - t0) * 1000), transport="browser")546547548_browser_sem: asyncio.Semaphore | None = None549550551def _browser_slots() -> asyncio.Semaphore:552    global _browser_sem553    if _browser_sem is None:554        _browser_sem = asyncio.Semaphore(settings.browser_concurrency)555    return _browser_sem556557558def looks_like_challenge(content: bytes) -> bool:559    if len(content) > 80_000:560        return False561    head = content[:20000].lower()562    markers = (b"just a moment", b"cf-chl-", b"challenge-platform", b"attention required", b"verify you are human", b"access denied",563               b"captcha", b"perimeterx", b"_pxappid", b"datadome", b"enable javascript and cookies to continue", b"request unsuccessful. incapsula",564               b"bot detection", b"are you a robot")565    return sum(m in head for m in markers) >= 2566567568def file_result(path: str, *, url: str, content_type: str = "text/html") -> FetchResult:569    """Wrap a fixture file as a FetchResult (tests, `catlas run-sensor --file`)."""570    with open(path, "rb") as fh:571        content = fh.read()572    return FetchResult(url=url, final_url=url, status=200, headers={}, content=content, content_type=content_type,573                       fetched_at=datetime.now(UTC), duration_ms=0, transport="file")574575576__all__ = [577    "BlockedDestination",578    "BlockedError",579    "FetchError",580    "FetchResult",581    "Fetcher",582    "NotModified",583    "classify_exception",584    "decode_text",585    "file_result",586    "governor",587    "looks_binary",588    "looks_like_challenge",589    "robots",590    "text_quality",591    "validate_destination",592    "validate_destination_async",593]594