diff --git a/core/douyu/proxy.py b/core/douyu/proxy.py index 7ec518e..1d31110 100644 --- a/core/douyu/proxy.py +++ b/core/douyu/proxy.py @@ -1,16 +1,17 @@ """代理管理模块""" import re +import time import requests -from typing import Optional, List +from typing import Optional from loguru import logger class ProxyManager: """代理管理器""" - def __init__(self, api_url: str = None): - self.api_url = api_url or "http://api.xiequ.cn/VAD/GetIp.aspx?act=get&uid=106015&vkey=97111DB5379E38E3BC2FF09A1B00A0C7&num=1&time=30&plat=1&re=0&type=0&so=1&ow=1&spl=1&addr=&db=1" + def __init__(self, api_url: str = ""): + self.api_url = api_url self.current_proxy: Optional[str] = None def get_proxy(self) -> Optional[str]: @@ -37,46 +38,26 @@ class ProxyManager: self.current_proxy = proxy logger.info(f"获取到代理: {proxy}") return proxy - else: - logger.warning(f"无法解析代理地址: {text}") - return None + + logger.warning(f"无法解析代理地址: {text}") + return None except Exception as e: logger.error(f"获取代理失败: {e}") return None def get_proxies_dict(self, proxy: str = None) -> dict: - """ - 获取requests使用的proxies字典 - - Args: - proxy: 代理URL，如果不提供则使用当前代理 - - Returns: - proxies字典 - """ + """获取requests使用的proxies字典""" proxy = proxy or self.current_proxy if proxy: - return { - 'http': proxy, - 'https': proxy, - } + return {'http': proxy, 'https': proxy} return {} def verify_proxy(self, proxy: str = None) -> bool: - """ - 验证代理是否可用 - - Args: - proxy: 代理URL - - Returns: - 是否可用 - """ + """验证代理是否可用""" proxy = proxy or self.current_proxy if not proxy: return False - try: response = requests.get( 'https://httpbin.org/ip', @@ -93,23 +74,138 @@ class ProxyManager: return False -# 全局代理管理器实例 -_proxy_manager: Optional[ProxyManager] = None +def get_proxy_manager(api_url: str = "") -> ProxyManager: + """获取代理管理器实例（每次传入 api_url 时创建新实例，避免全局状态污染）""" + return ProxyManager(api_url) -def get_proxy_manager(api_url: str = None) -> ProxyManager: - """获取全局代理管理器实例""" - global _proxy_manager - if _proxy_manager is None: - _proxy_manager = ProxyManager(api_url) - return _proxy_manager +def parse_proxy_response(text: str) -> tuple[Optional[str], Optional[str]]: + """ + 解析代理API响应。 + + Returns: + (proxy_url, whitelist_ip) + - proxy_url: 解析到的代理地址（http://ip:port），无法解析时为 None + - whitelist_ip: 需要添加到白名单的IP（当API返回白名单错误时），无错误时为 None + """ + text = text.strip() + + # 正常代理地址 + match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text) + if match: + ip, port = match.group(1), match.group(2) + # 排除 "请先添加白名单：1.2.3.4" 中误匹配到 ip:port 的情况 + if '白名单' not in text: + return f"http://{ip}:{port}", None + + # 白名单错误：DB1.请先添加白名单：39.144.114.76 + if '添加白名单' in text or '白名单' in text: + ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', text) + if ip_match: + return None, ip_match.group(1) + + return None, None -def get_proxy() -> Optional[str]: - """获取代理URL的便捷函数""" - return get_proxy_manager().get_proxy() +def verify_proxy_url(proxy_url: str, timeout: tuple = (4, 6)) -> tuple[bool, str]: + """ + 验证代理是否可用。 + + Returns: + (是否可用, 消息) + """ + proxies = {'http': proxy_url, 'https': proxy_url} + targets = [ + 'https://qifu-api.baidubce.com/ip/local/geo/v1/district', + 'https://myip.ipip.net', + 'https://4.ipw.cn', + ] + + for url in targets: + try: + response = requests.get( + url, proxies=proxies, timeout=timeout, + headers={'User-Agent': 'Mozilla/5.0'}, + ) + response.raise_for_status() + return True, f'代理可用: {url.split("/")[2]}' + 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"代理验证 {url} 失败: {detail}") + continue + + return False, '代理验证失败（所有目标不可达）' -def get_proxies_dict() -> dict: - """获取proxies字典的便捷函数""" - return get_proxy_manager().get_proxies_dict() +def resolve_working_proxy( + api_url: str, + whitelist_uid: str = "", + whitelist_ukey: str = "", + max_attempts: int = 5, + log_func=None, +) -> tuple[Optional[str], str]: + """ + 从代理API获取可用代理，自动处理白名单同步。 + + Args: + api_url: 代理API地址 + whitelist_uid: 白名单UID（启用白名单时传入） + whitelist_ukey: 白名单UKEY + max_attempts: 最大尝试次数 + log_func: 日志回调函数 (level, message) + + Returns: + (代理URL, 消息) + """ + def log(level, msg): + if log_func: + log_func(level, msg) + else: + getattr(logger, level if level in ('info', 'warning', 'error', 'success') else 'info', logger.info)(msg) + + synced_whitelist = False + + for attempt in range(1, max_attempts + 1): + log('info', f'代理预检 {attempt}/{max_attempts}: 正在获取代理') + try: + response = requests.get(api_url, timeout=10) + response.raise_for_status() + text = response.text.strip() + + proxy_url, whitelist_ip = parse_proxy_response(text) + + if proxy_url: + # 验证代理 + ok, msg = verify_proxy_url(proxy_url) + if ok: + log('success', f'代理预检成功: {proxy_url}') + return proxy_url, msg + log('warning', f'代理预检 {attempt}/{max_attempts}: {msg}') + continue + + # 代理API返回白名单错误 + if whitelist_ip and not synced_whitelist and whitelist_uid and whitelist_ukey: + log('warning', f'代理需要白名单IP: {whitelist_ip}，自动同步...') + from core.douyu.whitelist import WhitelistManager + manager = WhitelistManager(whitelist_uid, whitelist_ukey) + ok, sync_msg = manager.sync_ip(whitelist_ip) + log('success' if ok else 'error', f'白名单同步: {sync_msg}') + if ok: + synced_whitelist = True + log('info', '白名单已更新，等待2秒后重试...') + time.sleep(2) + continue + return None, f'白名单同步失败: {sync_msg}' + + log('warning', f'代理预检 {attempt}/{max_attempts}: 代理API响应无法解析: {text[:80]}') + + except Exception as exc: + log('warning', f'代理预检 {attempt}/{max_attempts}: 代理API请求失败: {exc}') + + return None, f'代理预检失败，已尝试 {max_attempts} 次' diff --git a/core/douyu/whitelist.py b/core/douyu/whitelist.py index 07cdb53..3865605 100644 --- a/core/douyu/whitelist.py +++ b/core/douyu/whitelist.py @@ -2,6 +2,7 @@ import json import re +import time from typing import Optional from urllib.parse import urlencode @@ -65,12 +66,12 @@ class WhitelistManager: logger.error(f"获取白名单失败: {e}") return [] - def add_ip(self, ip: str) -> bool: + def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]: """ 添加IP到白名单 - Args: - ip: 要添加的IP地址 + Returns: + (是否成功, API原始响应) """ try: url = self._build_url(act="add", ip=ip, meno=self._memo) @@ -80,22 +81,40 @@ class WhitelistManager: text = response.text.strip() logger.debug(f"添加白名单响应: {text}") - # 成功通常返回 "ok" 或类似信息 if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text: logger.info(f"白名单添加成功: {ip} (备注: {self._memo})") - return True + return True, text - # 检查是否已存在 - if "已存在" in text or "exist" in text.lower(): + if "已存在" in text or "exist" in text.lower() or "IpRep" in text: logger.info(f"白名单已存在: {ip}") - return True + 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 + return False, text except Exception as e: logger.error(f"添加白名单失败: {e}") - return False + return False, str(e) def delete_ip(self, ip: str) -> bool: """ @@ -175,15 +194,18 @@ class WhitelistManager: if existing_ip: logger.info(f"白名单IP变化: {existing_ip} -> {current_ip}") self.delete_ip(existing_ip) + time.sleep(1) # 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) # 添加新IP - if self.add_ip(current_ip): + ok, resp = self.add_ip(current_ip) + if ok: if existing_ip: msg = f"白名单IP已更新: {existing_ip} -> {current_ip}" else: @@ -191,7 +213,7 @@ class WhitelistManager: logger.info(msg) return True, msg - return False, "白名单添加失败" + return False, f"白名单添加失败，API响应: {resp}" except Exception as e: msg = f"白名单同步失败: {e}" diff --git a/data/web.db b/data/web.db index a8a724e..f29ea73 100644 Binary files a/data/web.db and b/data/web.db differ diff --git a/web/backend/routers/__pycache__/login.cpython-312.pyc b/web/backend/routers/__pycache__/login.cpython-312.pyc index 47a99b1..4f77e01 100644 Binary files a/web/backend/routers/__pycache__/login.cpython-312.pyc and b/web/backend/routers/__pycache__/login.cpython-312.pyc differ diff --git a/web/backend/routers/__pycache__/proxy.cpython-312.pyc b/web/backend/routers/__pycache__/proxy.cpython-312.pyc index 78be19e..51b9606 100644 Binary files a/web/backend/routers/__pycache__/proxy.cpython-312.pyc and b/web/backend/routers/__pycache__/proxy.cpython-312.pyc differ diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index c909390..096829b 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -115,7 +115,7 @@ async def ws_login_logs(websocket: WebSocket, batch_id: str): await websocket.accept() log_queue = asyncio.Queue() - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() # 查找已运行的批次，或等待新批次 # 简化：直接把 log_queue 注册到全局，前端创建批次后连 ws diff --git a/web/backend/routers/proxy.py b/web/backend/routers/proxy.py index 807a3c7..d306629 100644 --- a/web/backend/routers/proxy.py +++ b/web/backend/routers/proxy.py @@ -1,17 +1,22 @@ -"""代理 & 白名单配置路由""" +"""代理 & 白名单配置路由 + WebSocket 实时日志""" -import re -import time -from fastapi import APIRouter, Depends, HTTPException +import asyncio +import threading +import uuid +from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect from sqlalchemy.orm import Session from ..database import get_db from ..models import User, ProxyConfig as ProxyConfigModel, AuditLog -from ..schemas import ProxyConfigOut, ProxyConfigUpdate, MessageResponse +from ..schemas import ProxyConfigOut, ProxyConfigUpdate from ..deps import require_permission +from core.douyu.proxy import resolve_working_proxy, verify_proxy_url, parse_proxy_response router = APIRouter(prefix="/api/proxy", tags=["代理与白名单"]) +# 运行中的测试: test_id -> {log_queue, loop, result} +_active_tests: dict[str, dict] = {} + def _get_or_create(db: Session) -> ProxyConfigModel: cfg = db.query(ProxyConfigModel).first() @@ -54,93 +59,233 @@ def update_proxy_config( return cfg +# ---- WebSocket 日志推送 ---- + +@router.websocket("/ws/test/{test_id}") +async def ws_test_logs(websocket: WebSocket, test_id: str): + """WebSocket 推送代理/白名单测试实时日志。""" + await websocket.accept() + + test = _active_tests.get(test_id) + if not test: + await websocket.send_json({"level": "error", "message": "测试任务不存在"}) + await websocket.close() + return + + log_queue: asyncio.Queue = test["log_queue"] + + try: + while True: + try: + msg = await asyncio.wait_for(log_queue.get(), timeout=30) + await websocket.send_json(msg) + # 收到 result 消息表示测试结束 + if msg.get("level") == "result": + await asyncio.sleep(0.1) + break + except asyncio.TimeoutError: + await websocket.send_json({"level": "heartbeat", "message": ""}) + except WebSocketDisconnect: + pass + finally: + _active_tests.pop(test_id, None) + + +# ---- 异步测试执行 ---- + +def _run_proxy_test( + cfg: ProxyConfigModel, + log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, +): + """在线程中执行代理测试。""" + + def push(level, message): + asyncio.run_coroutine_threadsafe( + log_queue.put({"level": level, "message": message}), + loop, + ) + + try: + if not cfg.enabled: + push("error", "代理未启用") + push("result", "") + return + + # 静态代理 + if cfg.http or cfg.https: + proxy_url = cfg.http or cfg.https + push("info", f"验证静态代理: {proxy_url}") + ok, msg = verify_proxy_url(proxy_url) + push("success" if ok else "error", msg) + push("result", "") + return + + # 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 "" + + proxy_url, msg = resolve_working_proxy( + api_url=cfg.api_url, + whitelist_uid=whitelist_uid, + whitelist_ukey=whitelist_ukey, + max_attempts=3, + log_func=push, + ) + if proxy_url: + push("success", f"代理可用: {proxy_url}") + else: + push("error", msg) + push("result", "") + return + + push("error", "未配置代理地址或API") + push("result", "") + except Exception as e: + push("error", f"测试异常: {e}") + push("result", "") + + +def _run_whitelist_test( + cfg: ProxyConfigModel, + log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, +): + """在线程中执行白名单测试。""" + import requests as req_lib + import re + from core.douyu.whitelist import WhitelistManager, get_exit_ip_via_proxy + + def push(level, message): + asyncio.run_coroutine_threadsafe( + log_queue.put({"level": level, "message": message}), + loop, + ) + + try: + if not cfg.whitelist_enabled: + push("error", "白名单未启用") + push("result", "") + return + if not cfg.whitelist_uid or not cfg.whitelist_ukey: + push("error", "未配置白名单UID/UKEY") + push("result", "") + return + + manager = WhitelistManager(cfg.whitelist_uid, cfg.whitelist_ukey) + + # 1. 测试API连接 + push("info", "测试白名单API连接...") + ok, msg = manager.test_connection() + push("info" if ok else "error", f"白名单API: {msg}") + if not ok: + push("result", "") + return + + # 2. 获取本机公网IP（白名单需要的是本机IP，不是代理出口IP） + push("info", "正在获取本机公网IP...") + + # 优先：从代理API响应中提取（代理API返回"请先添加白名单：xxx"时，xxx就是本机IP） + local_ip = None + if cfg.api_url: + try: + resp = req_lib.get(cfg.api_url, timeout=10) + text = resp.text.strip() + push("info", f"代理API响应: {text[:80]}") + _, whitelist_ip = parse_proxy_response(text) + if whitelist_ip: + local_ip = whitelist_ip + push("info", f"从代理API获取到本机IP: {local_ip}") + except Exception as e: + push("warning", f"代理API请求失败: {e}") + + # 备用：直接访问IP检测服务获取本机公网IP + 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 + + if not local_ip: + push("error", "无法获取本机公网IP") + push("result", "") + return + + 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)} 条记录") + + if in_list: + record = next((r for r in records if r.get('IP') == local_ip), {}) + memo = record.get('MEMO', '') + if memo == manager.memo: + push("success", f"本机IP {local_ip} 已在白名单中 (备注正确)") + else: + push("warning", f'本机IP {local_ip} 备注不匹配 (当前: "{memo}")，更新中...') + sync_ok, sync_msg = manager.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) + push("success" if sync_ok else "error", f"白名单同步: {sync_msg}") + + push("result", "") + except Exception as e: + push("error", f"测试异常: {e}") + push("result", "") + + +# ---- API 端点 ---- + @router.post("/test") -def test_proxy( +async def test_proxy( db: Session = Depends(get_db), current: User = Depends(require_permission("proxy:manage")), ): - """测试代理连通性。""" - import requests as req_lib + """启动代理测试（异步执行，通过 WebSocket 推送日志）。""" cfg = _get_or_create(db) - if not cfg.enabled: - return {"success": False, "message": "代理未启用"} - proxy_url = cfg.http or cfg.https - if cfg.api_url and not proxy_url: - try: - resp = req_lib.get(cfg.api_url, timeout=10) - match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', resp.text) - if match: - proxy_url = f"http://{match.group(1)}:{match.group(2)}" - except Exception as e: - return {"success": False, "message": f"代理API请求失败: {e}"} + test_id = uuid.uuid4().hex[:12] + log_queue = asyncio.Queue() + loop = asyncio.get_running_loop() - if not proxy_url: - return {"success": False, "message": "无可用代理地址"} + _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} - try: - resp = req_lib.get( - "https://qifu-api.baidubce.com/ip/local/geo/v1/district", - proxies={"http": proxy_url, "https": proxy_url}, - timeout=(4, 6), - headers={"User-Agent": "Mozilla/5.0"}, - ) - resp.raise_for_status() - return {"success": True, "message": f"代理可用，响应: {resp.text[:100]}"} - except Exception as e: - return {"success": False, "message": f"代理验证失败: {e}"} + thread = threading.Thread(target=_run_proxy_test, args=(cfg, log_queue, loop), daemon=True) + thread.start() + + return {"test_id": test_id, "success": True} @router.post("/whitelist/test") -def test_whitelist( +async def test_whitelist( db: Session = Depends(get_db), current: User = Depends(require_permission("whitelist:test")), ): - """测试白名单连接并自动同步出口IP。""" - from core.douyu.whitelist import WhitelistManager, get_exit_ip_via_proxy - import requests as req_lib - + """启动白名单测试（异步执行，通过 WebSocket 推送日志）。""" cfg = _get_or_create(db) - if not cfg.whitelist_enabled: - return {"success": False, "message": "白名单未启用"} - if not cfg.whitelist_uid or not cfg.whitelist_ukey: - return {"success": False, "message": "未配置白名单UID/UKEY"} - manager = WhitelistManager(cfg.whitelist_uid, cfg.whitelist_ukey) + test_id = uuid.uuid4().hex[:12] + log_queue = asyncio.Queue() + loop = asyncio.get_running_loop() - # 测试API连接 - ok, msg = manager.test_connection() - if not ok: - return {"success": False, "message": msg} + _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} - # 获取出口IP - proxy_url = cfg.http or cfg.https - if cfg.api_url and not proxy_url: - try: - resp = req_lib.get(cfg.api_url, timeout=10) - match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', resp.text) - if match: - proxy_url = f"http://{match.group(1)}:{match.group(2)}" - except Exception: - pass + thread = threading.Thread(target=_run_whitelist_test, args=(cfg, log_queue, loop), daemon=True) + thread.start() - exit_ip = None - if proxy_url: - exit_ip = get_exit_ip_via_proxy(proxy_url) - if not exit_ip: - try: - resp = req_lib.get("https://4.ipw.cn", 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: - exit_ip = match.group(1) - except Exception: - pass - - if not exit_ip: - return {"success": False, "message": "API连接正常但无法获取出口IP"} - - # 同步白名单 - sync_ok, sync_msg = manager.sync_ip(exit_ip) - if sync_ok: - return {"success": True, "message": f"出口IP {exit_ip} 已同步: {sync_msg}"} - return {"success": False, "message": f"同步失败: {sync_msg}"} + return {"test_id": test_id, "success": True} diff --git a/web/backend/services/__pycache__/login_service.cpython-312.pyc b/web/backend/services/__pycache__/login_service.cpython-312.pyc index 65207c9..4432412 100644 Binary files a/web/backend/services/__pycache__/login_service.cpython-312.pyc and b/web/backend/services/__pycache__/login_service.cpython-312.pyc differ diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 516da12..469ceec 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -1,8 +1,7 @@ -"""登录服务：复用 douyu/ 核心模块，在线程池中执行登录并推送日志。""" +"""登录服务：复用 core/ 核心模块，在线程池中执行登录并推送日志。""" import asyncio import threading -import time import uuid from datetime import datetime from typing import Optional @@ -11,6 +10,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 ..models import Account as AccountModel, LoginTask, ProxyConfig as ProxyConfigModel from ..permissions import has_permission @@ -50,6 +50,42 @@ class LoginBatchRunner: self.loop, ) + def _resolve_proxy(self) -> tuple[Optional[dict], str]: + """ + 解析代理配置，返回 (proxy_dict, message)。 + + - 静态代理：直接返回 dict + - API代理：调用 resolve_working_proxy 预检，自动同步白名单 + - 无代理：返回 (None, '') + """ + if not self.proxy_config or not self.proxy_config.enabled: + return None, '' + + # 静态代理 + if self.proxy_config.http or self.proxy_config.https: + 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, + ) + if proxy_url: + return {'http': proxy_url, 'https': proxy_url}, msg + return None, msg + + return None, '' + def run(self): """在线程中执行批量登录。""" batch_id = self.batch_id @@ -78,17 +114,20 @@ class LoginBatchRunner: self.db.commit() - # 构建代理配置 - proxy_url = None - proxy_api_url = None - if self.proxy_config and self.proxy_config.enabled: - if self.proxy_config.http or self.proxy_config.https: - proxy_url = { - 'http': self.proxy_config.http or self.proxy_config.https, - 'https': self.proxy_config.https or self.proxy_config.http, - } - elif self.proxy_config.api_url: - proxy_api_url = self.proxy_config.api_url + # 代理预检 + proxy_dict, proxy_msg = self._resolve_proxy() + if proxy_msg: + self._push_log("info", proxy_msg) + + # 如果启用了代理但预检失败，终止任务 + if self.proxy_config and self.proxy_config.enabled and not proxy_dict: + self._push_log("error", f"代理不可用，任务终止: {proxy_msg}") + for task, _ in tasks: + task.status = "error" + task.message = f"代理不可用: {proxy_msg}" + task.finished_at = datetime.utcnow() + self.db.commit() + return for i, (task, acc) in enumerate(tasks): if self._stop.is_set(): @@ -112,8 +151,7 @@ class LoginBatchRunner: loginer = DouyuLogin( account, - proxy=proxy_url, - proxy_api_url=proxy_api_url, + proxy=proxy_dict, max_geetest_retries=self.max_geetest_retries, ) result = loginer.login() diff --git a/web/frontend/src/pages/ProxyPage.tsx b/web/frontend/src/pages/ProxyPage.tsx index c7b08e3..87f30ba 100644 --- a/web/frontend/src/pages/ProxyPage.tsx +++ b/web/frontend/src/pages/ProxyPage.tsx @@ -1,26 +1,64 @@ -import { useEffect, useState } from 'react'; -import { Form, Input, Switch, Button, Card, message, Divider, Space } from 'antd'; +import { useEffect, useState, useRef } from 'react'; +import { Form, Input, Switch, Button, Card, message, Row, Col } from 'antd'; import { proxyApi } from '../api/modules'; +const WS_BASE = `ws://${window.location.hostname}:8000`; + export default function ProxyPage() { const [form] = Form.useForm(); const [loading, setLoading] = useState(false); const [testing, setTesting] = useState(false); const [testingWl, setTestingWl] = useState(false); + const [configLoaded, setConfigLoaded] = useState(false); + const [logs, setLogs] = useState<{ level: string; message: string }[]>([]); + const wsRef = useRef(null); const loadConfig = async () => { try { const data = await proxyApi.get(); - form.setFieldsValue(data); + form.setFieldsValue({ + enabled: data.enabled ?? false, + api_url: data.api_url ?? '', + http: data.http ?? '', + https: data.https ?? '', + whitelist_enabled: data.whitelist_enabled ?? false, + whitelist_uid: data.whitelist_uid ?? '', + whitelist_ukey: data.whitelist_ukey ?? '', + }); } catch (e: any) { message.error(e.message); + } finally { + setConfigLoaded(true); } }; useEffect(() => { loadConfig(); + return () => { + wsRef.current?.close(); + }; }, []); + const appendLog = (level: string, msg: string) => { + setLogs((prev) => [...prev, { level, message: msg }]); + }; + + const connectWs = (testId: string) => { + wsRef.current?.close(); + setLogs([]); + const ws = new WebSocket(`${WS_BASE}/api/proxy/ws/test/${testId}`); + wsRef.current = ws; + ws.onmessage = (event) => { + const msg = JSON.parse(event.data); + if (msg.level === 'heartbeat') return; + if (msg.level === 'result') return; + appendLog(msg.level, msg.message); + }; + ws.onclose = () => { + wsRef.current = null; + }; + }; + const handleSave = async () => { setLoading(true); try { @@ -36,74 +74,115 @@ export default function ProxyPage() { const handleTestProxy = async () => { setTesting(true); + setLogs([]); try { const result = await proxyApi.test(); - if (result.success) { - message.success(result.message); - } else { - message.warning(result.message); - } + if (result.test_id) connectWs(result.test_id); } catch (e: any) { message.error(e.message); - } finally { setTesting(false); } }; const handleTestWhitelist = async () => { setTestingWl(true); + setLogs([]); try { const result = await proxyApi.testWhitelist(); - if (result.success) { - message.success(result.message); - } else { - message.warning(result.message); - } + if (result.test_id) connectWs(result.test_id); } catch (e: any) { message.error(e.message); - } finally { setTestingWl(false); } }; + // WebSocket 关闭后恢复按钮 + useEffect(() => { + if (!wsRef.current) { + setTesting(false); + setTestingWl(false); + } + }, [logs.length === 0]); + + const logColors: Record = { + error: '#ff4d4f', + success: '#52c41a', + warning: '#faad14', + info: '#333', + }; + return ( -
-

代理配置

-
- - - - - - - - - - - - - - - +
+
+

代理配置

+ +
- - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + {logs.length === 0 ? ( + 点击"测试代理"或"测试白名单"查看日志 + ) : ( + logs.map((log, i) => ( +
+ {log.message} +
+ )) + )} +
); }