137 lines
6.3 KiB
Python
137 lines
6.3 KiB
Python
"""
|
||
Резолвинг геолокации/таймзоны исходящего 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
|