robotex/src/engine/geoip.py

"""
Резолвинг геолокации/таймзоны исходящего 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 patchright.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 — не кэшируем навечно
DEFAULT_LOCALE = "en-US"

# Страна exit-IP -> основная локаль. Раньше это (как и timezone/geolocation) под капотом
# делал Camoufox(geoip=True); теперь резолвим сами и прокидываем в browserforge (locale
# согласованно выставляет navigator.language/languages и Accept-Language header) и в
# launch-флаги Chromium --lang=/--accept-lang= (engine/browser_launch.py). Не более чем
# точечная таблица под реально используемые прокси-гео — не претендует на полноту ISO-3166.
_COUNTRY_LOCALE: dict[str, str] = {
	"RU": "ru-RU", "BY": "ru-BY", "KZ": "ru-KZ", "UA": "uk-UA",
	"US": "en-US", "GB": "en-GB", "CA": "en-CA", "AU": "en-AU", "IE": "en-IE", "IN": "en-IN",
	"DE": "de-DE", "AT": "de-AT", "CH": "de-CH",
	"FR": "fr-FR", "BE": "fr-BE",
	"ES": "es-ES", "MX": "es-MX", "AR": "es-AR",
	"IT": "it-IT", "PT": "pt-PT", "BR": "pt-BR", "NL": "nl-NL",
	"PL": "pl-PL", "CZ": "cs-CZ", "SK": "sk-SK", "RO": "ro-RO", "HU": "hu-HU",
	"SE": "sv-SE", "NO": "nb-NO", "FI": "fi-FI", "DK": "da-DK",
	"TR": "tr-TR", "GR": "el-GR",
	"CN": "zh-CN", "TW": "zh-TW", "JP": "ja-JP", "KR": "ko-KR",
}

_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", "locale"}
	по IP, который виден снаружи через прокси, или None, если резолв не удался и в кэше
	тоже ничего нет — в этом случае вызывающий код просто не переопределяет timezone/geolocation/locale.

	"locale" — основная локаль страны exit-IP (см. _COUNTRY_LOCALE), с фолбэком DEFAULT_LOCALE
	для неизвестных/отсутствующих кодов страны. Реальный человек с российским прокси не сидит
	с английской раскладкой браузера, поэтому locale — такой же производный от IP сигнал, как
	timezone/geolocation, а не независимое "предпочтение пользователя". Прокидывается вызывающим
	кодом в browserforge (согласованно выставляет navigator.language/languages и Accept-Language,
	см. engine/fingerprint.py) и в launch-флаги Chromium (engine/browser_launch.py)."""
	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

	country = data.get("country")  # "RU", "US", ...
	result = {
		"ip": data.get("ip"),
		"country": country,
		"timezone_id": data.get("timezone"),
		"geolocation": geolocation,
		"locale": _COUNTRY_LOCALE.get(country, DEFAULT_LOCALE),
	}

	async with _cache_lock:
		cache = _load_cache(cache_file)
		cache[key] = {"ts": time.time(), "data": result}
		_save_cache(cache_file, cache)

	return result