feat: X(Twitter) resolve 透出全部 mp4 variants（码率升序），供下游 ASR 自选低码率流

README.md

@@ -197,6 +197,7 @@ audio 模式不代表 MP3 或其他容器格式，服务不做转码；请使用
     "height": 1920,
     "duration": 30,
     "quality": "1080p",
+    "variants": null,
     "view_count": 100000,
     "like_count": 5000,
     "comment_count": 200,
@@ -229,6 +230,7 @@ audio 模式不代表 MP3 或其他容器格式，服务不做转码；请使用
 | `data.height` | int | 视频高度（像素） |
 | `data.duration` | int \| null | 视频时长（秒） |
 | `data.quality` | string \| null | 选中流的实际清晰度档位；Cobalt 链没有逐档元数据时为 `"unknown"` |
+| `data.variants` | array \| null | 仅 X（Twitter）video 模式透出全部 mp4 variants，按码率升序（`bitrate` 单位为 bps），供下游自行选择低码率流；其他平台和 audio 模式为 `null` |
 | `data.view_count` | int \| null | 播放量。视频号恒为 `null`（TikHub 的 `read_count` 恒为 0），不是偶尔缺失 |
 | `data.like_count` | int \| null | 点赞数 |
 | `data.comment_count` | int \| null | 评论数 |

app/api/resolve.py

@@ -81,6 +81,16 @@ def reject_quality_for_audio(self):
         return self
 
 
+class VideoVariant(BaseModel):
+    """A playable MP4 variant exposed for downstream stream selection."""
+
+    url: str
+    bitrate: Optional[int] = None
+    width: int = 0
+    height: int = 0
+    quality: Optional[str] = None
+
+
 class VideoInfoResponse(BaseModel):
     """Video info in the response."""
 
@@ -98,6 +108,7 @@ class VideoInfoResponse(BaseModel):
     height: int
     duration: Optional[int] = None
     quality: Optional[str] = None
+    variants: Optional[list[VideoVariant]] = None
     view_count: Optional[int] = None
     like_count: Optional[int] = None
     comment_count: Optional[int] = None
