解析思路
ShayuAgent 现在用的是增强版抖音解析链路。它不再只依赖 playwm 替换，而是优先走分享页 SSR 数据里的 video.play_addr token：
分享文案 / v.douyin.com 短链 / douyin.com/video/<id> / aweme_id
-> 跟随短链跳转
-> https://www.iesdouyin.com/share/video/<aweme_id>/
-> 解析 window._ROUTER_DATA 或 RENDER_DATA
-> 读取 video.play_addr.uri / url_key / url_list 里的 video_id token
-> aweme.snssdk.com/aweme/v1/play/?video_id=<token>&ratio=1080p
-> 302 / CDN
-> mp4这条链路零签名、零登录 cookie。ttwid 只是匿名设备 cookie，用来让分享页更稳定返回 SSR 数据，不是账号登录态。
关键点
v.douyin.com短链最终会跳到/video/<aweme_id>、/note/<aweme_id>或带aweme_id的页面。- 分享页里优先解析
window._ROUTER_DATA，新旧页面结构不一致时再兜底RENDER_DATA。 - 视频 token 优先取
video.play_addr.uri或url_key；没有时从url_list查询参数里的video_id/vid提取。 - 清晰度按
1080p -> 720p -> 540p -> 360p探测，用Range: bytes=0-1避免把整段视频下载下来。 - 用
Content-Range或Content-Length判断文件大小，相同大小视作同一档清晰度，去重后选最高档。 - token play 端点不可用时，才退回旧方案：
playwm替换成play，再取一次跳转地址。
ShayuAgent 里的核心实现
下面是从 ShayuAgent/mcp-server/src/shayu_mcp/douyin.py 整理出来的核心版本，保留真实域名、真实端点和真实请求方式。
import html as html_lib
import json
import re
from typing import Any
from urllib.parse import parse_qs, quote, unquote, urlparse
import requests
DESKTOP_UA = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
MOBILE_UA = (
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
)
PLAY_BASE = "https://aweme.snssdk.com/aweme/v1/play/"
SHARE_BASE = "https://www.iesdouyin.com/share/video/{aweme_id}/"
TTWID_REGISTER_URL = "https://ttwid.bytedance.com/ttwid/union/register/"
PLAY_RATIOS = ("1080p", "720p", "540p", "360p")
def extract_first_url(value: str) -> str:
match = re.search(r"https?://[^\s\"'<>，,。]+", value or "")
if match:
return match.group(0).rstrip("，,。.!！)")
return (value or "").strip()
def new_douyin_session(timeout: float = 20.0) -> requests.Session:
session = requests.Session()
session.headers.update({
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
"Connection": "keep-alive",
"User-Agent": MOBILE_UA,
})
ensure_ttwid(session, timeout)
return session
def ensure_ttwid(session: requests.Session, timeout: float = 20.0) -> str | None:
payload = {
"aid": 6383,
"service": "www.douyin.com",
"region": "cn",
"union": True,
"needFid": False,
"fid": "",
"migrate_info": {"ticket": "", "source": "node"},
"is_enable_migrate": True,
}
headers = {
"Accept": "application/json, text/plain, */*",
"Content-Type": "application/json",
"Origin": "https://www.douyin.com",
"Referer": "https://www.douyin.com/",
}
try:
response = session.post(TTWID_REGISTER_URL, headers=headers, json=payload, timeout=timeout)
if not response.ok:
return None
data = response.json()
ttwid = data.get("ttwid")
if not ttwid:
return None
for domain in (".douyin.com", ".iesdouyin.com", ".bytedance.com"):
session.cookies.set("ttwid", ttwid, domain=domain, path="/")
return ttwid
except Exception:
return None
def follow_url(url: str, session: requests.Session, timeout: float = 20.0) -> str:
response = session.get(
url,
headers={"User-Agent": DESKTOP_UA, "Accept-Language": "zh-CN,zh;q=0.9"},
timeout=timeout,
allow_redirects=True,
)
response.raise_for_status()
return response.url
def extract_aweme_id(input_value: str, final_url: str) -> tuple[str, str | None]:
for value in (input_value, final_url):
for pattern, mode in (
(r"/video/(\d{10,25})", "video"),
(r"/share/video/(\d{10,25})", "video"),
(r"/note/(\d{10,25})", "note"),
(r"/share/slides/(\d{10,25})", "note"),
):
match = re.search(pattern, value)
if match:
return match.group(1), mode
query = parse_qs(urlparse(value).query)
for key in ("aweme_id", "modal_id", "item_id", "video_id"):
candidate = query.get(key, [None])[0]
if candidate and candidate.isdigit():
return candidate, None
if re.fullmatch(r"\d{10,25}", input_value):
return input_value, None
raise ValueError(f"无法从输入中提取 aweme_id：{final_url}")
def get_share_html(aweme_id: str, session: requests.Session, timeout: float = 20.0) -> str:
response = session.get(
SHARE_BASE.format(aweme_id=aweme_id),
headers={
"User-Agent": MOBILE_UA,
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://www.douyin.com/",
"Upgrade-Insecure-Requests": "1",
},
timeout=timeout,
)
response.raise_for_status()
return response.text
def parse_router_data(text: str) -> dict[str, Any]:
data = extract_window_json(text, "window._ROUTER_DATA") or extract_render_data(text)
if not isinstance(data, dict):
raise RuntimeError("share 页没有解析到 SSR 数据")
return data
def extract_render_data(text: str) -> dict[str, Any] | None:
match = re.search(r'<script[^>]+id=["\']RENDER_DATA["\'][^>]*>(.*?)</script>', text, re.S)
if not match:
return None
raw = html_lib.unescape(match.group(1).strip())
for candidate in (raw, unquote(raw)):
try:
data = json.loads(candidate)
except json.JSONDecodeError:
continue
if isinstance(data, dict):
return data
return None
def extract_window_json(text: str, marker: str) -> dict[str, Any] | None:
index = text.find(marker)
if index < 0:
return None
brace_index = text.find("{", text.find("=", index))
if brace_index < 0:
return None
depth = 0
in_string = False
escaped = False
for i in range(brace_index, len(text)):
char = text[i]
if in_string:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == '"':
in_string = False
continue
if char == '"':
in_string = True
elif char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return json.loads(text[brace_index:i + 1])
return None
def walk_dicts(value: Any):
if isinstance(value, dict):
yield value
for child in value.values():
yield from walk_dicts(child)
elif isinstance(value, list):
for child in value:
yield from walk_dicts(child)
def token_from_url(url: Any) -> str | None:
if not isinstance(url, str):
return None
query = parse_qs(urlparse(html_lib.unescape(url)).query)
for key in ("video_id", "vid"):
value = query.get(key, [None])[0]
if value:
return value
return None
def token_from_play_addr(value: Any) -> str | None:
if not isinstance(value, dict):
return None
for key in ("uri", "url_key"):
token = value.get(key)
if isinstance(token, str) and token:
return token
for url in value.get("url_list") or []:
token = token_from_url(url)
if token:
return token
return None
def extract_video_detail(router_data: dict[str, Any], aweme_id: str) -> dict[str, Any]:
aweme = None
for node in walk_dicts(router_data):
if str(node.get("aweme_id") or node.get("awemeId") or "") == aweme_id:
aweme = node
break
if aweme is None:
for node in walk_dicts(router_data):
video = node.get("video")
if isinstance(video, dict) and token_from_play_addr(video.get("play_addr")):
aweme = node
break
if not isinstance(aweme, dict):
raise RuntimeError("ROUTER_DATA 里没有找到 aweme 数据")
video = aweme.get("video") if isinstance(aweme.get("video"), dict) else {}
token = None
for node in walk_dicts(video):
token = token_from_play_addr(node.get("play_addr")) or token_from_play_addr(node)
if token:
break
if not token:
raise RuntimeError("ROUTER_DATA 里没有找到 video.play_addr token")
cover = None
for key in ("cover", "origin_cover", "dynamic_cover"):
urls = (video.get(key) or {}).get("url_list")
if isinstance(urls, list) and urls:
cover = urls[0]
break
return {
"id": aweme_id,
"token": token,
"title": aweme.get("desc") or "",
"author": (aweme.get("author") or {}).get("nickname"),
"cover": cover,
}
def build_play_url(token: str, ratio: str) -> str:
return f"{PLAY_BASE}?video_id={quote(token)}&ratio={quote(ratio)}"
def size_from_headers(headers: requests.structures.CaseInsensitiveDict[str]) -> int | None:
content_range = headers.get("Content-Range")
if content_range:
match = re.search(r"/(\d+)\s*$", content_range)
if match:
return int(match.group(1))
content_length = headers.get("Content-Length")
if content_length and content_length.isdigit():
value = int(content_length)
return None if value <= 2 else value
return None
def probe_play_candidates(token: str, share_url: str, session: requests.Session, timeout: float = 20.0) -> list[dict]:
candidates = []
seen_sizes = set()
for ratio in PLAY_RATIOS:
request_url = build_play_url(token, ratio)
response = session.get(
request_url,
headers={
"Accept": "*/*",
"Range": "bytes=0-1",
"Referer": share_url,
"User-Agent": session.headers.get("User-Agent", MOBILE_UA),
},
allow_redirects=True,
stream=True,
timeout=timeout,
)
try:
if not (200 <= response.status_code < 400):
continue
size = size_from_headers(response.headers)
if isinstance(size, int):
if size in seen_sizes:
continue
seen_sizes.add(size)
candidates.append({
"ratio": ratio,
"request_url": request_url,
"final_url": response.url,
"status_code": response.status_code,
"size_bytes": size,
"content_type": response.headers.get("Content-Type"),
})
finally:
response.close()
return sorted(candidates, key=lambda item: (int(re.search(r"\d+", item["ratio"]).group()), item.get("size_bytes") or 0), reverse=True)
def parse_douyin_video(input_value: str, timeout: float = 20.0) -> dict:
raw = extract_first_url(input_value)
session = new_douyin_session(timeout)
if re.match(r"^https?://", raw, re.I):
final_url = follow_url(raw, session, timeout)
elif re.fullmatch(r"\d{10,25}", raw):
final_url = SHARE_BASE.format(aweme_id=raw)
else:
raise ValueError(f"输入里没有可解析的抖音链接或 aweme_id：{input_value}")
aweme_id, mode = extract_aweme_id(raw, final_url)
share_url = SHARE_BASE.format(aweme_id=aweme_id)
html = get_share_html(aweme_id, session, timeout)
router_data = parse_router_data(html)
if mode == "note" or "/note/" in final_url or "/share/slides/" in final_url:
raise RuntimeError("这里省略图文分支，ShayuAgent 源码里会返回 images 列表")
detail = extract_video_detail(router_data, aweme_id)
candidates = probe_play_candidates(detail["token"], share_url, session, timeout)
if not candidates:
raise RuntimeError("没有探测到可用的 play CDN 地址")
best = candidates[0]
return {
"success": True,
"platform": "douyin",
"mode": "video",
"id": aweme_id,
"title": detail["title"],
"author": detail["author"],
"cover": detail["cover"],
"video_url": best["final_url"],
"quality": best["ratio"],
"file_size": best["size_bytes"],
"content_type": best["content_type"],
"play_url": best["request_url"],
"request_headers": {
"Referer": share_url,
"User-Agent": session.headers.get("User-Agent", MOBILE_UA),
},
"raw": {"token": detail["token"], "candidates": candidates},
}并发和风控
真实使用时，下载链路本身是普通 CDN mp4；更容易触发限制的是分享页元数据获取。低频本地工具基本无感，高频批量解析建议：
if index > 0 and index % 20 == 0:
time.sleep(1.0)Range: bytes=0-1 探测只拿响应头和前两个字节，不下载完整文件，适合先判断清晰度和文件大小。真正下载时再用 video_url 发普通流式 GET。
和旧方案的区别
旧方案是从 play_addr.url_list 里挑一个 /playwm/ 地址，然后替换成 /play/，再取 302。它简单，但依赖具体 URL 形态。
增强版更接近分享页自身的 SSR 渲染路径：先拿 video_id token，再请求 aweme/v1/play。如果这条主路径失败，ShayuAgent 仍然保留旧 playwm -> play 作为兜底，所以线上工具不会因为单一路径波动就立刻不可用。