feature: переезд на хром

src/engine/fingerprint.py

@@ -0,0 +1,67 @@
+"""
+Генерация и persist согласованного browser fingerprint на профиль (user_data_dir).
+
+Используем browserforge — ту же библиотеку, которую под капотом использовал Camoufox — но
+поверх голого Playwright Chromium: генерируем Fingerprint (navigator/screen/WebGL/canvas/шрифты/
+headers, всё согласовано друг с другом и с выбранным UA) и инжектим его через add_init_script.
+
+Важно: генерируем именно под browser=['chrome'] — раз движок у нас реальный Chromium, подделывать
+navigator под Firefox/Safari было бы хуже, чем не подделывать вообще (несоответствие между
+заявленным UA и реальными JS-фичами движка — сильный сигнал для антибота).
+"""
+import json
+from pathlib import Path
+
+from browserforge.fingerprints import Fingerprint, FingerprintGenerator, Screen
+from browserforge.fingerprints.generator import NavigatorFingerprint, ScreenFingerprint, VideoCard
+from browserforge.injectors.utils import InjectFunction, only_injectable_headers
+
+FINGERPRINT_FILE = "fingerprint.json"
+
+
+def _fingerprint_from_dict(data: dict) -> Fingerprint:
+	data = dict(data)
+	data["screen"] = ScreenFingerprint(**data["screen"])
+	data["navigator"] = NavigatorFingerprint(**data["navigator"])
+	if data.get("videoCard"):
+		data["videoCard"] = VideoCard(**data["videoCard"])
+	return Fingerprint(**data)
+
+
+def load_or_create_fingerprint(user_data_dir: Path, **generator_kwargs) -> Fingerprint:
+	"""Держит fingerprint стабильным для профиля: генерируется один раз при первом запросе
+	этой сессии (user_data_dir) и переиспользуется во всех последующих — так же, как реальный
+	человек не меняет видеокарту/шрифты/UA между визитами на сайт."""
+	fp_file = user_data_dir / FINGERPRINT_FILE
+	if fp_file.exists():
+		return _fingerprint_from_dict(json.loads(fp_file.read_text()))
+
+	generator_kwargs.setdefault("browser", ["chrome"])
+	fingerprint = FingerprintGenerator().generate(**generator_kwargs)
+	fp_file.write_text(fingerprint.dumps())
+	return fingerprint
+
+
+def context_options_for(fingerprint: Fingerprint, **overrides) -> dict:
+	"""Опции контекста (launch_persistent_context/new_context), согласованные с fingerprint'ом.
+	Headers сюда намеренно не кладём — их полный набор выставляет apply_fingerprint() через
+	set_extra_http_headers() уже после создания контекста, а он не мёржит, а заменяет,
+	так что выставлять их дважды бессмысленно."""
+	return {
+		"user_agent": fingerprint.navigator.userAgent,
+		"viewport": {
+			"width": fingerprint.screen.width,
+			"height": fingerprint.screen.height,
+			**overrides.pop("viewport", {}),
+		},
+		"device_scale_factor": fingerprint.screen.devicePixelRatio,
+		**overrides,
+	}
+
+
+async def apply_fingerprint(context, fingerprint: Fingerprint, browser_name: str = "chromium"):
+	"""Довешивает на уже созданный контекст заголовки и init-script с подделкой
+	navigator/screen/WebGL/codecs/battery. Вызывать сразу после launch_persistent_context/new_context,
+	до первого page.goto()."""
+	await context.set_extra_http_headers(only_injectable_headers(fingerprint.headers, browser_name))
+	await context.add_init_script(InjectFunction(fingerprint))

src/engine/geoip.py

