"""抖音 Web 接口客户端。 实测要点（都是打真实接口验出来的）： 1. 空 body 有两种成因，必须分开看： - 新 session 的第一个请求必然空，第二个才通 —— cookie 预热，见 _warm_up() - 服务端主动拦截 —— 响应头 X-Whale-Throughput-Abort-Data 里有 base64 原因 2. /aweme/detail/ 对匿名用户返回 abort={"name":"强制登录"}，必须带登录 Cookie。 3. /aweme/related/ 匿名可用，返回完整 play_addr，但不含目标视频自身。 4. 请求头 UA 必须与签名时的 UA 完全一致，params 里的 browser_version 也要跟着对齐。 """ from __future__ import annotations import base64 import json import random import string import time from dataclasses import dataclass from typing import Any, Dict, List, Optional from urllib.parse import urlencode import requests from .signer import Signer DETAIL_URL = "https://www.douyin.com/aweme/v1/web/aweme/detail/" RELATED_URL = "https://www.douyin.com/aweme/v1/web/aweme/related/" TTWID_URL = "https://ttwid.bytedance.com/ttwid/union/register/" BASE_PARAMS: Dict[str, str] = { "device_platform": "webapp", "aid": "6383", "channel": "channel_pc_web", "pc_client_type": "1", "version_code": "190500", "version_name": "19.5.0", "cookie_enabled": "true", "screen_width": "1920", "screen_height": "1080", "browser_language": "zh-CN", "browser_platform": "Win32", "browser_name": "Chrome", "browser_version": "126.0.0.0", "browser_online": "true", "engine_name": "Blink", "engine_version": "126.0.0.0", "os_name": "Windows", "os_version": "10", "cpu_core_num": "8", "device_memory": "8", "platform": "PC", "downlink": "10", "effective_type": "4g", "round_trip_time": "50", } PLAY_URL = "https://www.douyin.com/aweme/v1/play/" def play_url_from_uri(uri: str, ratio: str = "1080p", watermark: bool = False) -> str: """由 video URI 拼出播放直链。 实测这个接口完全免鉴权：无 a_bogus、无 Cookie、无登录，裸 requests 即可， 302 跳到 douyinvod.com 的 CDN 地址。所以只要拿到了 URI，后续下载不再需要 任何身份——URI 值得缓存复用。 """ path = "playwm" if watermark else "play" return f"https://www.douyin.com/aweme/v1/{path}/?video_id={uri}&ratio={ratio}&line=0" class DouyinError(RuntimeError): pass class LoginRequired(DouyinError): """服务端明确要求登录。""" @dataclass class Aweme: aweme_id: str desc: str author: str create_time: int video_urls: List[str] cover_urls: List[str] music_url: Optional[str] uri: str raw: Dict[str, Any] @property def is_video(self) -> bool: return bool(self.video_urls or self.uri) def play_url(self, ratio: str = "1080p") -> str: """免鉴权直链。play_addr 里的地址带时效，URI 换出来的是现算的。""" if not self.uri: raise DouyinError("该作品没有 video URI") return play_url_from_uri(self.uri, ratio) def _rand(n: int, pool: str) -> str: return "".join(random.choice(pool) for _ in range(n)) def gen_ms_token(length: int = 107) -> str: return _rand(length, string.ascii_letters + string.digits + "=-_") def gen_s_v_web_id() -> str: body = "".join( "_" if i in (8, 13, 18, 23) else random.choice("0123456789abcdef") for i in range(32) ) return f"verify_{format(int(time.time()), 'x')}_{body}" def decode_abort(resp: requests.Response) -> str: """解出服务端拦截原因；没有则返回空串。""" raw = resp.headers.get("X-Whale-Throughput-Abort-Data") if not raw: return "" try: info = json.loads(base64.b64decode(raw)) return info.get("name") or info.get("content") or "" except (ValueError, TypeError): return "unknown" class DouyinClient: def __init__(self, signer: Signer, cookie: str = "", timeout: float = 25.0, warm_up: bool = True) -> None: self.signer = signer self.timeout = timeout self.user_agent = signer.user_agent self.fp = gen_s_v_web_id() # params 里的浏览器版本必须跟签名用的 UA 对上，否则服务端会判不一致 self.params = dict(BASE_PARAMS) if "Chrome/" in self.user_agent: ver = self.user_agent.split("Chrome/")[1].split(" ")[0] self.params["browser_version"] = ver self.params["engine_version"] = ver self.session = requests.Session() self.session.headers.update({ "User-Agent": self.user_agent, "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": "https://www.douyin.com/", }) self._logged_in = bool(cookie) if cookie: self.session.headers["Cookie"] = cookie else: self._bootstrap_cookies() self._warmed = False if warm_up: self._warm_up() # ------------------------------------------------------------------ # def _bootstrap_cookies(self) -> None: try: self.session.post(TTWID_URL, timeout=self.timeout, json={ "region": "cn", "aid": 1768, "needFid": False, "service": "www.ixigua.com", "migrate_info": {"ticket": "", "source": "node"}, "cbUrlProtocol": "https", "union": True, }) except requests.RequestException: pass self.session.cookies.set("s_v_web_id", self.fp, domain=".douyin.com") self.session.cookies.set("odin_tt", _rand(160, "0123456789abcdef"), domain=".douyin.com") def _warm_up(self) -> None: """新 session 的第一个接口请求必然返回空，先打一发把它消耗掉。""" if self._warmed: return try: self._get(RELATED_URL, {"aweme_id": "1", "count": "1", "filterGids": "1"}) except (DouyinError, requests.RequestException): pass self._warmed = True def _build_url(self, base: str, extra: Dict[str, str]) -> str: p = dict(self.params) p.update(extra) p["msToken"] = gen_ms_token() return f"{base}?{urlencode(p)}" def _get(self, base: str, extra: Dict[str, str], retries: int = 3) -> Dict[str, Any]: """签名 + 请求 + 解析。 「空 body 且无拦截原因」是可重试的瞬时态（session 还没热起来）， 「空 body 且原因含登录」是硬门槛，立即抛错不重试。 """ last = "" for attempt in range(1, retries + 1): url = self.signer.sign_url(self._build_url(base, extra), self.fp) resp = self.session.get(url, timeout=self.timeout) if resp.content: try: return resp.json() except ValueError as exc: raise DouyinError(f"响应不是 JSON: {resp.text[:200]!r}") from exc reason = decode_abort(resp) if "登录" in reason: raise LoginRequired( f"接口要求登录（服务端返回 {reason!r}）。" "请用 --cookie 传入已登录的抖音 Cookie。" ) last = (f"HTTP {resp.status_code}" + (f"，拦截原因 {reason!r}" if reason else "，无拦截原因")) if attempt < retries: time.sleep(2 * attempt) raise DouyinError(f"接口连续 {retries} 次返回空 body（{last}）") # ------------------------------------------------------------------ # def fetch_detail(self, aweme_id: str) -> Aweme: """按 aweme_id 取作品详情。匿名调用会抛 LoginRequired。""" self.session.headers["Referer"] = f"https://www.douyin.com/video/{aweme_id}" data = self._get(DETAIL_URL, {"aweme_id": aweme_id}) detail = data.get("aweme_detail") if not detail: raise DouyinError( f"响应无 aweme_detail，status_code={data.get('status_code')}，" "作品可能已删除或为私密" ) return self.parse(detail) def fetch_related(self, aweme_id: str, count: int = 10) -> List[Aweme]: """相关推荐。匿名可用，但不含 aweme_id 自身。""" data = self._get(RELATED_URL, { "aweme_id": aweme_id, "count": str(count), "filterGids": aweme_id, }) return [self.parse(a) for a in (data.get("aweme_list") or [])] # ------------------------------------------------------------------ # @staticmethod def parse(detail: Dict[str, Any]) -> Aweme: video = detail.get("video") or {} # play_addr 带水印，playwm -> play 即无水印源。 # bit_rate 里有多档清晰度，优先挑码率最高的。 urls: List[str] = [] gears = sorted( (b for b in (video.get("bit_rate") or []) if b.get("play_addr")), key=lambda b: b.get("bit_rate") or 0, reverse=True, ) for gear in gears: urls.extend((gear["play_addr"].get("url_list") or [])) urls.extend((video.get("play_addr") or {}).get("url_list") or []) seen: set[str] = set() video_urls = [] for u in urls: u = u.replace("playwm", "play") if u not in seen: seen.add(u) video_urls.append(u) cover = video.get("origin_cover") or video.get("cover") or {} music_list = ((detail.get("music") or {}).get("play_url") or {}).get("url_list") or [] return Aweme( aweme_id=str(detail.get("aweme_id", "")), desc=(detail.get("desc") or "").strip(), author=((detail.get("author") or {}).get("nickname") or "").strip(), create_time=int(detail.get("create_time") or 0), video_urls=video_urls, cover_urls=list(cover.get("url_list") or []), music_url=music_list[0] if music_list else None, uri=(video.get("play_addr") or {}).get("uri") or "", raw=detail, )