omni-parse/omni/net.py

"""Сетевой слой: httpx -> curl_cffi (TLS Chrome) -> прокси. Троттлинг по хосту."""
import random
import threading
import time
import urllib.parse

from . import config

_cfg = config.load()
_lock = threading.Lock()
_last = {}          # host -> ts

BLOCK_MARKS = ("just a moment", "cf-browser-verification", "attention required",
               "ddos-guard", "checking your browser", "enable javascript and cookies")


class Resp:
    __slots__ = ("url", "status", "text", "mode", "error", "headers", "elapsed")

    def __init__(self, url, status=0, text="", mode="", error="", headers=None, elapsed=0.0):
        self.url, self.status, self.text = url, status, text
        self.mode, self.error, self.headers, self.elapsed = mode, error, headers or {}, elapsed

    @property
    def ok(self):
        return self.status == 200 and len(self.text) > 50

    @property
    def blocked(self):
        if self.status in (403, 429, 503):
            return True
        low = self.text[:4000].lower()
        return any(m in low for m in BLOCK_MARKS)

    def json(self):
        try:
            import json
            return json.loads(self.text)
        except Exception:
            return None

    def __repr__(self):
        return f"<Resp {self.status} {self.mode} {len(self.text)}b {self.url[:60]}>"


def _throttle(host):
    """Слот резервируется под локом ДО сна: иначе N потоков проходят одновременно.
    Замер до фикса: 24 вызова на хост при 0.3с == 0.9с вместо 6.9с."""
    delay = float(_cfg.get("per_host_delay", 0.6))
    with _lock:
        now = time.time()
        prev = _last.get(host, 0)
        wait = max(0.0, delay - (now - prev))
        _last[host] = now + wait
    if wait > 0:
        time.sleep(wait)


def _pick_proxy():
    ps = config.proxies()
    return random.choice(ps) if ps else None


def _try_httpx(url, headers, timeout, proxy):
    try:
        import httpx
        kw = {"headers": headers, "timeout": timeout, "follow_redirects": True}
        if proxy:
            kw["proxy"] = proxy
        with httpx.Client(**kw) as c:
            r = c.get(url)
            return Resp(url, r.status_code, r.text[: _cfg.get("max_bytes", 4_000_000)],
                        "proxy+httpx" if proxy else "httpx", headers=dict(r.headers))
    except Exception as e:
        return Resp(url, 0, "", "httpx", f"{type(e).__name__}: {e}"[:160])


def _try_curl(url, headers, timeout, proxy, impersonate="chrome131"):
    try:
        from curl_cffi import requests as cr
        kw = {"headers": headers, "timeout": timeout, "impersonate": impersonate, "allow_redirects": True}
        if proxy:
            kw["proxies"] = {"http": proxy, "https": proxy}
        r = cr.get(url, **kw)
        return Resp(url, r.status_code, (r.text or "")[: _cfg.get("max_bytes", 4_000_000)],
                    "proxy+curl" if proxy else "curl", headers=dict(r.headers))
    except Exception as e:
        return Resp(url, 0, "", "curl", f"{type(e).__name__}: {e}"[:160])


def post_json(url, payload, headers=None, timeout=None):
    """POST JSON: httpx, при блоке/ошибке повтор через curl_cffi."""
    import json
    timeout = timeout or _cfg.get("timeout", 25)
    host = urllib.parse.urlsplit(url).netloc
    _throttle(host)
    h = {"User-Agent": _cfg["ua"], "Content-Type": "application/json", "Accept": "application/json"}
    h.update(headers or {})
    body = json.dumps(payload).encode()
    try:
        import httpx
        with httpx.Client(timeout=timeout, follow_redirects=True) as c:
            r = c.post(url, content=body, headers=h)
            txt = r.text[:_cfg.get("max_bytes", 4_000_000)]
            if r.status_code == 200:
                return Resp(url, r.status_code, txt, "httpx-post", headers=dict(r.headers))
    except Exception as e:
        txt, code = "", 0
        err = f"{type(e).__name__}: {e}"[:160]
        _ = err
    try:
        from curl_cffi import requests as cr
        r = cr.post(url, data=body, headers=h, timeout=timeout, impersonate="chrome131")
        return Resp(url, r.status_code, (r.text or "")[:_cfg.get("max_bytes", 4_000_000)], "curl-post",
                    headers=dict(r.headers))
    except Exception as e:
        return Resp(url, 0, "", "curl-post", f"{type(e).__name__}: {e}"[:160])


def get(url, headers=None, timeout=None, json_mode=False, proxy=None):
    """Лестница: httpx -> curl_cffi -> curl_cffi с прокси. Возвращает первый осмысленный ответ.

    proxy=... — принудительный прокси для конкретного источника (напр. CF-форумы через DE).
    """
    timeout = timeout or _cfg.get("timeout", 25)
    host = urllib.parse.urlsplit(url).netloc
    _throttle(host)
    h = {"User-Agent": _cfg["ua"], "Accept": "text/html,application/json;q=0.9,*/*;q=0.8",
         "Accept-Language": "ru,en;q=0.8"}
    if json_mode:
        h["Accept"] = "application/json"
        h["Content-Type"] = "application/json"
    h.update(headers or {})
    ck = config.cookies_for(host)
    if ck:
        h["Cookie"] = ck

    if proxy:
        forced = _try_curl(url, h, timeout, proxy)
        if forced.ok and not forced.blocked:
            return forced
        alt = _try_httpx(url, h, timeout, proxy)
        return alt if alt.status == 200 else forced

    first = _try_httpx(url, h, timeout, None)
    if first.ok and not first.blocked:
        return first
    second = _try_curl(url, h, timeout, None)
    if second.ok and not second.blocked:
        return second
    proxy = _pick_proxy()
    if proxy:
        third = _try_curl(url, h, timeout, proxy)
        if third.status == 200:
            return third
        return third
    return second if second.status else first