@@ -0,0 +1,113 @@
+"""
+Резолвинг геолокации/таймзоны исходящего IP через сам прокси (не локальный IP), с кэшем в файле.
+
+Раньше это делал Camoufox (geoip=True) под капотом. Смысл: если браузер репортит IP из
+одной страны/города (через прокси), а navigator.timezone/geolocation — из другой, это
+явное рассогласование, которое легко ловится антиботом. Поэтому таймзону/координаты нужно
+брать по тому же IP, что увидит целевой сайт — то есть запрашивать через прокси, а не с
+локальной машины.
+
+Кэш нужен по двум причинам: у ipinfo.io есть квота на бесплатные запросы, и результат для
+одного и того же прокси (пока он не сменил exit-IP) не меняется от запроса к запросу — незачем
+дёргать внешний сервис на каждый /v1/solve.
+"""
+import asyncio
+import json
+import logging
+import time
+from pathlib import Path
+
+import httpx
+from playwright.async_api import ProxySettings
+
+log = logging.getLogger(__name__)
+
+IPINFO_URL = "https://ipinfo.io/json"
+DEFAULT_CACHE_FILE = Path(__file__).resolve().parent.parent.parent / "extra" / "geoip_cache.json"
+DEFAULT_TTL = 24 * 3600  # прокси может сменить exit-IP — не кэшируем навечно
+
+_cache_lock = asyncio.Lock()
+
+
+def proxy_settings_to_url(proxy: ProxySettings) -> str:
+	server = proxy["server"]
+	username = proxy.get("username")
+	password = proxy.get("password")
+	if not username:
+		return server
+	scheme, _, rest = server.partition("://")
+	return f"{scheme}://{username}:{password}@{rest}"
+
+
+def _load_cache(cache_file: Path) -> dict:
+	if not cache_file.exists():
+		return {}
+	try:
+		return json.loads(cache_file.read_text())
+	except (json.JSONDecodeError, OSError):
+		return {}
+
+
+def _save_cache(cache_file: Path, cache: dict) -> None:
+	cache_file.parent.mkdir(parents=True, exist_ok=True)
+	cache_file.write_text(json.dumps(cache, indent=2, ensure_ascii=False))
+
+
+async def resolve_geo(
+	proxy: ProxySettings,
+	timeout: float = 10.0,
+	cache_file: Path = DEFAULT_CACHE_FILE,
+	ttl: float = DEFAULT_TTL,
+) -> dict | None:
+	"""Возвращает {"timezone_id", "geolocation": {"latitude", "longitude"}, "country", "ip"}
+	по IP, который виден снаружи через прокси, или None, если резолв не удался и в кэше
+	тоже ничего нет — в этом случае вызывающий код просто не переопределяет timezone/geolocation.
+
+	Язык (locale) сюда намеренно не подмешиваем: Accept-Language — это предпочтение пользователя,
+	а не производная от геолокации IP, и его уже согласованно генерирует fingerprint (см.
+	engine/fingerprint.py). Подменять его на "en-<COUNTRY>" было бы отдельным, никак не
+	связанным с реальным поведением допущением."""
+	key = proxy_settings_to_url(proxy)
+
+	async with _cache_lock:
+		cache = _load_cache(cache_file)
+		entry = cache.get(key)
+		if entry and (time.time() - entry["ts"]) < ttl:
+			return entry["data"]
+
+	try:
+		async with httpx.AsyncClient(proxy=key, timeout=timeout) as client:
+			resp = await client.get(IPINFO_URL)
+			resp.raise_for_status()
+			data = resp.json()
+	except Exception as e:
+		log.warning("не удалось резолвнуть geoip через прокси: %r", e)
+		# сеть/квота ipinfo могли отказать разово — лучше отдать протухший кэш, чем ничего,
+		# если прокси тот же самый (скорее всего и IP тот же)
+		if entry:
+			log.info("используем протухший geoip-кэш из-за сбоя запроса")
+			return entry["data"]
+		return None
+
+	loc = data.get("loc")  # "lat,lon"
+	geolocation = None
+	if loc and "," in loc:
+		lat_str, lon_str = loc.split(",", 1)
+		try:
+			geolocation = {"latitude": float(lat_str), "longitude": float(lon_str)}
+		except ValueError:
+			geolocation = None
+
+	result = {
+		"ip": data.get("ip"),
+		"country": data.get("country"),  # "RU", "US", ...
+		"timezone_id": data.get("timezone"),
+		"geolocation": geolocation,
+	}
+
+	async with _cache_lock:
+		cache = _load_cache(cache_file)
+		cache[key] = {"ts": time.time(), "data": result}
+		_save_cache(cache_file, cache)
+
+	return result