chore: попытка сделать так же как у scrapling

This commit is contained in:
protokey committed 2026-09-23 14:40:50 +04:00
1 parent 8a464391a2
commit 8f8bc62ae7
5 files changed
+215 -87

No files matched your search

+8
View File
@@ -37,6 +37,14 @@ def load_or_create_fingerprint(user_data_dir: Path, **generator_kwargs) -> Finge
return _fingerprint_from_dict(json.loads(fp_file.read_text()))
generator_kwargs.setdefault("browser", ["chrome"])
# strict=True: без него browserforge при отсутствии точного совпадения в датасете
# тихо ослабляет даже нижнюю границу screen (min_width/min_height) — на практике
# так может прилететь мобильный экран под десктопный viewport (screen.availWidth
# меньше самого окна браузера — физически невозможно и легко детектится), а в
# части случаев цикл релаксации виснет на неопределённое время вместо быстрого
# ответа. Со strict=True generate() либо отдаёт честно подходящий сэмпл, либо
# сразу кидает ValueError — и то и другое быстро (~0.1с)
generator_kwargs.setdefault("strict", True)
fingerprint = FingerprintGenerator().generate(**generator_kwargs)
fp_file.write_text(fingerprint.dumps())
return fingerprint
+69
View File
@@ -0,0 +1,69 @@
"""
Мелкие хелперы навигации, портированные из scrapling.engines.toolbelt.navigation
и scrapling.engines.toolbelt.proxy_rotation — адаптированы под patchright/async.
В отличие от полного Session/PagePool из scrapling.engines._browsers._base, тут
браузер запускается и закрывается на каждый запрос (см. main.py), поэтому
переиспользовать нечего — берём только то, что относится к одной навигации:
блокировку ресурсов/доменов на странице и детект прокси-ошибок для retry.
"""
from typing import Callable, Optional
from urllib.parse import urlparse
from patchright.async_api import Route
# ресурсы, которые можно дропать ради скорости — портировано из
# scrapling.engines.constants.EXTRA_RESOURCES
EXTRA_RESOURCES = {
"font", "image", "media", "beacon", "object",
"imageset", "texttrack", "websocket", "csp_report", "stylesheet",
}
# признаки прокси-ошибки в тексте исключения — портировано из
# scrapling.engines.toolbelt.proxy_rotation._PROXY_ERROR_INDICATORS
_PROXY_ERROR_INDICATORS = (
"net::err_proxy", "net::err_tunnel", "connection refused",
"connection reset", "connection timed out", "failed to connect",
"could not resolve proxy",
)
def is_proxy_error(error: Exception) -> bool:
"""Похоже ли исключение на сбой прокси (а не на обычную ошибку навигации)."""
msg = str(error).lower()
return any(indicator in msg for indicator in _PROXY_ERROR_INDICATORS)
def _is_domain_blocked(hostname: str, domains: frozenset) -> bool:
"""Матчинг хоста и его родительских доменов за O(1) на каждый уровень —
портировано из scrapling.engines.toolbelt.navigation._is_domain_blocked."""
if hostname in domains:
return True
idx = hostname.find(".")
while idx != -1:
suffix = hostname[idx + 1:]
if "." in suffix and suffix in domains:
return True
idx = hostname.find(".", idx + 1)
return False
def create_intercept_handler(disable_resources: bool, blocked_domains: Optional[set] = None) -> Callable:
"""Обработчик route, блокирующий типы ресурсов и/или домены — портировано из
scrapling.engines.toolbelt.navigation.create_async_intercept_handler."""
disabled_resources = EXTRA_RESOURCES if disable_resources else set()
domains = frozenset(blocked_domains) if blocked_domains else frozenset()
async def handler(route: Route) -> None:
if route.request.resource_type in disabled_resources:
await route.abort()
elif domains:
hostname = urlparse(route.request.url).hostname or ""
if _is_domain_blocked(hostname, domains):
await route.abort()
else:
await route.continue_()
else:
await route.continue_()
return handler
+9
View File
@@ -35,3 +35,12 @@ class SolveRequest(BaseModel):
captcha_detect_locator: str | None = None
captcha_success_locator: str | None = None
captcha_type: str | None = None
# ниже — опции навигации, портированные из scrapling.engines._browsers (StealthySession.fetch),
# все выключены/нейтральны по умолчанию, чтобы поведение для текущих вызывающих не менялось
disable_resources: bool = False
blocked_domains: list[str] | None = None
retries: int = 1
retry_delay: float = 1.0
wait: int = 0
google_search: bool = False