freedakgmail
237 lines
8.8 KiB
Python
237 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
"""视频号批量下载管理器。
|
||
|
||
基于 wx_channels_download 工具的 API 接口，实现：
|
||
1. 检查工具是否运行
|
||
2. 获取视频号创作者的视频列表
|
||
3. 与本地已下载文件比对，发现新视频
|
||
4. 触发批量下载
|
||
5. 下载完成后自动触发 ASR 转写入库
|
||
|
||
工具 GitHub: https://github.com/ltaoo/wx_channels_download
|
||
API 基地址: http://127.0.0.1:2022
|
||
|
||
使用前提：
|
||
- wx_channels_download 以 sudo 运行
|
||
- IPv6 已禁用（sudo networksetup -setv6off Wi-Fi）
|
||
- VPN 已关闭
|
||
- 微信保持运行
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
load_dotenv(BASE_DIR / ".env")
|
||
|
||
TOOL_API = os.getenv("WX_VIDEO_API", "http://127.0.0.1:2022")
|
||
VIDEO_CONFIG_PATH = BASE_DIR / "video_config.json"
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
||
|
||
|
||
def api_get(endpoint: str, params: dict[str, str] | None = None, timeout: int = 30) -> dict[str, Any]:
|
||
"""调用 wx_channels_download 工具的 API。"""
|
||
url = f"{TOOL_API}{endpoint}"
|
||
if params:
|
||
qs = "&".join(f"{k}={v}" for k, v in params.items())
|
||
url += f"?{qs}"
|
||
req = urllib.request.Request(url, method="GET")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as response:
|
||
return json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", errors="replace")[:500]
|
||
raise RuntimeError(f"工具 API 返回 {exc.code}：{detail}") from exc
|
||
except urllib.error.URLError as exc:
|
||
raise RuntimeError(f"无法连接工具 API（{TOOL_API}）：{exc.reason}\n请确认 wx_channels_download 已以 sudo 启动。") from exc
|
||
|
||
|
||
def check_tool_running() -> bool:
|
||
"""检查下载工具是否在运行。"""
|
||
try:
|
||
result = api_get("/api/channels/version", timeout=5)
|
||
return bool(result.get("version") or result.get("data"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def get_video_list(finder: str) -> list[dict[str, Any]]:
|
||
"""获取视频号创作者的视频列表。
|
||
|
||
需要微信客户端已连接到工具（WebSocket）。
|
||
"""
|
||
result = api_get("/api/channels/contact/feed/list", {"finder": finder}, timeout=60)
|
||
# 适配不同版本的返回格式
|
||
if isinstance(result, list):
|
||
return result
|
||
if isinstance(result, dict):
|
||
return result.get("data", result.get("list", result.get("feeds", [])))
|
||
return []
|
||
|
||
|
||
def list_local_videos(directory: Path) -> set[str]:
|
||
"""列出本地已下载的视频文件名（不含扩展名）。"""
|
||
if not directory.is_dir():
|
||
return set()
|
||
suffixes = {".mp4", ".mov", ".m4v", ".mkv", ".webm"}
|
||
return {f.stem for f in directory.iterdir() if f.is_file() and f.suffix.lower() in suffixes}
|
||
|
||
|
||
def normalize_title(title: str) -> str:
|
||
"""将视频标题规范化，用于与本地文件名比对。"""
|
||
# 去除 wx_channels_download 添加的 _xWT 后缀和画质标记
|
||
value = re.sub(r"[_ ]?xWT\d+$", "", title, flags=re.I)
|
||
value = re.sub(r"[\x00-\x1f/:]", "_", value).strip(" .")
|
||
return value[:180] or "untitled"
|
||
|
||
|
||
def find_new_videos(video_list: list[dict[str, Any]], local_dir: Path) -> list[dict[str, Any]]:
|
||
"""比对线上视频列表与本地已下载文件，返回未下载的视频。"""
|
||
local_names = list_local_videos(local_dir)
|
||
new_videos = []
|
||
for video in video_list:
|
||
title = video.get("title", video.get("desc", ""))
|
||
normalized = normalize_title(title)
|
||
# 检查本地是否已有该视频（模糊匹配）
|
||
if not any(normalized in local_name or local_name in normalized for local_name in local_names):
|
||
new_videos.append(video)
|
||
return new_videos
|
||
|
||
|
||
def create_batch_download(finder: str, video_ids: list[str]) -> dict[str, Any]:
|
||
"""创建批量下载任务。"""
|
||
params = {"finder": finder, "ids": ",".join(video_ids)}
|
||
return api_get("/api/task/create_batch", params, timeout=120)
|
||
|
||
|
||
def wait_for_downloads(local_dir: Path, expected_count: int, timeout: int = 600) -> int:
|
||
"""等待下载完成，返回实际新增的文件数。"""
|
||
initial = list_local_videos(local_dir)
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
time.sleep(10)
|
||
current = list_local_videos(local_dir)
|
||
new_files = current - initial
|
||
if len(new_files) >= expected_count:
|
||
return len(new_files)
|
||
return len(list_local_videos(local_dir) - initial)
|
||
|
||
|
||
def run_download(auto_transcribe: bool = True) -> int:
|
||
"""主流程：检查工具 → 获取视频列表 → 发现新视频 → 触发下载。"""
|
||
config = json.loads(VIDEO_CONFIG_PATH.read_text(encoding="utf-8")) if VIDEO_CONFIG_PATH.exists() else {"sources": []}
|
||
sources = config.get("sources", [])
|
||
|
||
if not sources:
|
||
print(json.dumps({"status": "skipped", "reason": "video_config.json 中未配置视频源"}, ensure_ascii=False))
|
||
return 0
|
||
|
||
# 1. 检查工具是否运行
|
||
if not check_tool_running():
|
||
print(json.dumps({
|
||
"status": "error",
|
||
"reason": f"wx_channels_download 工具未运行（{TOOL_API}）",
|
||
"hint": "请执行: sudo <工具路径>/wx_video_download，并确保 IPv6 已禁用、VPN 已关闭"
|
||
}, ensure_ascii=False))
|
||
return 1
|
||
|
||
print(json.dumps({"status": "tool_running", "api": TOOL_API}, ensure_ascii=False), flush=True)
|
||
|
||
total_new = 0
|
||
total_downloaded = 0
|
||
|
||
for source in sources:
|
||
account = source.get("account", "未知")
|
||
finder = source.get("finder", "")
|
||
download_dir = Path(source.get("directory", ""))
|
||
|
||
if not finder:
|
||
print(f"SKIP {account}: 未配置 finder", file=sys.stderr)
|
||
continue
|
||
|
||
# 2. 获取视频列表
|
||
try:
|
||
video_list = get_video_list(finder)
|
||
except Exception as exc:
|
||
print(f"FAIL {account}: 获取视频列表失败 - {exc}", file=sys.stderr, flush=True)
|
||
continue
|
||
|
||
print(f"INFO {account}: 线上视频 {len(video_list)} 个", flush=True)
|
||
|
||
# 3. 发现新视频
|
||
new_videos = find_new_videos(video_list, download_dir)
|
||
total_new += len(new_videos)
|
||
|
||
if not new_videos:
|
||
print(f"INFO {account}: 无新视频", flush=True)
|
||
continue
|
||
|
||
print(f"INFO {account}: 发现 {len(new_videos)} 个新视频", flush=True)
|
||
|
||
# 4. 触发批量下载
|
||
video_ids = [v.get("id", v.get("objectId", "")) for v in new_videos if v.get("id") or v.get("objectId")]
|
||
if video_ids:
|
||
try:
|
||
result = create_batch_download(finder, video_ids)
|
||
print(f"INFO {account}: 批量下载任务已创建 - {result}", flush=True)
|
||
# 5. 等待下载完成
|
||
downloaded = wait_for_downloads(download_dir, len(new_videos), timeout=600)
|
||
total_downloaded += downloaded
|
||
print(f"INFO {account}: 新增下载 {downloaded} 个文件", flush=True)
|
||
except Exception as exc:
|
||
print(f"FAIL {account}: 下载失败 - {exc}", file=sys.stderr, flush=True)
|
||
else:
|
||
# 如果无法获取 video_id，提示用户手动下载
|
||
print(f"WARN {account}: 无法自动获取视频ID，请通过微信内批量下载或 Web 管理页面手动下载", flush=True)
|
||
print(f" Web 管理页面: {TOOL_API}/download", flush=True)
|
||
print(f" 下载目录: {download_dir}", flush=True)
|
||
|
||
print(json.dumps({
|
||
"status": "success",
|
||
"total_new": total_new,
|
||
"total_downloaded": total_downloaded,
|
||
}, ensure_ascii=False), flush=True)
|
||
|
||
# 6. 自动触发 ASR 转写入库
|
||
if auto_transcribe and total_downloaded > 0:
|
||
print("INFO 自动触发视频号 ASR 转写...", flush=True)
|
||
import subprocess
|
||
subprocess.run(
|
||
[sys.executable, str(BASE_DIR / "crawler.py"), "--source-type", "视频号"],
|
||
cwd=str(BASE_DIR),
|
||
)
|
||
|
||
return 0
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="视频号批量下载管理器")
|
||
parser.add_argument("--no-transcribe", action="store_true", help="仅下载，不自动触发 ASR 转写")
|
||
parser.add_argument("--check", action="store_true", help="仅检查工具状态")
|
||
return parser
|
||
|
||
|
||
if __name__ == "__main__":
|
||
args = build_parser().parse_args()
|
||
if args.check:
|
||
running = check_tool_running()
|
||
print(json.dumps({"tool_running": running, "api": TOOL_API}, ensure_ascii=False))
|
||
raise SystemExit(0 if running else 1)
|
||
raise SystemExit(run_download(auto_transcribe=not args.no_transcribe))
|