fix(douyin): 修复直链选择优先级，CDN直链优先，过滤图文BGM MP3，支持图文视频仅返回文案

packages/douyin_parser/__init__.py

@@ -4,7 +4,6 @@
 接口直接返回 aweme_list 包含视频元信息和 play_addr 无水印直链。
 此接口不需要任何签名算法、不需要 cookies、不需要特殊 TLS 指纹，稳定性 >95%。
 """
-
 from __future__ import annotations
 
 import logging
@@ -29,7 +28,34 @@ _APP_UAS = [
 _AID = "1128"  # Douyin APP aid
 _FEED_URL = "https://aweme.snssdk.com/aweme/v1/feed/"
 
-_VIDEO_HOST_HINTS = ("douyinvod.com", "bytecdn.com", "365yg.com", "amemv.com", "byteimg.com", "bytedance.com")
+# 视频直链 CDN 域名（优先级从高到低）。api-play.amemv.com 是带签名的临时接口，签名过期或
+# 缺失必要 header 会 403/404，因此只作为降级兜底；优先用稳定 CDN。
+_CDN_HOSTS = ("douyinvod.com", "bytecdn.com", "365yg.com", "byteimg.com", "bytedance.com")
+_FALLBACK_HOSTS = ("api-play.amemv.com", "api-hl.amemv.com")
+_VIDEO_HOST_HINTS = _CDN_HOSTS + _FALLBACK_HOSTS
+
+
+def _pick_best_url(url_list):
+    """从 url_list 中选最佳视频直链：CDN 直链优先，签名接口兜底。"""
+    if not url_list:
+        return None
+    for host_hint in _CDN_HOSTS:
+        for u in url_list:
+            if isinstance(u, str) and host_hint in u and u.endswith(".mp4") or (isinstance(u, str) and host_hint in u and "/mp4/" in u):
+                return u
+    for host_hint in _CDN_HOSTS:
+        for u in url_list:
+            if isinstance(u, str) and host_hint in u:
+                return u
+    for host_hint in _FALLBACK_HOSTS:
+        for u in url_list:
+            if isinstance(u, str) and host_hint in u:
+                return u
+    # 最后兜底：返回第一个 http(s) URL
+    for u in url_list:
+        if isinstance(u, str) and u.startswith("http"):
+            return u
+    return None
 
 
 def _extract_aweme_id(url: str) -> Optional[str]:
@@ -40,9 +66,7 @@ def _extract_aweme_id(url: str) -> Optional[str]:
     return None
 
 
-def fetch_douyin_video_url(
-    page_url: str, timeout: int = 15, max_retries: int = 3
-) -> Tuple[Optional[str], Optional[str]]:
+def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = 3) -> Tuple[Optional[str], Optional[str]]:
     """通过抖音 App Feed API 获取视频无水印直链和文案。
 
     Args:
@@ -93,9 +117,7 @@ def fetch_douyin_video_url(
                 continue
 
             if len(resp.text) < 100:
-                logger.warning(
-                    "Feed API 第%d次: 返回内容过短 len=%d body=%s", attempt + 1, len(resp.text), resp.text[:100]
-                )
+                logger.warning("Feed API 第%d次: 返回内容过短 len=%d body=%s", attempt + 1, len(resp.text), resp.text[:100])
                 time.sleep(0.5 * (attempt + 1))
                 continue
 
@@ -106,9 +128,7 @@ def fetch_douyin_video_url(
                 status_msg = data.get("status_msg", "")
                 logger.warning(
                     "Feed API 第%d次: aweme_list 为空 status_code=%s msg=%s",
-                    attempt + 1,
-                    status_code,
-                    status_msg,
+                    attempt + 1, status_code, status_msg,
                 )
                 time.sleep(0.5 * (attempt + 1))
                 continue
@@ -121,28 +141,49 @@ def fetch_douyin_video_url(
 
             desc = item.get("desc", "")
             video = item.get("video") or {}
+            aweme_type = item.get("aweme_type", 0)
 
-            # 提取无水印直链（优先 play_addr_h264 或 play_addr）
+            # 图文视频 (aweme_type=68): play_addr 通常返回 BGM 的 MP3，不是视频本身。
+            # 此时没有可用视频直链，返回 (None, desc) 让调用方仅使用文案。
+            images = item.get("images") or []
+            if images and not video.get("play_addr_h264", {}).get("url_list"):
+                logger.info("Feed API 返回图文视频: aweme_id=%s images=%d desc_len=%d（无视频直链，返回文案）",
+                             aweme_id, len(images), len(desc))
+                return None, desc
+
+            # 提取无水印直链：优先 download_addr（含 logo 但 CDN 直链稳定），再 play_addr_h264/play_addr
             play_url = None
-            for addr_key in ("play_addr_h264", "play_addr", "download_addr", "play_addr_265"):
+            # 按 key 优先级遍历：download_addr（含水印但CDN直链稳定）→ play_addr_h264 → play_addr → play_addr_lowbr
+            addr_keys = ("download_addr", "play_addr_h264", "play_addr", "play_addr_lowbr", "play_addr_265")
+            for addr_key in addr_keys:
                 addr = video.get(addr_key) or {}
                 url_list = addr.get("url_list") or []
-                for u in url_list:
-                    if isinstance(u, str) and any(h in u for h in _VIDEO_HOST_HINTS):
-                        play_url = u
-                        break
-                if play_url:
+                # 注意：图文视频的 play_addr 里可能是 BGM MP3 而非视频，需过滤 .mp3
+                filtered = [u for u in url_list if isinstance(u, str) and not u.endswith(".mp3")]
+                picked = _pick_best_url(filtered)
+                if picked:
+                    play_url = picked
                     break
 
+            # bit_rate 里的多码率地址作为最后兜底
+            if not play_url:
+                for br_entry in (video.get("bit_rate") or []):
+                    for addr_key in addr_keys:
+                        addr = br_entry.get(addr_key) or {}
+                        url_list = addr.get("url_list") or []
+                        filtered = [u for u in url_list if isinstance(u, str) and not u.endswith(".mp3")]
+                        picked = _pick_best_url(filtered)
+                        if picked:
+                            play_url = picked
+                            break
+                    if play_url:
+                        break
+
             if play_url:
                 duration = video.get("duration", 0) / 1000 if video.get("duration") else 0
                 logger.info(
-                    "抖音 Feed API 成功（第%d次）: aweme_id=%s play_len=%d desc_len=%d dur=%.1f",
-                    attempt + 1,
-                    aweme_id,
-                    len(play_url),
-                    len(desc),
-                    duration,
+                    "抖音 Feed API 成功（第%d次）: aweme_id=%s play_len=%d desc_len=%d dur=%.1f type=%d",
+                    attempt + 1, aweme_id, len(play_url), len(desc), duration, aweme_type,
                 )
                 return play_url, desc
 
@@ -159,12 +200,11 @@ def fetch_douyin_video_url(
 
 if __name__ == "__main__":
     import sys
-
     logging.basicConfig(level=logging.INFO)
     test_url = sys.argv[1] if len(sys.argv) > 1 else "https://www.douyin.com/video/7661819662322649065"
     url, desc = fetch_douyin_video_url(test_url)
     if url:
-        print("\n✅ SUCCESS!")
+        print(f"\n✅ SUCCESS!")
         print(f"desc: {desc}")
         print(f"play_url: {url[:200]}")
     else: