426 lines
14 KiB
Python
426 lines
14 KiB
Python
 
|
"""抖音 API 响应解析 — 纯数据提取，不涉及网络/下载"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import re
|
|||
|
|
import urllib.parse
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Dict, List, Optional
|
|||
|
|
|
|||
|
|
from nonebot import logger
|
|||
|
|
|
|||
|
|
from ..models import DouyinFetchError
|
|||
|
|
from ..utils import slugify
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 数据结构 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ParsedDouyinContent:
|
|||
|
|
"""从 API 响应解析出的抖音内容元数据"""
|
|||
|
|
raw_title: str
|
|||
|
|
raw_nickname: str
|
|||
|
|
media_type: str # "视频" | "图片"
|
|||
|
|
file_name: str # 构建好的文件名 stem
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= URL 工具 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def classify_url(url: str) -> str:
|
|||
|
|
"""返回 FULL / VIDEO / AUDIO / UNKNOWN"""
|
|||
|
|
if "/media-video-" in url:
|
|||
|
|
return "VIDEO"
|
|||
|
|
if "/media-audio-" in url:
|
|||
|
|
return "AUDIO"
|
|||
|
|
if "mime_type=video_mp4" in url:
|
|||
|
|
return "FULL"
|
|||
|
|
return "UNKNOWN"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_group_key(url: str) -> str:
|
|||
|
|
"""提取分组 key（视频ID + l参数）"""
|
|||
|
|
try:
|
|||
|
|
m = re.search(r"/([0-9a-f]{8})/", url)
|
|||
|
|
vid = m.group(1) if m else "unknown"
|
|||
|
|
parsed = urllib.parse.urlparse(url)
|
|||
|
|
qs = urllib.parse.parse_qs(parsed.query)
|
|||
|
|
l = qs.get("l", [""])[0]
|
|||
|
|
return f"{vid}_{l}"
|
|||
|
|
except Exception:
|
|||
|
|
return url
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_br(url: str) -> int:
|
|||
|
|
"""提取码率"""
|
|||
|
|
try:
|
|||
|
|
parsed = urllib.parse.urlparse(url)
|
|||
|
|
qs = urllib.parse.parse_qs(parsed.query)
|
|||
|
|
return int(qs.get("br", [0])[0])
|
|||
|
|
except Exception:
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_trailing_digits(url):
|
|||
|
|
"""从 URL 末尾提取数字 ID"""
|
|||
|
|
match = re.search(r"/(\d+)(?:\?|#|$)", url)
|
|||
|
|
return match.group(1) if match else None
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 内容提取 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_title_from_api(api_response: dict) -> str:
|
|||
|
|
"""从 API 响应提取标题，优先级: preview_title > desc > caption > aweme_list[0].desc"""
|
|||
|
|
aweme_detail = api_response.get("aweme_detail") or {}
|
|||
|
|
title = aweme_detail.get("preview_title")
|
|||
|
|
if not title:
|
|||
|
|
title = aweme_detail.get("desc")
|
|||
|
|
if not title:
|
|||
|
|
title = aweme_detail.get("caption")
|
|||
|
|
if not title:
|
|||
|
|
aweme_list = api_response.get("aweme_list") or []
|
|||
|
|
if aweme_list and isinstance(aweme_list, list):
|
|||
|
|
first = aweme_list[0] or {}
|
|||
|
|
title = first.get("desc")
|
|||
|
|
title_str = (title or "").strip()
|
|||
|
|
logger.info(f"RAW标题：{title_str}")
|
|||
|
|
if not title_str:
|
|||
|
|
logger.info("未获取到标题")
|
|||
|
|
return title_str
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_author_nickname(api_response: dict) -> str:
|
|||
|
|
"""从 API 响应提取作者昵称，优先级: aweme_detail.author.nickname > aweme_list[0].author.nickname"""
|
|||
|
|
aweme_detail = api_response.get("aweme_detail") or {}
|
|||
|
|
author = aweme_detail.get("author") or {}
|
|||
|
|
nickname = author.get("nickname")
|
|||
|
|
if not nickname:
|
|||
|
|
aweme_list = api_response.get("aweme_list") or []
|
|||
|
|
if aweme_list and isinstance(aweme_list, list):
|
|||
|
|
first = aweme_list[0] or {}
|
|||
|
|
author = first.get("author") or {}
|
|||
|
|
nickname = author.get("nickname")
|
|||
|
|
logger.info(f"RAW作者昵称：{nickname}")
|
|||
|
|
if not nickname:
|
|||
|
|
nickname = "未知作者"
|
|||
|
|
logger.info("未获取到作者昵称，使用默认值")
|
|||
|
|
return nickname
|
|||
|
|
|
|||
|
|
|
|||
|
|
def detect_media_type(referer_url: str | None) -> str | None:
|
|||
|
|
"""根据页面 URL 检测媒体类型（视频/图片）"""
|
|||
|
|
if referer_url is None:
|
|||
|
|
return None
|
|||
|
|
if "video" in referer_url:
|
|||
|
|
return "视频"
|
|||
|
|
elif "note" in referer_url:
|
|||
|
|
return "图片"
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 媒体 URL 解析 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_video_urls(api_response: dict) -> Dict[str, Dict[str, List[dict]]]:
|
|||
|
|
"""从 API 响应解析视频 URL，返回分组结构: {group_id: {VIDEO: [...], AUDIO: [...], FULL: [...]}}"""
|
|||
|
|
captured = []
|
|||
|
|
try:
|
|||
|
|
video_data = api_response.get("aweme_detail", {}).get("video", {})
|
|||
|
|
bit_rates = video_data.get("bit_rate", [])
|
|||
|
|
for item in bit_rates:
|
|||
|
|
play_addr = item.get("play_addr", {})
|
|||
|
|
url_list = play_addr.get("url_list", [])
|
|||
|
|
br = item.get("bit_rate", 0)
|
|||
|
|
if url_list:
|
|||
|
|
url = url_list[0]
|
|||
|
|
captured.append(
|
|||
|
|
{
|
|||
|
|
"url": url,
|
|||
|
|
"br": br,
|
|||
|
|
"type": classify_url(url),
|
|||
|
|
"group": extract_group_key(url),
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
logger.info(f"从 API 解析出 {len(captured)} 个视频链接")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"解析视频 URL 失败: {e}")
|
|||
|
|
raise DouyinFetchError(f"解析视频 URL 失败: {e}")
|
|||
|
|
|
|||
|
|
groups: Dict[str, Dict[str, List[dict]]] = {}
|
|||
|
|
for item in captured:
|
|||
|
|
g = item["group"]
|
|||
|
|
if g not in groups:
|
|||
|
|
groups[g] = {"FULL": [], "VIDEO": [], "AUDIO": []}
|
|||
|
|
if item["type"] in groups[g]:
|
|||
|
|
groups[g][item["type"]].append(item)
|
|||
|
|
return groups
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_note_images(
|
|||
|
|
api_response: dict, api_response_favorite: dict, aweme_id: str
|
|||
|
|
) -> tuple[List[List[str]], Optional[str]]:
|
|||
|
|
"""
|
|||
|
|
从 API 响应解析图文/图+视频链接
|
|||
|
|
|
|||
|
|
images 列表中每项有两种结构:
|
|||
|
|
- 纯图片: 取 url_list
|
|||
|
|
- 视频: 取 video.play_addr.url_list
|
|||
|
|
|
|||
|
|
返回:
|
|||
|
|
(images_list, video_url)
|
|||
|
|
- images_list: 图片 URL 列表 [[url1, ...], ...]
|
|||
|
|
- video_url: 混合作品中的视频 URL（无视频时为 None）
|
|||
|
|
"""
|
|||
|
|
images_list: List[List[str]] = []
|
|||
|
|
video_url: Optional[str] = None
|
|||
|
|
try:
|
|||
|
|
aweme_list = api_response.get("aweme_list", None)
|
|||
|
|
if not aweme_list:
|
|||
|
|
raise DouyinFetchError(
|
|||
|
|
f"获取到的作品图片列表为空,解析图文链接失败:{aweme_list}"
|
|||
|
|
)
|
|||
|
|
images_part = aweme_list[0]
|
|||
|
|
images = images_part.get("images", [])
|
|||
|
|
if images:
|
|||
|
|
for image in images:
|
|||
|
|
if not isinstance(image, dict):
|
|||
|
|
continue
|
|||
|
|
# 优先检查 video.play_addr（有视频的项）
|
|||
|
|
video = image.get("video")
|
|||
|
|
if video and isinstance(video, dict):
|
|||
|
|
play_addr = video.get("play_addr")
|
|||
|
|
if play_addr and isinstance(play_addr, dict):
|
|||
|
|
play_urls = play_addr.get("url_list") or []
|
|||
|
|
if play_urls:
|
|||
|
|
video_url = play_urls[0]
|
|||
|
|
logger.info(
|
|||
|
|
f"从 images 中发现视频: {video_url[:60]}..."
|
|||
|
|
)
|
|||
|
|
continue
|
|||
|
|
# 无视频的项取 url_list
|
|||
|
|
url_list = image.get("url_list", [])
|
|||
|
|
if url_list:
|
|||
|
|
images_list.append(url_list)
|
|||
|
|
logger.info(f"从 API 解析出 {len(images_list)} 张图片")
|
|||
|
|
if video_url:
|
|||
|
|
logger.info("混合作品包含视频")
|
|||
|
|
else:
|
|||
|
|
raise DouyinFetchError(
|
|||
|
|
f"获取到的作品图片列表为空,解析图文链接失败:{aweme_list}"
|
|||
|
|
)
|
|||
|
|
except DouyinFetchError:
|
|||
|
|
raise
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"解析图文链接失败: {e}")
|
|||
|
|
raise DouyinFetchError(f"解析图文链接失败: {e}")
|
|||
|
|
return images_list, video_url
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= SSR 页面回退解析 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _extract_rsc_payload(html: str) -> Optional[str]:
|
|||
|
|
"""从 HTML 中提取 __pace_f.push 的 RSC 流数据（7:[...] 格式）"""
|
|||
|
|
pos = 0
|
|||
|
|
while True:
|
|||
|
|
m = re.search(r'self\.__pace_f\.push\(\[(\d+),"', html[pos:])
|
|||
|
|
if not m:
|
|||
|
|
break
|
|||
|
|
start = pos + m.end()
|
|||
|
|
# 手动找闭合引号，处理转义
|
|||
|
|
end = start
|
|||
|
|
while end < len(html):
|
|||
|
|
ch = html[end]
|
|||
|
|
if ch == "\\":
|
|||
|
|
end += 2
|
|||
|
|
elif ch == '"':
|
|||
|
|
break
|
|||
|
|
else:
|
|||
|
|
end += 1
|
|||
|
|
# RSC payload 是 JS 字符串字面量（\" \\n \\\\ 等转义），
|
|||
|
|
# 用 json.loads 按字符串语义一次解码，避免手动替换顺序错误
|
|||
|
|
# （2026-08-13 实测：双重转义页手动替换会产出非法 JSON）
|
|||
|
|
try:
|
|||
|
|
raw = json.loads('"' + html[start:end] + '"')
|
|||
|
|
except json.JSONDecodeError:
|
|||
|
|
pos = end + 1
|
|||
|
|
continue
|
|||
|
|
# 找 RSC 流块: "7:[...]"
|
|||
|
|
if raw.startswith("7:["):
|
|||
|
|
return raw[2:] # 去掉 "7:" 前缀
|
|||
|
|
pos = end + 1
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _rsc_to_json(rsc: str) -> dict:
|
|||
|
|
"""将 React Flight 格式转为普通 JSON dict，处理 $undefined / $Lx 等特殊 token"""
|
|||
|
|
# $undefined -> null
|
|||
|
|
text = re.sub(r'\$undefined', 'null', rsc)
|
|||
|
|
# $L9 等 lazy ref -> null（避免解析报错）
|
|||
|
|
text = re.sub(r'\$L\d+', 'null', text)
|
|||
|
|
try:
|
|||
|
|
return json.loads(text)
|
|||
|
|
except json.JSONDecodeError:
|
|||
|
|
logger.error(f"SSR 页面 JSON 解析失败，数据前 500 字: {text[:500]}")
|
|||
|
|
raise DouyinFetchError("SSR 页面数据解析失败")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_ssr_page(html: str) -> Optional[dict]:
|
|||
|
|
"""
|
|||
|
|
从 SSR 页面 HTML 中提取图文数据，转换为 aweme_list 格式
|
|||
|
|
当 API 拦截未触发时作为回退方案
|
|||
|
|
|
|||
|
|
返回的 dict 可直接用于 parse_note_images / parse_animated_note_videos /
|
|||
|
|
extract_title_from_api / extract_author_nickname
|
|||
|
|
"""
|
|||
|
|
rsc = _extract_rsc_payload(html)
|
|||
|
|
if not rsc:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
data = _rsc_to_json(rsc)
|
|||
|
|
if not isinstance(data, list) or len(data) < 4:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# React Flight 格式: ["$","$L9",null,{awemeId,aweme:{detail:{...}}}]
|
|||
|
|
page_data = data[3]
|
|||
|
|
if not isinstance(page_data, dict):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
aweme = page_data.get("aweme") or {}
|
|||
|
|
if isinstance(aweme, str):
|
|||
|
|
aweme = {}
|
|||
|
|
detail = aweme.get("detail") or {}
|
|||
|
|
if isinstance(detail, str):
|
|||
|
|
detail = {}
|
|||
|
|
|
|||
|
|
author_info = detail.get("authorInfo") or {}
|
|||
|
|
|
|||
|
|
# 标准化: 构建 aweme_list 格式（图文解析函数依赖此结构）
|
|||
|
|
item = {
|
|||
|
|
"desc": detail.get("desc") or "",
|
|||
|
|
"images": detail.get("images") or [],
|
|||
|
|
"media_type": detail.get("awemeType") or 0,
|
|||
|
|
"author": {
|
|||
|
|
"nickname": author_info.get("nickname") or "",
|
|||
|
|
"uid": author_info.get("uid") or "",
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 标准化 urlList -> url_list, downloadUrlList -> download_url_list
|
|||
|
|
for img in item["images"]:
|
|||
|
|
if "urlList" in img and "url_list" not in img:
|
|||
|
|
img["url_list"] = img.pop("urlList")
|
|||
|
|
if "downloadUrlList" in img and "download_url_list" not in img:
|
|||
|
|
img["download_url_list"] = img.pop("downloadUrlList")
|
|||
|
|
# 动图的 video 字段
|
|||
|
|
video = img.get("video")
|
|||
|
|
if isinstance(video, dict):
|
|||
|
|
if "playAddr" in video and "play_addr" not in video:
|
|||
|
|
video["play_addr"] = video.pop("playAddr")
|
|||
|
|
|
|||
|
|
logger.info(f"SSR 页面解析成功: {len(item['images'])} 张图片, "
|
|||
|
|
f"media_type={item['media_type']}")
|
|||
|
|
|
|||
|
|
return {"aweme_list": [item]}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 文件名构建 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def build_file_name(
|
|||
|
|
raw_title: str,
|
|||
|
|
raw_nickname: str,
|
|||
|
|
media_type: str,
|
|||
|
|
) -> str:
|
|||
|
|
"""
|
|||
|
|
构建文件名 stem，格式: {作者}_{标题}_{类型}_{时间戳}
|
|||
|
|
|
|||
|
|
昵称不限长，标题最多 15 字符（slugify 后），末尾 HHMMSS 防覆盖。
|
|||
|
|
"""
|
|||
|
|
slug_nickname = slugify(raw_nickname)
|
|||
|
|
|
|||
|
|
if raw_title:
|
|||
|
|
slug_title = slugify(raw_title, max_length=15)
|
|||
|
|
else:
|
|||
|
|
slug_title = ""
|
|||
|
|
|
|||
|
|
if not slug_title:
|
|||
|
|
slug_title = datetime.now().strftime("%H%M%S")
|
|||
|
|
logger.info(f"标题为空，使用短时间戳: {slug_title}")
|
|||
|
|
|
|||
|
|
slug_type = slugify(media_type)
|
|||
|
|
time_suffix = datetime.now().strftime("%H%M%S")
|
|||
|
|
|
|||
|
|
return f"{slug_nickname}_{slug_title}_{slug_type}_{time_suffix}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 动图检测 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_animated_note(api_response: dict) -> bool:
|
|||
|
|
"""检测图文是否为动图（media_type == 42），动图每张图片内含无声 mp4 视频"""
|
|||
|
|
aweme_list = api_response.get("aweme_list") or []
|
|||
|
|
if aweme_list and isinstance(aweme_list, list):
|
|||
|
|
return aweme_list[0].get("media_type") == 42
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_animated_note_videos(api_response: dict) -> List[str]:
|
|||
|
|
"""从动图 API 响应解析视频链接（无音轨），返回最佳质量的视频 URL 列表"""
|
|||
|
|
video_urls = []
|
|||
|
|
try:
|
|||
|
|
aweme_list = api_response.get("aweme_list") or []
|
|||
|
|
if not aweme_list:
|
|||
|
|
raise DouyinFetchError("动图作品列表为空")
|
|||
|
|
|
|||
|
|
images = aweme_list[0].get("images") or []
|
|||
|
|
for image in images:
|
|||
|
|
video = image.get("video") or {}
|
|||
|
|
play_addr = video.get("play_addr") or {}
|
|||
|
|
url_list = play_addr.get("url_list") or []
|
|||
|
|
if url_list:
|
|||
|
|
video_urls.append(url_list[0])
|
|||
|
|
|
|||
|
|
logger.info(f"从动图 API 解析出 {len(video_urls)} 个视频")
|
|||
|
|
except DouyinFetchError:
|
|||
|
|
raise
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"解析动图视频链接失败: {e}")
|
|||
|
|
raise DouyinFetchError(f"解析动图视频链接失败: {e}")
|
|||
|
|
|
|||
|
|
if not video_urls:
|
|||
|
|
raise DouyinFetchError("未找到动图视频链接")
|
|||
|
|
return video_urls
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================= 一站式解析 =============================
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_douyin_response(
|
|||
|
|
api_response: dict, referer_url: str | None
|
|||
|
|
) -> ParsedDouyinContent:
|
|||
|
|
"""
|
|||
|
|
一站式解析: 从 API 响应 + referer URL 中提取所有元数据并构建文件名
|
|||
|
|
"""
|
|||
|
|
media_type = detect_media_type(referer_url)
|
|||
|
|
if not media_type:
|
|||
|
|
raise DouyinFetchError("媒体类型无法获取，请检查url是否正确")
|
|||
|
|
|
|||
|
|
raw_title = extract_title_from_api(api_response)
|
|||
|
|
raw_nickname = extract_author_nickname(api_response)
|
|||
|
|
file_name = build_file_name(raw_title, raw_nickname, media_type)
|
|||
|
|
|
|||
|
|
logger.info(
|
|||
|
|
f"内容标题: {raw_title}, 作者: {raw_nickname}, "
|
|||
|
|
f"类型: {media_type}, 文件名: {file_name}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
return ParsedDouyinContent(
|
|||
|
|
raw_title=raw_title,
|
|||
|
|
raw_nickname=raw_nickname,
|
|||
|
|
media_type=media_type,
|
|||
|
|
file_name=file_name,
|
|||
|
|
)
|