尝试通过浏览器刷新 Session Token 使用常驻 tab 获取新的 __Secure-next-auth.session-token Args: token_id: Token ID token: Token 对象 Returns: 新的 ST 字符串，如果失败返回 None
(self, token_id: int, token)
| 732 | return None |
| 733 | |
| 734 | async def _try_refresh_st(self, token_id: int, token) -> Optional[str]: |
| 735 | """尝试通过浏览器刷新 Session Token |
| 736 | |
| 737 | 使用常驻 tab 获取新的 __Secure-next-auth.session-token |
| 738 | |
| 739 | Args: |
| 740 | token_id: Token ID |
| 741 | token: Token 对象 |
| 742 | |
| 743 | Returns: |
| 744 | 新的 ST 字符串，如果失败返回 None |
| 745 | """ |
| 746 | try: |
| 747 | from ..core.config import config |
| 748 | |
| 749 | protocol_st = await self._try_protocol_refresh_st(token_id, token) |
| 750 | if protocol_st: |
| 751 | return protocol_st |
| 752 | |
| 753 | # 仅在 personal 模式下支持 ST 自动刷新 |
| 754 | if config.captcha_method != "personal": |
| 755 | debug_logger.log_info(f"[ST_REFRESH] 非 personal 模式，跳过 ST 自动刷新") |
| 756 | return None |
| 757 | |
| 758 | if not token.current_project_id: |
| 759 | debug_logger.log_warning(f"[ST_REFRESH] Token {token_id} 没有 project_id，无法刷新 ST") |
| 760 | return None |
| 761 | |
| 762 | debug_logger.log_info(f"[ST_REFRESH] Token {token_id}: 尝试通过浏览器刷新 ST...") |
| 763 | |
| 764 | from .browser_captcha_personal import BrowserCaptchaService |
| 765 | service = await BrowserCaptchaService.get_instance(self.db) |
| 766 | |
| 767 | refresh_timeout_seconds = 45.0 |
| 768 | try: |
| 769 | new_st = await asyncio.wait_for( |
| 770 | service.refresh_session_token(token.current_project_id), |
| 771 | timeout=refresh_timeout_seconds, |
| 772 | ) |
| 773 | except asyncio.TimeoutError: |
| 774 | debug_logger.log_error( |
| 775 | f"[ST_REFRESH] Token {token_id}: 刷新 ST 超时 ({refresh_timeout_seconds:.0f}s)" |
| 776 | ) |
| 777 | record_token_refresh("st", "failure") |
| 778 | return None |
| 779 | if new_st and new_st != token.st: |
| 780 | # 更新数据库中的 ST |
| 781 | await self.db.update_token(token_id, st=new_st) |
| 782 | debug_logger.log_info(f"[ST_REFRESH] Token {token_id}: ST 已自动更新") |
| 783 | record_token_refresh("st", "success") |
| 784 | return new_st |
| 785 | elif new_st == token.st: |
| 786 | debug_logger.log_warning(f"[ST_REFRESH] Token {token_id}: 获取到的 ST 与原 ST 相同，可能登录已失效") |
| 787 | record_token_refresh("st", "failure") |
| 788 | return None |
| 789 | else: |
| 790 | debug_logger.log_warning(f"[ST_REFRESH] Token {token_id}: 无法获取新 ST") |
| 791 | record_token_refresh("st", "failure") |
no test coverage detected