优化客服CK重登流程

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