@@ -367,6 +378,20 @@ def _build_response(
         height=video_info.height,
         duration=video_info.duration,
         quality=video_info.quality,
+        variants=(
+            [
+                VideoVariant(
+                    url=_absolutize_video_url(variant["url"], public_origin),
+                    bitrate=variant.get("bitrate"),
+                    width=variant.get("width", 0),
+                    height=variant.get("height", 0),
+                    quality=variant.get("quality"),
+                )
+                for variant in video_info.variants
+            ]
+            if video_info.variants is not None
+            else None
+        ),
         view_count=video_info.view_count,
         like_count=video_info.like_count,
         comment_count=video_info.comment_count,

app/services/cache.py

@@ -87,6 +87,7 @@ def get_cached_video(
                 duration=cached_data.get("duration"),
                 quality=cached_data.get("quality"),
                 media_type=cached_data.get("media_type", "video"),
+                variants=cached_data.get("variants"),
                 view_count=cached_data.get("view_count"),
                 like_count=cached_data.get("like_count"),
                 comment_count=cached_data.get("comment_count"),

app/services/platforms/base.py

@@ -31,6 +31,7 @@ class VideoInfo:
     duration: Optional[int] = None
     quality: Optional[str] = None
     media_type: str = "video"
+    variants: Optional[list[Dict[str, Any]]] = None
 
     # 统计信息
     view_count: Optional[int] = None
@@ -64,6 +65,7 @@ def to_dict(self) -> Dict[str, Any]:
             'duration': self.duration,
             'quality': self.quality,
             'media_type': self.media_type,
+            'variants': self.variants,
             'view_count': self.view_count,
             'like_count': self.like_count,
             'comment_count': self.comment_count,

app/services/platforms/twitter.py

@@ -49,6 +49,24 @@ def _parse_response(
         if not candidates:
             return None
 
+        variants = [
+            {
+                "url": str(candidate[0]["url"]),
+                "bitrate": candidate[5] if candidate[5] > 0 else None,
+                "width": candidate[2],
+                "height": candidate[3],
+                "quality": f"{candidate[3]}p" if candidate[3] > 0 else None,
+            }
+            for candidate in sorted(
+                candidates,
+                key=lambda candidate: (
+                    candidate[5] <= 0,
+                    candidate[5] if candidate[5] > 0 else 0,
+                    candidate[4],
+                ),
+            )
+        ]
+
         if quality is None:
             selected = self._select_default_variant(candidates)
         else:
@@ -97,6 +115,7 @@ def _parse_response(
             height=height,
             duration=duration,
             quality=f"{height}p",
+            variants=variants,
             view_count=self._parse_count(data.get("views")),
             like_count=self._parse_count(data.get("likes")),
             comment_count=self._parse_count(data.get("replies")),

docs/sessions/resolve-intent/progress/twitter-variants-progress.md

@@ -0,0 +1,41 @@
+## 2026-09-14 — VideoInfo contract
+
+- 当前阶段：implementing — VideoInfo + to_dict
+- 本段结论：为统一视频信息增加可选 `variants` 字段，并让缓存序列化带出该键；既有构造方不受影响。
+- 关键决策与已否决方案：沿用任务卡的附加可选字段契约，不增加开关或兼容分支。
+- 下一步唯一动作：在 Twitter 解析层从既有 candidates 构造全部 mp4 variants。
+
+## 2026-09-14 — Twitter parsing
+
+- 当前阶段：implementing — Twitter 解析层
+- 本段结论：Twitter 解析现在把 candidates 中的全部 mp4 variant 映射为 url、bitrate、width、height、quality，并按码率升序输出；HLS 仍不进入 candidates，主选流逻辑未变。
+- 关键决策与已否决方案：无
+- 下一步唯一动作：把 variants 接入缓存命中重建路径。
+
+## 2026-09-14 — Cache round-trip
+
+- 当前阶段：implementing — cache 回环
+- 本段结论：缓存重建从序列化数据透传 `variants`，旧缓存行没有该键时自然得到 `None`；新缓存数据可完整回环。
+- 关键决策与已否决方案：沿用任务卡约定，不添加旧行回填或额外兼容分支。
+- 下一步唯一动作：增加 API 响应模型和 `_build_response` 的 variants 映射。
+
+## 2026-09-14 — API response contract
+
+- 当前阶段：implementing — API 模型与映射
+- 本段结论：新增 `VideoVariant` 和可选 `VideoInfoResponse.variants`，响应映射逐条绝对化 URL；`VideoInfo.variants is None` 时响应字段保持 `null`。
+- 关键决策与已否决方案：variants 只作为常驻附加字段，不增加请求参数或平台分支。
+- 下一步唯一动作：补解析层与 API/缓存回环测试，并先执行未改代码红验。
+
+## 2026-09-14 — Tests
+
+- 当前阶段：implementing — 测试
+- 本段结论：新增解析层完整 mp4、排序/非法码率测试，以及 API force-refresh 与缓存命中逐一相等的 X variants 断言；同时锁定非 Twitter 响应为 `null`。
+- 关键决策与已否决方案：无
+- 下一步唯一动作：完成 README 字段表与示例，并执行红验和全量测试。
+
+## 2026-09-14 — Acceptance
+
+- 当前阶段：completed — 交付文档、红验与全量测试
+- 本段结论：README 的成功响应示例补充 `data.variants: null`，字段表说明 X video 的 mp4 variants、bps 单位、码率升序和下游自选低码率语义；两处红验均以断言失败验证约束有效，全量测试 413 passed。
+- 关键决策与已否决方案：无
+- 下一步唯一动作：写入执行器报告并交给主脑验收。

tests/test_resolve_quality.py

@@ -341,6 +341,7 @@ async def tikhub_fetch(*args, **kwargs):
     assert response_720.json()["data"]["video_url"].endswith("youtube-720-high.mp4")
     assert response_1080.status_code == 200
     assert response_1080.json()["data"]["video_url"].endswith("youtube-1080.mp4")
+    assert response_720.json()["data"]["variants"] is None
     assert cached_720.json()["data"]["video_url"].endswith("youtube-720-high.mp4")
     assert calls == ["720p", "1080p"]
     assert {
@@ -349,6 +350,62 @@ async def tikhub_fetch(*args, **kwargs):
     } == {"720p", "1080p"}
 
 
+def test_api_twitter_variants_survive_cache_round_trip(
+    authed_client, db, monkeypatch
+):
+    monkeypatch.setattr(resolve_mod.url_parser, "is_short_url", lambda _url: False)
+    monkeypatch.setattr(
+        resolve_mod.url_parser,
+        "parse_url",
+        lambda _url: ("twitter", "twitter-variants"),
+    )
+    resolver = VideoResolver()
+    calls = []
+
+    async def tikhub_fetch(*args, **kwargs):
+        calls.append(kwargs)
+        return _load_quality("twitter_multi")
+
+    monkeypatch.setattr(resolver.tikhub_provider, "fetch_video_info", tikhub_fetch)
+    monkeypatch.setattr(resolve_mod, "get_video_resolver", lambda: resolver)
+
+    first = authed_client.post(
+        "/api/resolve",
+        json={
+            "url": "https://x.com/0xCodez/status/twitter-variants",
+            "translate": False,
+            "force_refresh": True,
+        },
+    )
+    cached = authed_client.post(
+        "/api/resolve",
+        json={
+            "url": "https://x.com/0xCodez/status/twitter-variants",
+            "translate": False,
+        },
+    )
+
+    assert first.status_code == 200
+    assert cached.status_code == 200
+    first_variants = first.json()["data"]["variants"]
+    cached_payload = db.query(VideoCache).filter(
+        VideoCache.video_id == "twitter-variants"
+    ).one().video_data
+    expected_keys = ("url", "bitrate", "width", "height", "quality")
+    expected_values = [
+        ("https://video.twimg.com/fake/640x360/twitter-360.mp4", 500000, 640, 360, "360p"),
+        ("https://video.twimg.com/fake/1280x720/twitter-720-low.mp4", 1000000, 1280, 720, "720p"),
+        ("https://video.twimg.com/fake/1280x720/twitter-720-high.mp4", 2000000, 1280, 720, "720p"),
+        ("https://video.twimg.com/fake/1920x1080/twitter-1080.mp4", 3000000, 1920, 1080, "1080p"),
+        ("https://video.twimg.com/fake/2560x1440/twitter-1440.mp4", 4000000, 2560, 1440, "1440p"),
+        ("https://video.twimg.com/fake/3840x2160/twitter-2160.mp4", 5000000, 3840, 2160, "2160p"),
+    ]
+    assert first_variants == [dict(zip(expected_keys, values)) for values in expected_values]
+    assert cached_payload["variants"] == first_variants
+    assert cached.json()["data"]["variants"] == first_variants
+    assert len(calls) == 1
+
+
 def test_douyin_fixture_quality_cap_selects_available_lower_stream():
     raw = _load_platform("douyin", "app_v3")
     default_info = DouyinService("key", "base")._parse_response(raw)

tests/test_tikhub_provider_twitter.py

@@ -30,6 +30,81 @@ def test_parser_picks_highest_mp4():
     assert info.height == 1080
 
 
+def test_parser_exposes_all_mp4_variants_sorted_by_bitrate():
+    info = TwitterService("k", "b")._parse_response(load("tweet_video"))
+
+    assert info is not None
+    assert info.variants == [
+        {
+            "url": "https://video.twimg.com/amplify_video/fake/avc1/480x270/twitter_270.mp4",
+            "bitrate": 256000,
+            "width": 480,
+            "height": 270,
+            "quality": "270p",
+        },
+        {
+            "url": "https://video.twimg.com/amplify_video/fake/avc1/640x360/twitter_360.mp4",
+            "bitrate": 832000,
+            "width": 640,
+            "height": 360,
+            "quality": "360p",
+        },
+        {
+            "url": "https://video.twimg.com/amplify_video/fake/avc1/1280x720/twitter_720.mp4",
+            "bitrate": 2176000,
+            "width": 1280,
+            "height": 720,
+            "quality": "720p",
+        },
+        {
+            "url": "https://video.twimg.com/amplify_video/fake/avc1/1920x1080/twitter_1080.mp4",
+            "bitrate": 10368000,
+            "width": 1920,
+            "height": 1080,
+            "quality": "1080p",
+        },
+    ]
+
+
+def test_parser_sorts_invalid_bitrates_last_and_ties_by_short_side():
+    payload = load("tweet_video")
+    payload["data"]["media"]["video"][0]["variants"] = [
+        {
+            "content_type": "video/mp4",
+            "bitrate": 0,
+            "url": "https://video.twimg.com/fake/1920x1080/zero.mp4",
+        },
+        {
+            "content_type": "video/mp4",
+            "bitrate": 1000,
+            "url": "https://video.twimg.com/fake/1280x720/high.mp4",
+        },
+        {
+            "content_type": "video/mp4",
+            "bitrate": 1000,
+            "url": "https://video.twimg.com/fake/640x360/low.mp4",
+        },
+        {
+            "content_type": "video/mp4",
+            "bitrate": "unknown",
+            "url": "https://video.twimg.com/fake/480x270/unknown.mp4",
+        },
+    ]
+
+    info = TwitterService("k", "b")._parse_response(payload)
+
+    assert info is not None
+    assert [
+        (variant["bitrate"], variant["width"], variant["height"])
+        for variant in info.variants
+    ] == [
+        (1000, 640, 360),
+        (1000, 1280, 720),
+        (None, 480, 270),
+        (None, 1920, 1080),
+    ]
+
+
 def test_parser_picks_highest_mp4_across_videos():
     info = TwitterService("k", "b")._parse_response(load("tweet_multi_video"))
     assert info is not None
@@ -219,6 +294,13 @@ def test_parser_malformed_bitrate_falls_soft_and_prefers_valid():
     info = TwitterService("k", "b")._parse_response(payload)
     assert info is not None
     assert info.video_url == "https://video.twimg.com/fake_unknown_bitrate.mp4"
+    assert info.variants == [{
+        "url": "https://video.twimg.com/fake_unknown_bitrate.mp4",
+        "bitrate": None,
+        "width": 1920,
+        "height": 1080,
+        "quality": "1080p",
+    }]
 
     payload["data"]["media"]["video"][0]["variants"].append({
         "content_type": "video/mp4",