feat(token): add automatic ST refresh functionality for personal mode

- Implement automatic ST (Session Token) refresh when AT refresh fails
- Add refresh_session_token method in BrowserCaptchaService to extract
  __Secure-next-auth.session-token from browser cookies
- Enhance token_manager to automatically retry AT refresh after ST update
- Add validation for AT tokens to ensure they are actually working
- Update README to document AT/ST automatic refresh feature
- Improve error handling and logging for token refresh operations
- Add conditional ST refresh support only available in personal mode

README.md

@@ -16,7 +16,7 @@
 - 🎨 **文生图** / **图生图**
 - 🎬 **文生视频** / **图生视频**
 - 🎞️ **首尾帧视频**
-- 🔄 **AT自动刷新**
+- 🔄 **AT/ST自动刷新** - AT 过期自动刷新，ST 过期时自动通过浏览器更新（personal 模式）
 - 📊 **余额显示** - 实时查询和显示 VideoFX Credits
 - 🚀 **负载均衡** - 多 Token 轮询和并发控制
 - 🌐 **代理支持** - 支持 HTTP/SOCKS5 代理

src/api/admin.py

@@ -353,17 +353,32 @@ async def refresh_at(
     token_id: int,
     token: str = Depends(verify_admin_token)
 ):
-    """手动刷新Token的AT (使用ST转换) 🆕"""
+    """手动刷新Token的AT (使用ST转换) 🆕
+    
+    如果 AT 刷新失败且处于 personal 模式，会自动尝试通过浏览器刷新 ST
+    """
+    from ..core.logger import debug_logger
+    from ..core.config import config
+    
+    debug_logger.log_info(f"[API] 手动刷新 AT 请求: token_id={token_id}, captcha_method={config.captcha_method}")
+    
     try:
-        # 调用token_manager的内部刷新方法
+        # 调用token_manager的内部刷新方法（包含 ST 自动刷新逻辑）
         success = await token_manager._refresh_at(token_id)
 
         if success:
             # 获取更新后的token信息
             updated_token = await token_manager.get_token(token_id)
