|
|
@@ -32,35 +32,90 @@ def resolve_share_url(url: str) -> str:
|
|
|
|
|
|
|
|
|
def extract_aweme_id(final_url: str) -> str:
|
|
|
+ # 抖音: /video/123 或 aweme_id=123
|
|
|
m = re.search(r"/(?:video|note)/(\d+)", final_url) or re.search(r"aweme_id=(\d+)", final_url)
|
|
|
- if not m:
|
|
|
- raise SystemExit(f"无法从 URL 提取视频ID: {final_url}")
|
|
|
- return m.group(1)
|
|
|
+ if m:
|
|
|
+ return m.group(1)
|
|
|
+ # 微信视频号: channels/weixin/v/XXX 或 finder.video.qq.com 页面里的 wxv_xxx
|
|
|
+ m = re.search(r"wxv_(?:[A-Za-z0-9_\-]{10,})", final_url)
|
|
|
+ if m:
|
|
|
+ return m.group(0)
|
|
|
+ # 兜底: 任意 20+ 位纯数字(部分分享链把视频 id 放 query)
|
|
|
+ m = re.search(r"(\d{18,25})", final_url)
|
|
|
+ if m:
|
|
|
+ return m.group(1)
|
|
|
+ raise SystemExit(f"无法从 URL 提取视频ID: {final_url}")
|
|
|
+
|
|
|
+
|
|
|
+def detect_platform(url: str) -> str:
|
|
|
+ """按 URL 形态自动判平台(用户只贴链接不用选)。"""
|
|
|
+ u = url.lower()
|
|
|
+ if "douyin.com" in u or "iesdouyin.com" in u:
|
|
|
+ return "douyin"
|
|
|
+ if "channels.weixin" in u or "finder.video.qq.com" in u or "weixin" in u and "wxv" in u:
|
|
|
+ return "weixin"
|
|
|
+ if "xiaohongshu" in u or "xhslink" in u:
|
|
|
+ return "xiaohongshu"
|
|
|
+ if "bilibili" in u or "b23.tv" in u:
|
|
|
+ return "bilibili"
|
|
|
+ return "douyin"
|
|
|
|
|
|
|
|
|
def fetch_meta(aweme_id: str, platform: str) -> dict:
|
|
|
- """VOC 网关取视频详情(抖音)。其他平台留 TODO。"""
|
|
|
- if platform != "douyin":
|
|
|
- print(f"[fetch] WARN 平台 {platform} 详情接口未接入, 仅下载/登记")
|
|
|
+ """VOC 网关取视频详情。douyin=fetch_one_video; weixin=视频号通道。"""
|
|
|
+ if platform == "douyin":
|
|
|
+ resp = voc_call("douyin/web/fetch_one_video", {"aweme_id": aweme_id})
|
|
|
+ if resp.get("error"):
|
|
|
+ raise SystemExit(f"[fetch] VOC fetch_one_video 失败: {resp['error']}")
|
|
|
+ return aweme_summary(extract_aweme_detail(resp))
|
|
|
+ if platform == "weixin":
|
|
|
+ # 视频号: 先试专用通道,404/不支持时降级为直连下载(纯拉片不需要 meta 也能跑)
|
|
|
+ for proxy_path, params in (
|
|
|
+ ("weixin/channels/fetch_one_video", {"video_id": aweme_id, "aweme_id": aweme_id}),
|
|
|
+ ("weixin/channels/fetch", {"id": aweme_id}),
|
|
|
+ ):
|
|
|
+ try:
|
|
|
+ resp = voc_call(proxy_path, params)
|
|
|
+ if resp and not resp.get("error"):
|
|
|
+ data = resp.get("data") or resp
|
|
|
+ summary = {
|
|
|
+ "video_id": aweme_id,
|
|
|
+ "platform": "weixin",
|
|
|
+ "title": (data.get("title") or data.get("desc") or "")[:120],
|
|
|
+ "author": data.get("nickname") or data.get("author") or "",
|
|
|
+ "play_urls": data.get("play_urls") or data.get("urls") or [],
|
|
|
+ "download_urls": data.get("download_urls") or data.get("video_url") or [],
|
|
|
+ "raw": {k: data.get(k) for k in ("duration", "cover_url") if data.get(k)},
|
|
|
+ }
|
|
|
+ print(f"[fetch] 视频号详情经 {proxy_path} 获取")
|
|
|
+ return summary
|
|
|
+ except SystemExit:
|
|
|
+ continue
|
|
|
+ except Exception as exc: # noqa: BLE001
|
|
|
+ print(f"[fetch] {proxy_path} 不可用: {str(exc)[:80]}")
|
|
|
+ print("[fetch] 视频号详情接口未命中, 降级直连模式(仅下载视频做拉片)")
|
|
|
return {}
|
|
|
- resp = voc_call("douyin/web/fetch_one_video", {"aweme_id": aweme_id})
|
|
|
- if resp.get("error"):
|
|
|
- raise SystemExit(f"[fetch] VOC fetch_one_video 失败: {resp['error']}")
|
|
|
- return aweme_summary(extract_aweme_detail(resp))
|
|
|
+ print(f"[fetch] WARN 平台 {platform} 详情接口未接入, 仅下载/登记")
|
|
|
+ return {}
|
|
|
|
|
|
|
|
|
def main():
|
|
|
ap = argparse.ArgumentParser(description="下载/登记待拉片视频")
|
|
|
- ap.add_argument("--url", help="抖音分享短链或含视频ID的页面URL")
|
|
|
- ap.add_argument("--aweme-id", help="视频ID(数字)")
|
|
|
+ ap.add_argument("--url", help="抖音/视频号分享短链或含视频ID的页面URL")
|
|
|
+ ap.add_argument("--aweme-id", help="视频ID")
|
|
|
ap.add_argument("--local", help="已有本地视频文件路径(跳过下载)")
|
|
|
- ap.add_argument("--platform", default="douyin",
|
|
|
- help="平台英文简称: douyin/xiaohongshu/weixin/bilibili/tiktok/shortdrama")
|
|
|
+ ap.add_argument("--platform", default="auto",
|
|
|
+ help="平台: auto(按URL自动判)/douyin/weixin/xiaohongshu/bilibili/tiktok/shortdrama")
|
|
|
ap.add_argument("--out-dir", required=True, help="工作目录(产物放这里)")
|
|
|
args = ap.parse_args()
|
|
|
|
|
|
os.makedirs(args.out_dir, exist_ok=True)
|
|
|
- platform = slugify_platform(args.platform)
|
|
|
+ if args.platform == "auto":
|
|
|
+ src = args.url or ""
|
|
|
+ platform = detect_platform(src) if src else "douyin"
|
|
|
+ print(f"[fetch] 自动识别平台: {platform}")
|
|
|
+ else:
|
|
|
+ platform = slugify_platform(args.platform)
|
|
|
meta, video_path = {}, os.path.join(args.out_dir, "video.mp4")
|
|
|
|
|
|
if args.local:
|
fmode