From cd2d02c86eedb50aa1fd84feb2c869547e051ffa Mon Sep 17 00:00:00 2001 From: yml2213 Date: Thu, 9 Jul 2026 23:12:39 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=96=97=E9=B1=BC=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E6=A3=80=E6=B5=8B=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/douyu/login.py | 215 ++++++++-- web/backend/main.py | 3 +- web/backend/permissions.py | 2 + web/backend/routers/account_check.py | 112 ++++++ web/backend/routers/login.py | 6 +- web/backend/schemas.py | 37 ++ web/backend/services/account_check_service.py | 371 ++++++++++++++++++ web/backend/services/login_service.py | 42 +- web/frontend/src/App.tsx | 2 + web/frontend/src/api/accountCheck.ts | 13 + web/frontend/src/api/login.ts | 1 + web/frontend/src/api/modules.ts | 1 + web/frontend/src/api/types.ts | 37 ++ web/frontend/src/layouts/MainLayout.tsx | 4 + web/frontend/src/pages/AccountCheckPage.tsx | 365 +++++++++++++++++ web/frontend/src/pages/LoginTasksPage.tsx | 48 ++- 16 files changed, 1206 insertions(+), 53 deletions(-) create mode 100644 web/backend/routers/account_check.py create mode 100644 web/backend/services/account_check_service.py create mode 100644 web/frontend/src/api/accountCheck.ts create mode 100644 web/frontend/src/pages/AccountCheckPage.tsx diff --git a/core/douyu/login.py b/core/douyu/login.py index 086b95a..130731b 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -30,7 +30,16 @@ _geetest_semaphore = threading.Semaphore(2) class CredentialError(ValueError): """账号或密码错误，不应重试。""" - pass + + def __init__( + self, + message: str, + code: str = "credential_error", + status_message: str = "", + ): + super().__init__(message) + self.code = code + self.status_message = status_message or message class AccountLike(Protocol): @@ -62,6 +71,8 @@ class DouyuLogin: SEND_EMAIL_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/sendemail" VERIFY_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/verify" WEBLOGIN_API = "https://msg.douyu.com/webLogin" + CP_RPC_API = "https://www.douyu.com/member/cp/cp_rpc_ajax" + CP_REFERER = "https://www.douyu.com/member/cp" LOGIN_REFERER = ( "https://passport.douyu.com/index/login?" "passport_reg_callback=PASSPORT_REG_SUCCESS_CALLBACK&" @@ -242,6 +253,64 @@ class DouyuLogin: preview = body[:200].replace("\n", "\\n") raise ValueError(f"{source} 返回的不是有效 JSON: {preview}") from exc + @staticmethod + def _classify_credential_payload(payload: dict) -> tuple[str, str] | None: + """识别登录接口返回的账号类终态。""" + error_code = payload.get('error') + error_msg = str(payload.get('msg') or '') + + if error_code == 110022 or '账号不存在' in error_msg: + return "account_cancelled", "账号已注销" + + password_keywords = ['密码错误', '账号或密码', '账号或者密码'] + if error_code == 110018 or any(kw in error_msg for kw in password_keywords): + return "password_wrong", "账号密码错误" + + return None + + def _credential_error_from_payload(self, stage: str, payload: dict) -> CredentialError | None: + """把账号类错误转换为不重试的 CredentialError。""" + classified = self._classify_credential_payload(payload) + if not classified: + return None + code, status_message = classified + return CredentialError( + f"{stage}失败: {status_message}", + code=code, + status_message=status_message, + ) + + def _run_login_steps(self, deadline: float = 0) -> str: + """执行一次完整登录链路，成功后返回 Cookie。""" + # 1️⃣ 第一次登录（获取极验参数） + logger.info("步骤1: 第一次登录，获取极验参数...") + gt, challenge, code_token, _ = self._first_login() + + # 2️⃣ 极验 fullpage 验证 + logger.info("步骤2: 极验 fullpage 验证...") + validate, seccode = self._solve_geetest(gt, challenge, deadline=deadline) + + # 3️⃣ 第二次登录（带极验） + logger.info("步骤3: 第二次登录（带极验验证）...") + remote_code = self._second_login(gt, challenge, validate, seccode, code_token) + + # 4️⃣ 发送邮箱验证 + logger.info("步骤4: 发送邮箱验证...") + email_sent_at = time.time() + self._send_email_verify(remote_code) + + # 5️⃣ IMAP获取验证码 + logger.info("步骤5: 获取邮箱验证码...") + verify_code = self._get_email_code(after_timestamp=email_sent_at) + + # 6️⃣ 提交验证码 + logger.info("步骤6: 提交验证码...") + login_url = self._submit_verify_code(remote_code, verify_code) + + # 7️⃣ 完成登录获取Cookie + logger.info("步骤7: 完成登录，获取Cookie...") + return self._complete_login(login_url) + def login(self) -> LoginResult: """ 完整登录流程（带整体重试）。 @@ -280,34 +349,7 @@ class DouyuLogin: ) try: - # 1️⃣ 第一次登录（获取极验参数） - logger.info("步骤1: 第一次登录，获取极验参数...") - gt, challenge, code_token, _ = self._first_login() - - # 2️⃣ 极验 fullpage 验证 - logger.info("步骤2: 极验 fullpage 验证...") - validate, seccode = self._solve_geetest(gt, challenge, deadline=deadline) - - # 3️⃣ 第二次登录（带极验） - logger.info("步骤3: 第二次登录（带极验验证）...") - remote_code = self._second_login(gt, challenge, validate, seccode, code_token) - - # 4️⃣ 发送邮箱验证 - logger.info("步骤4: 发送邮箱验证...") - email_sent_at = time.time() - self._send_email_verify(remote_code) - - # 5️⃣ IMAP获取验证码 - logger.info("步骤5: 获取邮箱验证码...") - verify_code = self._get_email_code(after_timestamp=email_sent_at) - - # 6️⃣ 提交验证码 - logger.info("步骤6: 提交验证码...") - login_url = self._submit_verify_code(remote_code, verify_code) - - # 7️⃣ 完成登录获取Cookie - logger.info("步骤7: 完成登录，获取Cookie...") - cookie = self._complete_login(login_url) + cookie = self._run_login_steps(deadline=deadline) message = "登录成功" if self._cookie_enrich_error: message = f"登录成功，补CK失败: {self._cookie_enrich_error}" @@ -320,7 +362,7 @@ class DouyuLogin: return LoginResult(success=False, message=str(e)) except CredentialError as e: logger.error(f"登录失败（凭据错误，不再重试）: {e}") - return LoginResult(success=False, message=str(e), code="credential_error") + return LoginResult(success=False, message=str(e), code=e.code) except EmailLoginError as e: logger.error(f"登录失败（邮箱登录失败，不再重试）: {e}") return LoginResult(success=False, message=str(e), code="email_login_failed") @@ -339,6 +381,107 @@ class DouyuLogin: # 所有重试耗尽或超时 return LoginResult(success=False, message=str(e)) + def check_account(self) -> LoginResult: + """ + 检测账号状态。 + + 可直接识别注销/密码错误；登录成功后请求个人中心接口判断实名状态。 + """ + logger.info(f"开始检测账号: {self.account.username}") + start_time = time.monotonic() + deadline = start_time + self.max_total_time if self.max_total_time > 0 else 0 + + attempt = 0 + while True: + attempt += 1 + if self._is_stopped(): + logger.warning("账号检测任务已停止") + return LoginResult(success=False, message="任务已停止") + elapsed = time.monotonic() - start_time + if self.max_total_time > 0 and elapsed > self.max_total_time: + logger.warning(f"账号检测总耗时 {elapsed:.0f}s 超过上限 {self.max_total_time}s，放弃") + return LoginResult(success=False, message=f"账号检测超时（{elapsed:.0f}s > {self.max_total_time}s）") + + if attempt > 1: + if self.max_login_retries > 0: + logger.info(f"账号检测整体重试 {attempt}/{self.max_login_retries}，换新代理从头开始") + else: + logger.info(f"账号检测整体重试 {attempt} (无限重试)，换新代理从头开始") + if not self._prepare_retry(): + return LoginResult( + success=False, + message=f"静态代理连续失败 {self.MAX_STATIC_RETRY} 次，无法切换代理", + ) + + try: + self._run_login_steps(deadline=deadline) + return self._check_certification_status() + + except InterruptedError as e: + logger.warning(f"账号检测任务已停止: {e}") + return LoginResult(success=False, message=str(e)) + except CredentialError as e: + logger.info(f"账号检测完成: {e.status_message}") + return LoginResult(success=True, message=e.status_message, code=e.code) + except EmailLoginError as e: + logger.error(f"账号检测失败（邮箱登录失败，不再重试）: {e}") + return LoginResult(success=False, message=str(e), code="email_login_failed") + except Exception as e: + elapsed = time.monotonic() - start_time + if self.max_login_retries > 0: + logger.error(f"账号检测失败（尝试 {attempt}/{self.max_login_retries}，已耗时 {elapsed:.0f}s）: {e}") + else: + logger.error(f"账号检测失败（尝试 {attempt}，已耗时 {elapsed:.0f}s）: {e}") + + has_retry = self.max_login_retries <= 0 or attempt < self.max_login_retries + has_time = self.max_total_time <= 0 or elapsed < self.max_total_time + if has_retry and has_time: + self._sleep_interruptible(1) + continue + return LoginResult(success=False, message=str(e)) + + def _check_certification_status(self) -> LoginResult: + """请求个人中心接口并判断实名状态。""" + headers = { + 'Accept': 'application/json, text/javascript, */*; q=0.01', + 'Accept-Language': 'zh-CN,zh;q=0.9,fr;q=0.8,de;q=0.7,en;q=0.6', + 'Cache-Control': 'no-cache', + 'Pragma': 'no-cache', + 'Priority': 'u=1, i', + 'Referer': self.CP_REFERER, + 'Sec-CH-UA': '"Google Chrome";v="149", "Chromium";v="149", "Not)A;Brand";v="24"', + 'Sec-CH-UA-Mobile': '?0', + 'Sec-CH-UA-Platform': '"macOS"', + 'Sec-Fetch-Dest': 'empty', + 'Sec-Fetch-Mode': 'cors', + 'Sec-Fetch-Site': 'same-origin', + 'X-Requested-With': 'XMLHttpRequest', + 'Origin': None, + 'Content-Type': None, + } + payload = self._request_json( + 'get', + self.CP_RPC_API, + '账号认证状态接口', + headers=headers, + timeout=(5, 10), + ) + info = payload.get('info') or {} + ident_status = str(info.get('ident_status', '')) + ident_type = str(info.get('ident_type', '')) + + if ident_status == '0' and ident_type == '0': + logger.success("账号检测完成: 账号未认证") + return LoginResult(success=True, message="账号未认证", code="account_unverified") + + if ident_status == '2' and ident_type == '2': + logger.success("账号检测完成: 账号已认证") + return LoginResult(success=True, message="账号已认证", code="account_verified") + + message = f"账号认证状态未知: ident_status={ident_status or '-'}, ident_type={ident_type or '-'}" + logger.warning(message) + return LoginResult(success=True, message=message, code="account_auth_unknown") + def _prepare_retry(self) -> bool: """重试前准备：取新代理、重置 session cookies。 @@ -389,9 +532,9 @@ class DouyuLogin: if payload.get('error') != 81: error_msg = payload.get('msg', '未知错误') - # 密码错误等凭据问题不应重试 - if any(kw in error_msg for kw in ['密码错误', '账号或密码', '账号或者密码', '账号不存在']): - raise CredentialError(f"第一次登录失败: {error_msg}") + credential_error = self._credential_error_from_payload("第一次登录", payload) + if credential_error: + raise credential_error raise ValueError(f"第一次登录失败: {error_msg}") # 提取极验参数 @@ -534,9 +677,9 @@ class DouyuLogin: if payload.get('error') != 130014: error_msg = payload.get('msg', '未知错误') - # 密码错误等凭据问题不应重试 - if any(kw in error_msg for kw in ['密码错误', '账号或密码', '账号或者密码', '账号不存在']): - raise CredentialError(f"第二次登录失败: {error_msg}") + credential_error = self._credential_error_from_payload("第二次登录", payload) + if credential_error: + raise credential_error raise ValueError(f"第二次登录失败: {error_msg}") # 提取remote_code diff --git a/web/backend/main.py b/web/backend/main.py index 08cd896..0456174 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -11,7 +11,7 @@ from fastapi.responses import FileResponse from starlette.middleware.base import BaseHTTPMiddleware from .database import init_db -from .routers import auth, users, accounts, login, proxy, cookies, huya +from .routers import auth, users, accounts, account_check, login, proxy, cookies, huya from .schemas import AppInfo from .version import get_app_version from utils import setup_logger @@ -66,6 +66,7 @@ app.add_middleware(SecurityHeadersMiddleware) app.include_router(auth.router) app.include_router(users.router) app.include_router(accounts.router) +app.include_router(account_check.router) app.include_router(login.router) app.include_router(proxy.router) app.include_router(cookies.router) diff --git a/web/backend/permissions.py b/web/backend/permissions.py index baa11ae..bef9350 100644 --- a/web/backend/permissions.py +++ b/web/backend/permissions.py @@ -13,6 +13,7 @@ PERMISSIONS = { "account:view_full": "查看账号完整字段（密码/邮箱，仅管理员）", "account:view_assigned": "查看分配给自己的账号", "account:import": "导入账号", + "account:check": "账号检测", "account:assign": "分配账号给客服", "account:delete": "删除账号", # 登录任务 @@ -50,6 +51,7 @@ ROLE_PERMISSIONS = { "operation": [ "account:view_all", "account:import", + "account:check", "account:assign", "login:batch", "login:view_all", diff --git a/web/backend/routers/account_check.py b/web/backend/routers/account_check.py new file mode 100644 index 0000000..bb3b6f5 --- /dev/null +++ b/web/backend/routers/account_check.py @@ -0,0 +1,112 @@ +"""斗鱼账号检测路由。""" + +from __future__ import annotations + +import threading + +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import StreamingResponse +from sqlalchemy.orm import Session + +from ..database import get_db +from ..deps import get_current_user +from ..models import ProxyConfig as ProxyConfigModel, User +from ..permissions import user_has_permission +from ..schemas import AccountCheckBatchOut, AccountCheckBatchRequest +from ..services.account_check_service import ( + account_check_registry, + parse_account_check_lines, +) + + +router = APIRouter(prefix="/api/account-check", tags=["账号检测"]) + + +def _get_runner_or_404(batch_id: str): + """读取账号检测批次。""" + runner = account_check_registry.get(batch_id) + if not runner: + raise HTTPException(status_code=404, detail="批次不存在或服务已重启") + return runner + + +def _require_account_check_perm(user: User) -> None: + """账号检测权限；兼容已有批量登录权限。""" + if not ( + user_has_permission(user, "account:check") + or user_has_permission(user, "login:batch") + ): + raise HTTPException(status_code=403, detail="权限不足") + + +@router.post("/batches", response_model=AccountCheckBatchOut) +def create_account_check_batch( + req: AccountCheckBatchRequest, + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """启动斗鱼账号检测批次。""" + _require_account_check_perm(current) + try: + accounts = parse_account_check_lines(req.text) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + if not accounts: + raise HTTPException(status_code=400, detail="没有识别到有效账号") + + proxy_config = db.query(ProxyConfigModel).first() + runner = account_check_registry.create( + accounts=accounts, + created_by=current.id, + concurrency=req.concurrency, + max_login_retries=req.max_login_retries, + max_total_time=req.max_total_time, + proxy_config=proxy_config, + ) + thread = threading.Thread(target=runner.run, daemon=True) + thread.start() + return runner.snapshot() + + +@router.get("/batches/{batch_id}", response_model=AccountCheckBatchOut) +def get_account_check_batch( + batch_id: str, + current: User = Depends(get_current_user), +): + """查询斗鱼账号检测批次。""" + _require_account_check_perm(current) + runner = _get_runner_or_404(batch_id) + return runner.snapshot() + + +@router.post("/batches/{batch_id}/stop") +def stop_account_check_batch( + batch_id: str, + current: User = Depends(get_current_user), +): + """停止斗鱼账号检测批次。""" + _require_account_check_perm(current) + runner = _get_runner_or_404(batch_id) + runner.stop() + return {"message": "已发送停止信号", "success": True} + + +@router.get("/batches/{batch_id}/download") +def download_account_check_batch( + batch_id: str, + current: User = Depends(get_current_user), +): + """下载斗鱼账号检测分类结果 zip。""" + _require_account_check_perm(current) + runner = _get_runner_or_404(batch_id) + snapshot = runner.snapshot() + if snapshot["status"] in {"pending", "running"}: + raise HTTPException(status_code=400, detail="批次尚未完成") + + content, filename = runner.build_zip() + return StreamingResponse( + iter([content]), + media_type="application/zip", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index 7f00d4c..9e390ab 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -22,7 +22,7 @@ async def create_batch( db: Session = Depends(get_db), current: User = Depends(require_permission("login:batch")), ): - """创建批量登录任务。""" + """创建批量登录或账号检测任务。""" if not req.account_ids: raise HTTPException(status_code=400, detail="请选择账号") @@ -41,7 +41,8 @@ async def create_batch( valid_ids.append(aid) if not valid_ids: - raise HTTPException(status_code=403, detail="没有可登录的账号") + action_name = "检测" if req.mode == "check" else "登录" + raise HTTPException(status_code=403, detail=f"没有可{action_name}的账号") # 在主事件循环中创建 log_queue，传给后台线程 log_queue = asyncio.Queue() @@ -61,6 +62,7 @@ async def create_batch( loop=loop, concurrency=req.concurrency, api_strategy=req.api_strategy, + mode=req.mode, ) batch_id = runner.batch_id diff --git a/web/backend/schemas.py b/web/backend/schemas.py index 5456c2f..523dd8f 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -136,6 +136,7 @@ class LoginBatchRequest(BaseModel): max_total_time: float = 300 # 单账号登录总时长上限（秒），0=不限制 concurrency: int = 3 # 并发数，1-10 api_strategy: str = "wgapi" # 接口策略: wgapi（新版）或 iframe（旧版备选） + mode: str = Field("login", pattern="^(login|check)$") # login=登录取CK，check=账号状态检测 class LoginTaskOut(BaseModel): @@ -168,6 +169,42 @@ class LoginTaskOut(BaseModel): } +# ---- 账号检测 ---- +class AccountCheckBatchRequest(BaseModel): + text: str = Field(..., min_length=1) + concurrency: int = Field(3, ge=1, le=10) + max_login_retries: int = Field(0, ge=0, le=50) + max_total_time: float = Field(300, ge=0, le=3600) + + +class AccountCheckItemOut(BaseModel): + line: int + username: str + email: str + status: str + message: str + started_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + + +class AccountCheckBatchOut(BaseModel): + batch_id: str + status: str + message: str + created_by: int + concurrency: int + max_login_retries: int + max_total_time: float + total: int + finished_count: int + running_count: int + status_counts: dict[str, int] + created_at: Optional[datetime] = None + started_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + items: list[AccountCheckItemOut] + + # ---- 虎牙 ---- class HuyaCookieImport(BaseModel): """批量导入虎牙 Cookie。支持纯 CK 或 账号----密码----CK。""" diff --git a/web/backend/services/account_check_service.py b/web/backend/services/account_check_service.py new file mode 100644 index 0000000..e6a43d9 --- /dev/null +++ b/web/backend/services/account_check_service.py @@ -0,0 +1,371 @@ +"""斗鱼账号检测批次执行器。""" + +from __future__ import annotations + +import io +import threading +import uuid +import zipfile +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from datetime import datetime, timezone +from types import SimpleNamespace +from typing import Optional + +from core.douyu import DouyuLogin, WgapiLoginAPI +from core.douyu.proxy_fetcher import ProxyFetcher + +from ..models import ProxyConfig as ProxyConfigModel + + +def _now() -> datetime: + """返回时区感知 UTC 时间。""" + return datetime.now(timezone.utc) + + +STATUS_LABELS = { + "account_cancelled": "账号已注销", + "password_wrong": "账号密码错误", + "account_unverified": "账号未认证", + "account_verified": "账号已认证", + "account_auth_unknown": "认证状态未知", + "error": "检测失败", + "stopped": "已停止", +} + +EXPORT_STATUS_ORDER = [ + "account_cancelled", + "password_wrong", + "account_unverified", + "account_verified", + "account_auth_unknown", + "error", + "stopped", +] + + +@dataclass +class AccountCheckInput: + """导入的一行斗鱼账号。""" + + line: int + username: str + password: str + email: str + email_password: str + + def export_line(self) -> str: + """导出为统一四段格式。""" + return f"{self.username}----{self.password}----{self.email}----{self.email_password}" + + +@dataclass +class AccountCheckItemState: + """单个账号检测状态。""" + + line: int + username: str + email: str + export_text: str + status: str = "pending" + message: str = "等待开始" + started_at: datetime | None = None + finished_at: datetime | None = None + + def to_dict(self) -> dict: + return { + "line": self.line, + "username": self.username, + "email": self.email, + "status": self.status, + "message": self.message, + "started_at": self.started_at, + "finished_at": self.finished_at, + } + + +@dataclass +class AccountCheckBatch: + """账号检测批次内存快照。""" + + batch_id: str + created_by: int + concurrency: int + max_login_retries: int + max_total_time: float + items: list[AccountCheckItemState] + status: str = "pending" + message: str = "等待开始" + created_at: datetime = field(default_factory=_now) + started_at: datetime | None = None + finished_at: datetime | None = None + + +def parse_account_check_lines(text: str) -> list[AccountCheckInput]: + """解析账号检测导入文本，支持 ---- 和 | 两种分隔符。""" + accounts: list[AccountCheckInput] = [] + for line_no, raw_line in enumerate(text.splitlines(), 1): + line = raw_line.strip() + if not line: + continue + + separator = "----" if "----" in line else "|" + parts = [part.strip() for part in line.split(separator)] + if len(parts) != 4 or any(not part for part in parts): + raise ValueError( + f"第 {line_no} 行格式错误，请使用：账号----密码----邮箱----邮箱密码 " + f"或 账号|密码|邮箱|邮箱密码" + ) + + accounts.append(AccountCheckInput( + line=line_no, + username=parts[0], + password=parts[1], + email=parts[2], + email_password=parts[3], + )) + + return accounts + + +class AccountCheckRunner: + """在后台线程中批量检测斗鱼账号。""" + + def __init__( + self, + batch: AccountCheckBatch, + accounts: list[AccountCheckInput], + proxy_config: Optional[ProxyConfigModel] = None, + ): + self.batch = batch + self.accounts = accounts + self.proxy_config = proxy_config + self._lock = threading.Lock() + self._stop = threading.Event() + self._shared_proxy_fetcher = self._create_proxy_fetcher() + + def _create_proxy_fetcher(self) -> ProxyFetcher | None: + """按全局代理配置创建 API 代理获取器。""" + if not self.proxy_config or not self.proxy_config.enabled or not self.proxy_config.api_url: + return None + + wl_platform = "xiequ" + wl_credentials = None + if self.proxy_config.whitelist_enabled: + wl_platform = getattr(self.proxy_config, "whitelist_platform", None) or "xiequ" + wl_credentials = getattr(self.proxy_config, "whitelist_credentials", None) + if not wl_credentials and self.proxy_config.whitelist_uid and self.proxy_config.whitelist_ukey: + wl_credentials = { + "uid": self.proxy_config.whitelist_uid, + "ukey": self.proxy_config.whitelist_ukey, + } + + return ProxyFetcher( + api_url=self.proxy_config.api_url, + whitelist_platform=wl_platform, + whitelist_credentials=wl_credentials, + stop_event=self._stop, + ) + + def _resolve_static_proxy(self) -> tuple[dict[str, str] | None, str]: + """解析静态代理；API 代理由 DouyuLogin 内部通过 proxy_fetcher 获取。""" + 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}, "" + + if self._shared_proxy_fetcher: + return None, "" + + return None, "代理不可用: 未配置代理" + + def stop(self): + self._stop.set() + with self._lock: + if self.batch.status == "running": + self.batch.message = "正在停止" + + def _set_item(self, index: int, **updates): + with self._lock: + item = self.batch.items[index] + for key, value in updates.items(): + setattr(item, key, value) + + def _set_batch(self, **updates): + with self._lock: + for key, value in updates.items(): + setattr(self.batch, key, value) + + def _run_one(self, index: int, account: AccountCheckInput): + if self._stop.is_set(): + self._set_item(index, status="stopped", message="已停止", finished_at=_now()) + return + + proxy_dict, proxy_error = self._resolve_static_proxy() + if proxy_error: + self._set_item(index, status="error", message=proxy_error, finished_at=_now()) + return + + self._set_item(index, status="running", message="检测中", started_at=_now(), finished_at=None) + try: + result = DouyuLogin( + SimpleNamespace( + username=account.username, + password=account.password, + email=account.email, + email_password=account.email_password, + email_imap_server="", + email_imap_port=993, + email_imap_ssl=True, + ), + proxy=proxy_dict, + max_login_retries=self.batch.max_login_retries, + max_total_time=self.batch.max_total_time, + proxy_fetcher=self._shared_proxy_fetcher, + stop_event=self._stop, + api_strategy=WgapiLoginAPI(), + ).check_account() + except Exception as exc: + self._set_item(index, status="error", message=f"检测异常: {exc}", finished_at=_now()) + return + + if self._stop.is_set() and not result.success: + self._set_item(index, status="stopped", message=result.message or "已停止", finished_at=_now()) + return + + if result.success: + status = result.code if result.code in STATUS_LABELS else "account_auth_unknown" + message = result.message or STATUS_LABELS.get(status, "认证状态未知") + else: + status = "error" + message = result.message or "检测失败" + + self._set_item(index, status=status, message=message, finished_at=_now()) + + def snapshot(self) -> dict: + with self._lock: + items = [item.to_dict() for item in self.batch.items] + status_counts = { + status: sum(1 for item in self.batch.items if item.status == status) + for status in STATUS_LABELS + } + running_count = sum(1 for item in self.batch.items if item.status in {"pending", "running"}) + finished_count = len(self.batch.items) - running_count + return { + "batch_id": self.batch.batch_id, + "status": self.batch.status, + "message": self.batch.message, + "created_by": self.batch.created_by, + "concurrency": self.batch.concurrency, + "max_login_retries": self.batch.max_login_retries, + "max_total_time": self.batch.max_total_time, + "total": len(self.batch.items), + "finished_count": finished_count, + "running_count": running_count, + "status_counts": status_counts, + "created_at": self.batch.created_at, + "started_at": self.batch.started_at, + "finished_at": self.batch.finished_at, + "items": items, + } + + def build_zip(self) -> tuple[bytes, str]: + """按检测状态生成 zip 包。""" + with self._lock: + items = list(self.batch.items) + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf: + for status in EXPORT_STATUS_ORDER: + label = STATUS_LABELS[status] + lines: list[str] = [] + for item in items: + if item.status != status: + continue + line = item.export_text + if status in {"error", "stopped", "account_auth_unknown"} and item.message: + line = f"{line}----{item.message}" + lines.append(line) + content = "\n".join(lines) + if content: + content += "\n" + zf.writestr(f"{timestamp}_{label}.txt", content.encode("utf-8")) + + filename = f"account_check_{timestamp}.zip" + return buffer.getvalue(), filename + + def run(self): + """线程入口。""" + self._set_batch(status="running", message="批次运行中", started_at=_now(), finished_at=None) + try: + if self._shared_proxy_fetcher: + self._shared_proxy_fetcher.warmup_whitelist() + + with ThreadPoolExecutor(max_workers=self.batch.concurrency) as executor: + futures = [] + for index, account in enumerate(self.accounts): + if self._stop.is_set(): + self._set_item(index, status="stopped", message="已停止", finished_at=_now()) + continue + futures.append(executor.submit(self._run_one, index, account)) + + for future in as_completed(futures): + future.result() + except Exception as exc: + self._set_batch(status="error", message=f"批次执行异常: {exc}", finished_at=_now()) + return + + if self._stop.is_set(): + self._set_batch(status="stopped", message="批次已停止", finished_at=_now()) + else: + self._set_batch(status="finished", message="批次已完成", finished_at=_now()) + + +class AccountCheckRegistry: + """管理账号检测批次。""" + + def __init__(self): + self._lock = threading.Lock() + self._runners: dict[str, AccountCheckRunner] = {} + + def create( + self, + accounts: list[AccountCheckInput], + created_by: int, + concurrency: int, + max_login_retries: int, + max_total_time: float, + proxy_config: Optional[ProxyConfigModel] = None, + ) -> AccountCheckRunner: + batch_id = uuid.uuid4().hex[:12] + normalized_retries = max_login_retries if max_login_retries > 0 else 20 + batch = AccountCheckBatch( + batch_id=batch_id, + created_by=created_by, + concurrency=max(1, min(int(concurrency or 1), 10)), + max_login_retries=normalized_retries, + max_total_time=max(0.0, float(max_total_time or 0)), + items=[ + AccountCheckItemState( + line=account.line, + username=account.username, + email=account.email, + export_text=account.export_line(), + ) + for account in accounts + ], + ) + runner = AccountCheckRunner(batch=batch, accounts=accounts, proxy_config=proxy_config) + with self._lock: + self._runners[batch_id] = runner + return runner + + def get(self, batch_id: str) -> AccountCheckRunner | None: + with self._lock: + return self._runners.get(batch_id) + + +account_check_registry = AccountCheckRegistry() diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index b48c7f2..16ee011 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -23,6 +23,23 @@ def _create_api_strategy(strategy_name: str): return WgapiLoginAPI() +CHECK_STATUS_MESSAGES = { + "account_cancelled": "账号已注销", + "password_wrong": "账号密码错误", + "account_unverified": "账号未认证", + "account_verified": "账号已认证", + "account_auth_unknown": "账号认证状态未知", +} + +CHECK_STATUS_LOG_LEVELS = { + "account_cancelled": "warning", + "password_wrong": "error", + "account_unverified": "warning", + "account_verified": "success", + "account_auth_unknown": "warning", +} + + class LoginBatchRunner: """批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。""" @@ -39,6 +56,7 @@ class LoginBatchRunner: loop: Optional[asyncio.AbstractEventLoop] = None, concurrency: int = 3, api_strategy: str = "wgapi", + mode: str = "login", ): self.db = db self.account_ids = account_ids @@ -54,6 +72,7 @@ class LoginBatchRunner: self.batch_id = uuid.uuid4().hex[:12] self.concurrency = max(1, min(concurrency, 10)) # 限制 1-10 self.api_strategy = _create_api_strategy(api_strategy) + self.mode = "check" if mode == "check" else "login" self._stop = threading.Event() self._counter_lock = threading.Lock() self._completed = 0 @@ -129,7 +148,8 @@ class LoginBatchRunner: self._completed += 1 current = self._completed - self._push_log("info", f"[{current}/{total}] 开始登录: {acc_info['username']}") + action_name = "检测" if self.mode == "check" else "登录" + self._push_log("info", f"[{current}/{total}] 开始{action_name}: {acc_info['username']}") # 解析代理配置 proxy_dict, proxy_msg = self._resolve_static_proxy() @@ -165,9 +185,16 @@ class LoginBatchRunner: stop_event=self._stop, api_strategy=self.api_strategy, ) - result = loginer.login() + result = loginer.check_account() if self.mode == "check" else loginer.login() - if result.success: + if self.mode == "check" and result.success: + status = result.code if result.code in CHECK_STATUS_MESSAGES else "account_auth_unknown" + task.status = status + task.cookie = "" + task.message = result.message or CHECK_STATUS_MESSAGES[status] + level = CHECK_STATUS_LOG_LEVELS.get(status, "info") + self._push_log(level, f"[{current}] {acc_info['username']} 检测结果: {task.message}") + elif result.success: task.status = "success" task.cookie = result.cookie task.message = result.message or "登录成功" @@ -175,12 +202,12 @@ class LoginBatchRunner: else: task.status = "failed" task.message = result.message - self._push_log("error", f"[{current}] {acc_info['username']} 登录失败: {result.message}") + self._push_log("error", f"[{current}] {acc_info['username']} {action_name}失败: {result.message}") except Exception as e: task.status = "error" task.message = str(e) - self._push_log("error", f"[{current}] {acc_info['username']} 登录异常: {e}") + self._push_log("error", f"[{current}] {acc_info['username']} {action_name}异常: {e}") task.finished_at = datetime.now(timezone.utc) worker_db.commit() @@ -192,7 +219,8 @@ class LoginBatchRunner: """在线程中执行批量登录。""" batch_id = self.batch_id concurrency = self.concurrency - self._push_log("info", f"批量登录任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号，并发数: {concurrency}") + action_name = "账号检测" if self.mode == "check" else "登录" + self._push_log("info", f"批量{action_name}任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号，并发数: {concurrency}") # 批次开始前同步一次出口 IP 到白名单，后续 fetch_new_proxy 不再主动同步 if self._shared_proxy_fetcher: @@ -280,7 +308,7 @@ class LoginBatchRunner: except Exception as e: self._push_log("error", f"Worker 异常: {e}") - self._push_log("info", f"批量登录任务 {batch_id} 完成") + self._push_log("info", f"批量{action_name}任务 {batch_id} 完成") self._push_log("result", "") finally: # 确保 DB Session 被关闭，避免连接泄漏 diff --git a/web/frontend/src/App.tsx b/web/frontend/src/App.tsx index db2d17d..143c0a2 100644 --- a/web/frontend/src/App.tsx +++ b/web/frontend/src/App.tsx @@ -10,6 +10,7 @@ const LoginPage = lazy(() => import('./pages/LoginPage')); const MainLayout = lazy(() => import('./layouts/MainLayout')); const DashboardPage = lazy(() => import('./pages/DashboardPage')); const AccountsPage = lazy(() => import('./pages/AccountsPage')); +const AccountCheckPage = lazy(() => import('./pages/AccountCheckPage')); const AssignmentsPage = lazy(() => import('./pages/AssignmentsPage')); const LoginTasksPage = lazy(() => import('./pages/LoginTasksPage')); const ProxyPage = lazy(() => import('./pages/ProxyPage')); @@ -63,6 +64,7 @@ function AppContent() { > )} /> )} /> + )} /> )} /> )} /> )} /> diff --git a/web/frontend/src/api/accountCheck.ts b/web/frontend/src/api/accountCheck.ts new file mode 100644 index 0000000..fc5d7f5 --- /dev/null +++ b/web/frontend/src/api/accountCheck.ts @@ -0,0 +1,13 @@ +import api from './client'; +import type { AccountCheckBatch, AccountCheckBatchRequest, MessageResponse } from './types'; + +export const accountCheckApi = { + start: (data: AccountCheckBatchRequest) => + api.post('/account-check/batches', data), + getBatch: (batchId: string) => + api.get(`/account-check/batches/${batchId}`), + stop: (batchId: string) => + api.post(`/account-check/batches/${batchId}/stop`), + download: (batchId: string) => + api.get(`/account-check/batches/${batchId}/download`, { responseType: 'blob' }), +}; diff --git a/web/frontend/src/api/login.ts b/web/frontend/src/api/login.ts index 0f261eb..017d936 100644 --- a/web/frontend/src/api/login.ts +++ b/web/frontend/src/api/login.ts @@ -7,6 +7,7 @@ interface CreateBatchParams { max_login_retries?: number; max_total_time?: number; api_strategy?: string; + mode?: 'login' | 'check'; } export const loginApi = { diff --git a/web/frontend/src/api/modules.ts b/web/frontend/src/api/modules.ts index db7076c..6a7c637 100644 --- a/web/frontend/src/api/modules.ts +++ b/web/frontend/src/api/modules.ts @@ -1,4 +1,5 @@ export * from './types'; +export { accountCheckApi } from './accountCheck'; export { accountApi } from './accounts'; export { appApi } from './app'; export { authApi } from './auth'; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index 20fa159..a9172be 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -94,6 +94,43 @@ export interface BatchLoginResult { success: boolean; } +// ==================== Account Check ==================== + +export interface AccountCheckBatchRequest { + text: string; + concurrency?: number; + max_login_retries?: number; + max_total_time?: number; +} + +export interface AccountCheckItem { + line: number; + username: string; + email: string; + status: string; + message: string; + started_at: string | null; + finished_at: string | null; +} + +export interface AccountCheckBatch { + batch_id: string; + status: string; + message: string; + created_by: number; + concurrency: number; + max_login_retries: number; + max_total_time: number; + total: number; + finished_count: number; + running_count: number; + status_counts: Record; + created_at: string | null; + started_at: string | null; + finished_at: string | null; + items: AccountCheckItem[]; +} + // ==================== Cookie ==================== export interface CookieItem { diff --git a/web/frontend/src/layouts/MainLayout.tsx b/web/frontend/src/layouts/MainLayout.tsx index d651cfc..cb5e9d4 100644 --- a/web/frontend/src/layouts/MainLayout.tsx +++ b/web/frontend/src/layouts/MainLayout.tsx @@ -6,6 +6,7 @@ import { CloudServerOutlined, TeamOutlined, ApiOutlined, KeyOutlined, MenuFoldOutlined, MenuUnfoldOutlined, SwapOutlined, SunOutlined, MoonOutlined, DesktopOutlined, GiftOutlined, ShoppingCartOutlined, ApartmentOutlined, MobileOutlined, + SafetyCertificateOutlined, } from '@ant-design/icons'; import { useNavigate, useLocation, Outlet } from 'react-router-dom'; import { getUser, clearAuth, type AuthUser } from '../store/auth'; @@ -57,6 +58,9 @@ export default function MainLayout({ onLogout }: { onLogout?: () => void }) { if (canAny(['account:view_all', 'account:view_assigned'])) { douyuItems.push({ key: '/accounts', label: '账号管理', icon: }); } + if (canAny(['account:check', 'login:batch'])) { + douyuItems.push({ key: '/account-check', label: '账号检测', icon: }); + } if (can('account:assign')) { douyuItems.push({ key: '/assignments', label: '分配管理', icon: }); } diff --git a/web/frontend/src/pages/AccountCheckPage.tsx b/web/frontend/src/pages/AccountCheckPage.tsx new file mode 100644 index 0000000..d45c7db --- /dev/null +++ b/web/frontend/src/pages/AccountCheckPage.tsx @@ -0,0 +1,365 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { + Button, Card, Col, Input, InputNumber, message, Popconfirm, Row, Space, Statistic, Table, Tag, Typography, +} from 'antd'; +import type { TableProps } from 'antd'; +import { DownloadOutlined, PlayCircleOutlined, ReloadOutlined, StopOutlined } from '@ant-design/icons'; +import { accountCheckApi, type AccountCheckBatch, type AccountCheckItem } from '../api/modules'; +import { formatTime } from '../utils/time'; +import { getErrorMessage } from '../utils/error'; + +const { TextArea } = Input; +const { Text, Title } = Typography; + +const FORM_STORAGE_KEY = 'douyu_account_check_form'; +const BATCH_STORAGE_KEY = 'douyu_account_check_batch_id'; + +const STATUS_LABELS: Record = { + pending: '等待', + running: '检测中', + account_cancelled: '账号已注销', + password_wrong: '账号密码错误', + account_unverified: '账号未认证', + account_verified: '账号已认证', + account_auth_unknown: '认证状态未知', + error: '检测失败', + stopped: '已停止', +}; + +const STATUS_COLORS: Record = { + pending: 'default', + running: 'processing', + account_cancelled: 'error', + password_wrong: 'error', + account_unverified: 'warning', + account_verified: 'success', + account_auth_unknown: 'warning', + error: 'error', + stopped: 'warning', +}; + +const FINISHED_BATCH_STATUS = new Set(['finished', 'stopped', 'error']); +const RUNNING_BATCH_STATUS = new Set(['pending', 'running']); + +function readStoredText() { + try { + return localStorage.getItem(FORM_STORAGE_KEY) || ''; + } catch { + return ''; + } +} + +function readStoredBatchId() { + try { + return localStorage.getItem(BATCH_STORAGE_KEY) || ''; + } catch { + return ''; + } +} + +function readStoredNumber(key: string, fallback: number) { + try { + const raw = localStorage.getItem(key); + if (raw === null) return fallback; + const value = Number(raw); + return Number.isFinite(value) ? value : fallback; + } catch { + return fallback; + } +} + +function downloadBlob(blob: Blob, filename: string) { + const url = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = url; + link.download = filename; + link.click(); + URL.revokeObjectURL(url); +} + +function fileTimestamp() { + const d = new Date(); + const pad = (n: number) => String(n).padStart(2, '0'); + return `${d.getFullYear()}${pad(d.getMonth() + 1)}${pad(d.getDate())}_${pad(d.getHours())}${pad(d.getMinutes())}${pad(d.getSeconds())}`; +} + +export default function AccountCheckPage() { + const [text, setText] = useState(readStoredText); + const [concurrency, setConcurrency] = useState(() => readStoredNumber('account_check_concurrency', 3)); + const [maxTotalTime, setMaxTotalTime] = useState(() => readStoredNumber('account_check_max_total_time', 300)); + const [maxLoginRetries, setMaxLoginRetries] = useState(() => readStoredNumber('account_check_max_login_retries', 0)); + const [batch, setBatch] = useState(null); + const [starting, setStarting] = useState(false); + const [refreshing, setRefreshing] = useState(false); + const [stopping, setStopping] = useState(false); + const [downloading, setDownloading] = useState(false); + const restoredBatchRef = useRef(false); + + const batchId = batch?.batch_id || ''; + const isRunning = !!batch && RUNNING_BATCH_STATUS.has(batch.status); + const canDownload = !!batch && FINISHED_BATCH_STATUS.has(batch.status); + + const loadBatchById = useCallback(async (id: string) => { + if (!id) return; + setRefreshing(true); + try { + const data = await accountCheckApi.getBatch(id); + setBatch(data); + localStorage.setItem(BATCH_STORAGE_KEY, data.batch_id); + } catch (e: unknown) { + const err = getErrorMessage(e); + if (err.includes('批次不存在') || err.includes('服务已重启')) { + localStorage.removeItem(BATCH_STORAGE_KEY); + setBatch(null); + message.warning('上次账号检测批次已不存在，已清除恢复记录'); + } else { + message.error(err); + } + } finally { + setRefreshing(false); + } + }, []); + + const refreshBatch = useCallback(async () => { + if (!batchId) return; + await loadBatchById(batchId); + }, [batchId, loadBatchById]); + + useEffect(() => { + localStorage.setItem(FORM_STORAGE_KEY, text); + }, [text]); + + useEffect(() => { + localStorage.setItem('account_check_concurrency', String(concurrency)); + }, [concurrency]); + + useEffect(() => { + localStorage.setItem('account_check_max_total_time', String(maxTotalTime)); + }, [maxTotalTime]); + + useEffect(() => { + localStorage.setItem('account_check_max_login_retries', String(maxLoginRetries)); + }, [maxLoginRetries]); + + useEffect(() => { + if (restoredBatchRef.current) return; + restoredBatchRef.current = true; + const storedBatchId = readStoredBatchId(); + if (storedBatchId) { + loadBatchById(storedBatchId); + } + }, [loadBatchById]); + + useEffect(() => { + if (!isRunning || !batchId) return undefined; + const timer = window.setInterval(() => { + refreshBatch(); + }, 2000); + return () => window.clearInterval(timer); + }, [batchId, isRunning, refreshBatch]); + + const statusCounts = batch?.status_counts || {}; + const categoryStats = useMemo(() => [ + { title: '已注销', value: statusCounts.account_cancelled || 0 }, + { title: '密码错误', value: statusCounts.password_wrong || 0 }, + { title: '未认证', value: statusCounts.account_unverified || 0 }, + { title: '已认证', value: statusCounts.account_verified || 0 }, + ], [statusCounts]); + + const handleStart = async () => { + if (!text.trim()) { + message.warning('请先导入账号'); + return; + } + setStarting(true); + try { + const data = await accountCheckApi.start({ + text, + concurrency, + max_total_time: maxTotalTime, + max_login_retries: maxLoginRetries, + }); + setBatch(data); + localStorage.setItem(BATCH_STORAGE_KEY, data.batch_id); + message.success(`账号检测批次已启动，共 ${data.total} 个账号`); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setStarting(false); + } + }; + + const handleStop = async () => { + if (!batchId) return; + setStopping(true); + try { + const result = await accountCheckApi.stop(batchId); + message.success(result.message); + refreshBatch(); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setStopping(false); + } + }; + + const handleDownload = async () => { + if (!batchId) return; + setDownloading(true); + try { + const blob = await accountCheckApi.download(batchId); + downloadBlob(blob instanceof Blob ? blob : new Blob([blob]), `account-check-${fileTimestamp()}.zip`); + message.success('已下载压缩包'); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setDownloading(false); + } + }; + + const columns: TableProps['columns'] = [ + { title: '行', dataIndex: 'line', width: 70 }, + { title: '账号', dataIndex: 'username', width: 180, ellipsis: true }, + { title: '邮箱', dataIndex: 'email', width: 220, ellipsis: true }, + { + title: '状态', + dataIndex: 'status', + width: 130, + render: (status: string) => ( + + {STATUS_LABELS[status] || status} + + ), + }, + { title: '消息', dataIndex: 'message', ellipsis: true }, + { + title: '完成时间', + dataIndex: 'finished_at', + width: 180, + render: (value: string | null) => formatTime(value), + }, + ]; + + return ( + + 账号检测 + + + +