+            
+            message = "AT刷新成功"
+            if config.captcha_method == "personal":
+                message += "（支持ST自动刷新）"
+            
+            debug_logger.log_info(f"[API] AT 刷新成功: token_id={token_id}")
+            
             return {
                 "success": True,
-                "message": "AT刷新成功",
+                "message": message,
                 "token": {
                     "id": updated_token.id,
                     "email": updated_token.email,
@@ -371,8 +386,17 @@ async def refresh_at(
                 }
             }
         else:
-            raise HTTPException(status_code=500, detail="AT刷新失败")
+            debug_logger.log_error(f"[API] AT 刷新失败: token_id={token_id}")
+            
+            error_detail = "AT刷新失败"
+            if config.captcha_method != "personal":
+                error_detail += f"（当前打码模式: {config.captcha_method}，ST自动刷新仅在 personal 模式下可用）"
+            
+            raise HTTPException(status_code=500, detail=error_detail)
+    except HTTPException:
+        raise
     except Exception as e:
+        debug_logger.log_error(f"[API] 刷新AT异常: {str(e)}")
         raise HTTPException(status_code=500, detail=f"刷新AT失败: {str(e)}")
 
 

src/services/browser_captcha_personal.py

@@ -541,6 +541,121 @@ class BrowserCaptchaService:
         debug_logger.log_info("[BrowserCaptcha] 请在打开的浏览器中登录账号。登录完成后，无需关闭浏览器，脚本下次运行时会自动使用此状态。")
         print("请在打开的浏览器中登录账号。登录完成后，无需关闭浏览器，脚本下次运行时会自动使用此状态。")
 
+    # ========== Session Token 刷新 ==========
+
+    async def refresh_session_token(self, project_id: str) -> Optional[str]:
+        """从常驻标签页获取最新的 Session Token
+        
+        复用 reCAPTCHA 常驻标签页，通过刷新页面并从 cookies 中提取
+        __Secure-next-auth.session-token
+        
+        Args:
+            project_id: 项目ID，用于定位常驻标签页
+            
+        Returns:
+            新的 Session Token，如果获取失败返回 None
+        """
+        # 确保浏览器已初始化
+        await self.initialize()
+        
+        start_time = time.time()
+        debug_logger.log_info(f"[BrowserCaptcha] 开始刷新 Session Token (project: {project_id})...")
+        
+        # 尝试获取或创建常驻标签页
+        async with self._resident_lock:
+            resident_info = self._resident_tabs.get(project_id)
+            
+            # 如果该 project_id 没有常驻标签页，则创建
+            if resident_info is None:
+                debug_logger.log_info(f"[BrowserCaptcha] project_id={project_id} 没有常驻标签页，正在创建...")
+                resident_info = await self._create_resident_tab(project_id)
+                if resident_info is None:
+                    debug_logger.log_warning(f"[BrowserCaptcha] 无法为 project_id={project_id} 创建常驻标签页")
+                    return None
+                self._resident_tabs[project_id] = resident_info
+        
+        if not resident_info or not resident_info.tab:
+            debug_logger.log_error(f"[BrowserCaptcha] 无法获取常驻标签页")
+            return None
+        
+        tab = resident_info.tab
+        
+        try:
+            # 刷新页面以获取最新的 cookies
+            debug_logger.log_info(f"[BrowserCaptcha] 刷新常驻标签页以获取最新 cookies...")
+            await tab.reload()
+            
+            # 等待页面加载完成
+            for i in range(30):
+                await asyncio.sleep(1)
+                try:
+                    ready_state = await tab.evaluate("document.readyState")
+                    if ready_state == "complete":
+                        break
+                except Exception:
+                    pass
+            
+            # 额外等待确保 cookies 已设置
+            await asyncio.sleep(2)
+            
+            # 从 cookies 中提取 __Secure-next-auth.session-token
+            # nodriver 可以通过 browser 获取 cookies
+            session_token = None
+            
+            try:
+                # 使用 nodriver 的 cookies API 获取所有 cookies
+                cookies = await self.browser.cookies.get_all()
+                
+                for cookie in cookies:
+                    if cookie.name == "__Secure-next-auth.session-token":
+                        session_token = cookie.value
+                        break
+                        
+            except Exception as e:
+                debug_logger.log_warning(f"[BrowserCaptcha] 通过 cookies API 获取失败: {e}，尝试从 document.cookie 获取...")
+                
+                # 备选方案：通过 JavaScript 获取 (注意：HttpOnly cookies 可能无法通过此方式获取)
+                try:
+                    all_cookies = await tab.evaluate("document.cookie")
+                    if all_cookies:
+                        for part in all_cookies.split(";"):
+                            part = part.strip()
+                            if part.startswith("__Secure-next-auth.session-token="):
+                                session_token = part.split("=", 1)[1]
+                                break
+                except Exception as e2:
+                    debug_logger.log_error(f"[BrowserCaptcha] document.cookie 获取失败: {e2}")
+            
+            duration_ms = (time.time() - start_time) * 1000
+            
+            if session_token:
+                debug_logger.log_info(f"[BrowserCaptcha] ✅ Session Token 获取成功（耗时 {duration_ms:.0f}ms）")
+                return session_token
+            else:
+                debug_logger.log_error(f"[BrowserCaptcha] ❌ 未找到 __Secure-next-auth.session-token cookie")
+                return None
+                
+        except Exception as e:
+            debug_logger.log_error(f"[BrowserCaptcha] 刷新 Session Token 异常: {str(e)}")
+            
+            # 常驻标签页可能已失效，尝试重建
+            async with self._resident_lock:
+                await self._close_resident_tab(project_id)
+                resident_info = await self._create_resident_tab(project_id)
+                if resident_info:
+                    self._resident_tabs[project_id] = resident_info
+                    # 重建后再次尝试获取
+                    try:
+                        cookies = await self.browser.cookies.get_all()
+                        for cookie in cookies:
+                            if cookie.name == "__Secure-next-auth.session-token":
+                                debug_logger.log_info(f"[BrowserCaptcha] ✅ 重建后 Session Token 获取成功")
+                                return cookie.value
+                    except Exception:
+                        pass
+            
+            return None
+
     # ========== 状态查询 ==========
 
     def is_resident_mode_active(self) -> bool:

src/services/token_manager.py

@@ -268,9 +268,13 @@ class TokenManager:
         # AT有效
         return True
 
+
     async def _refresh_at(self, token_id: int) -> bool:
         """内部方法: 刷新AT
 
+        如果 AT 刷新失败（ST 可能过期），会尝试通过浏览器自动刷新 ST，
+        然后重试 AT 刷新。
+
         Returns:
             True if refresh successful, False otherwise
         """
@@ -279,49 +283,132 @@ class TokenManager:
             if not token:
                 return False
 
-            try:
-                debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: 开始刷新AT...")
+            # 第一次尝试刷新 AT
+            result = await self._do_refresh_at(token_id, token.st)
+            if result:
+                return True
 
-                # 使用ST转AT
-                result = await self.flow_client.st_to_at(token.st)
-                new_at = result["access_token"]
-                expires = result.get("expires")
+            # AT 刷新失败，尝试自动更新 ST
+            debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: 第一次 AT 刷新失败，尝试自动更新 ST...")
+            
+            new_st = await self._try_refresh_st(token_id, token)
+            if new_st:
+                # ST 更新成功，重试 AT 刷新
+                debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: ST 已更新，重试 AT 刷新...")
+                result = await self._do_refresh_at(token_id, new_st)
+                if result:
+                    return True
 
-                # 解析过期时间
-                new_at_expires = None
-                if expires:
-                    try:
-                        new_at_expires = datetime.fromisoformat(expires.replace('Z', '+00:00'))
-                    except:
-                        pass
+            # 所有刷新尝试都失败，禁用 Token
+            debug_logger.log_error(f"[AT_REFRESH] Token {token_id}: 所有刷新尝试失败，禁用 Token")
+            await self.disable_token(token_id)
+            return False
 
-                # 更新数据库
-                await self.db.update_token(
-                    token_id,
-                    at=new_at,
-                    at_expires=new_at_expires
-                )
+    async def _do_refresh_at(self, token_id: int, st: str) -> bool:
+        """执行 AT 刷新的核心逻辑
 
-                debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: AT刷新成功")
-                debug_logger.log_info(f"  - 新过期时间: {new_at_expires}")
+        Args:
+            token_id: Token ID
+            st: Session Token
 
-                # 同时刷新credits
+        Returns:
+            True if refresh successful AND AT is valid, False otherwise
+        """
+        try:
+            debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: 开始刷新AT...")
+
+            # 使用ST转AT
+            result = await self.flow_client.st_to_at(st)
+            new_at = result["access_token"]
+            expires = result.get("expires")
+
+            # 解析过期时间
+            new_at_expires = None
+            if expires:
                 try:
-                    credits_result = await self.flow_client.get_credits(new_at)
-                    await self.db.update_token(
-                        token_id,
-                        credits=credits_result.get("credits", 0)
-                    )
+                    new_at_expires = datetime.fromisoformat(expires.replace('Z', '+00:00'))
                 except:
                     pass
 
-                return True
+            # 更新数据库
+            await self.db.update_token(
+                token_id,
+                at=new_at,
+                at_expires=new_at_expires
+            )
 
-            except Exception as e:
-                debug_logger.log_error(f"[AT_REFRESH] Token {token_id}: AT刷新失败 - {str(e)}")
-                # 刷新失败,禁用Token
-                await self.disable_token(token_id)
-                return False
+            debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: AT刷新成功")
+            debug_logger.log_info(f"  - 新过期时间: {new_at_expires}")
+
+            # 验证 AT 有效性：通过 get_credits 测试
+            try:
+                credits_result = await self.flow_client.get_credits(new_at)
+                await self.db.update_token(
+                    token_id,
+                    credits=credits_result.get("credits", 0)
+                )
+                debug_logger.log_info(f"[AT_REFRESH] Token {token_id}: AT 验证成功（余额: {credits_result.get('credits', 0)}）")
+                return True
+            except Exception as verify_err:
+                # AT 验证失败（可能返回 401），说明 ST 已过期
+                error_msg = str(verify_err)
+                if "401" in error_msg or "UNAUTHENTICATED" in error_msg:
+                    debug_logger.log_warning(f"[AT_REFRESH] Token {token_id}: AT 验证失败 (401)，ST 可能已过期")
+                    return False
+                else:
+                    # 其他错误（如网络问题），仍视为成功
+                    debug_logger.log_warning(f"[AT_REFRESH] Token {token_id}: AT 验证时发生非认证错误: {error_msg}")
+                    return True
+
+        except Exception as e:
+            debug_logger.log_error(f"[AT_REFRESH] Token {token_id}: AT刷新失败 - {str(e)}")
+            return False
+
+    async def _try_refresh_st(self, token_id: int, token) -> Optional[str]:
+        """尝试通过浏览器刷新 Session Token
+
+        使用常驻 tab 获取新的 __Secure-next-auth.session-token
+
+        Args:
+            token_id: Token ID
+            token: Token 对象
+
+        Returns:
+            新的 ST 字符串，如果失败返回 None
+        """
+        try:
+            from ..core.config import config
+
+            # 仅在 personal 模式下支持 ST 自动刷新
+            if config.captcha_method != "personal":
+                debug_logger.log_info(f"[ST_REFRESH] 非 personal 模式，跳过 ST 自动刷新")
+                return None
+
+            if not token.current_project_id:
+                debug_logger.log_warning(f"[ST_REFRESH] Token {token_id} 没有 project_id，无法刷新 ST")
+                return None
+
+            debug_logger.log_info(f"[ST_REFRESH] Token {token_id}: 尝试通过浏览器刷新 ST...")
+
+            from .browser_captcha_personal import BrowserCaptchaService
+            service = await BrowserCaptchaService.get_instance(self.db)
+
+            new_st = await service.refresh_session_token(token.current_project_id)
+            if new_st and new_st != token.st:
+                # 更新数据库中的 ST
+                await self.db.update_token(token_id, st=new_st)
+                debug_logger.log_info(f"[ST_REFRESH] Token {token_id}: ST 已自动更新")
+                return new_st
+            elif new_st == token.st:
+                debug_logger.log_warning(f"[ST_REFRESH] Token {token_id}: 获取到的 ST 与原 ST 相同，可能登录已失效")
+                return None
+            else:
+                debug_logger.log_warning(f"[ST_REFRESH] Token {token_id}: 无法获取新 ST")
+                return None
+
+        except Exception as e:
+            debug_logger.log_error(f"[ST_REFRESH] Token {token_id}: 刷新 ST 失败 - {str(e)}")
+            return None
 
     async def ensure_project_exists(self, token_id: int) -> str:
         """确保Token有可用的Project