autoChatGPTPlus-Paypal/geo_fingerprint.py

"""根据代理出口的地理位置自动生成匹配的浏览器指纹参数。

支持通过不同的探测域名拿到不同分流链路上的真实出口信息，
例如 OpenAI 走日本节点、PayPal 走美国节点时，即使都共用同一个
本地 Clash 端口，也能分别得到对应链路的 locale / timezone。
"""
from __future__ import annotations

import json
import urllib.request
import urllib.error
from typing import Callable
from dataclasses import dataclass

# 国家代码 → (locale, Accept-Language header)
_COUNTRY_LOCALE_MAP: dict[str, tuple[str, str]] = {
    "JP": ("ja-JP", "ja-JP,ja;q=0.9,en;q=0.5"),
    "US": ("en-US", "en-US,en;q=0.9"),
    "GB": ("en-GB", "en-GB,en;q=0.9"),
    "CA": ("en-CA", "en-CA,en;q=0.9,fr;q=0.5"),
    "AU": ("en-AU", "en-AU,en;q=0.9"),
    "DE": ("de-DE", "de-DE,de;q=0.9,en;q=0.5"),
    "FR": ("fr-FR", "fr-FR,fr;q=0.9,en;q=0.5"),
    "KR": ("ko-KR", "ko-KR,ko;q=0.9,en;q=0.5"),
    "TW": ("zh-TW", "zh-TW,zh;q=0.9,en;q=0.5"),
    "HK": ("zh-HK", "zh-HK,zh;q=0.9,en;q=0.5"),
    "SG": ("en-SG", "en-SG,en;q=0.9,zh;q=0.5"),
    "IN": ("en-IN", "en-IN,en;q=0.9,hi;q=0.5"),
    "BR": ("pt-BR", "pt-BR,pt;q=0.9,en;q=0.5"),
    "MX": ("es-MX", "es-MX,es;q=0.9,en;q=0.5"),
    "ES": ("es-ES", "es-ES,es;q=0.9,en;q=0.5"),
    "IT": ("it-IT", "it-IT,it;q=0.9,en;q=0.5"),
    "NL": ("nl-NL", "nl-NL,nl;q=0.9,en;q=0.5"),
    "RU": ("ru-RU", "ru-RU,ru;q=0.9,en;q=0.5"),
    "TR": ("tr-TR", "tr-TR,tr;q=0.9,en;q=0.5"),
    "TH": ("th-TH", "th-TH,th;q=0.9,en;q=0.5"),
    "VN": ("vi-VN", "vi-VN,vi;q=0.9,en;q=0.5"),
    "PH": ("en-PH", "en-PH,en;q=0.9,fil;q=0.5"),
    "ID": ("id-ID", "id-ID,id;q=0.9,en;q=0.5"),
    "MY": ("ms-MY", "ms-MY,ms;q=0.9,en;q=0.5"),
    "CN": ("zh-CN", "zh-CN,zh;q=0.9,en;q=0.5"),
}

_DEFAULT_LOCALE = "en-US"
_DEFAULT_ACCEPT_LANG = "en-US,en;q=0.9"
_DEFAULT_TIMEZONE = "America/New_York"
_DEFAULT_PROBE_URL = "http://ip-api.com/json/?fields=status,countryCode,country,city,timezone,query"

# 通过 Clash 规则把这两个探测域名分别固定到日本 / 美国出口，
# 从而为不同业务链路生成匹配的浏览器指纹。
OPENAI_GEO_PROBE_URL = "https://ipinfo.io/json"
PAYPAL_GEO_PROBE_URL = "https://api.ip.sb/geoip"


@dataclass
class GeoFingerprint:
    country_code: str
    locale: str
    timezone_id: str
    accept_language: str
    ip: str = ""
    city: str = ""


