增加斗鱼账号检测功能

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)

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",

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}"'},
+    )

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

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。"""

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()

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 被关闭，避免连接泄漏