e1fbd31c64
- 高危词库从 process.py 抽到 config.toml [security] section - process.py 改为 _get_high_risk_words() 动态加载（配置优先） - 新增 15 个 QQ 操作脚本（已脱敏：QQ号/IP/Token → 占位符） - 新增 8 个 SKILL.md（qq-messenger/management/resolver/napcat-extras 等） - 新增 SKILL.md 入口（完整部署方案文档） - 保留 SDK 框架（plugin_modules.py, file_store_api.py, package.py 等）
260 lines
8.7 KiB
Python
260 lines
8.7 KiB
Python
#!/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()
|