diff --git a/core/douyu/login.py b/core/douyu/login.py index f2b21a7..2b3430a 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -23,6 +23,9 @@ from core.geetest.common.network import ( req_fullpage_validate, ) +# ── 全局极验并发限制：同一时刻最多2个线程做极验验证 ── +_geetest_semaphore = threading.Semaphore(2) + class AccountLike(Protocol): """DouyuLogin 所需的最小账号接口，ORM 对象或 SimpleNamespace 均可满足。""" @@ -75,6 +78,8 @@ class DouyuLogin: max_total_time: float = 0, whitelist_uid: str = "", whitelist_ukey: str = "", + whitelist_platform: str = "xiequ", + whitelist_credentials: dict = None, proxy_manager: Optional[ProxyManager] = None, stop_event: Optional[threading.Event] = None, ): @@ -95,6 +100,8 @@ class DouyuLogin: proxy_api_url, whitelist_uid=whitelist_uid, whitelist_ukey=whitelist_ukey, + whitelist_platform=whitelist_platform, + whitelist_credentials=whitelist_credentials, ) else: self.proxy_manager = None @@ -220,6 +227,22 @@ class DouyuLogin: return True return False + @staticmethod + def _is_geetest_network_error(err_str: str) -> bool: + """ + 判断异常是否为极验接口的网络类错误（代理到极验不可达/被限流）。 + 这类错误说明当前代理已被极验识别或限流，应立即换代理而非软重试。 + """ + err_lower = err_str.lower() + network_error_keywords = [ + '网络不给力', # 极验 get.php 返回的网络错误 + 'connection aborted', # 远程断开连接 + 'remotedisconnected', # requests RemoteDisconnected + 'read timed out', # 读取超时（极验接口） + 'api.geetest.com', # 极验接口相关错误 + ] + return any(kw in err_lower for kw in network_error_keywords) + @staticmethod def _truncate_error(err_str: str, max_len: int = 80) -> str: """截断错误信息，避免日志刷屏。""" @@ -461,28 +484,41 @@ class DouyuLogin: """ 解决极验 fullpage 验证（带重试机制） - 区分两类错误： - - 临时波动（slide、网络不给力、KeyError）：先用原代理重试，连续失败才换 + 区分三类错误： - 代理死亡（ProxyError、连接超时）：立即换代理 + - 网络类错误（网络不给力、Connection aborted）：立即换代理 + - 极验逻辑失败（slide等）：原代理重试，连续2次才换 Args: gt: 极验gt参数 challenge: 极验challenge参数（第一次登录返回的） deadline: 登录整体超时截止时间（monotonic），0=不限 - challenge: 极验challenge参数（第一次登录返回的） Returns: (validate, seccode) """ logger.info("开始极验 fullpage 验证...") - # max_proxy_retries=0 表示不限代理切换次数。 + # ── 全局并发限制：同一时刻最多2个线程做极验，避免被极验限流 ── + acquired = _geetest_semaphore.acquire(timeout=120) + if not acquired: + raise ValueError("极验验证等待超时（并发排队120秒未获得信号量）") + try: + return self._solve_geetest_inner(gt, challenge, deadline) + finally: + _geetest_semaphore.release() + + def _solve_geetest_inner(self, gt: str, challenge: str, deadline: float = 0) -> Tuple[str, str]: + """极验验证内部实现（已获取并发信号量）。""" max_proxy_switches = self.max_proxy_retries if self.max_proxy_retries > 0 else 999999 proxy_switches = 0 + # ── 极验验证最大尝试次数（超出后回到 login 整体重试）── + _MAX_GEETEST_ATTEMPTS = 15 + # 连续临时失败计数（同一代理下），超过阈值才换代理 _soft_fail_streak = 0 - _SOFT_FAIL_THRESHOLD = 2 # 同一代理连续临时失败2次才换 + _SOFT_FAIL_THRESHOLD = 2 def refresh_proxy(mark_bad: bool) -> None: """按代理切换上限刷新代理。""" @@ -499,12 +535,20 @@ class DouyuLogin: while True: attempt += 1 self._ensure_not_stopped() + + # ── 极验验证尝试上限 ── + if attempt > _MAX_GEETEST_ATTEMPTS: + raise ValueError( + f"极验验证已尝试 {attempt} 次，超过上限 {_MAX_GEETEST_ATTEMPTS}，" + f"回到 login 整体重试换新代理" + ) + # 超时兜底：极验验证不应超过登录整体时间上限 if deadline and time.monotonic() > deadline: raise ValueError(f"极验验证超时（登录整体时间耗尽）") try: - logger.info(f"极验验证尝试 {attempt} (无限重试)") + logger.info(f"极验验证尝试 {attempt}/{_MAX_GEETEST_ATTEMPTS}") # 按斗鱼登录页 HAR：fullpage 智能检测流程，不进入图片滑块。 str_16 = _generate_seed() @@ -556,17 +600,23 @@ class DouyuLogin: except Exception as e: err_str = str(e) - if "代理切换次数已达上限" in err_str: + if "代理切换次数已达上限" in err_str or "超过上限" in err_str: raise is_proxy_dead = self._is_proxy_connection_error(err_str) + is_network_error = self._is_geetest_network_error(err_str) if is_proxy_dead: # 代理确实不可用：立即换，标记坏 logger.warning(f"极验验证代理连接失败: {self._truncate_error(err_str)}，换代理") refresh_proxy(mark_bad=True) _soft_fail_streak = 0 + elif is_network_error: + # 网络类错误（极验限流/网络不给力/Connection aborted）：立即换代理 + logger.warning(f"极验网络错误: {self._truncate_error(err_str)}，立即换代理") + refresh_proxy(mark_bad=False) + _soft_fail_streak = 0 else: - # 临时异常（KeyError、网络不给力等）：先原代理重试 + # 其他临时异常（KeyError 等）：先原代理重试 _soft_fail_streak += 1 if _soft_fail_streak >= _SOFT_FAIL_THRESHOLD: logger.warning(f"极验验证临时异常: {self._truncate_error(err_str)}，连续 {_soft_fail_streak} 次，换代理") diff --git a/core/douyu/proxy_manager.py b/core/douyu/proxy_manager.py index aaad9c7..9cc464b 100644 --- a/core/douyu/proxy_manager.py +++ b/core/douyu/proxy_manager.py @@ -19,10 +19,19 @@ class ProxyManager: _COOLDOWN_MAX = 120 # 最大冷却秒数 _MAX_FAIL_COUNT = 3 # 连续失败此次数后永久移出池 - def __init__(self, api_url: str = "", whitelist_uid: str = "", whitelist_ukey: str = ""): + def __init__( + self, + api_url: str = "", + whitelist_uid: str = "", + whitelist_ukey: str = "", + whitelist_platform: str = "xiequ", + whitelist_credentials: dict = None, + ): self.api_url = api_url self.whitelist_uid = whitelist_uid self.whitelist_ukey = whitelist_ukey + self.whitelist_platform = whitelist_platform + self.whitelist_credentials = whitelist_credentials self.current_proxy: Optional[str] = None self._cond = threading.Condition() # 已验证可用的代理池: {proxy_url: validated_timestamp} @@ -31,11 +40,19 @@ class ProxyManager: self._in_use: set[str] = set() # 代理失败记录: {proxy_url: {"count": 失败次数, "cooldown_until": 冷却到期时间戳}} self._fail_records: dict[str, dict] = {} - self._pool_ttl = 90 + self._pool_ttl = 180 self._fetching = False + + # 构建白名单同步器（优先使用新参数，向后兼容旧参数） + _wl_platform = whitelist_platform or "xiequ" + _wl_credentials = whitelist_credentials + if not _wl_credentials and whitelist_uid and whitelist_ukey: + _wl_platform = "xiequ" + _wl_credentials = {"uid": whitelist_uid, "ukey": whitelist_ukey} + self._whitelist_syncer = ( - DouyuWhitelistSyncer(whitelist_uid, whitelist_ukey) - if whitelist_uid and whitelist_ukey + DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials) + if _wl_credentials else None ) @@ -70,49 +87,71 @@ class ProxyManager: return proxy return None - def get_proxy(self, max_attempts: int = 5) -> Optional[str]: + def get_proxy(self, max_attempts: int = 5, max_retries: int = 3) -> Optional[str]: """ 获取可用代理 IP，优先从已验证代理池复用。 线程安全：池空时只有一个线程调 API 获取并验证所有代理入池， - 其他线程等待后复用。 + 其他线程等待后复用。等待线程被唤醒后会循环尝试取代理， + 避免多个线程竞争拿到同一个代理。 + + Args: + max_attempts: 代理 API 获取尝试次数 + max_retries: 等待线程获取代理的重试次数（防止唤醒后池仍为空直接放弃） """ + # ── 快速路径：池中有可用代理 ── with self._cond: proxy = self._pick_from_pool_locked() if proxy: logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})") return proxy - if self._fetching: - logger.debug("代理池空，等待其他线程获取...") - self._cond.wait(timeout=60) + # ── 池空：需要获取新代理 ── + for retry in range(max_retries): + with self._cond: + # 再次检查（可能其他线程刚充盈了池） proxy = self._pick_from_pool_locked() if proxy: - logger.debug(f"等待后从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})") + logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})") return proxy - return None - self._fetching = True + if self._fetching: + logger.debug("代理池空，等待其他线程获取...") + self._cond.wait(timeout=60) + # 等待后循环重试取代理（不再只尝试一次） + proxy = self._pick_from_pool_locked() + if proxy: + logger.debug(f"等待后从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})") + return proxy + # 池仍为空，但可能有其他等待线程会去获取，继续重试 + continue - try: - proxies_all, msg = self._fetch_and_verify_all(max_attempts) - finally: - with self._cond: - self._fetching = False - self._cond.notify_all() + # 当前线程负责获取 + self._fetching = True - if proxies_all: - with self._cond: - now = time.time() - for proxy in proxies_all: - self._verified_pool[proxy] = now - first = proxies_all[0] - self._in_use.add(first) - self.current_proxy = first - logger.info(f"代理池充盈 {len(proxies_all)} 个可用代理: {first} ...") - return first + try: + proxies_all, msg = self._fetch_and_verify_all(max_attempts) + finally: + with self._cond: + self._fetching = False + self._cond.notify_all() - logger.warning(f"获取代理失败: {msg}") + if proxies_all: + with self._cond: + now = time.time() + for proxy in proxies_all: + self._verified_pool[proxy] = now + first = proxies_all[0] + self._in_use.add(first) + self.current_proxy = first + logger.info(f"代理池充盈 {len(proxies_all)} 个可用代理: {first} ...") + return first + + logger.warning(f"获取代理失败: {msg}") + # 获取失败，等一小段时间后重试 + time.sleep(2) + + logger.warning(f"获取代理失败（已重试 {max_retries} 次）") return None def _fetch_and_verify_all(self, max_attempts: int) -> tuple[Optional[list[str]], str]: @@ -206,6 +245,14 @@ def get_proxy_manager( api_url: str = "", whitelist_uid: str = "", whitelist_ukey: str = "", + whitelist_platform: str = "xiequ", + whitelist_credentials: dict = None, ) -> ProxyManager: """获取代理管理器实例。""" - return ProxyManager(api_url, whitelist_uid=whitelist_uid, whitelist_ukey=whitelist_ukey) + return ProxyManager( + api_url, + whitelist_uid=whitelist_uid, + whitelist_ukey=whitelist_ukey, + whitelist_platform=whitelist_platform, + whitelist_credentials=whitelist_credentials, + ) diff --git a/core/douyu/proxy_platforms/__init__.py b/core/douyu/proxy_platforms/__init__.py new file mode 100644 index 0000000..e5fe918 --- /dev/null +++ b/core/douyu/proxy_platforms/__init__.py @@ -0,0 +1,61 @@ +"""代理平台适配器注册表与工厂函数。 + +新增平台适配器： + 1. 在 proxy_platforms/ 下创建 .py，继承 BaseWhitelistAdapter + 2. 在本文件导入并注册到 _PLATFORM_REGISTRY + 3. 在 _PLATFORM_CREDENTIAL_FIELDS 中定义凭据字段 +""" + +from typing import Optional, Type + +from .base import BaseWhitelistAdapter, get_exit_ip_via_proxy +from .xiequ import XiequAdapter +from .xkdaili import XkdailiAdapter + +# ── 平台注册表 ── +_PLATFORM_REGISTRY: dict[str, Type[BaseWhitelistAdapter]] = { + "xiequ": XiequAdapter, + "xkdaili": XkdailiAdapter, +} + +# ── 平台凭据字段定义（前端动态渲染用）── +_PLATFORM_CREDENTIAL_FIELDS: dict[str, list[dict]] = { + "xiequ": [ + {"key": "uid", "label": "UID", "placeholder": "如: 99769"}, + {"key": "ukey", "label": "UKEY", "placeholder": "如: C99371082B965B70F46DCAA87A04618B"}, + ], + "xkdaili": [ + {"key": "apikey", "label": "API Key", "placeholder": "如: XK86872CC0C85A415461"}, + {"key": "wl_sign", "label": "白名单签名(Sign)", "placeholder": "白名单接口专用，如: 8953b3082173aa4188e7403e81249027"}, + {"key": "flag", "label": "Flag", "placeholder": "套餐标识，如: 8"}, + ], +} + +# ── 平台显示名 ── +_PLATFORM_LABELS: dict[str, str] = { + name: cls(dict()).platform_label + for name, cls in _PLATFORM_REGISTRY.items() +} + + +def create_adapter(platform: str, credentials: dict) -> Optional[BaseWhitelistAdapter]: + """根据平台标识符和凭据创建适配器实例。""" + cls = _PLATFORM_REGISTRY.get(platform) + if not cls: + return None + return cls(credentials) + + +def get_platform_names() -> list[str]: + """获取所有已注册平台标识符。""" + return list(_PLATFORM_REGISTRY.keys()) + + +def get_platform_labels() -> dict[str, str]: + """获取所有平台标识符到中文显示名的映射。""" + return dict(_PLATFORM_LABELS) + + +def get_credential_fields(platform: str) -> list[dict]: + """获取指定平台的凭据字段定义。""" + return _PLATFORM_CREDENTIAL_FIELDS.get(platform, []) diff --git a/core/douyu/proxy_platforms/base.py b/core/douyu/proxy_platforms/base.py new file mode 100644 index 0000000..9819751 --- /dev/null +++ b/core/douyu/proxy_platforms/base.py @@ -0,0 +1,193 @@ +"""代理平台白名单适配器基类。 + +所有平台适配器继承 BaseWhitelistAdapter，实现平台特定的 API 调用。 +基类提供通用的 sync_ip 逻辑，子类只需实现增删查等原子操作。 +""" + +import re +import threading +import time +from abc import ABC, abstractmethod +from typing import Optional + +import requests +from loguru import logger + +# 全局锁：防止并发白名单同步触发代理服务商限流 +_whitelist_sync_lock = threading.Lock() + + +class BaseWhitelistAdapter(ABC): + """白名单适配器基类。 + + 子类必须实现: + - platform_name / platform_label 属性 + - add_ip / delete_ip / get_whitelist / test_connection 方法 + 基类提供: + - sync_ip 通用同步逻辑 + - get_local_exit_ip 静态方法 + """ + + MEMO_PREFIX = "douyu_auto" + + @property + def memo(self) -> str: + return self.MEMO_PREFIX + + @property + @abstractmethod + def platform_name(self) -> str: + """平台标识符，如 'xiequ', 'xkdaili'""" + ... + + @property + @abstractmethod + def platform_label(self) -> str: + """平台中文显示名，如 '协固', '星客'""" + ... + + @abstractmethod + def add_ip(self, ip: str) -> tuple[bool, str]: + """添加IP到白名单。返回 (是否成功, 消息)""" + ... + + @abstractmethod + def delete_ip(self, ip: str) -> tuple[bool, str]: + """删除白名单IP。返回 (是否成功, 消息)""" + ... + + @abstractmethod + def get_whitelist(self) -> list[dict]: + """获取白名单列表。返回 [{"ip": "x.x.x.x", "memo": "备注"}, ...] + + 如果平台不支持查询接口，返回空列表。 + """ + ... + + @abstractmethod + def test_connection(self) -> tuple[bool, str]: + """测试API连接。返回 (是否成功, 消息)""" + ... + + def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]: + """ + 通用白名单同步逻辑（线程安全）。 + + 策略： + - 当前IP已在白名单（同备注），无需操作 + - 当前IP不在白名单，添加 + - 超过 keep_recent 个同备注IP时，删除最老的 + - 当前IP已存在但备注不同，不处理 + + Args: + current_ip: 当前出口IP + keep_recent: 保留的同备注IP数量上限 + + Returns: + (是否成功, 消息) + """ + with _whitelist_sync_lock: + try: + records = self.get_whitelist() + + # 不支持查询的平台，直接尝试添加 + if records is None: + ok, msg = self.add_ip(current_ip) + if ok: + return True, f"白名单IP已添加: {current_ip}" + # 添加失败但提示已存在，视为成功 + if "已存在" in msg or "exist" in msg.lower(): + return True, f"白名单IP已存在: {current_ip}" + return False, f"白名单添加失败: {msg}" + + memo_records = [r for r in records if r.get("memo") == self.memo] + memo_ips = [r.get("ip") for r in memo_records] + + # 当前IP已在白名单（同备注），无需操作 + if current_ip in memo_ips: + msg = f"白名单IP已是最新的: {current_ip}" + logger.info(msg) + return True, msg + + # 当前IP已存在但备注不同，不处理 + if any(r.get('ip') == current_ip for r in records): + logger.info(f"白名单IP {current_ip} 已存在（备注不同），无需重复添加") + return True, f"白名单IP已存在: {current_ip}" + + # 超过上限，删除最老的 + if len(memo_ips) >= keep_recent: + to_delete = memo_ips[:len(memo_ips) - keep_recent + 1] + for old_ip in to_delete: + logger.info(f"白名单同备注IP超限，删除旧的: {old_ip}") + self.delete_ip(old_ip) + time.sleep(1) + + # 添加新IP + logger.info(f"白名单添加新出口IP: {current_ip} (当前 {len(memo_ips)} 个同备注)") + ok, resp = self.add_ip(current_ip) + if ok: + msg = f"白名单IP已添加: {current_ip}" + logger.info(msg) + return True, msg + + return False, f"白名单添加失败，API响应: {resp}" + + except Exception as e: + msg = f"白名单同步失败: {e}" + logger.error(msg) + return False, msg + + @staticmethod + def get_local_exit_ip() -> Optional[str]: + """获取本机当前公网出口IP（不走代理）。""" + return _get_local_exit_ip() + + +def _get_local_exit_ip() -> Optional[str]: + """获取本机当前公网出口IP（不走代理）。""" + targets = [ + "https://myip.ipip.net", + "https://4.ipw.cn", + "https://api.ipify.org", + ] + for url in targets: + try: + response = requests.get(url, timeout=8, headers={"User-Agent": "Mozilla/5.0"}) + response.raise_for_status() + text = response.text.strip() + match = re.search(r'(\d{1,3}(?:\.\d{1,3}){3})', text) + if match: + return match.group(1) + except Exception: + continue + return None + + +def get_exit_ip_via_proxy(proxy: str) -> Optional[str]: + """通过代理获取出口IP。""" + targets = [ + "https://qifu-api.baidubce.com/ip/local/geo/v1/district", + "https://myip.ipip.net", + "https://4.ipw.cn", + ] + proxies = {"http": proxy, "https": proxy} + for url in targets: + try: + response = requests.get( + url, proxies=proxies, timeout=6, + headers={"User-Agent": "Mozilla/5.0"}, + ) + response.raise_for_status() + content_type = response.headers.get("content-type", "").lower() + if "json" in content_type: + data = response.json() + ip = data.get("ip") or data.get("origin") + if ip: + return str(ip).split(",")[0].strip() + text = response.text.strip() + match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text) + if match: + return match.group(1) + except Exception: + continue + return None diff --git a/core/douyu/proxy_platforms/xiequ.py b/core/douyu/proxy_platforms/xiequ.py new file mode 100644 index 0000000..1734e11 --- /dev/null +++ b/core/douyu/proxy_platforms/xiequ.py @@ -0,0 +1,159 @@ +"""携趣代理(xiequ)白名单适配器。 + +从 core.douyu.whitelist.WhitelistManager 迁移而来。 +""" + +import json +import re +import time +from urllib.parse import urlencode + +import requests +from loguru import logger + +from .base import BaseWhitelistAdapter + + +class XiequAdapter(BaseWhitelistAdapter): + """携趣代理白名单适配器。 + + API格式： + 基础URL: http://op.xiequ.cn/IpWhiteList.aspx + 认证参数: uid + ukey + 操作区分: act=add / del / getjson + """ + + BASE_URL = "http://op.xiequ.cn/IpWhiteList.aspx" + + def __init__(self, credentials: dict): + self.uid = credentials.get("uid", "") + self.ukey = credentials.get("ukey", "") + + @property + def platform_name(self) -> str: + return "xiequ" + + @property + def platform_label(self) -> str: + return "携趣" + + def _build_url(self, **params) -> str: + """构建请求URL""" + base_params = {"uid": self.uid, "ukey": self.ukey} + base_params.update(params) + return f"{self.BASE_URL}?{urlencode(base_params)}" + + def get_whitelist(self) -> list[dict]: + """ + 获取白名单列表。 + + 协固原始格式: [{"IP": "x.x.x.x", "MEMO": "备注"}, ...] + 统一格式: [{"ip": "x.x.x.x", "memo": "备注"}, ...] + """ + try: + url = self._build_url(act="getjson") + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + if not text or text == "[]": + return [] + + data = json.loads(text) + if isinstance(data, dict): + data = data.get("data", []) + if not isinstance(data, list): + return [] + + # 映射到统一格式 + return [ + {"ip": r.get("IP", r.get("ip", "")), "memo": r.get("MEMO", r.get("memo", ""))} + for r in data + if r.get("IP") or r.get("ip") + ] + + except Exception as e: + logger.error(f"获取白名单失败: {e}") + return [] + + def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]: + """添加IP到白名单。""" + try: + url = self._build_url(act="add", ip=ip, meno=self.memo) + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + logger.debug(f"添加白名单响应: {text}") + + if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text: + logger.info(f"白名单添加成功: {ip} (备注: {self.memo})") + return True, text + + if "已存在" in text or "exist" in text.lower() or "IpRep" in text: + logger.info(f"白名单已存在: {ip}") + return True, text + + # 频率限制，等待后重试一次 + if retry and ("频率过快" in text or "稍后" in text): + wait = 5 + match = re.search(r'(\d+)\s*秒', text) + if match: + wait = int(match.group(1)) + logger.info(f"白名单添加被限流，等待 {wait} 秒后重试...") + time.sleep(wait) + return self.add_ip(ip, retry=False) + + # UKEY 错误 + if "Err:Key" in text: + logger.error("白名单UKEY错误，请检查配置") + return False, "UKEY错误，请检查白名单配置" + + # 超出白名单数量限制 + if "Err:Max" in text or "超过" in text or "上限" in text: + logger.error(f"白名单数量超限: {text}") + return False, f"白名单数量超限: {text}" + + logger.warning(f"白名单添加结果: {text}") + return False, text + + except Exception as e: + logger.error(f"添加白名单失败: {e}") + return False, str(e) + + def delete_ip(self, ip: str) -> tuple[bool, str]: + """删除白名单IP。""" + try: + url = self._build_url(act="del", ip=ip) + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + logger.debug(f"删除白名单响应: {text}") + + if "ok" in text.lower() or "success" in text.lower() or "删除成功" in text: + logger.info(f"白名单删除成功: {ip}") + return True, text + + logger.warning(f"白名单删除结果: {text}") + return False, text + + except Exception as e: + logger.error(f"删除白名单失败: {e}") + return False, str(e) + + def test_connection(self) -> tuple[bool, str]: + """测试白名单API连接。""" + try: + records = self.get_whitelist() + count = len(records) + my_records = [r for r in records if r.get("memo", "").startswith(self.MEMO_PREFIX)] + + msg = f"连接成功，白名单共 {count} 条记录，其中本机相关 {len(my_records)} 条" + logger.info(msg) + return True, msg + + except Exception as e: + msg = f"连接失败: {e}" + logger.error(msg) + return False, msg diff --git a/core/douyu/proxy_platforms/xkdaili.py b/core/douyu/proxy_platforms/xkdaili.py new file mode 100644 index 0000000..958941b --- /dev/null +++ b/core/douyu/proxy_platforms/xkdaili.py @@ -0,0 +1,162 @@ +"""星空代理(xkdaili)白名单适配器。 + +API格式： + 基础URL: http://api2.xkdaili.com/tools/XApi.ashx + 白名单认证参数: apikey + sign(白名单专用) + flag + 获取代理认证参数: apikey + sign(提取专用) — 由用户在代理API地址中填写完整URL + + 白名单接口的 sign 和获取代理的 sign 不同！ + +响应格式: + status=100: 操作成功 + status=508: 白名单已存在（视为成功） + status=202: 秘钥验证失败 + status=808: 其它错误 + +注意：没有查询白名单列表的接口 +""" + +import json + +import requests +from loguru import logger + +from .base import BaseWhitelistAdapter + + +class XkdailiAdapter(BaseWhitelistAdapter): + """星空代理(xkdaili)白名单适配器。""" + + BASE_URL = "http://api2.xkdaili.com/tools/XApi.ashx" + + def __init__(self, credentials: dict): + self.apikey = credentials.get("apikey", "") + self.wl_sign = credentials.get("wl_sign", "") + self.flag = credentials.get("flag", "") + + @property + def platform_name(self) -> str: + return "xkdaili" + + @property + def platform_label(self) -> str: + return "星空" + + def _build_url(self, **params) -> str: + """构建白名单请求URL""" + base_params = { + "apikey": self.apikey, + "sign": self.wl_sign, + "flag": self.flag, + } + base_params.update(params) + from urllib.parse import urlencode + return f"{self.BASE_URL}?{urlencode(base_params)}" + + def _parse_response(self, text: str) -> tuple[bool, str]: + """解析星空API响应。 + + status=100: 成功 + status=508: 白名单已存在（视为成功） + status=202: 秘钥验证失败 + status=808: 其它错误 + """ + try: + data = json.loads(text) + status = data.get("status", -1) + info = data.get("info", "") + + # 100=成功 + if status == 100: + return True, info + + # 508=已存在，视为成功 + if status == 508: + return True, info + + # 202=秘钥验证失败 + if status == 202: + return False, f"认证失败: {info}" + + # 其他错误 + return False, f"失败(status={status}): {info}" + + except (json.JSONDecodeError, ValueError): + # 非JSON响应，按文本判断 + if "成功" in text or "ok" in text.lower(): + return True, text + if "已存在" in text: + return True, text + return False, text + + def get_whitelist(self) -> list[dict]: + """ + 星空不支持查询白名单列表，返回空列表。 + + sync_ip 会检测到空列表后直接走添加逻辑。 + """ + return [] + + def add_ip(self, ip: str) -> tuple[bool, str]: + """添加IP到白名单。多个IP用英文逗号隔开。""" + try: + url = self._build_url(type="addwhiteip", ip=ip) + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + logger.debug(f"星空添加白名单响应: {text}") + + ok, msg = self._parse_response(text) + if ok: + logger.info(f"星空白名单添加成功: {ip} - {msg}") + else: + logger.warning(f"星空白名单添加失败: {msg}") + return ok, msg + + except Exception as e: + logger.error(f"星空添加白名单失败: {e}") + return False, str(e) + + def delete_ip(self, ip: str) -> tuple[bool, str]: + """删除白名单IP。多个IP用英文逗号隔开。""" + try: + url = self._build_url(type="delwhiteip", ip=ip) + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + logger.debug(f"星空删除白名单响应: {text}") + + ok, msg = self._parse_response(text) + if ok: + logger.info(f"星空白名单删除成功: {ip} - {msg}") + else: + logger.warning(f"星空白名单删除失败: {msg}") + return ok, msg + + except Exception as e: + logger.error(f"星空删除白名单失败: {e}") + return False, str(e) + + def test_connection(self) -> tuple[bool, str]: + """测试白名单API连接。""" + try: + # 尝试添加无效IP来验证认证 + url = self._build_url(type="addwhiteip", ip="0.0.0.0") + response = requests.get(url, timeout=10) + response.raise_for_status() + + text = response.text.strip() + ok, msg = self._parse_response(text) + + if ok: + return True, "连接成功，API认证正常" + # 认证失败 + if "认证失败" in msg: + return False, msg + # 其他错误（如无效IP），说明认证通过了 + return True, f"连接成功，API可访问: {msg}" + + except Exception as e: + return False, f"连接失败: {e}" diff --git a/core/douyu/proxy_resolver.py b/core/douyu/proxy_resolver.py index f7239ae..9d292de 100644 --- a/core/douyu/proxy_resolver.py +++ b/core/douyu/proxy_resolver.py @@ -149,17 +149,28 @@ def resolve_working_proxy( api_url: str, whitelist_uid: str = "", whitelist_ukey: str = "", + whitelist_platform: str = "xiequ", + whitelist_credentials: dict = None, max_attempts: int = 4, log_func: Optional[LogFunc] = None, ) -> tuple[Optional[str], str]: """ 从代理 API 获取可用代理，自动处理白名单同步。 - 保留旧函数签名，供 Web 层代理测试继续使用。 + 支持： + - 新参数：whitelist_platform + whitelist_credentials + - 旧参数：whitelist_uid + whitelist_ukey（向后兼容） """ + # 构建白名单同步器 + _wl_platform = whitelist_platform or "xiequ" + _wl_credentials = whitelist_credentials + if not _wl_credentials and whitelist_uid and whitelist_ukey: + _wl_platform = "xiequ" + _wl_credentials = {"uid": whitelist_uid, "ukey": whitelist_ukey} + syncer = ( - DouyuWhitelistSyncer(whitelist_uid, whitelist_ukey) - if whitelist_uid and whitelist_ukey + DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials) + if _wl_credentials else None ) resolver = ProxyResolver( diff --git a/core/douyu/proxy_verifier.py b/core/douyu/proxy_verifier.py index b486e1f..0b4183a 100644 --- a/core/douyu/proxy_verifier.py +++ b/core/douyu/proxy_verifier.py @@ -9,18 +9,19 @@ from loguru import logger def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str]: """ - 验证代理是否可用，依次验证： - 1. 斗鱼主站 www.douyu.com（登录目标） - 2. 极验接口 api.geetest.com（极验验证瓶颈） + 验证代理是否可用，只验证斗鱼主站可达。 - 两关都过才算可用。 + 不再验证极验接口，原因： + 1. 极验接口验证耗时 5-8秒，对短效代理是巨大浪费 + 2. 极验限流导致大量代理被误判为不可用 + 3. 登录流程中极验验证本身会检测代理到极验的连通性 Returns: (是否可用, 消息) """ proxies = {'http': proxy_url, 'https': proxy_url} - # ── 第1关：斗鱼主站 ── + # ── 斗鱼主站可达性验证 ── try: response = requests.get( 'https://www.douyu.com', @@ -29,6 +30,7 @@ def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str headers={'User-Agent': 'Mozilla/5.0'}, ) response.raise_for_status() + return True, '代理可用 → 斗鱼可达' except Exception as exc: err_msg = str(exc) if 'Tunnel connection failed' in err_msg or '503' in err_msg: @@ -40,26 +42,6 @@ def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str logger.debug(f"代理验证失败 [{proxy_url}]: 斗鱼主站不可达: {detail}") return False, detail - # ── 第2关：极验接口 ── - try: - response = requests.get( - 'https://api.geetest.com', - proxies=proxies, - timeout=timeout, - headers={'User-Agent': 'Mozilla/5.0'}, - # geetest 首页可能返回 4xx，只要能连上就算通 - allow_redirects=True, - ) - return True, '代理可用 → 斗鱼+极验' - except Exception as exc: - err_msg = str(exc) - if 'timed out' in err_msg.lower(): - detail = '极验接口超时' - else: - detail = f'极验不可达: {type(exc).__name__}' - logger.debug(f"代理验证失败 [{proxy_url}]: 斗鱼可达但{detail}") - return False, detail - def verify_proxies_concurrent( proxy_urls: list[str], diff --git a/core/douyu/proxy_whitelist.py b/core/douyu/proxy_whitelist.py index aa641a8..f45df2d 100644 --- a/core/douyu/proxy_whitelist.py +++ b/core/douyu/proxy_whitelist.py @@ -2,19 +2,41 @@ from typing import Optional -from .whitelist import WhitelistManager, get_local_exit_ip +from .proxy_platforms import create_adapter +from .proxy_platforms.base import BaseWhitelistAdapter, _get_local_exit_ip class DouyuWhitelistSyncer: - """把白名单 API 隔离成代理解析流程可调用的适配器。""" + """把白名单 API 隔离成代理解析流程可调用的适配器。 - def __init__(self, uid: str, ukey: str): - self.uid = uid - self.ukey = ukey + 支持： + - 新接口：platform + credentials（推荐） + - 旧接口：uid + ukey（向后兼容，自动映射为协固平台） + """ + + def __init__( + self, + platform: str = "xiequ", + credentials: dict = None, + uid: str = "", + ukey: str = "", + ): + # 向后兼容：如果传入 uid/ukey，转为 credentials + if uid and ukey and not credentials: + platform = "xiequ" + credentials = {"uid": uid, "ukey": ukey} + + self.platform = platform + self.credentials = credentials or {} + self._adapter: Optional[BaseWhitelistAdapter] = None + + if self.credentials: + self._adapter = create_adapter(platform, self.credentials) def sync_ip(self, ip: str) -> tuple[bool, str]: - manager = WhitelistManager(self.uid, self.ukey) - return manager.sync_ip(ip) + if not self._adapter: + return False, "未配置白名单凭据" + return self._adapter.sync_ip(ip) def get_local_exit_ip(self) -> Optional[str]: - return get_local_exit_ip() + return _get_local_exit_ip() diff --git a/core/douyu/whitelist.py b/core/douyu/whitelist.py index 01481c7..1bd4832 100644 --- a/core/douyu/whitelist.py +++ b/core/douyu/whitelist.py @@ -1,327 +1,89 @@ -"""代理IP白名单管理模块""" +"""代理IP白名单管理模块（兼容层）。 + +核心逻辑已迁移至 proxy_platforms/xiequ.py，本模块仅做向后兼容包装。 +旧代码仍可 from .whitelist import WhitelistManager 正常使用。 +""" -import json -import re -import threading -import time from typing import Optional -from urllib.parse import urlencode -import requests from loguru import logger -# 全局锁：防止多个并发登录任务同时同步白名单，触发代理服务商的30秒限流 -_whitelist_sync_lock = threading.Lock() +from .proxy_platforms.xiequ import XiequAdapter +from .proxy_platforms.base import ( + BaseWhitelistAdapter, + _get_local_exit_ip, + get_exit_ip_via_proxy, +) + +# 全局锁保留（基类内部使用，此处导出供旧代码引用） +from .proxy_platforms.base import _whitelist_sync_lock class WhitelistManager: - """协固代理IP白名单管理器""" + """携趣代理IP白名单管理器（兼容层，委托给 XiequAdapter）。""" - MEMO_PREFIX = "douyu_auto" - BASE_URL = "http://op.xiequ.cn/IpWhiteList.aspx" + MEMO_PREFIX = XiequAdapter.MEMO_PREFIX + BASE_URL = XiequAdapter.BASE_URL def __init__(self, uid: str, ukey: str): + self._adapter = XiequAdapter({"uid": uid, "ukey": ukey}) self.uid = uid self.ukey = ukey - self._memo = self.MEMO_PREFIX @property def memo(self) -> str: - """当前机器的固定备注""" - return self._memo + return self._adapter.memo def _build_url(self, **params) -> str: - """构建请求URL""" - base_params = { - "uid": self.uid, - "ukey": self.ukey, - } - base_params.update(params) - query = urlencode(base_params) - return f"{self.BASE_URL}?{query}" + """构建请求URL（兼容旧代码直接调用）。""" + return self._adapter._build_url(**params) def get_whitelist_json(self) -> list[dict]: """ - 获取白名单列表（JSON格式） + 获取白名单列表（JSON格式，协固原始格式）。 Returns: [{"IP": "x.x.x.x", "MEMO": "备注"}, ...] """ - try: - url = self._build_url(act="getjson") - response = requests.get(url, timeout=10) - response.raise_for_status() - - text = response.text.strip() - if not text or text == "[]": - return [] - - data = json.loads(text) - if isinstance(data, list): - return data - # 处理 {"data": [...]} 格式 - if isinstance(data, dict): - items = data.get("data", []) - if isinstance(items, list): - return items - return [] - - except Exception as e: - logger.error(f"获取白名单失败: {e}") - return [] + records = self._adapter.get_whitelist() + # 将统一格式映射回调固原始格式 + return [{"IP": r["ip"], "MEMO": r["memo"]} for r in records] def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]: - """ - 添加IP到白名单 - - Returns: - (是否成功, API原始响应) - """ - try: - url = self._build_url(act="add", ip=ip, meno=self._memo) - response = requests.get(url, timeout=10) - response.raise_for_status() - - text = response.text.strip() - logger.debug(f"添加白名单响应: {text}") - - if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text: - logger.info(f"白名单添加成功: {ip} (备注: {self._memo})") - return True, text - - if "已存在" in text or "exist" in text.lower() or "IpRep" in text: - logger.info(f"白名单已存在: {ip}") - return True, text - - # 频率限制，等待后重试一次 - if retry and ("频率过快" in text or "稍后" in text): - wait = 5 - match = re.search(r'(\d+)\s*秒', text) - if match: - wait = int(match.group(1)) - logger.info(f"白名单添加被限流，等待 {wait} 秒后重试...") - time.sleep(wait) - return self.add_ip(ip, retry=False) - - # UKEY 错误 - if "Err:Key" in text: - logger.error("白名单UKEY错误，请检查配置") - return False, "UKEY错误，请检查白名单配置" - - # 超出白名单数量限制 - if "Err:Max" in text or "超过" in text or "上限" in text: - logger.error(f"白名单数量超限: {text}") - return False, f"白名单数量超限: {text}" - - logger.warning(f"白名单添加结果: {text}") - return False, text - - except Exception as e: - logger.error(f"添加白名单失败: {e}") - return False, str(e) + """添加IP到白名单。""" + return self._adapter.add_ip(ip, retry=retry) def delete_ip(self, ip: str) -> bool: - """ - 删除指定IP - - Args: - ip: 要删除的IP地址 - """ - try: - url = self._build_url(act="del", ip=ip) - response = requests.get(url, timeout=10) - response.raise_for_status() - - text = response.text.strip() - logger.debug(f"删除白名单响应: {text}") - - if "ok" in text.lower() or "success" in text.lower() or "删除成功" in text: - logger.info(f"白名单删除成功: {ip}") - return True - - logger.warning(f"白名单删除结果: {text}") - return False - - except Exception as e: - logger.error(f"删除白名单失败: {e}") - return False + """删除指定IP。""" + ok, _ = self._adapter.delete_ip(ip) + return ok def get_memo_ip(self) -> Optional[str]: - """ - 获取当前备注对应的IP - - Returns: - IP地址，如果不存在返回None - """ - records = self.get_whitelist_json() + """获取当前备注对应的IP。""" + records = self._adapter.get_whitelist() for record in records: - if record.get("MEMO") == self._memo: - return record.get("IP") + if record.get("memo") == self.memo: + return record.get("ip") return None def get_memo_records(self) -> list[dict]: - """ - 获取当前备注的所有记录 - - Returns: - 匹配备注的记录列表 - """ - records = self.get_whitelist_json() - return [r for r in records if r.get("MEMO") == self._memo] + """获取当前备注的所有记录（协固原始格式）。""" + records = self._adapter.get_whitelist() + return [ + {"IP": r["ip"], "MEMO": r["memo"]} + for r in records + if r.get("memo") == self.memo + ] def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]: - """ - 同步白名单IP（保留多个近期出口IP，应对移动网络IP漂移） - - 策略： - - 当前IP已在白名单（同备注），无需操作 - - 当前IP不在白名单，添加（不删旧的，保留多个出口IP） - - 超过 keep_recent 个同备注IP时，删除最老的（按列表顺序） - - 当前IP已存在但备注不同，不处理（避免误删他人配置） - - 线程安全：使用全局锁串行化同步操作，避免并发调用触发代理服务商限流。 - - Args: - current_ip: 当前出口IP - keep_recent: 保留的同备注IP数量上限（默认3个，应对IP漂移） - - Returns: - (是否成功, 消息) - """ - with _whitelist_sync_lock: - try: - records = self.get_whitelist_json() - memo_records = [r for r in records if r.get("MEMO") == self._memo] - memo_ips = [r.get("IP") for r in memo_records] - - # 当前IP已在白名单（同备注），无需操作 - if current_ip in memo_ips: - msg = f"白名单IP已是最新的: {current_ip}" - logger.info(msg) - return True, msg - - # 当前IP已存在但备注不同，不处理（避免误删他人配置） - if any(r.get('IP') == current_ip for r in records): - logger.info(f"白名单IP {current_ip} 已存在（备注不同），无需重复添加") - return True, f"白名单IP已存在: {current_ip}" - - # 超过上限，删除最老的（列表前面的） - if len(memo_ips) >= keep_recent: - # 删除最早添加的（列表顺序） - to_delete = memo_ips[:len(memo_ips) - keep_recent + 1] - for old_ip in to_delete: - logger.info(f"白名单同备注IP超限，删除旧的: {old_ip}") - self.delete_ip(old_ip) - time.sleep(1) - - # 添加新IP - logger.info(f"白名单添加新出口IP: {current_ip} (当前 {len(memo_ips)} 个同备注)") - ok, resp = self.add_ip(current_ip) - if ok: - msg = f"白名单IP已添加: {current_ip}" - logger.info(msg) - return True, msg - - return False, f"白名单添加失败，API响应: {resp}" - - except Exception as e: - msg = f"白名单同步失败: {e}" - logger.error(msg) - return False, msg + """同步白名单IP。""" + return self._adapter.sync_ip(current_ip, keep_recent) def test_connection(self) -> tuple[bool, str]: - """ - 测试白名单API连接 - - Returns: - (是否成功, 消息) - """ - try: - records = self.get_whitelist_json() - count = len(records) - my_records = [r for r in records if r.get("MEMO", "").startswith(self.MEMO_PREFIX)] - - msg = f"连接成功，白名单共 {count} 条记录，其中本机相关 {len(my_records)} 条" - logger.info(msg) - return True, msg - - except Exception as e: - msg = f"连接失败: {e}" - logger.error(msg) - return False, msg + """测试白名单API连接。""" + return self._adapter.test_connection() def get_local_exit_ip() -> Optional[str]: - """ - 获取本机当前公网出口IP（不走代理）。 - - 用于在获取代理前主动同步白名单，避免出口IP漂移导致代理拒绝连接。 - - Returns: - 出口IP地址，获取失败返回 None - """ - targets = [ - "https://myip.ipip.net", - "https://4.ipw.cn", - "https://api.ipify.org", - ] - for url in targets: - try: - response = requests.get(url, timeout=8, headers={"User-Agent": "Mozilla/5.0"}) - response.raise_for_status() - text = response.text.strip() - # 从文本中提取IP - match = re.search(r'(\d{1,3}(?:\.\d{1,3}){3})', text) - if match: - return match.group(1) - except Exception: - continue - return None - - -def get_exit_ip_via_proxy(proxy: str) -> Optional[str]: - """ - 通过代理获取出口IP - - Args: - proxy: 代理URL，格式 http://ip:port - - Returns: - 出口IP地址 - """ - targets = [ - "https://qifu-api.baidubce.com/ip/local/geo/v1/district", - "https://myip.ipip.net", - "https://4.ipw.cn", - ] - - proxies = {"http": proxy, "https": proxy} - - for url in targets: - try: - response = requests.get( - url, - proxies=proxies, - timeout=6, - headers={"User-Agent": "Mozilla/5.0"}, - ) - response.raise_for_status() - - # 尝试解析IP - content_type = response.headers.get("content-type", "").lower() - if "json" in content_type: - data = response.json() - ip = data.get("ip") or data.get("origin") - if ip: - return str(ip).split(",")[0].strip() - - # 从文本中提取IP - text = response.text.strip() - match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text) - if match: - return match.group(1) - - except Exception: - continue - - return None + """获取本机当前公网出口IP（不走代理）。""" + return _get_local_exit_ip() diff --git a/utils/logger.py b/utils/logger.py index e4a402a..8c32806 100644 --- a/utils/logger.py +++ b/utils/logger.py @@ -1,17 +1,37 @@ """日志配置模块""" +import os import sys +from datetime import date from pathlib import Path + from loguru import logger -def setup_logger(level: str = "INFO", log_file: str = None) -> None: +def _should_rotate(message, file): + """自定义轮转条件：每天0点轮转，或单文件超过10M。""" + if not file: + return False + try: + # 单文件超过 10M 则轮转 + if os.path.getsize(file.name) > 10 * 1024 * 1024: + return True + # 跨天则轮转：日志日期 ≠ 文件最后修改日期 + file_date = date.fromtimestamp(os.path.getmtime(file.name)) + log_date = message.record["time"].date() + return file_date != log_date + except OSError: + return False + + +def setup_logger(level: str = "INFO", log_dir: str = None, log_file: str = None) -> None: """ 配置日志 Args: level: 日志级别 - log_file: 日志文件路径 + log_dir: 日志目录路径（文件名按日期自动生成，如 app-2026-06-24.log） + log_file: 日志文件路径（兼容旧接口，优先级低于 log_dir） """ # 移除默认handler logger.remove() @@ -27,16 +47,27 @@ def setup_logger(level: str = "INFO", log_file: str = None) -> None: colorize=True, ) - # 文件输出 - if log_file: - log_path = Path(log_file) - log_path.parent.mkdir(parents=True, exist_ok=True) + # 确定日志文件路径 + if log_dir: + # 新接口：按目录 + 日期命名 + dir_path = Path(log_dir) + dir_path.mkdir(parents=True, exist_ok=True) + file_path = str(dir_path / "app-{time:YYYY-MM-DD}.log") + elif log_file: + # 兼容旧接口 + file_path = Path(log_file) + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path = str(file_path) + else: + return - logger.add( - log_file, - level=level, - format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}", - rotation="10 MB", - retention="7 days", - encoding="utf-8", - ) + # 文件输出：按天命名 + 单文件最大10M轮转 + 保留7天 + logger.add( + file_path, + level=level, + format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}", + rotation=_should_rotate, # 超过10M轮转 + retention="7 days", # 保留7天 + compression="zip", # 旧日志自动压缩节省空间 + encoding="utf-8", + ) diff --git a/web/backend/main.py b/web/backend/main.py index 12e94db..9eff6f2 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -17,10 +17,10 @@ from utils import setup_logger @asynccontextmanager async def lifespan(app: FastAPI): - # 初始化日志（控制台 + 文件） + # 初始化日志（控制台 + 按天命名文件） _log_level = os.getenv("LOG_LEVEL", "DEBUG") _log_dir = Path(__file__).resolve().parents[2] / "logs" - setup_logger(level=_log_level, log_file=str(_log_dir / "app.log")) + setup_logger(level=_log_level, log_dir=str(_log_dir)) init_db() yield diff --git a/web/backend/migrations/versions/20260624_0003_proxy_platform.py b/web/backend/migrations/versions/20260624_0003_proxy_platform.py new file mode 100644 index 0000000..c0bfcab --- /dev/null +++ b/web/backend/migrations/versions/20260624_0003_proxy_platform.py @@ -0,0 +1,33 @@ +"""代理白名单平台抽象：新增 whitelist_platform / whitelist_credentials 字段 + +Revision ID: 20260624_0003 +Revises: 20260624_0002 +Create Date: 2026-06-24 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260624_0003" +down_revision: Union[str, None] = "20260624_0002" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("proxy_config") as batch: + batch.add_column( + sa.Column("whitelist_platform", sa.String(32), server_default="xiequ") + ) + batch.add_column( + sa.Column("whitelist_credentials", sa.JSON(), nullable=True) + ) + + +def downgrade() -> None: + with op.batch_alter_table("proxy_config") as batch: + batch.drop_column("whitelist_credentials") + batch.drop_column("whitelist_platform") diff --git a/web/backend/models.py b/web/backend/models.py index 9f03aac..2c41ea3 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -81,6 +81,9 @@ class ProxyConfig(Base): https = Column(EncryptedText(), default="") # 白名单 whitelist_enabled = Column(Boolean, default=False) + whitelist_platform = Column(String(32), default="xiequ") + whitelist_credentials = Column(JSON, nullable=True) + # 旧字段保留（向后兼容双写） whitelist_uid = Column(EncryptedText(), default="") whitelist_ukey = Column(EncryptedText(), default="") diff --git a/web/backend/routers/proxy.py b/web/backend/routers/proxy.py index 102235f..bc28b43 100644 --- a/web/backend/routers/proxy.py +++ b/web/backend/routers/proxy.py @@ -7,9 +7,12 @@ from sqlalchemy.orm import Session from ..database import get_db from ..models import User -from ..schemas import ProxyConfigOut, ProxyConfigUpdate +from ..schemas import ProxyConfigOut, ProxyConfigUpdate, PlatformInfo, PlatformFieldDef from ..deps import require_permission, authenticate_websocket from ..services.proxy_service import proxy_service +from core.douyu.proxy_platforms import ( + get_platform_names, get_platform_labels, get_credential_fields, +) router = APIRouter(prefix="/api/proxy", tags=["代理与白名单"]) @@ -30,17 +33,34 @@ def update_proxy_config( ): return proxy_service.update_config( db, - enabled=req.enabled, - api_url=req.api_url, - http=req.http, - https=req.https, - whitelist_enabled=req.whitelist_enabled, + enabled=req.enabled if req.enabled is not None else False, + api_url=req.api_url if req.api_url is not None else "", + http=req.http if req.http is not None else "", + https=req.https if req.https is not None else "", + whitelist_enabled=req.whitelist_enabled if req.whitelist_enabled is not None else False, + whitelist_platform=req.whitelist_platform or "xiequ", + whitelist_credentials=req.whitelist_credentials, whitelist_uid=req.whitelist_uid, whitelist_ukey=req.whitelist_ukey, current_user=current, ) +@router.get("/platforms", response_model=list[PlatformInfo]) +def list_platforms(): + """获取所有可用的代理白名单平台及其凭据字段定义。""" + labels = get_platform_labels() + result = [] + for name in get_platform_names(): + fields = get_credential_fields(name) + result.append(PlatformInfo( + name=name, + label=labels.get(name, name), + credential_fields=[PlatformFieldDef(**f) for f in fields], + )) + return result + + # ---- WebSocket 日志推送 ---- @router.websocket("/ws/test/{test_id}") diff --git a/web/backend/schemas.py b/web/backend/schemas.py index e673829..e3e1643 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -167,12 +167,39 @@ class ProxyConfigOut(BaseModel): http: str = "" https: str = "" whitelist_enabled: bool = False + whitelist_platform: str = "xiequ" + whitelist_credentials: Optional[dict] = None + # 旧字段保留（向后兼容） whitelist_uid: str = "" whitelist_ukey: str = "" -class ProxyConfigUpdate(ProxyConfigOut): - pass +class ProxyConfigUpdate(BaseModel): + enabled: Optional[bool] = None + api_url: Optional[str] = None + http: Optional[str] = None + https: Optional[str] = None + whitelist_enabled: Optional[bool] = None + whitelist_platform: Optional[str] = "xiequ" + whitelist_credentials: Optional[dict] = None + # 旧字段保留（向后兼容） + whitelist_uid: Optional[str] = None + whitelist_ukey: Optional[str] = None + + +# ---- 代理平台元信息 ---- +class PlatformFieldDef(BaseModel): + """平台凭据字段定义。""" + key: str + label: str + placeholder: str = "" + + +class PlatformInfo(BaseModel): + """平台元信息。""" + name: str + label: str + credential_fields: list[PlatformFieldDef] # ---- 通用 ---- diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index c3bcda4..bd6c069 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -55,12 +55,20 @@ class LoginBatchRunner: # 共享代理管理器（带锁，避免并发白名单限流；极验失败时可刷新代理） self._shared_proxy_manager = None if proxy_config and proxy_config.enabled and proxy_config.api_url: - wl_uid = proxy_config.whitelist_uid or "" if proxy_config.whitelist_enabled else "" - wl_ukey = proxy_config.whitelist_ukey or "" if proxy_config.whitelist_enabled else "" + wl_platform = "xiequ" + wl_credentials = None + if proxy_config.whitelist_enabled: + wl_platform = getattr(proxy_config, 'whitelist_platform', None) or "xiequ" + wl_credentials = getattr(proxy_config, 'whitelist_credentials', None) + # 向后兼容：旧字段有值但新字段为空 + if not wl_credentials and proxy_config.whitelist_uid and proxy_config.whitelist_ukey: + wl_platform = "xiequ" + wl_credentials = {"uid": proxy_config.whitelist_uid, "ukey": proxy_config.whitelist_ukey} + self._shared_proxy_manager = get_proxy_manager( proxy_config.api_url, - whitelist_uid=wl_uid, - whitelist_ukey=wl_ukey, + whitelist_platform=wl_platform, + whitelist_credentials=wl_credentials, ) def stop(self): diff --git a/web/backend/services/proxy_service.py b/web/backend/services/proxy_service.py index 3228ffe..4e40d4a 100644 --- a/web/backend/services/proxy_service.py +++ b/web/backend/services/proxy_service.py @@ -9,9 +9,31 @@ from typing import Optional from sqlalchemy.orm import Session from core.douyu.proxy import resolve_working_proxy, verify_proxy_url, parse_proxy_response +from core.douyu.proxy_platforms import create_adapter, get_platform_labels +from core.douyu.proxy_platforms.base import _get_local_exit_ip from ..models import ProxyConfig as ProxyConfigModel, AuditLog +def _build_whitelist_params(cfg: ProxyConfigModel) -> dict: + """从 ProxyConfig 构建白名单参数，自动处理新旧字段兼容。 + + Returns: + {"whitelist_platform": str, "whitelist_credentials": dict|None} + """ + platform = getattr(cfg, 'whitelist_platform', None) or "xiequ" + credentials = getattr(cfg, 'whitelist_credentials', None) + + # 向后兼容：旧字段有值但新字段为空时，自动迁移 + if not credentials and cfg.whitelist_uid and cfg.whitelist_ukey: + platform = "xiequ" + credentials = {"uid": cfg.whitelist_uid, "ukey": cfg.whitelist_ukey} + + return { + "whitelist_platform": platform, + "whitelist_credentials": credentials if cfg.whitelist_enabled else None, + } + + class ProxyService: """代理 & 白名单服务：管理代理配置、执行测试。""" @@ -31,12 +53,26 @@ class ProxyService: db.add(cfg) db.commit() db.refresh(cfg) + + # 应用层自动迁移：旧字段有值但新字段为空时，填充新字段 + if cfg.whitelist_uid and cfg.whitelist_ukey and not cfg.whitelist_credentials: + cfg.whitelist_platform = "xiequ" + cfg.whitelist_credentials = {"uid": cfg.whitelist_uid, "ukey": cfg.whitelist_ukey} + db.commit() + return cfg @staticmethod - def update_config(db: Session, enabled, api_url, http, https, - whitelist_enabled, whitelist_uid, whitelist_ukey, - current_user) -> ProxyConfigModel: + def update_config( + db: Session, + enabled, api_url, http, https, + whitelist_enabled, + whitelist_platform="xiequ", + whitelist_credentials=None, + whitelist_uid=None, + whitelist_ukey=None, + current_user=None, + ) -> ProxyConfigModel: """更新代理配置并记录审计日志。""" cfg = ProxyService.get_or_create(db) cfg.enabled = enabled @@ -44,18 +80,29 @@ class ProxyService: cfg.http = http cfg.https = https cfg.whitelist_enabled = whitelist_enabled - cfg.whitelist_uid = whitelist_uid - cfg.whitelist_ukey = whitelist_ukey + cfg.whitelist_platform = whitelist_platform + cfg.whitelist_credentials = whitelist_credentials + + # 双写：协固平台同步到旧字段，其他平台清空旧字段 + if whitelist_platform == "xiequ" and whitelist_credentials: + cfg.whitelist_uid = whitelist_credentials.get("uid", "") + cfg.whitelist_ukey = whitelist_credentials.get("ukey", "") + else: + # 非协固平台，旧字段使用传入值或清空 + cfg.whitelist_uid = whitelist_uid or "" + cfg.whitelist_ukey = whitelist_ukey or "" + db.commit() db.refresh(cfg) - db.add(AuditLog( - user_id=current_user.id, - username=current_user.username, - action="proxy:update", - target="proxy_config", - )) - db.commit() + if current_user: + db.add(AuditLog( + user_id=current_user.id, + username=current_user.username, + action="proxy:update", + target="proxy_config", + )) + db.commit() return cfg # ---- 测试任务管理 ---- @@ -112,13 +159,12 @@ class ProxyService: # API代理 if cfg.api_url: - whitelist_uid = cfg.whitelist_uid if cfg.whitelist_enabled else "" - whitelist_ukey = cfg.whitelist_ukey if cfg.whitelist_enabled else "" + wl_params = _build_whitelist_params(cfg) proxy_url, msg = resolve_working_proxy( api_url=cfg.api_url, - whitelist_uid=whitelist_uid, - whitelist_ukey=whitelist_ukey, + whitelist_platform=wl_params["whitelist_platform"], + whitelist_credentials=wl_params["whitelist_credentials"], max_attempts=3, log_func=push, ) @@ -145,7 +191,6 @@ class ProxyService: ): """在线程中执行白名单测试。""" import requests as req_lib - from core.douyu.whitelist import WhitelistManager def push(level, message): asyncio.run_coroutine_threadsafe( @@ -158,16 +203,28 @@ class ProxyService: push("error", "白名单未启用") push("result", "") return - if not cfg.whitelist_uid or not cfg.whitelist_ukey: - push("error", "未配置白名单UID/UKEY") + + # 构建适配器 + wl_params = _build_whitelist_params(cfg) + credentials = wl_params["whitelist_credentials"] + platform = wl_params["whitelist_platform"] + + if not credentials: + push("error", "未配置白名单凭据") push("result", "") return - manager = WhitelistManager(cfg.whitelist_uid, cfg.whitelist_ukey) + adapter = create_adapter(platform, credentials) + if not adapter: + push("error", f"不支持的白名单平台: {platform}") + push("result", "") + return + + push("info", f"当前白名单平台: {adapter.platform_label}") # 1. 测试API连接 push("info", "测试白名单API连接...") - ok, msg = manager.test_connection() + ok, msg = adapter.test_connection() push("info" if ok else "error", f"白名单API: {msg}") if not ok: push("result", "") @@ -190,19 +247,7 @@ class ProxyService: if not local_ip: push("info", "通过IP检测服务获取本机公网IP...") - for url in [ - 'https://qifu-api.baidubce.com/ip/local/geo/v1/district', - 'https://myip.ipip.net', - 'https://4.ipw.cn', - ]: - try: - resp = req_lib.get(url, timeout=6, headers={"User-Agent": "Mozilla/5.0"}) - match = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', resp.text) - if match: - local_ip = match.group(1) - break - except Exception: - continue + local_ip = _get_local_exit_ip() if not local_ip: push("error", "无法获取本机公网IP") @@ -212,22 +257,24 @@ class ProxyService: push("info", f"本机公网IP: {local_ip}") # 3. 检查并同步白名单 - records = manager.get_whitelist_json() - in_list = any(r.get('IP') == local_ip for r in records) - push("info", f"白名单共 {len(records)} 条记录") + records = adapter.get_whitelist() + in_list = any(r.get('ip') == local_ip for r in records) + + if records: + push("info", f"白名单共 {len(records)} 条记录") if in_list: - record = next((r for r in records if r.get('IP') == local_ip), {}) - memo = record.get('MEMO', '') - if memo == manager.memo: + record = next((r for r in records if r.get('ip') == local_ip), {}) + memo = record.get('memo', '') + if memo == adapter.memo: push("success", f"本机IP {local_ip} 已在白名单中 (备注正确)") else: push("warning", f'本机IP {local_ip} 备注不匹配 (当前: "{memo}")，更新中...') - sync_ok, sync_msg = manager.sync_ip(local_ip) + sync_ok, sync_msg = adapter.sync_ip(local_ip) push("success" if sync_ok else "error", f"白名单更新: {sync_msg}") else: push("info", f"正在将 {local_ip} 添加到白名单...") - sync_ok, sync_msg = manager.sync_ip(local_ip) + sync_ok, sync_msg = adapter.sync_ip(local_ip) push("success" if sync_ok else "error", f"白名单同步: {sync_msg}") push("result", "") diff --git a/web/frontend/src/api/proxy.ts b/web/frontend/src/api/proxy.ts index cfe797c..2670ed8 100644 --- a/web/frontend/src/api/proxy.ts +++ b/web/frontend/src/api/proxy.ts @@ -1,9 +1,10 @@ import api from './client'; -import type { ProxyConfig, ProxyTestResult } from './types'; +import type { ProxyConfig, ProxyTestResult, PlatformInfo } from './types'; export const proxyApi = { get: () => api.get('/proxy'), update: (data: ProxyConfig) => api.put('/proxy', data), test: () => api.post('/proxy/test'), testWhitelist: () => api.post('/proxy/whitelist/test'), + getPlatforms: () => api.get('/proxy/platforms'), }; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index aff53c3..0bf3a62 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -112,6 +112,9 @@ export interface ProxyConfig { http: string; https: string; whitelist_enabled: boolean; + whitelist_platform: string; + whitelist_credentials: Record | null; + // 旧字段保留（向后兼容） whitelist_uid: string; whitelist_ukey: string; } @@ -121,6 +124,18 @@ export interface ProxyTestResult { success: boolean; } +export interface PlatformFieldDef { + key: string; + label: string; + placeholder: string; +} + +export interface PlatformInfo { + name: string; + label: string; + credential_fields: PlatformFieldDef[]; +} + // ==================== Permissions ==================== export interface PermissionsListResult { diff --git a/web/frontend/src/pages/ProxyPage.tsx b/web/frontend/src/pages/ProxyPage.tsx index 1401f99..b909c8c 100644 --- a/web/frontend/src/pages/ProxyPage.tsx +++ b/web/frontend/src/pages/ProxyPage.tsx @@ -1,6 +1,6 @@ import { useEffect, useState, useCallback } from 'react'; -import { Form, Input, Switch, Button, Card, message, Row, Col } from 'antd'; -import { proxyApi, type ProxyConfig } from '../api/modules'; +import { Form, Input, Switch, Button, Card, Select, message, Row, Col } from 'antd'; +import { proxyApi, type ProxyConfig, type PlatformInfo } from '../api/modules'; import RealtimeLogPanel from '../components/RealtimeLogPanel'; import { useWebSocketLogs } from '../hooks/useWebSocketLogs'; import { getErrorMessage } from '../utils/error'; @@ -13,6 +13,19 @@ export default function ProxyPage() { const [configLoaded, setConfigLoaded] = useState(false); const { logs, connect: connectLogs, close: closeLogs } = useWebSocketLogs(); + // 平台相关状态 + const [platforms, setPlatforms] = useState([]); + const [selectedPlatform, setSelectedPlatform] = useState('xiequ'); + const [credentials, setCredentials] = useState>({}); + + // 加载平台列表 + useEffect(() => { + proxyApi.getPlatforms().then(setPlatforms).catch(() => {}); + }, []); + + // 当前平台的凭据字段定义 + const currentPlatformFields = platforms.find(p => p.name === selectedPlatform)?.credential_fields ?? []; + const loadConfig = useCallback(async () => { try { const data = await proxyApi.get(); @@ -22,9 +35,19 @@ export default function ProxyPage() { http: data.http ?? '', https: data.https ?? '', whitelist_enabled: data.whitelist_enabled ?? false, - whitelist_uid: data.whitelist_uid ?? '', - whitelist_ukey: data.whitelist_ukey ?? '', }); + + // 恢复平台和凭据 + const platform = data.whitelist_platform || 'xiequ'; + setSelectedPlatform(platform); + if (data.whitelist_credentials && Object.keys(data.whitelist_credentials).length > 0) { + setCredentials(data.whitelist_credentials); + } else if (data.whitelist_uid || data.whitelist_ukey) { + // 向后兼容：旧字段迁移到凭据 + setCredentials({ uid: data.whitelist_uid || '', ukey: data.whitelist_ukey || '' }); + } else { + setCredentials({}); + } } catch (e: unknown) { message.error(getErrorMessage(e)); } finally { @@ -56,7 +79,15 @@ export default function ProxyPage() { setLoading(true); try { const values = await form.validateFields(); - await proxyApi.update(values as ProxyConfig); + const submitData: ProxyConfig = { + ...values, + whitelist_platform: selectedPlatform, + whitelist_credentials: credentials, + // 旧字段：协固平台双写，其他平台清空 + whitelist_uid: selectedPlatform === 'xiequ' ? (credentials.uid || '') : '', + whitelist_ukey: selectedPlatform === 'xiequ' ? (credentials.ukey || '') : '', + }; + await proxyApi.update(submitData); message.success('已保存'); } catch (e: unknown) { message.error(getErrorMessage(e)); @@ -87,6 +118,15 @@ export default function ProxyPage() { } }; + const handlePlatformChange = (value: string) => { + setSelectedPlatform(value); + setCredentials({}); + }; + + const handleCredentialChange = (key: string, value: string) => { + setCredentials(prev => ({ ...prev, [key]: value })); + }; + return (
@@ -106,8 +146,6 @@ export default function ProxyPage() { api_url: '', http: '', https: '', - whitelist_uid: '', - whitelist_ukey: '', }} style={{ flexShrink: 0 }} > @@ -118,7 +156,7 @@ export default function ProxyPage() { - + @@ -134,12 +172,30 @@ export default function ProxyPage() { - - - - - - + + {/* 平台选择 */} +
+
代理平台
+ handleCredentialChange(field.key, e.target.value)} + /> +
+ ))} +