Files
omni-parse/omni/net.py
Vlad Gothbreach c3eec76775
Some checks failed
ci / test (3.10) (push) Has been cancelled
ci / test (3.11) (push) Has been cancelled
ci / test (3.12) (push) Has been cancelled
prod: чистая лента (дедуп перепечаток, анти-шум в score, категорийный вес), счётчик живых/мёртвых источников, таймауты net
2026-10-27 22:29:47 +03:00

254 lines
8.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Сетевой слой: 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
# ---------- условный GET: ETag / If-Modified-Since ----------
_DATA = __import__("pathlib").Path(__file__).resolve().parents[1] / "data"
_COND = _DATA / "http_cond.json"
_BODIES = _DATA / "http_cache"
_COND_DB = None
_cond_lock = threading.Lock()
def _cond_load():
global _COND_DB
if _COND_DB is None:
try:
import json as _j
_COND_DB = _j.loads(_COND.read_text(encoding="utf-8"))
except Exception:
_COND_DB = {}
return _COND_DB
def _cond_save():
try:
import json as _j
with _cond_lock:
_BODIES.mkdir(parents=True, exist_ok=True)
_COND.write_text(_j.dumps(_COND_DB, ensure_ascii=False), encoding="utf-8")
except Exception:
pass
def _body_path(key):
import hashlib
return _BODIES / (hashlib.sha1(key.encode("utf-8", "replace")).hexdigest() + ".body")
def _cond_ask(key, h):
rec = _cond_load().get(key) or {}
if rec.get("etag"):
h["If-None-Match"] = rec["etag"]
if rec.get("lm"):
h["If-Modified-Since"] = rec["lm"]
return bool(rec.get("etag") or rec.get("lm"))
def _cond_store(key, resp):
if resp.status != 200 or not resp.text or len(resp.text) > 1_500_000:
return
hdr = {str(k).lower(): v for k, v in (resp.headers or {}).items()}
etag, lm = hdr.get("etag"), hdr.get("last-modified")
if not etag and not lm:
return
try:
_BODIES.mkdir(parents=True, exist_ok=True)
p = _body_path(key)
p.write_text(resp.text, encoding="utf-8")
except Exception:
return
_cond_load()[key] = {"etag": etag, "lm": lm, "body": p.name,
"ts": int(time.time()), "size": len(resp.text)}
_cond_save()
def _cond_hit(key):
rec = _cond_load().get(key)
if not rec or not rec.get("body"):
return None
try:
txt = (_BODIES / rec["body"]).read_text(encoding="utf-8")
except Exception:
return None
return Resp(key.split("|")[0], 200, txt, "cache-304", headers={"x-omni-cache": "304"})
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, _min=10):
timeout = max(_min, min(timeout, 12))
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"):
timeout = max(timeout, 8)
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 с прокси + условный GET (ETag/304).
proxy=... — принудительный прокси для конкретного источника (напр. CF-форумы через DE).
Повторный запрос неизменного документа отдаётся из data/http_cache (304), это экономит
и трафик, и время прогона (RSS/HTML отдают ETag/Last-Modified почти всегда).
"""
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
key = url + ("|" + proxy if proxy else "")
conditional = _cond_ask(key, h)
def finish(resp):
if conditional and resp.status in (304,):
hit = _cond_hit(key)
if hit:
return hit
if resp.status == 200:
_cond_store(key, resp)
return resp
if proxy:
forced = _try_curl(url, h, timeout, proxy)
if forced.ok and not forced.blocked:
return finish(forced)
alt = _try_httpx(url, h, timeout, proxy)
return finish(alt if alt.status == 200 else forced)
first = _try_httpx(url, h, timeout, None)
if conditional and first.status == 304:
hit = _cond_hit(key)
if hit:
return hit
if first.ok and not first.blocked:
return finish(first)
second = _try_curl(url, h, timeout, None)
if conditional and second.status == 304:
hit = _cond_hit(key)
if hit:
return hit
if second.ok and not second.blocked:
return finish(second)
proxy = _pick_proxy()
if proxy:
third = _try_curl(url, h, timeout, proxy)
if third.status == 200:
return finish(third)
return finish(third)
return finish(second if second.status else first)