获取下一个可用的认证用户 Token(轮询算法) Returns: 可用的 Token 字符串,如果没有可用 Token 则返回 None
(self, exclude_tokens: Optional[Set[str]] = None)
| 230 | logger.warning("⚠️ Token 池为空,将依赖匿名模式") |
| 231 | |
| 232 | def get_next_token(self, exclude_tokens: Optional[Set[str]] = None) -> Optional[str]: |
| 233 | """ |
| 234 | 获取下一个可用的认证用户 Token(轮询算法) |
| 235 | |
| 236 | Returns: |
| 237 | 可用的 Token 字符串,如果没有可用 Token 则返回 None |
| 238 | """ |
| 239 | with self._lock: |
| 240 | if not self.token_statuses: |
| 241 | return None |
| 242 | |
| 243 | excluded = exclude_tokens or set() |
| 244 | |
| 245 | available_tokens = self._get_available_user_tokens() |
| 246 | if excluded: |
| 247 | available_tokens = [ |
| 248 | token for token in available_tokens if token not in excluded |
| 249 | ] |
| 250 | if not available_tokens: |
| 251 | # 尝试恢复过期的失败 Token |
| 252 | self._try_recover_failed_tokens() |
| 253 | available_tokens = self._get_available_user_tokens() |
| 254 | if excluded: |
| 255 | available_tokens = [ |
| 256 | token for token in available_tokens if token not in excluded |
| 257 | ] |
| 258 | |
| 259 | if not available_tokens: |
| 260 | logger.warning("⚠️ 没有可用的认证用户 Token") |
| 261 | return None |
| 262 | |
| 263 | # 轮询选择 |
| 264 | token = available_tokens[self._current_index % len(available_tokens)] |
| 265 | self._current_index = (self._current_index + 1) % len(available_tokens) |
| 266 | |
| 267 | return token |
| 268 | |
| 269 | def _get_available_user_tokens(self) -> List[str]: |
| 270 | """ |
no test coverage detected