def detect_geo_fingerprint(
    proxy_url: str = "",
    *,
    log: Callable[[str], None] = print,
    timeout: int = 10,
    probe_url: str = _DEFAULT_PROBE_URL,
) -> GeoFingerprint:
    """通过代理请求探测接口检测出口 IP 的地理信息，返回匹配的浏览器指纹参数。

    proxy_url 为空时直连检测。检测失败时回退到美国默认值。
    """
    result = _query_geo_payload(proxy_url, probe_url=probe_url, log=log, timeout=timeout)
    if not result:
        log(f"[geo] IP 检测失败 probe={probe_url}，使用默认指纹（美国）")
        return GeoFingerprint(
            country_code="US",
            locale=_DEFAULT_LOCALE,
            timezone_id=_DEFAULT_TIMEZONE,
            accept_language=_DEFAULT_ACCEPT_LANG,
        )

    country = _extract_country_code(result)
    timezone = str(result.get("timezone") or "")
    ip = str(result.get("query") or result.get("ip") or "")
    city = str(result.get("city") or "")

    locale_info = _COUNTRY_LOCALE_MAP.get(country, (_DEFAULT_LOCALE, _DEFAULT_ACCEPT_LANG))
    locale_str, accept_lang = locale_info

    tz = timezone if timezone else _DEFAULT_TIMEZONE

    fp = GeoFingerprint(
        country_code=country,
        locale=locale_str,
        timezone_id=tz,
        accept_language=accept_lang,
        ip=ip,
        city=city,
    )
    log(f"[geo] probe={probe_url} IP={ip} 国家={country} 城市={city} → locale={locale_str} timezone={tz}")
    return fp


def detect_openai_geo_fingerprint(
    proxy_url: str = "",
    *,
    log: Callable[[str], None] = print,
    timeout: int = 10,
) -> GeoFingerprint:
    """为 OpenAI 链路生成指纹。"""
    return detect_geo_fingerprint(
        proxy_url,
        log=log,
        timeout=timeout,
        probe_url=OPENAI_GEO_PROBE_URL,
    )


def detect_paypal_geo_fingerprint(
    proxy_url: str = "",
    *,
    log: Callable[[str], None] = print,
    timeout: int = 10,
) -> GeoFingerprint:
    """为 PayPal / Stripe 链路生成指纹。"""
    return detect_geo_fingerprint(
        proxy_url,
        log=log,
        timeout=timeout,
        probe_url=PAYPAL_GEO_PROBE_URL,
    )


def _extract_country_code(result: dict) -> str:
    country = (
        result.get("countryCode")
        or result.get("country_code")
        or result.get("country")
        or "US"
    )
    country_str = str(country).strip().upper()
    if len(country_str) == 2:
        return country_str
    return "US"


def _query_geo_payload(
    proxy_url: str,
    *,
    probe_url: str,
    log: Callable[[str], None],
    timeout: int = 10,
) -> dict | None:
    """调用探测接口获取地理信息。"""
    req = urllib.request.Request(probe_url, method="GET", headers={
        "Accept": "application/json",
        "User-Agent": "Mozilla/5.0",
    })

    opener = urllib.request.build_opener()
    if proxy_url and proxy_url.strip():
        proxy_handler = urllib.request.ProxyHandler({
            "http": proxy_url.strip(),
            "https": proxy_url.strip(),
        })
        opener = urllib.request.build_opener(proxy_handler)

    try:
        with opener.open(req, timeout=timeout) as resp:
            text = resp.read().decode("utf-8", errors="replace")
            data = json.loads(text)
            if not isinstance(data, dict):
                log(f"[geo] 探测接口返回非对象 probe={probe_url}: {text[:200]}")
                return None
            if data.get("status") == "fail":
                log(f"[geo] 探测接口返回失败 probe={probe_url}: {text[:200]}")
                return None
            if _extract_country_code(data):
                return data
            log(f"[geo] 探测接口缺少国家信息 probe={probe_url}: {text[:200]}")
            return None
    except Exception as exc:
        log(f"[geo] 探测接口请求失败 probe={probe_url}: {exc!r}")
        return None