#!/usr/bin/env python3 """ 视频链接解析与下载工具 — 基于 you-get 场景： AI 在聊天中检测到视频分享链接（B站等），调用此脚本下载到本地， 然后用 qq_upload_group_file.py 发送到对应群/私聊。 用法： # 查看视频信息（不下载） python3 qq_video_download.py --url "https://www.bilibili.com/video/BVxxxx" --info # 下载视频（自动选择最佳可用画质） python3 qq_video_download.py --url "https://www.bilibili.com/video/BVxxxx" # 指定画质下载 python3 qq_video_download.py --url "..." --format dash-flv480-AVC # JSON 输出（供 AI 解析） python3 qq_video_download.py --url "..." --info --json """ import json import sys import os import re import subprocess import argparse import requests from urllib.parse import urlparse FILE_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_OUTPUT = os.path.join(FILE_DIR, "..", "files", "videos") os.makedirs(DEFAULT_OUTPUT, exist_ok=True) # B站短链接域名 BILIBILI_SHORT_DOMAINS = ["b23.tv", "bili22.cn", "bili33.cn"] def normalize_url(url: str) -> str: """归一化视频链接：自动解析短链接到标准地址""" url = url.strip().strip('"').strip("'") # 短链接才需要解析，标准 URL 跳过 parsed = urlparse(url) if parsed.netloc not in BILIBILI_SHORT_DOMAINS: return url try: resp = requests.head(url, allow_redirects=True, timeout=10, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" }) final_url = resp.url if final_url and final_url != url: return final_url except Exception: pass return url # 安全文件名：去掉不安全的字符 def safe_filename(name: str) -> str: name = re.sub(r'[<>:"/\\|?*]', '_', name) name = re.sub(r'\s+', ' ', name).strip() return name or "video" def run_you_get(args: list, timeout=120) -> dict: """运行 you-get 并返回结构化结果""" cmd = ["you-get"] + args try: proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) except subprocess.TimeoutExpired: return {"ok": False, "error": "下载超时，视频太大或网络太慢"} stdout = proc.stdout or "" stderr = proc.stderr or "" exit_code = proc.returncode if exit_code != 0: # 有些 you-get 非零退出但实际成功了，检查输出 error_msg = stderr.strip() or stdout.strip() or f"you-get 退出码 {exit_code}" return {"ok": False, "error": error_msg} return {"ok": True, "exit_code": exit_code, "stdout": stdout, "stderr": stderr} def parse_info_json(raw_json: str) -> dict: """解析 you-get --json 输出""" try: data = json.loads(raw_json) except json.JSONDecodeError: return None info = { "site": data.get("site", ""), "title": data.get("title", ""), "url": data.get("url", ""), "streams": [] } for fmt_id, stream in data.get("streams", {}).items(): info["streams"].append({ "id": fmt_id, "container": stream.get("container", ""), "quality": stream.get("quality", ""), "size": stream.get("size", 0), "size_human": f"{stream.get('size', 0) / 1024 / 1024:.1f} MB" if stream.get("size", 0) > 0 else "未知" }) return info def cmd_info(url: str, json_output: bool) -> dict: """获取视频信息""" url = normalize_url(url) result = run_you_get(["--json", url], timeout=30) if not result["ok"]: return result info = parse_info_json(result["stdout"]) if not info: return {"ok": False, "error": "无法解析视频信息"} result["info"] = info if not json_output: # 人类可读格式 lines = [f"🎬 {info['title']}", f" 来源: {info['site']}", f" 链接: {info['url']}", ""] for s in info["streams"]: lines.append(f" [{s['id']}] {s['quality']} | {s['container']} | {s['size_human']}") result["text"] = "\n".join(lines) return result def _pick_best_avc_stream(info: dict) -> str | None: """从可用流中挑选最佳的 AVC (H.264) 格式，确保 QQ 客户端可播放""" avc_streams = [s for s in info.get("streams", []) if "AVC" in s.get("id", "")] if not avc_streams: return None # 按质量降序（480 比 360 高），取第一个 avc_streams.sort(key=lambda s: s.get("id", ""), reverse=True) return avc_streams[0]["id"] def cmd_download(url: str, output_dir: str, filename: str, fmt: str, no_merge: bool, no_caption: bool, json_output: bool) -> dict: """下载视频""" url = normalize_url(url) # 未指定格式时，自动选最佳的 AVC (H.264) 流（QQ 播放器不支持 AV1/HEVC） if not fmt: info_result = cmd_info(url, json_output=True) if info_result.get("ok"): best = _pick_best_avc_stream(info_result["info"]) if best: fmt = best else: # 没有 AVC 流，用第一个 streams = info_result["info"].get("streams", []) if streams: fmt = streams[0]["id"] args = ["--output-dir", output_dir, "--force"] if no_merge: args.append("--no-merge") if no_caption: args.append("--no-caption") if filename: args.extend(["--output-filename", filename]) if fmt: args.extend(["--format", fmt]) args.append(url) result = run_you_get(args, timeout=300) # 5 分钟超时 if not result["ok"]: return result # 解析下载后的文件 stderr = result.get("stderr", "") stdout = result.get("stdout", "") # you-get 会在 stdout/stderr 输出 Merged into xxx.mp4 merged_match = re.search(r'Merged into (.+\.mp4)', stderr or stdout) if merged_match: final_path = os.path.join(output_dir, merged_match.group(1)) else: # 查找输出目录中最新添加的 mp4 mp4_files = sorted( [f for f in os.listdir(output_dir) if f.endswith(".mp4")], key=lambda f: os.path.getmtime(os.path.join(output_dir, f)), reverse=True ) final_path = os.path.join(output_dir, mp4_files[0]) if mp4_files else "" file_size = os.path.getsize(final_path) if final_path and os.path.exists(final_path) else 0 result["file"] = { "path": final_path, "filename": os.path.basename(final_path) if final_path else "", "size": file_size, "size_human": f"{file_size / 1024 / 1024:.1f} MB" if file_size > 0 else "未知" } if not json_output: lines = [f"✅ 下载完成: {result['file']['filename']}"] lines.append(f" 大小: {result['file']['size_human']}") lines.append(f" 路径: {result['file']['path']}") result["text"] = "\n".join(lines) return result def main(): parser = argparse.ArgumentParser(description="视频链接解析与下载") parser.add_argument("--url", type=str, required=True, help="视频分享链接") parser.add_argument("--info", action="store_true", help="仅查看信息，不下载") parser.add_argument("--output", type=str, default=DEFAULT_OUTPUT, help="下载目录") parser.add_argument("--name", type=str, help="输出文件名（不含扩展名）") parser.add_argument("--format", type=str, help="画质格式 ID（如 dash-flv480-AVC）") parser.add_argument("--json", action="store_true", help="JSON 格式输出") parser.add_argument("--no-merge", action="store_true", help="不合并 DASH 流") parser.add_argument("--no-caption", action="store_true", help="不下字幕/弹幕") args = parser.parse_args() try: if args.info: result = cmd_info(args.url, args.json) else: result = cmd_download(args.url, args.output, args.name, args.format, args.no_merge, args.no_caption, args.json) if args.json: # JSON 输出中去掉 stdout/stderr 这种大字段 output = {k: v for k, v in result.items() if k not in ("stdout", "stderr", "text")} print(json.dumps(output, ensure_ascii=False, indent=2)) else: if result.get("text"): print(result["text"]) elif not result.get("ok"): print(f"❌ {result.get('error', '未知错误')}") else: print(result.get("text", "✅ 完成")) sys.exit(0 if result.get("ok") else 1) except Exception as e: if args.json: print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False)) else: print(f"❌ 错误: {e}") sys.exit(1) if __name__ == "__main__": main()