From 2ab67245439c7d48dcc23b91eac901b44c5e23bc Mon Sep 17 00:00:00 2001 From: yml2213 Date: Tue, 23 Jun 2026 07:34:17 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E8=A1=A5=E5=85=A8services=E5=B1=82?= =?UTF-8?q?+=E6=B6=88=E9=99=A4core/models.py=E5=86=97=E4=BD=99+=E4=BF=AE?= =?UTF-8?q?=E5=A4=8Dnetwork.py=E7=BC=A9=E8=BF=9B=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 高优先级问题1 - Router层业务逻辑泄漏: - 新建 services/proxy_service.py: 从routers/proxy.py提取ProxyService类， 封装代理/白名单测试逻辑、全局状态管理、所有core/ import - 新建 services/account_service.py: 从routers/accounts.py提取 split_account_line、cookie_account_ids_query、parse_and_build_accounts， 封装core.douyu.email_verifier的import - 重写routers/proxy.py为薄HTTP/WS层，移除所有core/直接依赖 - 精简routers/accounts.py，从account_service导入业务函数 - 将_active_batches全局状态移入login_service.py的BatchRegistry类 - routers/login.py改为使用batch_registry单例 高优先级问题2 - 两套models.py冗余: - 删除core/models.py（Account dataclass + ProxyConfig死代码） - 在core/douyu/login.py新增AccountLike Protocol替代Account dataclass - login_service.py中Account(...)构造改为SimpleNamespace(...) - 清理core/__init__.py导出 附带修复: - core/geetest/common/network.py的get_picture()函数缩进错误 （缺少if match:块和data赋值语句） --- core/__init__.py | 4 - core/douyu/login.py | 16 +- core/geetest/common/network.py | 4 + core/models.py | 28 --- web/backend/routers/accounts.py | 71 +------ web/backend/routers/login.py | 24 +-- web/backend/routers/proxy.py | 248 +++--------------------- web/backend/services/account_service.py | 76 ++++++++ web/backend/services/login_service.py | 33 +++- web/backend/services/proxy_service.py | 240 +++++++++++++++++++++++ 10 files changed, 411 insertions(+), 333 deletions(-) delete mode 100644 core/models.py create mode 100644 web/backend/services/account_service.py create mode 100644 web/backend/services/proxy_service.py diff --git a/core/__init__.py b/core/__init__.py index 5810f44..557a906 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,5 +1 @@ """核心业务逻辑""" - -from .models import Account, ProxyConfig - -__all__ = ["Account", "ProxyConfig"] diff --git a/core/douyu/login.py b/core/douyu/login.py index 700fb73..371f14a 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -4,11 +4,10 @@ import re import json import time import requests -from typing import Mapping, Optional, Tuple +from typing import Mapping, Optional, Protocol, Tuple from urllib.parse import urlsplit, urlunsplit from loguru import logger -from core.models import Account from .crypto import encrypt_password, encrypt_nickname_or_phone from .email_verifier import EmailVerifier from .proxy import ProxyManager, get_proxy_manager @@ -25,6 +24,17 @@ from core.geetest.common.network import ( from utils.http_logger import log_http +class AccountLike(Protocol): + """DouyuLogin 所需的最小账号接口，ORM 对象或 SimpleNamespace 均可满足。""" + username: str + password: str + email: str + email_password: str + email_imap_server: str + email_imap_port: int + email_imap_ssl: bool + + class LoginResult: """登录结果""" @@ -57,7 +67,7 @@ class DouyuLogin: def __init__( self, - account: Account, + account: AccountLike, proxy: Optional[str | Mapping[str, str]] = None, proxy_api_url: Optional[str] = None, timeout: tuple[float, float] = REQUEST_TIMEOUT, diff --git a/core/geetest/common/network.py b/core/geetest/common/network.py index 0ddd554..b621f96 100644 --- a/core/geetest/common/network.py +++ b/core/geetest/common/network.py @@ -266,6 +266,10 @@ def get_picture(gt:str, challenge:str) -> tuple[str, str, list[int], str, str, s response.raise_for_status() logger.debug("极验 get.php 原始响应: {}", response.text[:500]) match = re.search(r'\((.*)\)$', response.text) + if match: + json_str = match.group(1) + data = json.loads(json_str) + if 'data' in data and isinstance(data['data'], dict): # 新版极验格式 inner_data = data['data'] diff --git a/core/models.py b/core/models.py deleted file mode 100644 index 1602675..0000000 --- a/core/models.py +++ /dev/null @@ -1,28 +0,0 @@ -"""核心数据模型""" - -from dataclasses import dataclass - - -@dataclass -class Account: - """账号配置""" - username: str - password: str - email: str - email_password: str - email_imap_server: str - email_imap_port: int = 993 - email_imap_ssl: bool = True - - -@dataclass -class ProxyConfig: - """代理配置""" - enabled: bool = False - api_url: str = "" - http: str = "" - https: str = "" - # 白名单配置 - whitelist_enabled: bool = False - whitelist_uid: str = "" - whitelist_ukey: str = "" diff --git a/web/backend/routers/accounts.py b/web/backend/routers/accounts.py index b85ae57..ed185a3 100644 --- a/web/backend/routers/accounts.py +++ b/web/backend/routers/accounts.py @@ -1,41 +1,20 @@ """账号管理路由""" -import re -import csv from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy.orm import Session +from sqlalchemy import func +from sqlalchemy.orm import Session, joinedload from ..database import get_db from ..models import User, Account, AuditLog, LoginTask from ..schemas import AccountImport, AccountAssign, AccountTag, AccountOut, BatchAssign from ..deps import get_current_user, require_permission from ..permissions import user_has_permission -from sqlalchemy import func -from sqlalchemy.orm import joinedload +from ..services.account_service import ( + cookie_account_ids_query, parse_and_build_accounts, +) router = APIRouter(prefix="/api/accounts", tags=["账号管理"]) -EMAIL_PATTERN = re.compile(r'^[^\s@|]+@[^\s@|]+\.[^\s@|]+$') - - -def _split_account_line(line: str) -> list[str]: - if '|' in line: - return line.split('|') - if '\t' in line: - return line.split('\t') - if ',' in line: - return next(csv.reader([line])) - return line.split() - - -def _cookie_account_ids_query(db: Session): - """返回有成功登录记录(cookie非空)的账号ID子查询。""" - return db.query(LoginTask.account_id).filter( - LoginTask.status == 'success', - LoginTask.cookie != '', - LoginTask.cookie.isnot(None), - ).distinct() - @router.get("", response_model=list[AccountOut]) def list_accounts( @@ -50,7 +29,7 @@ def list_accounts( # 只展示已成功登录过的账号 if has_cookie: - query = query.filter(Account.id.in_(_cookie_account_ids_query(db))) + query = query.filter(Account.id.in_(cookie_account_ids_query(db))) # 权限控制：客服只能看分配给自己的 if not user_has_permission(current, "account:view_all"): @@ -91,39 +70,7 @@ def import_accounts( current: User = Depends(require_permission("account:import")), ): """批量导入账号。格式：用户名|密码|邮箱|邮箱密码|标签(可选)""" - from core.douyu.email_verifier import get_email_config_for_account - - accounts = [] - skipped = 0 - for line_num, line in enumerate(req.text.strip().split('\n'), 1): - line = line.strip() - if not line or line.startswith('#'): - continue - parts = _split_account_line(line) - if len(parts) < 4: - skipped += 1 - continue - - username, password, email, email_password = [p.strip() for p in parts[:4]] - tag = parts[4].strip() if len(parts) > 4 else "" - if not all([username, password, email, email_password]): - skipped += 1 - continue - if not EMAIL_PATTERN.match(email): - skipped += 1 - continue - - email_cfg = get_email_config_for_account(email) - accounts.append(Account( - username=username, - password=password, - email=email, - email_password=email_password, - email_imap_server=email_cfg['server'], - email_imap_port=email_cfg.get('port', 993), - email_imap_ssl=email_cfg.get('ssl', True), - tag=tag, - )) + accounts, skipped = parse_and_build_accounts(req.text) if accounts: db.add_all(accounts) @@ -187,7 +134,7 @@ def batch_assign_accounts( raise HTTPException(status_code=400, detail="只能分配给客服角色") # 只能分配已成功登录过的账号（有cookie） - cookie_ids_query = _cookie_account_ids_query(db).subquery() + cookie_ids_query = cookie_account_ids_query(db).subquery() invalid_ids = db.query(Account.id).filter( Account.id.in_(req.account_ids), Account.id.notin_(cookie_ids_query), @@ -220,7 +167,7 @@ def assignments_summary( current: User = Depends(require_permission("account:assign")), ): """分配概览：每个客服分配了多少账号（仅统计已成功登录的账号）。""" - cookie_subq = _cookie_account_ids_query(db).subquery() + cookie_subq = cookie_account_ids_query(db).subquery() cookie_accounts = db.query(Account).filter(Account.id.in_(cookie_subq)).subquery() results = ( diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index c89186e..a2d1b52 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -11,14 +11,10 @@ from ..models import User, Account, LoginTask, ProxyConfig as ProxyConfigModel from ..schemas import LoginBatchRequest, LoginTaskOut from ..deps import get_current_user, require_permission, authenticate_websocket from ..permissions import user_has_permission, get_user_permissions -from ..services.login_service import LoginBatchRunner +from ..services.login_service import LoginBatchRunner, batch_registry router = APIRouter(prefix="/api/login", tags=["登录任务"]) -# 运行中的批次: batch_id -> {log_queue, loop, runner} -_active_batches: dict[str, dict] = {} -_active_batches_lock = threading.Lock() - @router.post("/batch") async def create_batch( @@ -68,13 +64,8 @@ async def create_batch( batch_id = runner.batch_id - # 先注册到全局，再启动线程，确保 WebSocket 连接时能找到 - with _active_batches_lock: - _active_batches[batch_id] = { - "log_queue": log_queue, - "loop": loop, - "runner": runner, - } + # 先注册到批次注册表，再启动线程，确保 WebSocket 连接时能找到 + batch_registry.register(batch_id, log_queue, loop, runner) # 启动线程 thread = threading.Thread(target=runner.run, daemon=True) @@ -155,8 +146,7 @@ def stop_batch( batch_id: str, current: User = Depends(require_permission("login:batch")), ): - with _active_batches_lock: - batch = _active_batches.get(batch_id) + batch = batch_registry.get(batch_id) if batch: batch["runner"].stop() return {"message": "已发送停止信号", "success": True} @@ -175,8 +165,7 @@ async def ws_login_logs(websocket: WebSocket, batch_id: str): await websocket.accept() # 从已注册的批次中获取 log_queue（由 create_batch 创建） - with _active_batches_lock: - batch = _active_batches.get(batch_id) + batch = batch_registry.get(batch_id) if not batch: await websocket.send_json({"level": "error", "message": "批次不存在或已结束"}) await websocket.close() @@ -198,5 +187,4 @@ async def ws_login_logs(websocket: WebSocket, batch_id: str): except WebSocketDisconnect: pass finally: - with _active_batches_lock: - _active_batches.pop(batch_id, None) + batch_registry.pop(batch_id) diff --git a/web/backend/routers/proxy.py b/web/backend/routers/proxy.py index 4cfc233..102235f 100644 --- a/web/backend/routers/proxy.py +++ b/web/backend/routers/proxy.py @@ -2,39 +2,24 @@ import asyncio import threading -import uuid -from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect +from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect from sqlalchemy.orm import Session from ..database import get_db -from ..models import User, ProxyConfig as ProxyConfigModel, AuditLog +from ..models import User from ..schemas import ProxyConfigOut, ProxyConfigUpdate from ..deps import require_permission, authenticate_websocket -from core.douyu.proxy import resolve_working_proxy, verify_proxy_url, parse_proxy_response +from ..services.proxy_service import proxy_service router = APIRouter(prefix="/api/proxy", tags=["代理与白名单"]) -# 运行中的测试: test_id -> {log_queue, loop, result} -_active_tests: dict[str, dict] = {} -_active_tests_lock = threading.Lock() - - -def _get_or_create(db: Session) -> ProxyConfigModel: - cfg = db.query(ProxyConfigModel).first() - if not cfg: - cfg = ProxyConfigModel() - db.add(cfg) - db.commit() - db.refresh(cfg) - return cfg - @router.get("", response_model=ProxyConfigOut) def get_proxy_config( db: Session = Depends(get_db), _: User = Depends(require_permission("proxy:manage")), ): - return _get_or_create(db) + return proxy_service.get_or_create(db) @router.put("", response_model=ProxyConfigOut) @@ -43,21 +28,17 @@ def update_proxy_config( db: Session = Depends(get_db), current: User = Depends(require_permission("proxy:manage")), ): - cfg = _get_or_create(db) - cfg.enabled = req.enabled - cfg.api_url = req.api_url - cfg.http = req.http - cfg.https = req.https - cfg.whitelist_enabled = req.whitelist_enabled - cfg.whitelist_uid = req.whitelist_uid - cfg.whitelist_ukey = req.whitelist_ukey - db.commit() - db.refresh(cfg) - - db.add(AuditLog(user_id=current.id, username=current.username, - action="proxy:update", target="proxy_config")) - db.commit() - return cfg + 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, + whitelist_uid=req.whitelist_uid, + whitelist_ukey=req.whitelist_ukey, + current_user=current, + ) # ---- WebSocket 日志推送 ---- @@ -65,7 +46,6 @@ def update_proxy_config( @router.websocket("/ws/test/{test_id}") async def ws_test_logs(websocket: WebSocket, test_id: str): """WebSocket 推送代理/白名单测试实时日志（需认证）。""" - # 认证：从 cookie 或 token query param 验证用户身份 user = authenticate_websocket(websocket) if not user: await websocket.close(code=1008, reason="未授权") @@ -73,8 +53,7 @@ async def ws_test_logs(websocket: WebSocket, test_id: str): await websocket.accept() - with _active_tests_lock: - test = _active_tests.get(test_id) + test = proxy_service.get_test(test_id) if not test: await websocket.send_json({"level": "error", "message": "测试任务不存在"}) await websocket.close() @@ -96,169 +75,10 @@ async def ws_test_logs(websocket: WebSocket, test_id: str): except WebSocketDisconnect: pass finally: - with _active_tests_lock: - _active_tests.pop(test_id, None) + proxy_service.remove_test(test_id) -# ---- 异步测试执行 ---- - -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 端点 ---- +# ---- 异步测试 API 端点 ---- @router.post("/test") async def test_proxy( @@ -266,16 +86,14 @@ async def test_proxy( current: User = Depends(require_permission("proxy:manage")), ): """启动代理测试（异步执行，通过 WebSocket 推送日志）。""" - cfg = _get_or_create(db) + cfg = proxy_service.get_or_create(db) + test_id, log_queue, loop = proxy_service.register_test() - test_id = uuid.uuid4().hex[:12] - log_queue = asyncio.Queue() - loop = asyncio.get_running_loop() - - with _active_tests_lock: - _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} - - thread = threading.Thread(target=_run_proxy_test, args=(cfg, log_queue, loop), daemon=True) + thread = threading.Thread( + target=proxy_service.run_proxy_test, + args=(cfg, log_queue, loop), + daemon=True, + ) thread.start() return {"test_id": test_id, "success": True} @@ -287,16 +105,14 @@ async def test_whitelist( current: User = Depends(require_permission("whitelist:test")), ): """启动白名单测试（异步执行，通过 WebSocket 推送日志）。""" - cfg = _get_or_create(db) + cfg = proxy_service.get_or_create(db) + test_id, log_queue, loop = proxy_service.register_test() - test_id = uuid.uuid4().hex[:12] - log_queue = asyncio.Queue() - loop = asyncio.get_running_loop() - - with _active_tests_lock: - _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} - - thread = threading.Thread(target=_run_whitelist_test, args=(cfg, log_queue, loop), daemon=True) + thread = threading.Thread( + target=proxy_service.run_whitelist_test, + args=(cfg, log_queue, loop), + daemon=True, + ) thread.start() return {"test_id": test_id, "success": True} diff --git a/web/backend/services/account_service.py b/web/backend/services/account_service.py new file mode 100644 index 0000000..94d99b3 --- /dev/null +++ b/web/backend/services/account_service.py @@ -0,0 +1,76 @@ +"""账号管理服务层：封装账号导入、查询等业务逻辑，隔离 core/ 依赖。""" + +import csv +import re + +from sqlalchemy import func +from sqlalchemy.orm import Session, joinedload + +from ..models import Account, AuditLog, LoginTask +from ..permissions import user_has_permission + + +EMAIL_PATTERN = re.compile(r'^[^\s@|]+@[^\s@|]+\.[^\s@|]+$') + + +def split_account_line(line: str) -> list[str]: + """拆分一行账号文本，支持 |、tab、逗号、空格分隔。""" + if '|' in line: + return line.split('|') + if '\t' in line: + return line.split('\t') + if ',' in line: + return next(csv.reader([line])) + return line.split() + + +def cookie_account_ids_query(db: Session): + """返回有成功登录记录(cookie非空)的账号ID子查询。""" + return db.query(LoginTask.account_id).filter( + LoginTask.status == 'success', + LoginTask.cookie != '', + LoginTask.cookie.isnot(None), + ).distinct() + + +def parse_and_build_accounts(text: str) -> tuple[list[Account], int]: + """ + 解析批量导入文本，构建 Account ORM 对象列表。 + + Returns: + (accounts, skipped_count) + """ + from core.douyu.email_verifier import get_email_config_for_account + + accounts = [] + skipped = 0 + for line in text.strip().split('\n'): + line = line.strip() + if not line or line.startswith('#'): + continue + parts = split_account_line(line) + if len(parts) < 4: + skipped += 1 + continue + + username, password, email, email_password = [p.strip() for p in parts[:4]] + tag = parts[4].strip() if len(parts) > 4 else "" + if not all([username, password, email, email_password]): + skipped += 1 + continue + if not EMAIL_PATTERN.match(email): + skipped += 1 + continue + + email_cfg = get_email_config_for_account(email) + accounts.append(Account( + username=username, + password=password, + email=email, + email_password=email_password, + email_imap_server=email_cfg['server'], + email_imap_port=email_cfg.get('port', 993), + email_imap_ssl=email_cfg.get('ssl', True), + tag=tag, + )) + return accounts, skipped diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 637765a..8ae71c3 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -3,6 +3,7 @@ import asyncio import threading import uuid +from types import SimpleNamespace from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from typing import Optional @@ -10,7 +11,6 @@ from typing import Optional 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, get_proxy_manager from ..models import Account as AccountModel, LoginTask, ProxyConfig as ProxyConfigModel @@ -128,7 +128,7 @@ class LoginBatchRunner: return try: - account = Account( + account = SimpleNamespace( username=acc_info["username"], password=acc_info["password"], email=acc_info["email"], @@ -261,5 +261,34 @@ class LoginBatchRunner: self.db.close() +class BatchRegistry: + """管理运行中的登录批次状态。""" + + def __init__(self): + self._batches: dict[str, dict] = {} + self._lock = threading.Lock() + + def register(self, batch_id: str, log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, runner: LoginBatchRunner): + with self._lock: + self._batches[batch_id] = { + "log_queue": log_queue, + "loop": loop, + "runner": runner, + } + + def get(self, batch_id: str): + with self._lock: + return self._batches.get(batch_id) + + def pop(self, batch_id: str): + with self._lock: + return self._batches.pop(batch_id, None) + + +# 模块级单例 +batch_registry = BatchRegistry() + + # 在模块末尾导入 SessionLocal（避免循环导入） from ..database import SessionLocal diff --git a/web/backend/services/proxy_service.py b/web/backend/services/proxy_service.py new file mode 100644 index 0000000..8bd4cd2 --- /dev/null +++ b/web/backend/services/proxy_service.py @@ -0,0 +1,240 @@ +"""代理 & 白名单服务层：封装 core/ 代理/白名单逻辑，供路由调用。""" + +import asyncio +import re +import threading +import uuid +from typing import Optional + +import requests as req_lib +from sqlalchemy.orm import Session + +from core.douyu.proxy import resolve_working_proxy, verify_proxy_url, parse_proxy_response +from core.douyu.whitelist import WhitelistManager, get_exit_ip_via_proxy +from ..models import ProxyConfig as ProxyConfigModel, AuditLog + + +class ProxyService: + """代理 & 白名单服务：管理代理配置、执行测试。""" + + def __init__(self): + # 运行中的测试: test_id -> {log_queue, loop} + self._active_tests: dict[str, dict] = {} + self._active_tests_lock = threading.Lock() + + # ---- 配置 CRUD ---- + + @staticmethod + def get_or_create(db: Session) -> ProxyConfigModel: + """获取或创建全局代理配置（单条记录）。""" + cfg = db.query(ProxyConfigModel).first() + if not cfg: + cfg = ProxyConfigModel() + db.add(cfg) + db.commit() + db.refresh(cfg) + return cfg + + @staticmethod + def update_config(db: Session, enabled, api_url, http, https, + whitelist_enabled, whitelist_uid, whitelist_ukey, + current_user) -> ProxyConfigModel: + """更新代理配置并记录审计日志。""" + cfg = ProxyService.get_or_create(db) + cfg.enabled = enabled + cfg.api_url = api_url + cfg.http = http + cfg.https = https + cfg.whitelist_enabled = whitelist_enabled + cfg.whitelist_uid = whitelist_uid + cfg.whitelist_ukey = whitelist_ukey + 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() + return cfg + + # ---- 测试任务管理 ---- + + def register_test(self) -> tuple[str, asyncio.Queue, asyncio.AbstractEventLoop]: + """注册一个新测试任务，返回 (test_id, log_queue, loop)。""" + test_id = uuid.uuid4().hex[:12] + log_queue = asyncio.Queue() + loop = asyncio.get_running_loop() + with self._active_tests_lock: + self._active_tests[test_id] = {"log_queue": log_queue, "loop": loop} + return test_id, log_queue, loop + + def get_test(self, test_id: str) -> Optional[dict]: + """获取测试任务信息。""" + with self._active_tests_lock: + return self._active_tests.get(test_id) + + def remove_test(self, test_id: str) -> None: + """移除测试任务。""" + with self._active_tests_lock: + self._active_tests.pop(test_id, None) + + # ---- 代理测试 ---- + + def run_proxy_test( + self, + 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( + self, + 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.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 + push("info", "正在获取本机公网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}") + + 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", "") + + +# 模块级单例 +proxy_service = ProxyService()