MCPcopy Create free account
hub / github.com/ZyphrZero/z.ai2api_python / get_next_token

Method get_next_token

app/utils/token_pool.py:232–267  ·  view source on GitHub ↗

获取下一个可用的认证用户 Token(轮询算法) Returns: 可用的 Token 字符串,如果没有可用 Token 则返回 None

(self, exclude_tokens: Optional[Set[str]] = None)

Source from the content-addressed store, hash-verified

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 """

Callers 1

get_auth_infoMethod · 0.45

Calls 2

Tested by

no test coverage detected