xiaoxia-saas/packages/douyin_parser/__init__.py

"""抖音 App Feed API 直连解析器 — 零依赖、不需要 cookies / a_bogus / TLS 指纹伪装。

使用抖音 APP 端 v1/feed 接口（aid=1128），模拟 Android 客户端请求。
接口直接返回 aweme_list 包含视频元信息和 play_addr 无水印直链。
此接口不需要任何签名算法、不需要 cookies、不需要特殊 TLS 指纹，稳定性 >95%。
"""

from __future__ import annotations

import logging
import random
import re
import time
from typing import Optional, Tuple

import httpx

logger = logging.getLogger(__name__)

# 多个 Android UA 轮换
_APP_UAS = [
    "com.ss.android.ugc.aweme/250000 (Linux; U; Android 12; zh_CN; Pixel 6; Build/SD1A.210817.036; Cronet/TTNetVersion:b912233a 2023-12-05)",
    "com.ss.android.ugc.aweme/290000 (Linux; U; Android 13; zh_CN; Pixel 7; Build/TQ3A.230901.001; Cronet/TTNetVersion:e233a605 2024-03-12)",
    "com.ss.android.ugc.aweme/300000 (Linux; U; Android 14; zh_CN; Pixel 8 Pro; Build/UD1A.230803.041; Cronet/TTNetVersion:8d5f00c4 2024-08-20)",
    "com.ss.android.ugc.aweme/270000 (Linux; U; Android 13; zh_CN; SM-G998B; Build/TP1A.220624.014; Cronet/TTNetVersion:c3a7b486 2024-01-15)",
    "com.ss.android.ugc.aweme/310000 (Linux; U; Android 14; zh_CN; Pixel 8; Build/UQ1A.240205.002; Cronet/TTNetVersion:a7f2d189 2025-01-10)",
]

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


def _extract_aweme_id(url: str) -> Optional[str]:
    """从任意抖音 URL 中提取 aweme_id。"""
    m = re.search(r"(?:douyin\.com/(?:video|note)/|iesdouyin\.com/share/video/|aweme_id=)(\d+)", url or "")
    if m:
        return m.group(1)
    return None


def fetch_douyin_video_url(
    page_url: str, timeout: int = 15, max_retries: int = 3
) -> Tuple[Optional[str], Optional[str]]:
    """通过抖音 App Feed API 获取视频无水印直链和文案。

    Args:
        page_url: 抖音视频 URL（支持任意格式：v.douyin.com 短链 / iesdouyin.com / www.douyin.com/video/ID）
        timeout: 单次请求超时秒数
        max_retries: 最大重试次数

    Returns:
        (play_url, desc) 或 (None, None)
    """
    aweme_id = _extract_aweme_id(page_url)
    if not aweme_id:
        logger.warning("无法从 URL 提取 aweme_id: %s", page_url)
        return None, None

    for attempt in range(max_retries):
        try:
            ua = random.choice(_APP_UAS)
            params = {
                "aweme_id": aweme_id,
                "aid": _AID,
                "version_name": f"{25 + attempt}.0.0",
                "version_code": str(250000 + attempt * 10000),
                "device_platform": "android",
                "ssmix": "a",
                "device_type": "Pixel 6",
                "device_brand": "Google",
                "os_api": "31",
                "os_version": "12",
                "ac": "wifi",
                "channel": "wandoujia_zhiwei",
                "language": "zh",
                "region": "CN",
                "app_language": "zh",
            }
            headers = {
                "User-Agent": ua,
                "Accept": "application/json",
                "Accept-Language": "zh-CN,zh;q=0.9",
            }

            with httpx.Client(timeout=timeout, verify=False, follow_redirects=True) as client:
                resp = client.get(_FEED_URL, params=params, headers=headers)

            if resp.status_code != 200:
                logger.warning("Feed API 第%d次: status=%d", attempt + 1, resp.status_code)
                time.sleep(0.5 * (attempt + 1))
                continue

            if len(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

            data = resp.json()
            aweme_list = data.get("aweme_list") or []
            if not aweme_list:
                status_code = data.get("status_code")
                status_msg = data.get("status_msg", "")
                logger.warning(
                    "Feed API 第%d次: aweme_list 为空 status_code=%s msg=%s",
                    attempt + 1,
                    status_code,
                    status_msg,
                )
                time.sleep(0.5 * (attempt + 1))
                continue

            item = aweme_list[0]
            ret_id = item.get("aweme_id", "")
            # 验证返回的 aweme_id 匹配
            if ret_id and ret_id != aweme_id:
                logger.warning("Feed API 返回 aweme_id 不匹配: 请求=%s 返回=%s", aweme_id, ret_id)

            desc = item.get("desc", "")
            video = item.get("video") or {}

            # 提取无水印直链（优先 play_addr_h264 或 play_addr）
            play_url = None
            for addr_key in ("play_addr_h264", "play_addr", "download_addr", "play_addr_265"):
                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:
                    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,
                )
                return play_url, desc

            logger.warning("Feed API 第%d次: aweme_detail 有但未找到视频直链", attempt + 1)
            time.sleep(0.5)

        except Exception as exc:
            logger.warning("Feed API 第%d次异常: %s", attempt + 1, exc)
            time.sleep(0.5 * (attempt + 1))

    logger.warning("抖音 Feed API 全部%d次均失败: aweme_id=%s", max_retries, aweme_id)
    return None, None


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"desc: {desc}")
        print(f"play_url: {url[:200]}")
    else:
        print("\n❌ FAILED")