refactor: remove deprecated smart token routing configuration and related test

PR_DESCRIPTION.md

@@ -1,106 +0,0 @@
-# 功能实现总结：智能 Token 路由（视频生成）
-
-## 🎯 功能概述
-
-实现了视频生成的智能 Token 路由功能，根据视频需求自动选择最合适的 Token 池（ssoSuper vs ssoBasic）。
-
-## 🔧 实现内容
-
-### 1. Token 管理器增强 (`app/services/token/manager.py`)
-
-新增两个方法：
-
-#### `get_token_info(pool_name)` 
-- 返回完整的 TokenInfo 对象（而非仅 token 字符串）
-- 便于后续获取 Token 元数据
-
-#### `get_token_for_video(resolution, video_length)` ⭐ 核心功能
-智能路由逻辑：
-- **720p 或 >6s 视频** → 优先使用 `ssoSuper` Token
-- **480p 且 ≤6s 视频** → 使用 `ssoBasic` Token
-- **Super Token 不可用** → 自动回退到 Basic Token（带警告日志）
-
-### 2. 视频服务修改 (`app/services/grok/services/media.py`)
-
-- 使用新的 `get_token_for_video()` 方法替代原有的循环查找
-- 自动根据视频配置选择最佳 Token
-- 保持向后兼容
-
-### 3. 配置选项 (`config.defaults.toml`)
-
-新增 `[video]` 配置段：
-```toml
-[video]
-# 是否启用智能 Token 路由
-smart_token_routing = true
-
-# 当需要 Super Token 但不可用时是否允许回退
-allow_fallback_to_basic = true
-
-# 回退时的警告日志
-log_fallback_warning = true
-```
-
-## 📊 修改统计
-
-- **新增代码**: ~90 行
-- **修改文件**: 3 个
-- **测试文件**: 1 个 (`test_video_token_routing.py`)
-
-## ✅ 解决的问题
-
-1. **非 SSO Super Token 被静默降级**: 现在系统会主动选择正确的 Token 类型
-2. **720p 视频生成失败**: 自动使用 Super Token 避免降级
-3. **Token 选择不透明**: 详细的日志记录便于调试
-
-## 🧪 测试场景
-
-测试覆盖以下场景：
-1. ✅ 480p, 6s → 使用 Basic Token
-2. ✅ 720p, 6s → 使用 Super Token
-3. ✅ 480p, 10s → 使用 Super Token（>6s）
-4. ✅ 720p, Super 池空 → 回退到 Basic Token
-
-## 📝 使用说明
-
-无需修改现有代码，系统会自动根据视频请求参数选择 Token：
-
-```python
-# 请求 720p 视频
-{
-    "model": "grok-video",
-    "video_config": {
-        "resolution_name": "720p",
-        "video_length": 6
-    }
-}
-# → 自动使用 ssoSuper Token
-
-# 请求 480p 视频
-{
-    "model": "grok-video", 
-    "video_config": {
-        "resolution_name": "480p",
-        "video_length": 6
-    }
-}
-# → 使用 ssoBasic Token
-```
-
-## 🔍 日志示例
-
-```
-# 正常路由
-Video token routing: resolution=720p, length=6s -> pool=ssoSuper (token=eyJ0eXAiOi...)
-
-# 回退情况
-Video token routing: ssoSuper pool has no available token for resolution=720p, length=6s. 
-Falling back to ssoBasic pool.
-Video token routing: fallback from ssoSuper -> ssoBasic (token=eyJ0eXAiOi...)
-```
-
-## 🎉 预期效果
-
-- 720p 视频请求将正确使用 SSO Super Token
-- 避免因 Token 类型不匹配导致的静默降级
-- 当 Super Token 不足时自动回退，保证服务可用性

app/services/grok/services/media.py

@@ -17,7 +17,7 @@ from app.core.exceptions import (
     ErrorType,
 )
 from app.services.grok.models.model import ModelService
-from app.services.token import get_token_manager, EffortType, TokenInfo
+from app.services.token import get_token_manager, EffortType
 from app.services.grok.processors import VideoStreamProcessor, VideoCollectProcessor
 from app.services.grok.utils.headers import apply_statsig, build_sso_cookie
 from app.services.grok.utils.stream import wrap_stream_with_usage
