From bf2f78aa70a14e9d9052c7583d7e6494581d9d9d Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 14 Aug 2026 14:43:21 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=AE=A2=E6=9C=8DCK=E9=87=8D?= =?UTF-8?q?=E7=99=BB=E6=B5=81=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 4 + tests/test_cookie_operations.py | 89 +++++++ web/backend/main.py | 4 + web/backend/routers/cookies.py | 248 ++++++++++-------- web/backend/services/cookie_check_service.py | 92 +++++++ web/backend/services/login_service.py | 184 ++++++++++--- web/frontend/src/api/cookies.ts | 4 + web/frontend/src/api/types.ts | 3 + web/frontend/src/pages/AuditLogsPage.tsx | 4 + .../src/pages/CookieOperationsPage.tsx | 80 +++++- 10 files changed, 550 insertions(+), 162 deletions(-) create mode 100644 web/backend/services/cookie_check_service.py diff --git a/.env.example b/.env.example index e2709ce..58cb61a 100644 --- a/.env.example +++ b/.env.example @@ -21,6 +21,10 @@ APP_PORT=8000 # 应用日志等级：DEBUG、INFO、WARNING、ERROR、CRITICAL；生产环境建议保持 INFO。 LOG_LEVEL=INFO +# CK 重登保护：单账号最多尝试次数与总耗时秒数，超时/失败均保留旧 Cookie。 +COOKIE_RELOGIN_MAX_RETRIES=5 +COOKIE_RELOGIN_MAX_TOTAL_TIME=600 + # Docker 构建镜像源（网络可访问官方源时保持 false；国内服务器可改为 true） USE_CHINA_MIRRORS=false diff --git a/tests/test_cookie_operations.py b/tests/test_cookie_operations.py index 51767c6..4c7a7ba 100644 --- a/tests/test_cookie_operations.py +++ b/tests/test_cookie_operations.py @@ -11,12 +11,15 @@ from sqlalchemy.orm import sessionmaker from web.backend.database import Base from web.backend.models import Account, LoginTask, User +from web.backend.services.login_service import LoginBatchRunner, cleanup_orphan_relogin_tasks +from web.backend.services.cookie_check_service import check_douyu_cookie from web.backend.routers.cookies import ( check_cookie_operations, get_cookie, list_cookie_operation_tags, list_cookie_operations, list_cookies, + relogin_invalid_cookie_operations, ) @@ -78,6 +81,9 @@ class CookieOperationTests(unittest.TestCase): "ck_check_status": "", "ck_checked_at": None, "created_at": None, + "relogin_status": "", + "relogin_message": "", + "relogin_batch_id": "", }]) self.assertNotIn("owned-secret", str(result)) self.assertNotIn("owned-password", str(result)) @@ -168,6 +174,89 @@ class CookieOperationTests(unittest.TestCase): current=no_permission_user, ) + def test_relogin_failed_record_stays_visible_and_can_be_checked(self): + self.owned_task.status = "relogin_failed" + self.owned_task.message = "重新登录失败: 密码错误（旧 Cookie 已保留）" + self.session.commit() + + result = list_cookie_operations( + search="", tag="", page=1, page_size=20, db=self.session, current=self.support, + ) + + self.assertEqual(result["total"], 1) + self.assertEqual(result["items"][0]["relogin_status"], "relogin_failed") + self.assertIn("旧 Cookie 已保留", result["items"][0]["relogin_message"]) + + def test_service_restart_cleans_orphan_relogin_tasks(self): + self.owned_task.status = "relogin_running" + self.session.commit() + + cleaned = cleanup_orphan_relogin_tasks(self.session) + + self.assertEqual(cleaned, 1) + self.session.refresh(self.owned_task) + self.session.refresh(self.other_task) + self.assertEqual(self.owned_task.status, "relogin_failed") + self.assertIn("服务重启", self.owned_task.message) + self.assertEqual(self.other_task.status, "success") + + def test_runner_copies_proxy_config_before_background_execution(self): + proxy = SimpleNamespace( + enabled=True, + http="http://127.0.0.1:8080", + https="", + api_url="", + whitelist_enabled=False, + whitelist_platform=None, + whitelist_credentials=None, + whitelist_uid="", + whitelist_ukey="", + ) + runner = LoginBatchRunner( + db=self.session, + account_ids=[], + created_by=self.support.id, + creator_permissions=[], + proxy_config=proxy, + ) + proxy.http = "http://changed.example:8080" + + proxy_dict, _ = runner._resolve_static_proxy() + + self.assertEqual(proxy_dict, { + "http": "http://127.0.0.1:8080", + "https": "http://127.0.0.1:8080", + }) + + @patch("web.backend.services.cookie_check_service.requests.get") + def test_cookie_check_uses_new_cookie_result(self, mock_get): + mock_get.side_effect = [ + SimpleNamespace(json=lambda: {"error": 0, "data": {"count": 9}}), + SimpleNamespace(json=lambda: {"error": 0, "data": {"nn": "new-name", "lv": 12}}), + ] + + result = check_douyu_cookie("new-cookie") + + self.assertTrue(result["valid"]) + self.assertEqual(result["fish_ball"], 9) + self.assertEqual(result["nickname"], "new-name") + self.assertEqual(result["level"], 12) + + @patch("web.backend.routers.cookies._start_relogin_tasks") + def test_relogin_invalid_only_targets_support_visible_accounts(self, start_relogin): + self.owned_task.ck_check_status = "invalid" + self.other_task.ck_check_status = "invalid" + self.session.commit() + start_relogin.return_value = {"batch_id": "batch", "count": 1, "skipped": 0, "success": True} + + result = relogin_invalid_cookie_operations( + search="", tag="", db=self.session, current=self.support, + ) + + self.assertTrue(result["success"]) + selected_tasks = start_relogin.call_args.args[0] + self.assertEqual([task.id for task in selected_tasks], [self.owned_task.id]) + if __name__ == "__main__": unittest.main() diff --git a/web/backend/main.py b/web/backend/main.py index 5a1403d..5ae33c9 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -30,6 +30,7 @@ async def lifespan(app: FastAPI): from loguru import logger from .database import SessionLocal + from .services.login_service import cleanup_orphan_relogin_tasks from .services.huya_service import cleanup_orphan_huya_tasks from .services.douyu_service import cleanup_orphan_douyu_tasks @@ -41,6 +42,9 @@ async def lifespan(app: FastAPI): cleaned_douyu = cleanup_orphan_douyu_tasks(db, message="任务已中断（服务重启）") if cleaned_douyu: logger.info(f"启动清理斗鱼残留任务: {cleaned_douyu} 条") + cleaned_relogin = cleanup_orphan_relogin_tasks(db) + if cleaned_relogin: + logger.info(f"启动清理 CK 重登残留任务: {cleaned_relogin} 条") from .services.yyb_service import cleanup_orphan_yyb_tasks cleaned_yyb = cleanup_orphan_yyb_tasks(db, message="任务已中断（服务重启）") if cleaned_yyb: diff --git a/web/backend/routers/cookies.py b/web/backend/routers/cookies.py index 6681b02..019fbc1 100644 --- a/web/backend/routers/cookies.py +++ b/web/backend/routers/cookies.py @@ -3,7 +3,6 @@ import threading from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone -import requests from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from sqlalchemy import case, or_ @@ -16,7 +15,8 @@ from ..models import User, LoginTask, Account, AuditLog, ProxyConfig as ProxyCon from ..deps import get_current_user, require_permission from ..permissions import user_has_permission, get_user_permissions from ..schemas import CookieReloginRequest -from ..services.login_service import LoginBatchRunner +from ..services.login_service import BatchRegistry, LoginBatchRunner, get_relogin_limits +from ..services.cookie_check_service import check_douyu_cookie def _fmt_dt(dt) -> str | None: @@ -28,6 +28,7 @@ def _fmt_dt(dt) -> str | None: return dt.isoformat() router = APIRouter(prefix="/api/cookies", tags=["Cookie管理"]) +cookie_relogin_registry = BatchRegistry() def _require_cookie_operation_perm(current: User) -> None: @@ -63,89 +64,12 @@ def _order_cookie_tasks(query, selected_names: list[str]): return query.order_by(LoginTask.finished_at.desc(), LoginTask.id.desc()) -# 斗鱼 Cookie 有效性检测接口（抓包参考：钱包中心鱼丸余额 + 用户等级详情） -CHECK_UA = ( - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" -) -FISH_BALL_API = "https://www.douyu.com/wjapi/nc/exchange/fishBall" -USER_LEVEL_API = "https://www.douyu.com/japi/interactnc/web/userLevel/userLevelDetail" - - def _check_one_cookie(task: LoginTask) -> dict: - """检测单条斗鱼 Cookie 有效性：鱼丸 + 用户等级双接口均通过才算有效。 - - 附带鱼丸数、昵称/等级与检测时间；任一接口失败会说明原因。 - """ - cookie = task.cookie or "" - checked_at = datetime.now(timezone.utc).isoformat() - base = {"id": task.id, "checked_at": checked_at, "fish_ball": None, "nickname": None, "level": None} - if not cookie: - return {**base, "valid": False, "message": "Cookie 为空"} - - fish_ok = False - fish_msg = "" - fish_ball = None - try: - fish_data = requests.get( - FISH_BALL_API, - params={"appCode": "YJTX"}, - headers={ - "User-Agent": CHECK_UA, - "Accept": "application/json, text/plain, */*", - "Referer": "https://www.douyu.com/member/walletcenter", - "Cookie": cookie, - }, - timeout=(5, 10), - ).json() - if isinstance(fish_data, dict) and fish_data.get("error") in (0, "0"): - fish_ok = True - fish_ball = (fish_data.get("data") or {}).get("count") if isinstance(fish_data.get("data"), dict) else None - else: - fish_msg = str(fish_data.get("msg") or fish_data.get("error") or "响应异常") if isinstance(fish_data, dict) else "响应异常" - except Exception as exc: - fish_msg = f"请求失败: {exc}" - - level_ok = False - level_msg = "" - nickname = None - level = None - try: - level_data = requests.get( - USER_LEVEL_API, - params={"rid": "0"}, - headers={ - "User-Agent": CHECK_UA, - "Accept": "application/json, text/plain, */*", - "Referer": "https://www.douyu.com/pages/ord-user-level?clientType=web", - "Cookie": cookie, - }, - timeout=(5, 10), - ).json() - if isinstance(level_data, dict) and level_data.get("error") in (0, "0"): - level_ok = True - info = level_data.get("data") if isinstance(level_data.get("data"), dict) else {} - nickname = str(info.get("nn") or "") or None - level = info.get("lv") - else: - level_msg = str(level_data.get("msg") or level_data.get("error") or "响应异常") if isinstance(level_data, dict) else "响应异常" - except Exception as exc: - level_msg = f"请求失败: {exc}" - - valid = fish_ok and level_ok - if valid: - message = "有效" - else: - parts = [f"鱼丸接口: {'ok' if fish_ok else (fish_msg or '失败')}", f"等级接口: {'ok' if level_ok else (level_msg or '失败')}"] - message = "；".join(parts) - return { - **base, - "valid": valid, - "message": message[:200], - "fish_ball": fish_ball, - "nickname": nickname, - "level": level, - } + """将共享检测结果补充为 Cookie 记录的接口返回格式。""" + result = check_douyu_cookie(task.cookie or "") + result["id"] = task.id + result["checked_at"] = result["checked_at"].isoformat() + return result def _visible_cookie_tasks_query(db: Session, current: User): @@ -160,6 +84,19 @@ def _visible_cookie_tasks_query(db: Session, current: User): return query +def _visible_cookie_operation_tasks_query(db: Session, current: User): + """返回可检测/重登的 Cookie 记录，重登中或失败时仍保留在操作列表。""" + query = db.query(LoginTask).filter( + LoginTask.cookie != "", + LoginTask.status.in_(("success", "relogin_pending", "relogin_running", "relogin_failed")), + ) + if not user_has_permission(current, "login:view_all"): + query = query.join(Account, LoginTask.account_id == Account.id).filter( + Account.assigned_to == current.id + ) + return query + + @router.get("") def list_cookies( search: str = Query(""), @@ -292,7 +229,7 @@ def list_cookie_operations( ): """脱敏的 CK 操作列表，只用于检测与重登。""" _require_cookie_operation_perm(current) - query = _visible_cookie_tasks_query(db, current) + query = _visible_cookie_operation_tasks_query(db, current) if user_has_permission(current, "login:view_all"): query = query.join(Account, LoginTask.account_id == Account.id) search_text = (search or "").strip() @@ -321,6 +258,9 @@ def list_cookie_operations( "ck_check_status": task.ck_check_status or "", "ck_checked_at": _fmt_dt(task.ck_checked_at), "created_at": _fmt_dt(task.finished_at), + "relogin_status": task.status if task.status != "success" else "", + "relogin_message": task.message or "", + "relogin_batch_id": task.batch_id if task.status in {"relogin_pending", "relogin_running"} else "", } for task in tasks ], @@ -337,7 +277,7 @@ def list_cookie_operation_tags( ): """返回当前用户可操作 CK 记录对应的标签，不跨越账号分配范围。""" _require_cookie_operation_perm(current) - query = _visible_cookie_tasks_query(db, current) + query = _visible_cookie_operation_tasks_query(db, current) if user_has_permission(current, "login:view_all"): query = query.join(Account, LoginTask.account_id == Account.id) rows = ( @@ -487,7 +427,10 @@ def _check_cookies(ids: str, db: Session, current: User, *, detailed: bool) -> d id_list = [int(x) for x in ids.split(",") if x.strip().isdigit()] if not id_list: raise HTTPException(status_code=400, detail="无效的ID") - tasks = _visible_cookie_tasks_query(db, current).filter(LoginTask.id.in_(id_list)).all() + base_query = _visible_cookie_tasks_query(db, current) if detailed else _visible_cookie_operation_tasks_query(db, current) + tasks = base_query.filter(LoginTask.id.in_(id_list)).filter( + LoginTask.status.in_(("success", "relogin_failed")) + ).all() if not tasks: raise HTTPException(status_code=404, detail="记录不存在") @@ -562,48 +505,46 @@ def check_cookie_operations( return _check_cookies(ids, db, current, detailed=False) -def _start_relogin( - req: CookieReloginRequest, - db: Session, - current: User, -): - """重新登录失效的 Cookie：用账号信息重新登录，成功后替换旧 Cookie。 - - 复用原 LoginTask 记录，行不消失；重新登录失败保留旧 Cookie 并记录原因。 - """ - if not req.ids: - raise HTTPException(status_code=400, detail="请选择要重新登录的 Cookie") - - tasks = _visible_cookie_tasks_query(db, current).filter(LoginTask.id.in_(req.ids)).all() +def _start_relogin_tasks(tasks: list[LoginTask], db: Session, current: User, *, action: str = "cookie:relogin"): + """启动重登批次：旧 Cookie 保留到新登录成功后才替换。""" if not tasks: raise HTTPException(status_code=404, detail="记录不存在") - # 过滤缺少登录凭据的账号（密码/邮箱），避免启动后全部失败 task_ids: list[int] = [] - skipped: list[str] = [] - accounts_map = {t.account_id: t for t in tasks} - for acc_id, task in accounts_map.items(): - acc = db.query(Account).filter(Account.id == acc_id).first() + skipped: list[int] = [] + for task in tasks: + acc = db.query(Account).filter(Account.id == task.account_id).first() if not acc: - continue - if not acc.password or not acc.email: skipped.append(task.id) continue + if not acc.password or not acc.email: + task.status = "relogin_failed" + task.message = "重新登录失败: 账号缺少密码或邮箱，旧 Cookie 已保留" + task.finished_at = datetime.now(timezone.utc) + skipped.append(task.id) + continue + if task.status in {"relogin_pending", "relogin_running"}: + skipped.append(task.id) + continue + task.status = "relogin_pending" + task.message = "等待重新登录，旧 Cookie 已保留" + task.finished_at = None task_ids.append(task.id) if not task_ids: - raise HTTPException( - status_code=400, - detail="所选 Cookie 对应账号缺少密码或邮箱信息，无法重新登录，请先在账号管理中补充", - ) + db.commit() + raise HTTPException(status_code=400, detail="没有可重登的账号，请检查账号凭据或当前重登状态") proxy = db.query(ProxyConfigModel).first() thread_db = SessionLocal() + max_login_retries, max_total_time = get_relogin_limits() runner = LoginBatchRunner( db=thread_db, account_ids=[], created_by=current.id, creator_permissions=get_user_permissions(current), + max_login_retries=max_login_retries, + max_total_time=max_total_time, proxy_config=proxy, log_queue=None, loop=None, @@ -612,16 +553,29 @@ def _start_relogin( relogin_task_ids=task_ids, ) batch_id = runner.batch_id - thread = threading.Thread(target=runner.run, daemon=True) - thread.start() - db.add(AuditLog( user_id=current.id, username=current.username, - action="cookie:relogin", - target=f"重登 {len(task_ids)} 条已分配账号 CK", + action=action, + target=f"重登 {len(task_ids)} 条账号 CK", + detail="旧 Cookie 将在新登录成功后替换", )) db.commit() + cookie_relogin_registry.register(batch_id, None, None, runner, owner_id=current.id) + + def run_relogin_batch(): + try: + runner.run() + except Exception as exc: + runner._push_log("error", f"批量重新登录异常: {exc}") + message = f"重新登录异常: {exc}（旧 Cookie 已保留）" + for task_id in task_ids: + runner._mark_relogin_stopped(task_id, message) + runner.db.close() + finally: + cookie_relogin_registry.pop(batch_id) + + threading.Thread(target=run_relogin_batch, daemon=True).start() return { "batch_id": batch_id, "count": len(task_ids), @@ -630,6 +584,44 @@ def _start_relogin( } +@router.post("/operations/relogin/{batch_id}/stop") +def stop_cookie_relogin( + batch_id: str, + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """停止当前用户发起的 CK 重登批次。""" + _require_cookie_operation_perm(current) + batch = cookie_relogin_registry.get(batch_id) + if not batch: + raise HTTPException(status_code=404, detail="重登批次不存在、已结束或服务已重启") + if batch.get("owner_id") != current.id and not user_has_permission(current, "login:view_all"): + raise HTTPException(status_code=403, detail="无权限停止该重登批次") + + batch["runner"].stop() + db.add(AuditLog( + user_id=current.id, + username=current.username, + action="cookie:relogin_stop", + target=f"停止 CK 重登批次 {batch_id}", + )) + db.commit() + return {"message": "已发送停止信号，正在登录的账号会在当前请求结束后停止", "success": True} + + +def _start_relogin(req: CookieReloginRequest, db: Session, current: User): + """按选中的 Cookie 记录启动重新登录。""" + if not req.ids: + raise HTTPException(status_code=400, detail="请选择要重新登录的 Cookie") + tasks = ( + _visible_cookie_operation_tasks_query(db, current) + .filter(LoginTask.id.in_(req.ids)) + .filter(LoginTask.status.in_(("success", "relogin_failed"))) + .all() + ) + return _start_relogin_tasks(tasks, db, current) + + @router.post("/relogin") def relogin_cookies( req: CookieReloginRequest, @@ -651,6 +643,30 @@ def relogin_cookie_operations( return _start_relogin(req, db, current) +@router.post("/operations/relogin-invalid") +def relogin_invalid_cookie_operations( + search: str = Query(""), + tag: str = Query(""), + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """批量重新登录当前用户可见且最近检测为无效的账号。""" + _require_cookie_operation_perm(current) + query = _visible_cookie_operation_tasks_query(db, current).filter( + LoginTask.ck_check_status == "invalid", + LoginTask.status.in_(("success", "relogin_failed")), + ) + if user_has_permission(current, "login:view_all"): + query = query.join(Account, LoginTask.account_id == Account.id) + if tag.strip(): + query = query.filter(Account.tag == tag.strip()) + if search.strip(): + pattern = f"%{search.strip()}%" + query = query.filter(or_(Account.username.ilike(pattern), Account.tag.ilike(pattern))) + tasks = query.order_by(LoginTask.id.asc()).all() + return _start_relogin_tasks(tasks, db, current, action="cookie:relogin_invalid") + + @router.get("/{task_id}") def get_cookie( task_id: int, diff --git a/web/backend/services/cookie_check_service.py b/web/backend/services/cookie_check_service.py new file mode 100644 index 0000000..28c3e59 --- /dev/null +++ b/web/backend/services/cookie_check_service.py @@ -0,0 +1,92 @@ +"""斗鱼 Cookie 有效性检测服务。""" + +from datetime import datetime, timezone + +import requests + + +CHECK_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" +) +FISH_BALL_API = "https://www.douyu.com/wjapi/nc/exchange/fishBall" +USER_LEVEL_API = "https://www.douyu.com/japi/interactnc/web/userLevel/userLevelDetail" + + +def check_douyu_cookie(cookie: str) -> dict: + """检测斗鱼 Cookie 有效性，鱼丸与等级接口均通过才算有效。""" + checked_at = datetime.now(timezone.utc) + base = { + "checked_at": checked_at, + "fish_ball": None, + "nickname": None, + "level": None, + } + if not cookie: + return {**base, "valid": False, "message": "Cookie 为空"} + + fish_ok = False + fish_msg = "" + fish_ball = None + try: + fish_data = requests.get( + FISH_BALL_API, + params={"appCode": "YJTX"}, + headers={ + "User-Agent": CHECK_UA, + "Accept": "application/json, text/plain, */*", + "Referer": "https://www.douyu.com/member/walletcenter", + "Cookie": cookie, + }, + timeout=(5, 10), + ).json() + if isinstance(fish_data, dict) and fish_data.get("error") in (0, "0"): + fish_ok = True + fish_ball = (fish_data.get("data") or {}).get("count") if isinstance(fish_data.get("data"), dict) else None + else: + fish_msg = str(fish_data.get("msg") or fish_data.get("error") or "响应异常") if isinstance(fish_data, dict) else "响应异常" + except Exception as exc: + fish_msg = f"请求失败: {exc}" + + level_ok = False + level_msg = "" + nickname = None + level = None + try: + level_data = requests.get( + USER_LEVEL_API, + params={"rid": "0"}, + headers={ + "User-Agent": CHECK_UA, + "Accept": "application/json, text/plain, */*", + "Referer": "https://www.douyu.com/pages/ord-user-level?clientType=web", + "Cookie": cookie, + }, + timeout=(5, 10), + ).json() + if isinstance(level_data, dict) and level_data.get("error") in (0, "0"): + level_ok = True + info = level_data.get("data") if isinstance(level_data.get("data"), dict) else {} + nickname = str(info.get("nn") or "") or None + level = info.get("lv") + else: + level_msg = str(level_data.get("msg") or level_data.get("error") or "响应异常") if isinstance(level_data, dict) else "响应异常" + except Exception as exc: + level_msg = f"请求失败: {exc}" + + valid = fish_ok and level_ok + if valid: + message = "有效" + else: + message = "；".join([ + f"鱼丸接口: {'ok' if fish_ok else (fish_msg or '失败')}", + f"等级接口: {'ok' if level_ok else (level_msg or '失败')}", + ]) + return { + **base, + "valid": valid, + "message": message[:200], + "fish_ball": fish_ball, + "nickname": nickname, + "level": level, + } diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index f4c943a..6719ee7 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -1,6 +1,7 @@ """登录服务：复用 core/ 核心模块，在线程池中并发执行登录并推送日志。""" import asyncio +import os import threading import time import uuid @@ -10,10 +11,12 @@ from datetime import datetime, timezone from typing import Optional from sqlalchemy.orm import Session +from loguru import logger from core.douyu import DouyuLogin, WgapiLoginAPI, IframeLoginAPI from core.douyu.proxy_fetcher import ProxyFetcher from ..models import Account as AccountModel, LoginTask, ProxyConfig as ProxyConfigModel +from .cookie_check_service import check_douyu_cookie def _create_api_strategy(strategy_name: str): @@ -40,6 +43,41 @@ CHECK_STATUS_LOG_LEVELS = { } +def _positive_env_int(name: str, default: int) -> int: + """读取正整数环境变量，非法值回退到默认值。""" + try: + value = int(os.getenv(name, str(default))) + except ValueError: + return default + return value if value > 0 else default + + +def get_relogin_limits() -> tuple[int, int]: + """读取客服 CK 重登的重试与总时长限制。""" + return ( + _positive_env_int("COOKIE_RELOGIN_MAX_RETRIES", 5), + _positive_env_int("COOKIE_RELOGIN_MAX_TOTAL_TIME", 600), + ) + + +def _snapshot_proxy_config(proxy_config: Optional[ProxyConfigModel]) -> Optional[SimpleNamespace]: + """复制代理配置，避免后台线程访问已关闭会话中的 ORM 对象。""" + if proxy_config is None: + return None + credentials = getattr(proxy_config, "whitelist_credentials", None) + return SimpleNamespace( + enabled=bool(getattr(proxy_config, "enabled", False)), + http=getattr(proxy_config, "http", "") or "", + https=getattr(proxy_config, "https", "") or "", + api_url=getattr(proxy_config, "api_url", "") or "", + whitelist_enabled=bool(getattr(proxy_config, "whitelist_enabled", False)), + whitelist_platform=getattr(proxy_config, "whitelist_platform", None), + whitelist_credentials=dict(credentials) if isinstance(credentials, dict) else credentials, + whitelist_uid=getattr(proxy_config, "whitelist_uid", "") or "", + whitelist_ukey=getattr(proxy_config, "whitelist_ukey", "") or "", + ) + + class LoginBatchRunner: """批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。""" @@ -67,7 +105,7 @@ class LoginBatchRunner: # 0 表示使用默认值 20，其他值保持原样 self.max_login_retries = max_login_retries if max_login_retries > 0 else 20 self.max_total_time = max_total_time - self.proxy_config = proxy_config + self.proxy_config = _snapshot_proxy_config(proxy_config) self.log_queue = log_queue self.loop = loop self.batch_id = uuid.uuid4().hex[:12] @@ -81,6 +119,7 @@ class LoginBatchRunner: # 共享代理获取器（无池，每次取新代理） self._shared_proxy_fetcher = None + proxy_config = self.proxy_config if proxy_config and proxy_config.enabled and proxy_config.api_url: wl_platform = "xiequ" wl_credentials = None @@ -111,7 +150,28 @@ class LoginBatchRunner: time.sleep(min(0.2, deadline - time.monotonic())) return self._stop.is_set() + def _mark_relogin_stopped( + self, + task_id: int, + message: str = "重新登录已停止，旧 Cookie 已保留", + ) -> None: + """将未开始或已中断的重登任务收敛为保留旧 CK 的终态。""" + worker_db = SessionLocal() + try: + task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first() + if task and task.status in ("relogin_pending", "relogin_running"): + task.status = "relogin_failed" + task.message = message + task.finished_at = datetime.now(timezone.utc) + worker_db.commit() + finally: + worker_db.close() + def _push_log(self, level: str, message: str): + # 即使没有页面实时日志，也要保留批次进度到 app.log，便于排查卡点。 + if message: + log_level = level if level in {"debug", "info", "warning", "error", "success"} else "debug" + getattr(logger, log_level)(f"[登录批次 {self.batch_id}] {message}") if self.log_queue and self.loop: asyncio.run_coroutine_threadsafe( self.log_queue.put({"level": level, "message": message}), @@ -136,16 +196,7 @@ class LoginBatchRunner: if self._stop.is_set(): self._push_log("warning", f"任务已停止，跳过: {acc_info['username']}") if self.mode == "relogin": - # 重新登录被停止时恢复原 Cookie 记录，避免列表行消失 - worker_db = SessionLocal() - try: - task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first() - if task and task.status in ("pending", "running"): - task.status = "success" - task.finished_at = datetime.now(timezone.utc) - worker_db.commit() - finally: - worker_db.close() + self._mark_relogin_stopped(task_id) return worker_db = SessionLocal() @@ -154,7 +205,11 @@ class LoginBatchRunner: if not task: return - task.status = "running" + if self.mode == "relogin": + task.status = "relogin_running" + task.message = "正在重新登录，旧 Cookie 保留中" + else: + task.status = "running" worker_db.commit() with self._counter_lock: @@ -164,21 +219,25 @@ class LoginBatchRunner: action_name = "检测" if self.mode == "check" else "重新登录" if self.mode == "relogin" else "登录" self._push_log("info", f"[{current}/{total}] 开始{action_name}: {acc_info['username']}") - # 解析代理配置 - proxy_dict, proxy_msg = self._resolve_static_proxy() - if proxy_msg: - self._push_log("info", f"[{current}] {proxy_msg}") - - # 静态代理启用但配置为空 → 不可用 - if self.proxy_config and self.proxy_config.enabled and not (self.proxy_config.http or self.proxy_config.https) and not self._shared_proxy_fetcher and not proxy_dict: - task.status = "error" - task.message = "代理不可用: 未配置代理" - task.finished_at = datetime.now(timezone.utc) - worker_db.commit() - self._push_log("error", f"[{current}] {acc_info['username']} 代理不可用") - return - try: + # 代理配置也可能异常，必须由当前任务的失败处理收敛状态。 + proxy_dict, proxy_msg = self._resolve_static_proxy() + if proxy_msg: + self._push_log("info", f"[{current}] {proxy_msg}") + + # 静态代理启用但配置为空 → 不可用 + if self.proxy_config and self.proxy_config.enabled and not (self.proxy_config.http or self.proxy_config.https) and not self._shared_proxy_fetcher and not proxy_dict: + if self.mode == "relogin": + task.status = "relogin_failed" + task.message = "重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）" + else: + task.status = "error" + task.message = "代理不可用: 未配置代理" + task.finished_at = datetime.now(timezone.utc) + worker_db.commit() + self._push_log("error", f"[{current}] {acc_info['username']} 代理不可用") + return + account = SimpleNamespace( username=acc_info["username"], password=acc_info["password"], @@ -212,18 +271,28 @@ class LoginBatchRunner: task.cookie = result.cookie task.message = result.message or "登录成功" if self.mode == "relogin": - task.ck_check_status = "" - task.ck_check_result = None - task.ck_checked_at = None - task.message = "重新登录成功" - self._push_log("success", f"[{current}] {acc_info['username']} 重新登录成功，Cookie 已替换") + check_result = check_douyu_cookie(result.cookie) + task.ck_check_status = "valid" if check_result["valid"] else "invalid" + task.ck_check_result = { + "fish_ball": check_result["fish_ball"], + "nickname": check_result["nickname"], + "level": check_result["level"], + "message": check_result["message"], + } + task.ck_checked_at = check_result["checked_at"] + if check_result["valid"]: + task.message = "重新登录成功，Cookie 有效" + self._push_log("success", f"[{current}] {acc_info['username']} 重新登录成功，Cookie 已替换并验证有效") + else: + task.message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}" + self._push_log("warning", f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}") else: self._push_log("success", f"[{current}] {acc_info['username']} {task.message}") else: if self.mode == "relogin": # 重新登录失败时保留旧 Cookie 与成功状态，仅记录失败原因，行不消失 - task.status = "success" - task.message = f"重新登录失败: {result.message}" + task.status = "relogin_failed" + task.message = f"重新登录失败: {result.message}（旧 Cookie 已保留）" self._push_log("error", f"[{current}] {acc_info['username']} 重新登录失败: {result.message}") else: task.status = "failed" @@ -232,8 +301,8 @@ class LoginBatchRunner: except Exception as e: if self.mode == "relogin": - task.status = "success" - task.message = f"重新登录异常: {e}" + task.status = "relogin_failed" + task.message = f"重新登录异常: {e}（旧 Cookie 已保留）" self._push_log("error", f"[{current}] {acc_info['username']} 重新登录异常: {e}") else: task.status = "error" @@ -261,8 +330,13 @@ class LoginBatchRunner: def _append_task_info(task: LoginTask, acc: AccountModel): task.batch_id = batch_id - task.status = "pending" - task.message = "" + if self.mode == "relogin": + # 旧 Cookie 仅在 result.success 后由 _execute_one 原子替换。 + task.status = "relogin_pending" + task.message = "等待重新登录，旧 Cookie 已保留" + else: + task.status = "pending" + task.message = "" task.finished_at = None self.db.flush() task_infos.append({ @@ -295,9 +369,6 @@ class LoginBatchRunner: if acc.assigned_to != self.created_by: self._push_log("warning", f"跳过无权账号: {acc.username}") continue - task.ck_check_status = "" - task.ck_check_result = None - task.ck_checked_at = None _append_task_info(task, acc) else: seen_account_ids = set() @@ -371,9 +442,12 @@ class LoginBatchRunner: # 并发执行登录，每个账号独立获取代理 with ThreadPoolExecutor(max_workers=concurrency) as executor: futures = [] - for item in task_infos: + for item_index, item in enumerate(task_infos): if self._stop.is_set(): self._push_log("warning", "任务已停止，跳过剩余账号") + if self.mode == "relogin": + for pending_item in task_infos[item_index:]: + self._mark_relogin_stopped(pending_item["task_id"]) break future = executor.submit( self._execute_one, @@ -403,12 +477,14 @@ class BatchRegistry: def __init__(self): self._batches: dict[str, dict] = {} - def register(self, batch_id: str, log_queue: asyncio.Queue, - loop: asyncio.AbstractEventLoop, runner: LoginBatchRunner): + def register(self, batch_id: str, log_queue: Optional[asyncio.Queue], + loop: Optional[asyncio.AbstractEventLoop], runner: LoginBatchRunner, + owner_id: Optional[int] = None): self._batches[batch_id] = { "log_queue": log_queue, "loop": loop, "runner": runner, + "owner_id": owner_id, } def get(self, batch_id: str): @@ -422,5 +498,27 @@ class BatchRegistry: batch_registry = BatchRegistry() +def cleanup_orphan_relogin_tasks( + db: Session, + message: str = "重新登录已中断（服务重启），旧 Cookie 已保留", +) -> int: + """服务重启后收敛遗留重登状态，避免页面永久显示重登中。""" + tasks = ( + db.query(LoginTask) + .filter(LoginTask.status.in_(("relogin_pending", "relogin_running"))) + .all() + ) + if not tasks: + return 0 + + finished_at = datetime.now(timezone.utc) + for task in tasks: + task.status = "relogin_failed" + task.message = message + task.finished_at = finished_at + db.commit() + return len(tasks) + + # 在模块末尾导入 SessionLocal（避免循环导入） from ..database import SessionLocal diff --git a/web/frontend/src/api/cookies.ts b/web/frontend/src/api/cookies.ts index 02f785b..4051d17 100644 --- a/web/frontend/src/api/cookies.ts +++ b/web/frontend/src/api/cookies.ts @@ -26,6 +26,10 @@ export const cookieApi = { api.post<{ batch_id: string; count: number; skipped: number; success: boolean }, { batch_id: string; count: number; skipped: number; success: boolean }>('/cookies/relogin', { ids }), reloginOperations: (ids: number[]) => api.post<{ batch_id: string; count: number; skipped: number; success: boolean }, { batch_id: string; count: number; skipped: number; success: boolean }>('/cookies/operations/relogin', { ids }), + reloginInvalidOperations: (params?: { search?: string; tag?: string }) => + api.post<{ batch_id: string; count: number; skipped: number; success: boolean }, { batch_id: string; count: number; skipped: number; success: boolean }>('/cookies/operations/relogin-invalid', null, { params }), + stopOperationRelogin: (batchId: string) => + api.post(`/cookies/operations/relogin/${batchId}/stop`), loginTasks: (batchId: string) => api.get('/login/tasks', { params: { batch_id: batchId } }), delete: (id: number) => api.delete(`/cookies/${id}`), diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index 2e7a2cd..872eaee 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -280,6 +280,9 @@ export interface CookieOperationItem { ck_check_status: string; ck_checked_at: string | null; created_at: string | null; + relogin_status: '' | 'relogin_pending' | 'relogin_running' | 'relogin_failed'; + relogin_message: string; + relogin_batch_id: string; } export interface CookieOperationCheckResult { diff --git a/web/frontend/src/pages/AuditLogsPage.tsx b/web/frontend/src/pages/AuditLogsPage.tsx index 72e0218..f87e6bb 100644 --- a/web/frontend/src/pages/AuditLogsPage.tsx +++ b/web/frontend/src/pages/AuditLogsPage.tsx @@ -22,6 +22,10 @@ const ACTION_OPTIONS = [ { value: 'recharge:huya:create', label: '虎牙创建充值批次' }, { value: 'recharge:huya:stop', label: '虎牙停止充值批次' }, { value: 'recharge:huya:config', label: '虎牙充值配置' }, + { value: 'cookie:check', label: '检测账号 CK' }, + { value: 'cookie:relogin', label: '创建 CK 重登批次' }, + { value: 'cookie:relogin_invalid', label: '批量重登失效 CK' }, + { value: 'cookie:relogin_stop', label: '停止 CK 重登批次' }, ]; const ACTION_LABELS = new Map(ACTION_OPTIONS.map((item) => [item.value, item.label])); diff --git a/web/frontend/src/pages/CookieOperationsPage.tsx b/web/frontend/src/pages/CookieOperationsPage.tsx index 2ebd275..170a23d 100644 --- a/web/frontend/src/pages/CookieOperationsPage.tsx +++ b/web/frontend/src/pages/CookieOperationsPage.tsx @@ -1,7 +1,7 @@ import { useCallback, useEffect, useState } from 'react'; import { Button, Input, Popconfirm, Select, Space, Table, Tag, Typography } from 'antd'; import { message } from '../utils/antdMessage'; -import { LoadingOutlined, ReloadOutlined, SearchOutlined } from '@ant-design/icons'; +import { LoadingOutlined, ReloadOutlined, SearchOutlined, StopOutlined } from '@ant-design/icons'; import { cookieApi, type CookieOperationCheckResult, type CookieOperationItem } from '../api/modules'; import { formatTime } from '../utils/time'; import { getErrorMessage } from '../utils/error'; @@ -59,6 +59,20 @@ export default function CookieOperationsPage() { void loadItems(); }, [loadItems]); + const hasActiveRelogin = items.some((item) => ( + item.relogin_status === 'relogin_pending' || item.relogin_status === 'relogin_running' + )); + const activeReloginBatchIds = Array.from(new Set(items + .filter((item) => item.relogin_status === 'relogin_pending' || item.relogin_status === 'relogin_running') + .map((item) => item.relogin_batch_id) + .filter(Boolean))); + + useEffect(() => { + if (!hasActiveRelogin) return; + const timer = window.setInterval(() => void loadItems(), 2000); + return () => window.clearInterval(timer); + }, [hasActiveRelogin, loadItems]); + useEffect(() => { cookieApi.listOperationTags().then(setTags).catch(() => { // 标签加载失败不影响 CK 检测与重登。 @@ -108,8 +122,41 @@ export default function CookieOperationsPage() { } }; + const handleReloginInvalid = async () => { + try { + const response = await cookieApi.reloginInvalidOperations({ + search: search.trim() || undefined, + tag: tagFilter || undefined, + }); + const skipped = response.skipped ? `，${response.skipped} 条因缺少登录凭据跳过` : ''; + message.success(`已开始重登 ${response.count} 个失效账号${skipped}`); + setSelectedRowKeys([]); + void loadItems(); + } catch (error: unknown) { + message.error(getErrorMessage(error)); + } + }; + + const handleStopRelogin = async () => { + if (activeReloginBatchIds.length === 0) return; + try { + await Promise.all(activeReloginBatchIds.map((batchId) => cookieApi.stopOperationRelogin(batchId))); + message.success('已发送停止信号'); + void loadItems(); + } catch (error: unknown) { + message.error(getErrorMessage(error)); + } + }; + const resultFor = (item: CookieOperationItem) => checkResults.get(item.id); const selectedIds = selectedRowKeys.map((key) => Number(key)); + const reloginStatus = (item: CookieOperationItem) => { + if (item.relogin_status === 'relogin_pending') return 重登排队中; + if (item.relogin_status === 'relogin_running') return }>重登中; + if (item.relogin_status === 'relogin_failed') return 重登失败; + if (item.relogin_message.startsWith('重新登录成功')) return 重登成功; + return -; + }; const columns = [ { title: '账号', dataIndex: 'account_username', ellipsis: true }, { @@ -137,6 +184,20 @@ export default function CookieOperationsPage() { return checkedAt ? formatTime(checkedAt) : -; }, }, + { + title: '重登状态', + width: 130, + render: (_: unknown, item: CookieOperationItem) => ( + + {reloginStatus(item)} + {item.relogin_status === 'relogin_failed' && item.relogin_message ? ( + + {item.relogin_message} + + ) : null} + + ), + }, { title: '操作', width: 190, @@ -145,7 +206,7 @@ export default function CookieOperationsPage() { @@ -182,6 +243,19 @@ export default function CookieOperationsPage() { 重登选中 {selectedIds.length > 0 ? `(${selectedIds.length})` : ''} + void handleReloginInvalid()} + > + + + {activeReloginBatchIds.length > 0 ? ( + void handleStopRelogin()}> + + + ) : null}