""" 浏览器指纹生成与管理 - 生成逼真的浏览器指纹（UA、屏幕、Canvas、WebGL、TLS） - 指纹池轮换、冷却、黑名单机制 """ import asyncio import hashlib import random import time from datetime import datetime, timedelta from typing import Dict, Any, List, Optional, Set from loguru import logger from fake_useragent import UserAgent from src.core.config import get_config # ── 常量 ────────────────────────────────────────────── CHROME_VERSIONS = [ "120.0.6099.109", "121.0.6167.85", "122.0.6261.57", "123.0.6312.86", "124.0.6367.91", "125.0.6422.60", ] SCREEN_RESOLUTIONS = [ {"width": 1920, "height": 1080, "ratio": 1.0}, {"width": 1366, "height": 768, "ratio": 1.0}, {"width": 1536, "height": 864, "ratio": 1.25}, {"width": 2560, "height": 1440, "ratio": 1.5}, {"width": 1440, "height": 900, "ratio": 1.0}, {"width": 1680, "height": 1050, "ratio": 1.0}, {"width": 1280, "height": 800, "ratio": 1.0}, {"width": 3840, "height": 2160, "ratio": 2.0}, ] WEBGL_RENDERERS = [ "ANGLE (Intel, Intel(R) UHD Graphics 630 Direct3D11 vs_5_0 ps_5_0, D3D11)", "ANGLE (Intel, Intel(R) UHD Graphics Direct3D11 vs_5_0 ps_5_0, D3D11)", "ANGLE (AMD, AMD Radeon(TM) Graphics Direct3D11 vs_5_0 ps_5_0, D3D11)", "ANGLE (NVIDIA, NVIDIA GeForce GTX 1060 6GB Direct3D11 vs_5_0 ps_5_0, D3D11)", "ANGLE (NVIDIA, NVIDIA GeForce RTX 3060 Direct3D11 vs_5_0 ps_5_0, D3D11)", "ANGLE (Intel, Intel(R) Iris(R) Xe Graphics Direct3D11 vs_5_0 ps_5_0, D3D11)", ] FONTS_POOL = [ "Arial", "Arial Black", "Calibri", "Cambria", "Consolas", "Courier New", "Georgia", "Impact", "Segoe UI", "Tahoma", "Times New Roman", "Trebuchet MS", "Verdana", "Microsoft YaHei", "SimHei", "SimSun", "NSimSun", "FangSong", "KaiTi", ] SEC_CH_UA_TEMPLATES = { "chrome": '"Chromium";v="{major}", "Google Chrome";v="{major}", "Not-A.Brand";v="99"', "edge": '"Chromium";v="{major}", "Microsoft Edge";v="{major}", "Not-A.Brand";v="99"', } class FingerprintGenerator: """指纹生成器""" def __init__(self): try: self._ua = UserAgent(browsers=["chrome", "edge"], os=["windows", "macos"]) except Exception: self._ua = None def generate(self) -> Dict[str, Any]: browser = random.choice(["chrome", "edge"]) version = random.choice(CHROME_VERSIONS) major = version.split(".")[0] ua_string = self._build_ua(browser, version) screen = random.choice(SCREEN_RESOLUTIONS) fp_id = f"fp_{int(time.time())}_{random.randint(1000, 9999)}" return { "id": fp_id, "browser": browser, "version": version, "user_agent": ua_string, "screen": screen, "sec_ch_ua": SEC_CH_UA_TEMPLATES.get(browser, SEC_CH_UA_TEMPLATES["chrome"]).format(major=major), "http_headers": self._build_headers(ua_string, browser, major), "canvas_hash": hashlib.md5(f"canvas_{random.randint(100000, 999999)}".encode()).hexdigest()[:16], "webgl": { "vendor": "Google Inc. (Intel)", "renderer": random.choice(WEBGL_RENDERERS), }, "fonts": sorted(random.sample(FONTS_POOL, k=random.randint(10, len(FONTS_POOL)))), "timezone": "Asia/Shanghai", "language": "zh-CN", "created_at": datetime.now().isoformat(), "success_rate": 1.0, "usage_count": 0, "is_active": True, } def _build_ua(self, browser: str, version: str) -> str: os_strings = [ "Windows NT 10.0; Win64; x64", "Windows NT 10.0; Win64; x64", "Macintosh; Intel Mac OS X 10_15_7", ] os_str = random.choice(os_strings) major = version.split(".")[0] if browser == "edge": return ( f"Mozilla/5.0 ({os_str}) AppleWebKit/537.36 " f"(KHTML, like Gecko) Chrome/{version} Safari/537.36 Edg/{version}" ) return ( f"Mozilla/5.0 ({os_str}) AppleWebKit/537.36 " f"(KHTML, like Gecko) Chrome/{version} Safari/537.36" ) def _build_headers(self, ua: str, browser: str, major: str) -> Dict[str, str]: headers = { "User-Agent": ua, "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Referer": "https://www.dongqiudi.com/", "Origin": "https://www.dongqiudi.com", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site", "Sec-Ch-Ua": SEC_CH_UA_TEMPLATES.get(browser, SEC_CH_UA_TEMPLATES["chrome"]).format(major=major), "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Windows"', } return headers class FingerprintPool: """指纹池管理器""" def __init__(self): cfg = get_config().anti_detect.fingerprint self._pool_size = cfg.pool_size self._min_success_rate = cfg.min_success_rate self._cooldown = cfg.cooldown self._generator = FingerprintGenerator() self._pool: Dict[str, Dict[str, Any]] = {} self._blacklist: Set[str] = set() self._lock = asyncio.Lock() async def initialize(self, size: Optional[int] = None): size = size or self._pool_size async with self._lock: for _ in range(size): fp = self._generator.generate() self._pool[fp["id"]] = fp logger.info(f"指纹池初始化完成，共 {len(self._pool)} 个指纹") async def acquire(self) -> Dict[str, Any]: async with self._lock: now = datetime.now() candidates = [] for fp_id, fp in self._pool.items(): if fp_id in self._blacklist or not fp.get("is_active", True): continue if fp.get("success_rate", 1.0) < self._min_success_rate: fp["is_active"] = False continue last_used = fp.get("last_used") if last_used: last_dt = datetime.fromisoformat(last_used) if isinstance(last_used, str) else last_used if (now - last_dt).total_seconds() < self._cooldown: continue candidates.append(fp) if not candidates: logger.warning("指纹池耗尽，生成新指纹") fp = self._generator.generate() self._pool[fp["id"]] = fp candidates = [fp] candidates.sort(key=lambda x: (x.get("usage_count", 0), -x.get("success_rate", 1.0))) chosen = candidates[0] chosen["last_used"] = now.isoformat() chosen["usage_count"] = chosen.get("usage_count", 0) + 1 return chosen.copy() async def mark_success(self, fp_id: str): async with self._lock: fp = self._pool.get(fp_id) if not fp: return total = fp.get("usage_count", 1) fails = fp.get("fail_count", 0) fp["success_rate"] = 1.0 - (fails / total) if total > 0 else 1.0 async def mark_failure(self, fp_id: str, error_type: str = ""): async with self._lock: fp = self._pool.get(fp_id) if not fp: return fp["fail_count"] = fp.get("fail_count", 0) + 1 total = fp.get("usage_count", 1) fails = fp["fail_count"] fp["success_rate"] = 1.0 - (fails / total) if total > 0 else 0.0 if error_type in ("ip_blocked", "captcha", "fingerprint_detected"): self._blacklist.add(fp_id) fp["is_active"] = False logger.warning(f"指纹 {fp_id} 被加入黑名单: {error_type}") async def cleanup(self): async with self._lock: to_remove = [ fp_id for fp_id, fp in self._pool.items() if not fp.get("is_active") and fp.get("success_rate", 1.0) < self._min_success_rate * 0.5 ] for fp_id in to_remove: self._pool.pop(fp_id, None) self._blacklist.discard(fp_id) deficit = self._pool_size - len(self._pool) if deficit > 0: for _ in range(deficit): fp = self._generator.generate() self._pool[fp["id"]] = fp logger.info(f"指纹池补充 {deficit} 个新指纹") @property def stats(self) -> Dict[str, Any]: active = sum(1 for fp in self._pool.values() if fp.get("is_active", True)) rates = [fp.get("success_rate", 1.0) for fp in self._pool.values()] return { "total": len(self._pool), "active": active, "blacklisted": len(self._blacklist), "avg_success_rate": sum(rates) / len(rates) if rates else 1.0, }