diff --git a/.gitignore b/.gitignore index 19ca86c..747d4a3 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ __pycache__/ data/ logs/ *.db +*.log # geetest 临时图片 bg.jpg diff --git a/core/douyu/login.py b/core/douyu/login.py index e026582..c9e548a 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -22,6 +22,7 @@ from core.geetest.common.network import ( get_c_s, req_fullpage_validate, ) +from utils.http_logger import log_http class LoginResult: @@ -62,6 +63,9 @@ class DouyuLogin: timeout: tuple[float, float] = REQUEST_TIMEOUT, max_geetest_retries: int = 5, max_proxy_retries: int = 10, + whitelist_uid: str = "", + whitelist_ukey: str = "", + proxy_manager: Optional[ProxyManager] = None, ): self.account = account self.proxy = proxy @@ -70,9 +74,19 @@ class DouyuLogin: self.max_proxy_retries = max_proxy_retries # 0=无限切换直到成功 self.session = requests.Session() - # 初始化代理管理器 - self.proxy_manager = get_proxy_manager(proxy_api_url) if proxy_api_url else None + # 初始化代理管理器（优先使用外部传入的共享实例，避免并发刷新冲突） + if proxy_manager: + self.proxy_manager = proxy_manager + elif proxy_api_url: + self.proxy_manager = get_proxy_manager( + proxy_api_url, + whitelist_uid=whitelist_uid, + whitelist_ukey=whitelist_ukey, + ) + else: + self.proxy_manager = None + self._current_proxy_url: Optional[str] = None self._setup_session() def _setup_session(self) -> None: @@ -93,13 +107,14 @@ class DouyuLogin: self._apply_proxy() def _apply_proxy(self, proxy: str = None) -> None: - """应用代理到Session""" + """应用代理到Session，并记录当前代理URL供刷新时mark_bad""" if proxy: # 使用指定的代理 self.session.proxies = { 'http': proxy, 'https': proxy, } + self._current_proxy_url = proxy elif self.proxy: # 使用配置的代理 if isinstance(self.proxy, str): @@ -107,12 +122,14 @@ class DouyuLogin: 'http': self.proxy, 'https': self.proxy, } + self._current_proxy_url = self.proxy else: self.session.proxies = { scheme: url for scheme, url in self.proxy.items() if url } + self._current_proxy_url = self.proxy.get('http') or self.proxy.get('https') else: # 从代理管理器获取代理 if self.proxy_manager: @@ -122,12 +139,17 @@ class DouyuLogin: 'http': new_proxy, 'https': new_proxy, } + self._current_proxy_url = new_proxy def _refresh_proxy(self) -> Optional[str]: - """刷新代理IP""" + """刷新代理IP：先标记当前代理为坏（移出代理池），再获取新代理""" if not self.proxy_manager: return None + # 标记当前代理为不可用，避免再被复用 + if self._current_proxy_url: + self.proxy_manager.mark_bad(self._current_proxy_url) + new_proxy = self.proxy_manager.get_proxy() if new_proxy: self._apply_proxy(new_proxy) @@ -143,9 +165,14 @@ class DouyuLogin: """ 统一发送请求，附带分段超时和更明确的错误信息。 代理连接失败时自动重试获取新的代理IP。 + 所有请求详情会记录到 HTTP 详情日志。 """ timeout = kwargs.pop('timeout', self.timeout) safe_url = self._safe_url(url) + req_data = kwargs.get('data') + req_body = req_data if req_data else kwargs.get('json') + current_proxy = self._current_proxy_url + tag = self.account.username if self.account else "" for attempt in range(max_retries): started = time.monotonic() @@ -157,28 +184,56 @@ class DouyuLogin: f"{method.upper()} {safe_url} -> {response.status_code} " f"({elapsed:.2f}s)" ) + # 记录请求/响应详情 + resp_body = response.text[:500] if response.text else "" + log_http( + category="douyu_login", + method=method, + url=safe_url, + request_headers=dict(self.session.headers), + request_body=req_body, + status_code=response.status_code, + response_headers=dict(response.headers), + response_body=resp_body, + duration=elapsed, + proxy=current_proxy, + tag=tag, + ) return response except requests.Timeout as exc: elapsed = time.monotonic() - started + log_http( + category="douyu_login", method=method, url=safe_url, + request_headers=dict(self.session.headers), request_body=req_body, + duration=elapsed, error=f"超时 timeout={timeout}: {exc}", + proxy=current_proxy, tag=tag, + ) raise TimeoutError( f"{method.upper()} {safe_url} 超时，耗时 {elapsed:.1f}s，" f"timeout={timeout}" ) from exc except requests.ConnectionError as exc: - # requests 2.34+ 已移除 ProxyError，代理错误统一为 ConnectionError - is_proxy_err = "proxy" in str(exc).lower() or "Proxy" in type(exc).__name__ + elapsed = time.monotonic() - started + err_str = str(exc) + is_proxy_err = "proxy" in err_str.lower() or "Proxy" in type(exc).__name__ + log_http( + category="douyu_login", method=method, url=safe_url, + request_headers=dict(self.session.headers), request_body=req_body, + duration=elapsed, + error=f"{'代理' if is_proxy_err else ''}连接失败: {err_str[:300]}", + proxy=current_proxy, tag=tag, + ) if is_proxy_err: - elapsed = time.monotonic() - started logger.warning(f"代理连接失败，尝试 {attempt + 1}/{max_retries}: {exc}") if attempt < max_retries - 1: self._refresh_proxy() + current_proxy = self._current_proxy_url time.sleep(1) continue raise ConnectionError( f"{method.upper()} {safe_url} 代理连接失败，已重试 {max_retries} 次" ) from exc # 非代理的 ConnectionError 也重试 - elapsed = time.monotonic() - started logger.warning(f"连接失败，尝试 {attempt + 1}/{max_retries}: {exc}") if attempt < max_retries - 1: time.sleep(1) @@ -188,6 +243,12 @@ class DouyuLogin: ) from exc except requests.RequestException as exc: elapsed = time.monotonic() - started + log_http( + category="douyu_login", method=method, url=safe_url, + request_headers=dict(self.session.headers), request_body=req_body, + duration=elapsed, error=f"请求异常: {exc}", + proxy=current_proxy, tag=tag, + ) raise ConnectionError( f"{method.upper()} {safe_url} 请求失败，耗时 {elapsed:.1f}s: {exc}" ) from exc diff --git a/core/douyu/proxy.py b/core/douyu/proxy.py index e1ba608..e290331 100644 --- a/core/douyu/proxy.py +++ b/core/douyu/proxy.py @@ -1,6 +1,8 @@ """代理管理模块""" +import json import re +import threading import time import requests from typing import Optional @@ -8,56 +10,152 @@ from loguru import logger class ProxyManager: - """代理管理器""" + """代理管理器（带已验证代理池缓存，批次内共享复用）""" - def __init__(self, api_url: str = ""): + def __init__(self, api_url: str = "", whitelist_uid: str = "", whitelist_ukey: str = ""): self.api_url = api_url + self.whitelist_uid = whitelist_uid + self.whitelist_ukey = whitelist_ukey self.current_proxy: Optional[str] = None + self._cond = threading.Condition() + # 已验证可用的代理池: {proxy_url: validated_timestamp} + self._verified_pool: dict[str, float] = {} + self._pool_ttl = 90 # 代理验证后90秒内可复用 + self._fetching = False # 是否有线程正在获取代理 - def get_proxy(self) -> Optional[str]: + def _pick_from_pool_locked(self) -> Optional[str]: + """从池中取一个未过期的代理（调用前需持有锁）""" + now = time.time() + # 清理过期代理 + expired = [p for p, t in self._verified_pool.items() if now - t > self._pool_ttl] + for p in expired: + del self._verified_pool[p] + # 取第一个可用的 + for proxy in self._verified_pool: + self.current_proxy = proxy + return proxy + return None + + def get_proxy(self, max_attempts: int = 5) -> Optional[str]: """ - 从代理API获取代理IP，支持API返回多个IP（多行格式）， - 逐一验证返回第一个可用的。 + 获取可用代理IP，优先从已验证代理池复用。 + 线程安全：池空时只有一个线程调API获取并验证所有代理入池，其他线程等待后复用。 Returns: 代理URL，格式: http://ip:port """ - try: - logger.info("获取代理IP...") - response = requests.get(self.api_url, timeout=10) - response.raise_for_status() - - text = response.text.strip() - logger.debug(f"代理API响应: {text}") - - proxy_urls, _ = parse_proxy_response(text) - - if not proxy_urls: - logger.warning(f"无法解析代理地址: {text}") - return None - - if len(proxy_urls) == 1: - proxy = proxy_urls[0] - self.current_proxy = proxy - logger.info(f"获取到代理: {proxy}") + with self._cond: + # 1. 优先从池中取未过期的 + proxy = self._pick_from_pool_locked() + if proxy: + logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})") return proxy - - # 多个代理逐一验证，返回第一个可用的 - logger.info(f"获取到 {len(proxy_urls)} 个代理，逐一验证") - for proxy in proxy_urls: - if self.verify_proxy(proxy): - self.current_proxy = proxy - logger.success(f"可用代理: {proxy}") + # 2. 已有线程在获取，等待结果 + 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 - logger.warning(f"代理 {proxy} 不可用，尝试下一个") + return None + # 3. 自己去获取 + self._fetching = True - logger.warning("所有代理均不可用") - self.current_proxy = None - return None + # 释放锁后执行耗时的API调用+验证 + try: + proxies_all, msg = self._fetch_and_verify_all(max_attempts) + finally: + with self._cond: + self._fetching = False + self._cond.notify_all() - except Exception as e: - logger.error(f"获取代理失败: {e}") - return None + if proxies_all: + with self._cond: + now = time.time() + for p in proxies_all: + self._verified_pool[p] = now + first = proxies_all[0] + self.current_proxy = first + logger.info(f"代理池充盈 {len(proxies_all)} 个可用代理: {first} ...") + return first + logger.warning(f"获取代理失败: {msg}") + return None + + def _fetch_and_verify_all(self, max_attempts: int) -> tuple[Optional[list[str]], str]: + """调代理API获取一批代理，并发验证所有可用代理，自动处理白名单同步。""" + last_error = "" + last_synced_ip: Optional[str] = None + + for attempt in range(1, max_attempts + 1): + if attempt > 1: + time.sleep(min(attempt - 1, 2)) + + # 每次尝试前主动同步白名单（出口IP可能漂移，需重新同步） + if self.whitelist_uid and self.whitelist_ukey: + from core.douyu.whitelist import get_local_exit_ip, WhitelistManager + local_ip = get_local_exit_ip() + if local_ip and local_ip != last_synced_ip: + logger.info(f"[尝试 {attempt}] 检测出口IP: {local_ip}，同步白名单...") + manager = WhitelistManager(self.whitelist_uid, self.whitelist_ukey) + ok, sync_msg = manager.sync_ip(local_ip) + if ok: + last_synced_ip = local_ip + logger.info(f"白名单同步成功: {sync_msg}") + # 等待白名单生效 + time.sleep(2) + else: + logger.warning(f"白名单同步失败: {sync_msg}") + elif local_ip == last_synced_ip: + logger.debug(f"[尝试 {attempt}] 出口IP未变: {local_ip}") + + try: + response = requests.get(self.api_url, timeout=10) + response.raise_for_status() + text = response.text.strip() + + proxy_urls, whitelist_ip = parse_proxy_response(text) + + if proxy_urls: + logger.info(f"获取到 {len(proxy_urls)} 个代理，并发验证所有") + available, msg = verify_proxies_concurrent(proxy_urls, return_all=True) + if available: + return available, msg + last_error = msg + logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}") + continue + + # 代理API返回白名单错误 + if whitelist_ip and self.whitelist_uid and self.whitelist_ukey: + logger.warning(f"代理需要白名单IP: {whitelist_ip}，自动同步...") + from core.douyu.whitelist import WhitelistManager + manager = WhitelistManager(self.whitelist_uid, self.whitelist_ukey) + ok, sync_msg = manager.sync_ip(whitelist_ip) + if ok: + last_synced_ip = whitelist_ip + logger.info("白名单已更新，等待2秒后重试...") + time.sleep(2) + continue + return None, f'白名单同步失败: {sync_msg}' + + last_error = '代理API响应无法解析' + logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}: {text[:80]}") + + except Exception as exc: + last_error = f'代理API请求失败: {exc}' + logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}") + + return None, f'代理预检失败（{max_attempts}次尝试均失败）: {last_error}' + + def mark_bad(self, proxy_url: str) -> None: + """标记代理为不可用，从池中移除（极验失败/代理连接失败时调用）""" + with self._cond: + removed = self._verified_pool.pop(proxy_url, None) + if self.current_proxy == proxy_url: + self.current_proxy = None + if removed: + logger.info(f"代理标记为不可用并移出池: {proxy_url} (池剩余 {len(self._verified_pool)})") + self._cond.notify_all() def get_proxies_dict(self, proxy: str = None) -> dict: """获取requests使用的proxies字典""" @@ -87,94 +185,142 @@ class ProxyManager: return False -def get_proxy_manager(api_url: str = "") -> ProxyManager: - """获取代理管理器实例（每次传入 api_url 时创建新实例，避免全局状态污染）""" - return ProxyManager(api_url) +def get_proxy_manager( + api_url: str = "", + whitelist_uid: str = "", + whitelist_ukey: str = "", +) -> ProxyManager: + """获取代理管理器实例""" + return ProxyManager(api_url, whitelist_uid=whitelist_uid, whitelist_ukey=whitelist_ukey) def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]: """ - 解析代理API响应，支持返回多个代理地址。 + 解析代理API响应，支持 JSON 格式与旧版纯文本格式。 + + JSON 格式示例： + 正常: {"code":0,"success":"true","msg":"","data":[{"IP":"1.2.3.4","Port":5791,...}]} + 白名单错误: {"code":-1,"success":"true","msg":"51.请先添加白名单：39.144.109.21","data":""} + + 旧版纯文本格式（兼容）： + 正常: 1.2.3.4:5791\\n5.6.7.8:8080 + 白名单错误: 请先添加白名单：39.144.109.21 Returns: (proxy_urls, whitelist_ip) - proxy_urls: 解析到的所有代理地址列表（http://ip:port） - whitelist_ip: 需要添加到白名单的IP（当API返回白名单错误时），无错误时为 None """ - text = text.strip() + text = (text or "").strip() + if not text: + return [], None - # 白名单错误优先检测 + # 优先尝试 JSON 解析 + try: + data = json.loads(text) + if isinstance(data, dict): + # 白名单错误：code != 0 且 msg 含白名单提示 + code = data.get("code") + msg = data.get("msg", "") or "" + if code != 0 and ("白名单" in msg or "添加白名单" in msg): + ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', msg) + if ip_match: + return [], ip_match.group(1) + + # 正常返回：从 data 数组提取 IP/Port + data_field = data.get("data") + proxies: list[str] = [] + if isinstance(data_field, list): + for item in data_field: + if not isinstance(item, dict): + continue + ip = item.get("IP") or item.get("ip") + port = item.get("Port") or item.get("port") + if ip and port: + proxies.append(f"http://{ip}:{port}") + if proxies: + return proxies, None + + # data 为空但 code==0，可能代理暂时不可用 + if code == 0 and not proxies: + return [], None + except (json.JSONDecodeError, ValueError): + # 不是 JSON，回退到文本解析 + pass + + # 旧版文本格式：白名单错误优先检测 if '添加白名单' in text or '白名单' in text: ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', text) if ip_match: return [], ip_match.group(1) - # 解析所有 ip:port，支持多行格式 + # 旧版文本格式：解析所有 ip:port matches = re.findall(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text) - proxies = [] - for ip, port in matches: - # 排除白名单提示中误匹配的 - proxy = f"http://{ip}:{port}" - proxies.append(proxy) - + proxies = [f"http://{ip}:{port}" for ip, port in matches] if proxies: return proxies, None return [], None -def verify_proxy_url(proxy_url: str, timeout: tuple = (4, 6)) -> tuple[bool, str]: +def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str]: """ - 验证代理是否可用。 - - 验证优先级： - 1. 斗鱼主站（最相关，能访问斗鱼才是最终目的） - 2. myip（快速 IP 验证） - 3. 百度 IP 查询（备用） + 验证代理是否可用，只验证斗鱼主站（登录的最终目标）。 Returns: (是否可用, 消息) """ + from utils.http_logger import log_http + import time as _time + proxies = {'http': proxy_url, 'https': proxy_url} - targets = [ - ('https://www.douyu.com', '斗鱼主站'), - ('https://myip.ipip.net', 'IP验证'), - ('https://qifu-api.baidubce.com/ip/local/geo/v1/district', '百度IP查询'), - ] - - for url, label in targets: - try: - response = requests.get( - url, proxies=proxies, timeout=timeout, - headers={'User-Agent': 'Mozilla/5.0'}, - ) - response.raise_for_status() - return True, f'代理可用 → {label}' - except Exception as e: - err_msg = str(e) - if 'Tunnel connection failed' in err_msg or '503' in err_msg: - detail = '代理拒绝连接（白名单可能未生效）' - elif 'timed out' in err_msg.lower(): - detail = '连接超时' - else: - detail = type(e).__name__ - logger.debug(f"代理验证 {label} 失败: {detail}") - continue - - return False, '代理验证失败（所有目标不可达）' + started = _time.monotonic() + try: + response = requests.get( + 'https://www.douyu.com', + proxies=proxies, + timeout=timeout, + headers={'User-Agent': 'Mozilla/5.0'}, + ) + elapsed = _time.monotonic() - started + response.raise_for_status() + log_http( + category="proxy_verify", method="GET", url="https://www.douyu.com", + status_code=response.status_code, response_body=f"代理可用: {proxy_url}", + duration=elapsed, proxy=proxy_url, tag="proxy_verify", + ) + return True, '代理可用 → 斗鱼主站' + except Exception as e: + elapsed = _time.monotonic() - started + err_msg = str(e) + if 'Tunnel connection failed' in err_msg or '503' in err_msg: + detail = '代理拒绝连接（白名单可能未生效）' + elif 'timed out' in err_msg.lower(): + detail = '连接超时' + else: + detail = type(e).__name__ + logger.debug(f"代理验证失败 [{proxy_url}]: {detail}") + log_http( + category="proxy_verify", method="GET", url="https://www.douyu.com", + duration=elapsed, error=f"[{proxy_url}] {detail}: {err_msg[:200]}", + proxy=proxy_url, tag="proxy_verify", + ) + return False, detail -def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (4, 6), max_workers: int = 5) -> tuple[Optional[str], str]: +def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (5, 8), max_workers: int = 5, return_all: bool = False) -> tuple[Optional[str | list[str]], str]: """ - 并发验证多个代理URL，返回第一个可用的。 + 并发验证多个代理URL。 Args: proxy_urls: 代理URL列表 timeout: 验证超时 max_workers: 最大并发数 + return_all: True 时返回所有可用代理列表；False（默认）返回第一个可用的 Returns: - (可用的代理URL, 消息) + return_all=False: (可用代理URL或None, 消息) + return_all=True: (可用代理URL列表或None, 消息) """ if not proxy_urls: return None, '无代理可验证' @@ -182,11 +328,32 @@ def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (4, 6), ma if len(proxy_urls) == 1: ok, msg = verify_proxy_url(proxy_urls[0], timeout) if ok: - return proxy_urls[0], msg + return (proxy_urls if return_all else proxy_urls[0]), msg return None, msg from concurrent.futures import ThreadPoolExecutor, as_completed + if return_all: + # 收集所有可用代理（不取消任何任务） + available: list[str] = [] + with ThreadPoolExecutor(max_workers=min(max_workers, len(proxy_urls))) as executor: + future_map = { + executor.submit(verify_proxy_url, p, timeout): p + for p in proxy_urls + } + for future in as_completed(future_map): + try: + ok, _ = future.result() + if ok: + available.append(future_map[future]) + except Exception: + continue + if available: + logger.success(f"并发验证找到 {len(available)}/{len(proxy_urls)} 个可用代理") + return available, f'找到 {len(available)} 个可用代理' + return None, f'共 {len(proxy_urls)} 个代理均不可用' + + # 默认：返回第一个可用的，取消其余 with ThreadPoolExecutor(max_workers=min(max_workers, len(proxy_urls))) as executor: future_map = { executor.submit(verify_proxy_url, p, timeout): p @@ -198,7 +365,6 @@ def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (4, 6), ma ok, msg = future.result() if ok: logger.success(f"并发验证找到可用代理: {proxy_url}") - # 取消剩余任务 for f in future_map: if f != future: f.cancel() @@ -213,7 +379,7 @@ def resolve_working_proxy( api_url: str, whitelist_uid: str = "", whitelist_ukey: str = "", - max_attempts: int = 3, + max_attempts: int = 4, log_func=None, ) -> tuple[Optional[str], str]: """ @@ -224,7 +390,7 @@ def resolve_working_proxy( api_url: 代理API地址 whitelist_uid: 白名单UID（启用白名单时传入） whitelist_ukey: 白名单UKEY - max_attempts: 最大获取尝试次数（默认3次） + max_attempts: 最大获取尝试次数（默认4次） log_func: 日志回调函数 (level, message) Returns: @@ -242,7 +408,7 @@ def resolve_working_proxy( for attempt in range(1, max_attempts + 1): # 重试之间增加退避延迟，避免代理API返回同一个不可用IP if attempt > 1: - delay = min(attempt, 3) + delay = min(attempt - 1, 2) log('info', f'等待 {delay}s 后重试...') time.sleep(delay) diff --git a/core/douyu/whitelist.py b/core/douyu/whitelist.py index 3865605..01481c7 100644 --- a/core/douyu/whitelist.py +++ b/core/douyu/whitelist.py @@ -2,6 +2,7 @@ import json import re +import threading import time from typing import Optional from urllib.parse import urlencode @@ -9,6 +10,9 @@ from urllib.parse import urlencode import requests from loguru import logger +# 全局锁：防止多个并发登录任务同时同步白名单，触发代理服务商的30秒限流 +_whitelist_sync_lock = threading.Lock() + class WhitelistManager: """协固代理IP白名单管理器""" @@ -165,60 +169,65 @@ class WhitelistManager: records = self.get_whitelist_json() return [r for r in records if r.get("MEMO") == self._memo] - def sync_ip(self, current_ip: str) -> tuple[bool, str]: + def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]: """ - 同步白名单IP + 同步白名单IP（保留多个近期出口IP，应对移动网络IP漂移） - 检查当前备注是否有记录： - - 如果IP相同，无需操作 - - 如果IP不同，删除旧的并添加新的 - - 如果IP已存在但备注不同（如手动添加无备注），删除后重新添加 - - 如果无记录，添加新的 + 策略： + - 当前IP已在白名单（同备注），无需操作 + - 当前IP不在白名单，添加（不删旧的，保留多个出口IP） + - 超过 keep_recent 个同备注IP时，删除最老的（按列表顺序） + - 当前IP已存在但备注不同，不处理（避免误删他人配置） + + 线程安全：使用全局锁串行化同步操作，避免并发调用触发代理服务商限流。 Args: current_ip: 当前出口IP + keep_recent: 保留的同备注IP数量上限（默认3个，应对IP漂移） Returns: (是否成功, 消息) """ - try: - existing_ip = self.get_memo_ip() + 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 existing_ip == current_ip: - msg = f"白名单IP已是最新的: {current_ip}" - logger.info(msg) - return True, msg + # 当前IP已在白名单（同备注），无需操作 + if current_ip in memo_ips: + msg = f"白名单IP已是最新的: {current_ip}" + logger.info(msg) + return True, msg - # 有旧记录，先删除 - if existing_ip: - logger.info(f"白名单IP变化: {existing_ip} -> {current_ip}") - self.delete_ip(existing_ip) - time.sleep(1) + # 当前IP已存在但备注不同，不处理（避免误删他人配置） + if any(r.get('IP') == current_ip for r in records): + logger.info(f"白名单IP {current_ip} 已存在（备注不同），无需重复添加") + return True, f"白名单IP已存在: {current_ip}" - # IP已存在但备注不同（或备注为空），先删除后重新添加 - records = self.get_whitelist_json() - if any(r.get('IP') == current_ip for r in records): - logger.info(f"白名单IP {current_ip} 已存在但备注不同，删除后重新添加") - self.delete_ip(current_ip) - time.sleep(1) + # 超过上限，删除最老的（列表前面的） + 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 - ok, resp = self.add_ip(current_ip) - if ok: - if existing_ip: - msg = f"白名单IP已更新: {existing_ip} -> {current_ip}" - else: + # 添加新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 + logger.info(msg) + return True, msg - return False, f"白名单添加失败，API响应: {resp}" + return False, f"白名单添加失败，API响应: {resp}" - except Exception as e: - msg = f"白名单同步失败: {e}" - logger.error(msg) - return False, msg + except Exception as e: + msg = f"白名单同步失败: {e}" + logger.error(msg) + return False, msg def test_connection(self) -> tuple[bool, str]: """ @@ -242,6 +251,34 @@ class WhitelistManager: return False, msg +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 diff --git a/core/geetest/common/network.py b/core/geetest/common/network.py index 38df9ba..d329379 100644 --- a/core/geetest/common/network.py +++ b/core/geetest/common/network.py @@ -4,7 +4,7 @@ import json import re from typing import Mapping, Optional, Tuple -REQUEST_TIMEOUT = (3.05, 12) +REQUEST_TIMEOUT = (10, 30) PASSPORT_REFERER = "https://passport.douyu.com/" diff --git a/pyproject.toml b/pyproject.toml index 0653414..94bcc9b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ description = "斗鱼批量登录 Web 后台" readme = "README.md" requires-python = ">=3.12,<3.13" dependencies = [ - "requests>=2.31.0", + "requests[socks]>=2.31.0", "pycryptodome>=3.19.0", "numpy>=1.24.0", "opencv-python-headless>=4.8.0", diff --git a/utils/http_logger.py b/utils/http_logger.py new file mode 100644 index 0000000..dbc35c5 --- /dev/null +++ b/utils/http_logger.py @@ -0,0 +1,200 @@ +"""HTTP 请求/响应详情日志记录器 + +将关键 HTTP 请求的完整详情（method、url、headers、body、status、response、耗时） +以 JSON Lines 格式写入日志文件，便于在 Web 界面查看和排查问题。 +""" + +import json +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Optional + +from loguru import logger + +# 日志文件路径 +_LOG_DIR = Path("logs") +_LOG_FILE = _LOG_DIR / "http_detail.jsonl" +_MAX_BODY_LEN = 2000 # 单个 body 最大记录长度，避免过大 + +# 线程安全写锁 +_write_lock = threading.Lock() + +# 确保日志目录存在 +_LOG_DIR.mkdir(parents=True, exist_ok=True) + + +def _truncate(text: Any, max_len: int = _MAX_BODY_LEN) -> str: + """截断过长的文本""" + if text is None: + return "" + s = text if isinstance(text, str) else str(text) + if len(s) > max_len: + return s[:max_len] + f"...[truncated {len(s) - max_len} chars]" + return s + + +def _safe_headers(headers: Any) -> dict: + """清理 headers 中的敏感信息""" + if not headers: + return {} + if hasattr(headers, 'items'): + headers = dict(headers) + safe = {} + sensitive = {'authorization', 'cookie', 'set-cookie', 'password'} + for k, v in headers.items(): + if k.lower() in sensitive: + safe[k] = '***' + else: + safe[k] = v + return safe + + +def log_http( + category: str, + method: str, + url: str, + *, + request_headers: Any = None, + request_body: Any = None, + status_code: Optional[int] = None, + response_headers: Any = None, + response_body: Any = None, + duration: Optional[float] = None, + error: Optional[str] = None, + proxy: Optional[str] = None, + tag: str = "", +) -> None: + """ + 记录一条 HTTP 请求/响应详情日志。 + + Args: + category: 分类（如 'douyu_login', 'geetest', 'proxy_verify', 'whitelist'） + method: HTTP 方法 + url: 请求 URL（会被脱敏，移除敏感查询参数） + request_headers: 请求头 + request_body: 请求体 + status_code: 响应状态码 + response_headers: 响应头 + response_body: 响应体 + duration: 耗时（秒） + error: 错误信息 + proxy: 使用的代理 + tag: 额外标签（如账号名） + """ + entry = { + "timestamp": datetime.now().isoformat(timespec="milliseconds"), + "ts": time.time(), + "category": category, + "tag": tag, + "method": method.upper(), + "url": _truncate(url, 500), + "proxy": proxy, + "request": { + "headers": _safe_headers(request_headers), + "body": _truncate(request_body), + }, + "response": { + "status_code": status_code, + "headers": _safe_headers(response_headers), + "body": _truncate(response_body), + }, + "duration_ms": round(duration * 1000, 1) if duration is not None else None, + "error": _truncate(error, 500) if error else None, + } + + # 判断级别 + if error or (status_code and status_code >= 400): + entry["level"] = "error" + elif status_code and status_code >= 300: + entry["level"] = "warning" + else: + entry["level"] = "info" + + try: + line = json.dumps(entry, ensure_ascii=False) + with _write_lock: + with open(_LOG_FILE, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception as e: + logger.debug(f"写入HTTP详情日志失败: {e}") + + +def read_http_logs( + *, + limit: int = 100, + offset: int = 0, + category: Optional[str] = None, + level: Optional[str] = None, + keyword: Optional[str] = None, +) -> tuple[list[dict], int]: + """ + 读取 HTTP 详情日志，支持筛选和分页。 + + Args: + limit: 返回条数上限 + offset: 偏移量（从最新往前数） + category: 按分类筛选 + level: 按级别筛选（info/warning/error） + keyword: 关键词搜索（url、tag、error） + + Returns: + (日志条目列表, 总匹配条数)，列表按时间倒序（最新在前） + """ + if not _LOG_FILE.exists(): + return [], 0 + + entries: list[dict] = [] + try: + with open(_LOG_FILE, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + + # 筛选 + if category and entry.get("category") != category: + continue + if level and entry.get("level") != level: + continue + if keyword: + kw = keyword.lower() + searchable = " ".join([ + str(entry.get("url", "")), + str(entry.get("tag", "")), + str(entry.get("error", "")), + str(entry.get("method", "")), + ]).lower() + if kw not in searchable: + continue + + entries.append(entry) + except Exception as e: + logger.error(f"读取HTTP详情日志失败: {e}") + return [], 0 + + # 按时间倒序 + entries.sort(key=lambda x: x.get("ts", 0), reverse=True) + total = len(entries) + # 分页（offset 从最新开始算） + page = entries[offset:offset + limit] + return page, total + + +def clear_http_logs() -> int: + """清空 HTTP 详情日志，返回清空的条数""" + count = 0 + with _write_lock: + if _LOG_FILE.exists(): + try: + with open(_LOG_FILE, "r", encoding="utf-8") as f: + count = sum(1 for line in f if line.strip()) + except Exception: + pass + _LOG_FILE.write_text("", encoding="utf-8") + return count diff --git a/web/backend/main.py b/web/backend/main.py index c8fe990..7575cf3 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -6,7 +6,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from .database import init_db -from .routers import auth, users, accounts, login, proxy, cookies +from .routers import auth, users, accounts, login, proxy, cookies, logs @asynccontextmanager @@ -37,6 +37,7 @@ app.include_router(accounts.router) app.include_router(login.router) app.include_router(proxy.router) app.include_router(cookies.router) +app.include_router(logs.router) @app.get("/api/health") diff --git a/web/backend/routers/logs.py b/web/backend/routers/logs.py new file mode 100644 index 0000000..c4b1251 --- /dev/null +++ b/web/backend/routers/logs.py @@ -0,0 +1,46 @@ +"""请求日志路由 - 查看 HTTP 请求/响应详情日志""" + +from fastapi import APIRouter, Depends, Query +from typing import Optional + +from ..deps import get_current_user +from ..permissions import has_permission +from utils.http_logger import read_http_logs, clear_http_logs + +router = APIRouter(prefix="/api/logs", tags=["日志"]) + + +@router.get("/http") +def list_http_logs( + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + category: Optional[str] = None, + level: Optional[str] = None, + keyword: Optional[str] = None, + current=Depends(get_current_user), +): + """查看 HTTP 请求/响应详情日志（需要审计日志查看权限）""" + if not has_permission(current.role, "audit:view"): + # 运营也可以查看请求日志（用于排查登录问题） + if not has_permission(current.role, "login:batch"): + return {"items": [], "total": 0, "message": "无权限"} + + items, total = read_http_logs( + limit=limit, + offset=offset, + category=category, + level=level, + keyword=keyword, + ) + return {"items": items, "total": total} + + +@router.delete("/http") +def clear_http_logs_api(current=Depends(get_current_user)): + """清空 HTTP 请求/响应详情日志""" + if not has_permission(current.role, "audit:view"): + if not has_permission(current.role, "login:batch"): + return {"success": False, "message": "无权限"} + + count = clear_http_logs() + return {"success": True, "cleared": count} diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index febe8de..8047460 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -11,7 +11,7 @@ from sqlalchemy.orm import Session from core.douyu import DouyuLogin from core.models import Account, ProxyConfig as DouyuProxyConfig -from core.douyu.proxy import resolve_working_proxy +from core.douyu.proxy import resolve_working_proxy, get_proxy_manager from ..models import Account as AccountModel, LoginTask, ProxyConfig as ProxyConfigModel from ..permissions import has_permission @@ -47,6 +47,17 @@ class LoginBatchRunner: self._counter_lock = threading.Lock() self._completed = 0 + # 共享代理管理器（带锁，避免并发白名单限流；极验失败时可刷新代理） + 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 "" + self._shared_proxy_manager = get_proxy_manager( + proxy_config.api_url, + whitelist_uid=wl_uid, + whitelist_ukey=wl_ukey, + ) + def stop(self): self._stop.set() @@ -62,7 +73,7 @@ class LoginBatchRunner: 解析代理配置，返回 (proxy_dict, message)。 - 静态代理：直接返回 dict - - API代理：调用 resolve_working_proxy 预检，自动同步白名单 + - API代理：通过共享 ProxyManager（带已验证代理池缓存）获取，自动同步白名单 - 无代理：返回 (None, '') """ if not self.proxy_config or not self.proxy_config.enabled: @@ -73,23 +84,12 @@ class LoginBatchRunner: proxy_url = self.proxy_config.http or self.proxy_config.https return {'http': proxy_url, 'https': proxy_url}, f'使用静态代理: {proxy_url}' - # API代理：预检获取可用代理 - if self.proxy_config.api_url: - whitelist_uid = '' - whitelist_ukey = '' - if self.proxy_config.whitelist_enabled: - whitelist_uid = self.proxy_config.whitelist_uid or '' - whitelist_ukey = self.proxy_config.whitelist_ukey or '' - - proxy_url, msg = resolve_working_proxy( - api_url=self.proxy_config.api_url, - whitelist_uid=whitelist_uid, - whitelist_ukey=whitelist_ukey, - log_func=self._push_log, - ) + # API代理：通过共享代理管理器获取（优先从已验证代理池复用） + if self._shared_proxy_manager: + proxy_url = self._shared_proxy_manager.get_proxy() if proxy_url: - return {'http': proxy_url, 'https': proxy_url}, msg - return None, msg + return {'http': proxy_url, 'https': proxy_url}, f'使用API代理: {proxy_url}' + return None, '代理不可用' return None, '' @@ -144,6 +144,7 @@ class LoginBatchRunner: proxy=proxy_dict, max_geetest_retries=self.max_geetest_retries, max_proxy_retries=self.max_proxy_retries, + proxy_manager=self._shared_proxy_manager, ) result = loginer.login() diff --git a/web/frontend/src/App.tsx b/web/frontend/src/App.tsx index 651e7bc..8102b2b 100644 --- a/web/frontend/src/App.tsx +++ b/web/frontend/src/App.tsx @@ -11,6 +11,7 @@ import LoginTasksPage from './pages/LoginTasksPage'; import ProxyPage from './pages/ProxyPage'; import UsersPage from './pages/UsersPage'; import CookiePage from './pages/CookiePage'; +import HttpLogsPage from './pages/HttpLogsPage'; import { getToken } from './store/auth'; import { ThemeProvider, useTheme } from './store/theme'; @@ -44,6 +45,7 @@ function AppContent() { } /> } /> } /> + } /> } /> } /> diff --git a/web/frontend/src/api/modules.ts b/web/frontend/src/api/modules.ts index 73c3560..29987ff 100644 --- a/web/frontend/src/api/modules.ts +++ b/web/frontend/src/api/modules.ts @@ -77,3 +77,9 @@ export const proxyApi = { test: () => api.post('/proxy/test'), testWhitelist: () => api.post('/proxy/whitelist/test'), }; + +export const logApi = { + listHttp: (params?: { limit?: number; offset?: number; category?: string; level?: string; keyword?: string }) => + api.get('/logs/http', { params }), + clearHttp: () => api.delete('/logs/http'), +}; diff --git a/web/frontend/src/layouts/MainLayout.tsx b/web/frontend/src/layouts/MainLayout.tsx index 39930d4..0da85d9 100644 --- a/web/frontend/src/layouts/MainLayout.tsx +++ b/web/frontend/src/layouts/MainLayout.tsx @@ -4,7 +4,7 @@ import { DashboardOutlined, UserOutlined, LogoutOutlined, CloudServerOutlined, TeamOutlined, ApiOutlined, KeyOutlined, MenuFoldOutlined, MenuUnfoldOutlined, SwapOutlined, - SunOutlined, MoonOutlined, DesktopOutlined, + SunOutlined, MoonOutlined, DesktopOutlined, FileTextOutlined, } from '@ant-design/icons'; import { useNavigate, useLocation, Outlet } from 'react-router-dom'; import { getUser, clearAuth, hasPerm, type AuthUser } from '../store/auth'; @@ -69,6 +69,11 @@ export default function MainLayout({ onLogout }: { onLogout?: () => void }) { menuItems.push({ key: '/proxy', label: '代理配置', icon: }); } + // 请求日志（运营和管理员可见） + if (hasPerm(user, 'audit:view') || hasPerm(user, 'login:batch')) { + menuItems.push({ key: '/http-logs', label: '请求日志', icon: }); + } + // 用户管理 if (hasPerm(user, 'user:view')) { menuItems.push({ key: '/users', label: '用户管理', icon: }); diff --git a/web/frontend/src/pages/HttpLogsPage.tsx b/web/frontend/src/pages/HttpLogsPage.tsx new file mode 100644 index 0000000..0eb287a --- /dev/null +++ b/web/frontend/src/pages/HttpLogsPage.tsx @@ -0,0 +1,330 @@ +import { useState, useEffect, useCallback } from 'react'; +import { + Card, Table, Tag, Space, Button, Input, Select, Tooltip, Drawer, + Typography, message, Popconfirm, Empty, Segmented, +} from 'antd'; +import { + ReloadOutlined, DeleteOutlined, SearchOutlined, EyeOutlined, +} from '@ant-design/icons'; +import { logApi } from '../api/modules'; +import { hasPerm, getUser } from '../store/auth'; + +const { Text, Paragraph } = Typography; + +interface HttpLogEntry { + timestamp: string; + ts: number; + category: string; + tag: string; + method: string; + url: string; + proxy: string | null; + request: { headers: Record; body: string }; + response: { status_code: number | null; headers: Record; body: string }; + duration_ms: number | null; + error: string | null; + level: string; +} + +const CATEGORY_LABELS: Record = { + douyu_login: '斗鱼登录', + geetest: '极验', + proxy_verify: '代理验证', + whitelist: '白名单', +}; + +const LEVEL_COLORS: Record = { + info: 'green', + warning: 'orange', + error: 'red', +}; + +export default function HttpLogsPage() { + const [logs, setLogs] = useState([]); + const [total, setTotal] = useState(0); + const [loading, setLoading] = useState(false); + const [page, setPage] = useState(1); + const [pageSize, setPageSize] = useState(50); + const [category, setCategory] = useState(undefined); + const [level, setLevel] = useState(undefined); + const [keyword, setKeyword] = useState(''); + const [detailEntry, setDetailEntry] = useState(null); + const user = getUser(); + + const fetchLogs = useCallback(async () => { + setLoading(true); + try { + const res = await logApi.listHttp({ + limit: pageSize, + offset: (page - 1) * pageSize, + category, + level, + keyword: keyword || undefined, + }); + setLogs(res.items || []); + setTotal(res.total || 0); + } catch (e: any) { + message.error(e.message || '获取日志失败'); + } finally { + setLoading(false); + } + }, [page, pageSize, category, level, keyword]); + + useEffect(() => { + fetchLogs(); + }, [fetchLogs]); + + // 自动刷新 + useEffect(() => { + const timer = setInterval(() => { + if (!detailEntry) fetchLogs(); + }, 5000); + return () => clearInterval(timer); + }, [fetchLogs, detailEntry]); + + const handleClear = async () => { + try { + const res = await logApi.clearHttp(); + message.success(`已清空 ${res.cleared} 条日志`); + fetchLogs(); + } catch (e: any) { + message.error(e.message || '清空失败'); + } + }; + + const canManage = user && (hasPerm(user, 'audit:view') || hasPerm(user, 'login:batch')); + + const columns = [ + { + title: '时间', + dataIndex: 'timestamp', + width: 180, + render: (v: string) => {v}, + }, + { + title: '级别', + dataIndex: 'level', + width: 70, + render: (v: string) => {v}, + }, + { + title: '分类', + dataIndex: 'category', + width: 100, + render: (v: string) => CATEGORY_LABELS[v] || v, + }, + { + title: '方法', + dataIndex: 'method', + width: 60, + render: (v: string) => {v}, + }, + { + title: 'URL', + dataIndex: 'url', + ellipsis: true, + render: (v: string) => ( + + {v} + + ), + }, + { + title: '状态', + dataIndex: ['response', 'status_code'], + width: 70, + render: (v: number | null) => v ? ( + {v} + ) : -, + }, + { + title: '耗时', + dataIndex: 'duration_ms', + width: 80, + render: (v: number | null) => v != null ? ( + 5000 ? 'red' : v > 2000 ? 'orange' : undefined }}> + {v > 1000 ? `${(v / 1000).toFixed(1)}s` : `${v}ms`} + + ) : '-', + }, + { + title: '代理', + dataIndex: 'proxy', + width: 140, + ellipsis: true, + render: (v: string | null) => v ? ( + {v} + ) : null, + }, + { + title: '标签', + dataIndex: 'tag', + width: 100, + ellipsis: true, + render: (v: string) => v ? {v} : null, + }, + { + title: '操作', + width: 60, + render: (_: any, record: HttpLogEntry) => ( + + + + + + } + > + `${r.ts}-${r.url}`} + size="small" + loading={loading} + pagination={{ + current: page, + pageSize, + total, + showSizeChanger: true, + showTotal: (t) => `共 ${t} 条`, + onChange: (p, ps) => { setPage(p); setPageSize(ps); }, + }} + scroll={{ x: 1000 }} + /> + + + setDetailEntry(null)} + width={700} + > + {detailEntry && ( + +
+ 时间: + {detailEntry.timestamp} +
+
+ 级别: + {detailEntry.level} + 分类: + {CATEGORY_LABELS[detailEntry.category] || detailEntry.category} +
+
+ 请求: + {detailEntry.method} + {detailEntry.url} +
+ {detailEntry.proxy && ( +
+ 代理: + {detailEntry.proxy} +
+ )} + {detailEntry.tag && ( +
+ 标签: + {detailEntry.tag} +
+ )} +
+ 耗时: + {detailEntry.duration_ms != null ? `${detailEntry.duration_ms}ms` : '-'} +
+ + {detailEntry.error ? ( + + + {detailEntry.error} + + + ) : ( + +
+ 状态码: + {detailEntry.response.status_code ? ( + + {detailEntry.response.status_code} + + ) : -} +
+
+ 响应体: + + {detailEntry.response.body || '(空)'} + +
+ + )} + + + 
+                {JSON.stringify(detailEntry.request.headers, null, 2)}
+
+ + + {detailEntry.request.body && ( + + + {typeof detailEntry.request.body === 'object' + ? JSON.stringify(detailEntry.request.body, null, 2) + : detailEntry.request.body} + + + )} + + {Object.keys(detailEntry.response.headers || {}).length > 0 && ( + + 
+                  {JSON.stringify(detailEntry.response.headers, null, 2)}
+
+ + )} + + )} + + + ); +}