Function initialize_token_pool_from_db

app/utils/token_pool.py:581–635

从数据库初始化全局 Token 池 Args: provider: 提供商名称（当前仅使用 zai） failure_threshold: 失败阈值 recovery_timeout: 恢复超时时间（秒） Returns: TokenPool 实例（即使没有 Token 也会创建空池）

(
    provider: str = "zai",
    failure_threshold: int = 3,
    recovery_timeout: int = 1800
)

async def initialize_token_pool_from_db(
    provider: str = "zai",
    failure_threshold: int = 3,
    recovery_timeout: int = 1800
) -> Optional[TokenPool]:
    """
    从数据库初始化全局 Token 池

    Args:
        provider: 提供商名称（当前仅使用 zai）
        failure_threshold: 失败阈值
        recovery_timeout: 恢复超时时间（秒）

    Returns:
        TokenPool 实例（即使没有 Token 也会创建空池）
    """
    global _token_pool

    from app.services.token_dao import get_token_dao

    dao = get_token_dao()

    # 从数据库加载 Token（只加载启用的认证用户 Token）
    token_records = await dao.get_tokens_by_provider(provider, enabled_only=True)

    # 转换为 TokenPool 所需格式
    tokens = []
    if token_records:
        tokens = [
            (record["id"], record["token"], record.get("token_type", "unknown"))
            for record in token_records
        ]

        # 过滤掉 guest token（不应该在数据库中，但防御性检查）
        user_tokens = [
            (tid, tval, ttype) for tid, tval, ttype in tokens
            if ttype != "guest"
        ]

        if len(user_tokens) < len(tokens):
            guest_count = len(tokens) - len(user_tokens)
            logger.warning(f"⚠️ 过滤了 {guest_count} 个匿名用户 Token")

        tokens = user_tokens

    # 始终创建 Token 池实例（即使为空）
    with _pool_lock:
        _token_pool = TokenPool(tokens, failure_threshold, recovery_timeout)

        if not tokens:
            logger.warning(f"⚠️ {provider} 没有有效的认证用户 Token，已创建空 Token 池")
        else:
            logger.info(f"🔧 从数据库初始化 Token 池（{provider}），共 {len(tokens)} 个 Token")

        return _token_pool


async def sync_token_stats_to_db():