@@ -305,9 +305,12 @@ class VideoService:
         token_mgr = await get_token_manager()
         await token_mgr.reload_if_stale()
 
-        # 使用智能路由选择 token（根据视频需求）
+        # 使用智能路由选择 token（根据视频需求与候选池）
+        pool_candidates = ModelService.pool_candidates_for_model(model)
         token_info = token_mgr.get_token_for_video(
-            resolution=resolution, video_length=video_length
+            resolution=resolution,
+            video_length=video_length,
+            pool_candidates=pool_candidates,
         )
 
         if not token_info:

app/services/token/manager.py

@@ -233,19 +233,23 @@ class TokenManager:
         return token_info
 
     def get_token_for_video(
-        self, resolution: str = "480p", video_length: int = 6
+        self,
+        resolution: str = "480p",
+        video_length: int = 6,
+        pool_candidates: Optional[List[str]] = None,
     ) -> Optional["TokenInfo"]:
         """
         根据视频需求智能选择 Token 池
 
         路由策略:
         - 如果 resolution 是 "720p" 或 video_length > 6: 优先使用 "ssoSuper" 池
-        - 如果 "ssoSuper" 池没有可用 token: 记录警告并回退到 "ssoBasic" 池
-        - 如果 resolution 是 "480p" 且 video_length <= 6: 使用 "ssoBasic" 池
+        - 否则优先使用 "ssoBasic" 池
+        - 当提供 pool_candidates 时，按候选池顺序回退
 
         Args:
             resolution: 视频分辨率 ("480p" 或 "720p")
             video_length: 视频时长(秒)
+            pool_candidates: 候选 Token 池（按优先级）
 
         Returns:
             TokenInfo 对象或 None（无可用 token）
@@ -253,32 +257,39 @@ class TokenManager:
         # 确定首选池
         requires_super = resolution == "720p" or video_length > 6
         primary_pool = SUPER_POOL_NAME if requires_super else BASIC_POOL_NAME
-        fallback_pool = BASIC_POOL_NAME if requires_super else SUPER_POOL_NAME
 
-        # 优先从主池获取 token
-        primary_token = self.get_token_info(primary_pool)
-        if primary_token:
-            logger.info(
-                f"Video token routing: resolution={resolution}, length={video_length}s -> "
-                f"pool={primary_pool} (token={primary_token.token[:10]}...)"
-            )
-            return primary_token
+        if pool_candidates:
+            ordered_pools = list(pool_candidates)
+            if primary_pool in ordered_pools:
+                ordered_pools.remove(primary_pool)
+                ordered_pools.insert(0, primary_pool)
+        else:
+            fallback_pool = BASIC_POOL_NAME if requires_super else SUPER_POOL_NAME
+            ordered_pools = [primary_pool, fallback_pool]
 
-        # 主池无可用 token，尝试回退池
-        if requires_super:
-            logger.warning(
-                f"Video token routing: ssoSuper pool has no available token for "
-                f"resolution={resolution}, length={video_length}s. "
-                f"Falling back to ssoBasic pool."
-            )
+        for idx, pool_name in enumerate(ordered_pools):
+            token_info = self.get_token_info(pool_name)
+            if token_info:
+                if idx == 0:
+                    logger.info(
+                        f"Video token routing: resolution={resolution}, length={video_length}s -> "
+                        f"pool={pool_name} (token={token_info.token[:10]}...)"
+                    )
+                else:
+                    logger.info(
+                        f"Video token routing: fallback from {ordered_pools[0]} -> {pool_name} "
+                        f"(token={token_info.token[:10]}...)"
+                    )
+                return token_info
 
-        fallback_token = self.get_token_info(fallback_pool)
-        if fallback_token:
-            logger.info(
-                f"Video token routing: fallback from {primary_pool} -> {fallback_pool} "
-                f"(token={fallback_token.token[:10]}...)"
-            )
-            return fallback_token
+            if idx == 0 and requires_super and pool_name == primary_pool:
+                next_pool = ordered_pools[1] if len(ordered_pools) > 1 else None
+                if next_pool:
+                    logger.warning(
+                        f"Video token routing: {primary_pool} pool has no available token for "
+                        f"resolution={resolution}, length={video_length}s. "
+                        f"Falling back to {next_pool} pool."
+                    )
 
         # 两个池都没有可用 token
         logger.warning(

config.defaults.toml

@@ -138,16 +138,3 @@ nsfw_batch_size = 50
 # NSFW 单次最大数量
 nsfw_max_tokens = 1000
 
-
-# ==================== 视频 Token 路由 ====================
-[video]
-# 是否启用智能 Token 路由（根据视频需求自动选择 Token 池）
-# true:  720p 或 >6s 视频自动使用 ssoSuper Token
-# false: 所有视频使用默认 Token 池
-smart_token_routing = true
-
-# 当需要 Super Token 但不可用时是否允许回退到 Basic Token
-allow_fallback_to_basic = true
-
-# 回退时的警告日志
-log_fallback_warning = true

test_video_token_routing.py

@@ -1,75 +0,0 @@
-"""
-测试视频 Token 智能路由功能
-"""
-import sys
-sys.path.insert(0, '.')
-
-from app.services.token.models import TokenInfo, TokenStatus, BASIC__DEFAULT_QUOTA, SUPER_DEFAULT_QUOTA
-from app.services.token.pool import TokenPool
-from app.services.token.manager import TokenManager
-
-
-def test_token_routing():
-    """测试 Token 路由逻辑"""
-    print("🧪 测试视频 Token 智能路由功能\n")
-    
-    # 创建模拟 Token 池
-    basic_pool = TokenPool("ssoBasic")
-    super_pool = TokenPool("ssoSuper")
-    
-    # 添加测试 Token
-    basic_token = TokenInfo(
-        token="basic_test_token_123",
-        status=TokenStatus.ACTIVE,
-        quota=BASIC__DEFAULT_QUOTA
-    )
-    super_token = TokenInfo(
-        token="super_test_token_456",
-        status=TokenStatus.ACTIVE,
-        quota=SUPER_DEFAULT_QUOTA
-    )
-    
-    basic_pool.add(basic_token)
-    super_pool.add(super_token)
-    
-    # 创建 TokenManager 并设置池
-    manager = TokenManager.__new__(TokenManager)
-    manager.pools = {
-        "ssoBasic": basic_pool,
-        "ssoSuper": super_pool
-    }
-    
-    # 测试场景 1: 480p, 6s - 应该使用 Basic
-    print("场景 1: 480p, 6s")
-    result = manager.get_token_for_video("480p", 6)
-    assert result is not None, "应该获取到 Token"
-    assert result.token == "basic_test_token_123", f"应该使用 Basic Token, 但得到: {result.token}"
-    print("  ✅ 正确使用 Basic Token\n")
-    
-    # 测试场景 2: 720p, 6s - 应该使用 Super
-    print("场景 2: 720p, 6s")
-    result = manager.get_token_for_video("720p", 6)
-    assert result is not None, "应该获取到 Token"
-    assert result.token == "super_test_token_456", f"应该使用 Super Token, 但得到: {result.token}"
-    print("  ✅ 正确使用 Super Token\n")
-    
-    # 测试场景 3: 480p, 10s (>6s) - 应该使用 Super
-    print("场景 3: 480p, 10s (>6s)")
-    result = manager.get_token_for_video("480p", 10)
-    assert result is not None, "应该获取到 Token"
-    assert result.token == "super_test_token_456", f"应该使用 Super Token, 但得到: {result.token}"
-    print("  ✅ 正确使用 Super Token\n")
-    
-    # 测试场景 4: 720p, 但 Super 池为空 - 应该回退到 Basic
-    print("场景 4: 720p, 但 Super 池为空（回退测试）")
-    super_pool.remove(super_token.token)  # 清空 Super 池
-    result = manager.get_token_for_video("720p", 6)
-    assert result is not None, "应该获取到 Token（回退）"
-    assert result.token == "basic_test_token_123", f"应该回退到 Basic Token, 但得到: {result.token}"
-    print("  ✅ 正确回退到 Basic Token\n")
-    
-    print("🎉 所有测试通过！")
-
-
-if __name__ == "__main__":
-    test_token_routing()