diff --git a/web/backend/__pycache__/schemas.cpython-312.pyc b/web/backend/__pycache__/schemas.cpython-312.pyc index 6a6c628..9851552 100644 Binary files a/web/backend/__pycache__/schemas.cpython-312.pyc and b/web/backend/__pycache__/schemas.cpython-312.pyc differ diff --git a/web/backend/routers/__pycache__/login.cpython-312.pyc b/web/backend/routers/__pycache__/login.cpython-312.pyc index d66f986..3f5eaf1 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/login.py b/web/backend/routers/login.py index 3087a4e..b01700a 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -61,6 +61,7 @@ async def create_batch( proxy_config=proxy, log_queue=log_queue, loop=loop, + concurrency=req.concurrency, ) batch_id = runner.batch_id diff --git a/web/backend/schemas.py b/web/backend/schemas.py index 7494c13..17f6dfd 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -83,6 +83,7 @@ class AccountOut(BaseModel): class LoginBatchRequest(BaseModel): account_ids: list[int] max_geetest_retries: int = 5 + concurrency: int = 3 # 并发数，1-10 class LoginTaskOut(BaseModel): diff --git a/web/backend/services/__pycache__/login_service.cpython-312.pyc b/web/backend/services/__pycache__/login_service.cpython-312.pyc index 0833bb0..f7b5690 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 ff599f0..9027cbd 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -1,8 +1,9 @@ -"""登录服务：复用 core/ 核心模块，在线程池中执行登录并推送日志。""" +"""登录服务：复用 core/ 核心模块，在线程池中并发执行登录并推送日志。""" import asyncio import threading import uuid +from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import Optional @@ -16,7 +17,7 @@ from ..permissions import has_permission class LoginBatchRunner: - """批量登录执行器，在线程中运行，通过 asyncio.Queue 推送日志。""" + """批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。""" def __init__( self, @@ -28,6 +29,7 @@ class LoginBatchRunner: proxy_config: Optional[ProxyConfigModel] = None, log_queue: Optional[asyncio.Queue] = None, loop: Optional[asyncio.AbstractEventLoop] = None, + concurrency: int = 3, ): self.db = db self.account_ids = account_ids @@ -38,7 +40,10 @@ class LoginBatchRunner: self.log_queue = log_queue self.loop = loop self.batch_id = uuid.uuid4().hex[:12] + self.concurrency = max(1, min(concurrency, 10)) # 限制 1-10 self._stop = threading.Event() + self._counter_lock = threading.Lock() + self._completed = 0 def stop(self): self._stop.set() @@ -86,13 +91,73 @@ class LoginBatchRunner: return None, '' + def _execute_one(self, task_id: int, acc_info: dict, proxy_dict: Optional[dict], total: int): + """在独立线程中执行单个账号登录，使用独立的 DB 会话。""" + if self._stop.is_set(): + self._push_log("warning", f"任务已停止，跳过: {acc_info['username']}") + return + + worker_db = SessionLocal() + try: + task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first() + if not task: + return + + task.status = "running" + worker_db.commit() + + with self._counter_lock: + self._completed += 1 + current = self._completed + + self._push_log("info", f"[{current}/{total}] 开始登录: {acc_info['username']}") + + try: + account = Account( + username=acc_info["username"], + password=acc_info["password"], + email=acc_info["email"], + email_password=acc_info["email_password"], + email_imap_server=acc_info["email_imap_server"] or "", + email_imap_port=acc_info["email_imap_port"] or 993, + ) + + loginer = DouyuLogin( + account, + proxy=proxy_dict, + max_geetest_retries=self.max_geetest_retries, + ) + result = loginer.login() + + if result.success: + task.status = "success" + task.cookie = result.cookie + task.message = "登录成功" + self._push_log("success", f"[{current}] {acc_info['username']} 登录成功") + else: + task.status = "failed" + task.message = result.message + self._push_log("error", f"[{current}] {acc_info['username']} 登录失败: {result.message}") + + except Exception as e: + task.status = "error" + task.message = str(e) + self._push_log("error", f"[{current}] {acc_info['username']} 登录异常: {e}") + + task.finished_at = datetime.utcnow() + worker_db.commit() + + finally: + worker_db.close() + def run(self): """在线程中执行批量登录。""" batch_id = self.batch_id - self._push_log("info", f"批量登录任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号") + concurrency = self.concurrency + self._push_log("info", f"批量登录任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号，并发数: {concurrency}") - # 创建任务记录 - tasks = [] + # 创建任务记录（顺序执行，线程安全） + task_infos: list[dict] = [] # {task_id, acc_info} for aid in self.account_ids: acc = self.db.query(AccountModel).filter(AccountModel.id == aid).first() if not acc: @@ -110,9 +175,26 @@ class LoginBatchRunner: created_by=self.created_by, ) self.db.add(task) - tasks.append((task, acc)) + self.db.flush() # 获取 task.id + + task_infos.append({ + "task_id": task.id, + "acc_info": { + "username": acc.username, + "password": acc.password, + "email": acc.email, + "email_password": acc.email_password, + "email_imap_server": acc.email_imap_server or "", + "email_imap_port": acc.email_imap_port or 993, + }, + }) self.db.commit() + total = len(task_infos) + if total == 0: + self._push_log("warning", "没有可执行的账号") + self._push_log("result", "") + return # 代理预检 proxy_dict, proxy_msg = self._resolve_proxy() @@ -122,57 +204,46 @@ class LoginBatchRunner: # 如果启用了代理但预检失败，终止任务 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() + for item in task_infos: + worker_db = SessionLocal() + try: + task = worker_db.query(LoginTask).filter(LoginTask.id == item["task_id"]).first() + if task: + task.status = "error" + task.message = f"代理不可用: {proxy_msg}" + task.finished_at = datetime.utcnow() + worker_db.commit() + finally: + worker_db.close() + self._push_log("result", "") return - for i, (task, acc) in enumerate(tasks): - if self._stop.is_set(): - self._push_log("warning", "任务已停止") - break - - task.status = "running" - self.db.commit() - - self._push_log("info", f"[{i+1}/{len(tasks)}] 开始登录: {acc.username}") - - try: - account = Account( - username=acc.username, - password=acc.password, - email=acc.email, - email_password=acc.email_password, - email_imap_server=acc.email_imap_server or "", - email_imap_port=acc.email_imap_port or 993, + # 并发执行登录 + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = [] + for item in task_infos: + if self._stop.is_set(): + self._push_log("warning", "任务已停止，跳过剩余账号") + break + future = executor.submit( + self._execute_one, + item["task_id"], + item["acc_info"], + proxy_dict, + total, ) + futures.append(future) - loginer = DouyuLogin( - account, - proxy=proxy_dict, - max_geetest_retries=self.max_geetest_retries, - ) - result = loginer.login() - - if result.success: - task.status = "success" - task.cookie = result.cookie - task.message = "登录成功" - self._push_log("success", f"[{i+1}] {acc.username} 登录成功") - else: - task.status = "failed" - task.message = result.message - self._push_log("error", f"[{i+1}] {acc.username} 登录失败: {result.message}") - - except Exception as e: - task.status = "error" - task.message = str(e) - self._push_log("error", f"[{i+1}] {acc.username} 登录异常: {e}") - - task.finished_at = datetime.utcnow() - self.db.commit() + # 等待所有任务完成 + for future in as_completed(futures): + try: + future.result() + except Exception as e: + self._push_log("error", f"Worker 异常: {e}") self._push_log("info", f"批量登录任务 {batch_id} 完成") self._push_log("result", "") + + +# 在模块末尾导入 SessionLocal（避免循环导入） +from ..database import SessionLocal diff --git a/web/frontend/src/api/modules.ts b/web/frontend/src/api/modules.ts index b219a4f..2512f23 100644 --- a/web/frontend/src/api/modules.ts +++ b/web/frontend/src/api/modules.ts @@ -50,8 +50,8 @@ export const accountApi = { }; export const loginApi = { - createBatch: (account_ids: number[], max_geetest_retries?: number) => - api.post('/login/batch', { account_ids, max_geetest_retries }), + createBatch: (account_ids: number[], max_geetest_retries?: number, concurrency?: number) => + api.post('/login/batch', { account_ids, max_geetest_retries, concurrency }), listTasks: (batch_id?: string) => api.get('/login/tasks', { params: batch_id ? { batch_id } : {} }), stop: (batch_id: string) => api.post(`/login/stop/${batch_id}`), diff --git a/web/frontend/src/pages/LoginTasksPage.tsx b/web/frontend/src/pages/LoginTasksPage.tsx index 0a2fde7..3c067ea 100644 --- a/web/frontend/src/pages/LoginTasksPage.tsx +++ b/web/frontend/src/pages/LoginTasksPage.tsx @@ -1,8 +1,8 @@ import { useEffect, useState, useRef, useMemo, useCallback } from 'react'; import { - Table, Button, Select, message, Tag, Space, Card, Row, Col, Statistic, Spin, + Table, Button, Select, message, Tag, Space, Card, Row, Col, Statistic, Spin, InputNumber, Tooltip, } from 'antd'; -import { PlayCircleOutlined, StopOutlined, FilterOutlined } from '@ant-design/icons'; +import { PlayCircleOutlined, StopOutlined, FilterOutlined, ThunderboltOutlined, ReloadOutlined } from '@ant-design/icons'; import { accountApi, loginApi } from '../api/modules'; import { getUser, hasPerm } from '../store/auth'; @@ -31,6 +31,7 @@ export default function LoginTasksPage() { const [logs, setLogs] = useState<{ level: string; message: string }[]>([]); const [wsConnected, setWsConnected] = useState(false); const [selectedTags, setSelectedTags] = useState([]); + const [concurrency, setConcurrency] = useState(3); const wsRef = useRef(null); const user = getUser(); @@ -111,15 +112,16 @@ export default function LoginTasksPage() { return () => clearInterval(timer); }, [batchId]); - const handleBatchLogin = async () => { - if (selectedIds.length === 0) { + // 共享的批量登录启动逻辑 + const startBatch = async (accountIds: number[]) => { + if (accountIds.length === 0) { message.warning('请选择账号'); return; } setLoading(true); setLogs([]); try { - const result = await loginApi.createBatch(selectedIds); + const result = await loginApi.createBatch(accountIds, 5, concurrency); setBatchId(result.batch_id); message.success(`已创建登录任务，共 ${result.count} 个账号`); @@ -148,6 +150,27 @@ export default function LoginTasksPage() { } }; + const handleBatchLogin = () => startBatch(selectedIds); + + // 重试当前批次所有失败的任务 + const handleRetryFailed = () => { + const failedIds = tasks + .filter((t) => ['failed', 'error'].includes(t.status)) + .map((t) => t.account_id); + if (failedIds.length === 0) { + message.info('没有失败的任务'); + return; + } + startBatch(failedIds); + }; + + // 重试单个失败任务 + const handleRetryOne = (taskId: number) => { + const task = tasks.find((t) => t.id === taskId); + if (!task) return; + startBatch([task.account_id]); + }; + const handleStop = async () => { if (batchId) { try { @@ -172,6 +195,25 @@ export default function LoginTasksPage() { }, { title: '消息', dataIndex: 'message', ellipsis: true }, { title: '时间', dataIndex: 'created_at', width: 180 }, + { + title: '操作', + width: 80, + render: (_: any, record: any) => { + if (['failed', 'error'].includes(record.status) && !wsConnected) { + return ( + + ); + } + return null; + }, + }, ]; return ( @@ -252,6 +294,17 @@ export default function LoginTasksPage() { )} /> + + + + setConcurrency(v || 1)} + style={{ width: 60 }} + size="small" + /> + ) + } style={{ flex: 1, display: 'flex', flexDirection: 'column', minHeight: 0 }} bodyStyle={{ flex: 1, overflow: 'auto', padding: 0 }} >