scripts/monitor_youtube_playlist.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Monitor one or more YouTube playlists and prepare transcript packages for agent/XKB card generation.
This script intentionally does NOT call any LLM. It only:
1. loads playlist configs
2. lists playlist videos via yt-dlp
3. compares against existing youtube cards/raw packages
4. downloads subtitles for new videos
5. writes raw transcript JSON packages under memory/x-knowledge-base/youtube-raw/
Card generation is handled by OpenClaw cron agent, not by a direct model provider.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import parse_qs, urlparse
sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths
WORKSPACE = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
YOUTUBE_CARD_DIR = BOOKMARKS_DIR / "youtube"
RAW_DIR = WORKSPACE / "memory" / "x-knowledge-base" / "youtube-raw"
SKILL_DIR = Path(__file__).resolve().parent.parent
DEFAULT_PLAYLISTS_FILE = SKILL_DIR / "config" / "youtube-playlists.json"
COOKIES_FILE = Path.home() / ".config" / "yt-dlp" / "cookies.txt"
def yt_cmd(extra: list[str]) -> list[str]:
base = [
sys.executable,
"-m",
"yt_dlp",
"--js-runtimes",
"node",
"--remote-components",
"ejs:github",
]
if COOKIES_FILE.exists():
base += ["--cookies", str(COOKIES_FILE)]
return base + extra
def playlist_slug(url: str, fallback: str = "playlist") -> str:
try:
parsed = urlparse(url)
list_id = parse_qs(parsed.query).get("list", [""])[0]
if list_id:
return re.sub(r"[^A-Za-z0-9_-]+", "-", list_id).strip("-")[:80]
except Exception:
pass
return re.sub(r"[^A-Za-z0-9_-]+", "-", fallback).strip("-")[:80] or "playlist"
def normalize_playlist(raw: dict | str, idx: int = 0) -> dict:
if isinstance(raw, str):
raw = {"url": raw}
url = str(raw.get("url") or raw.get("playlist_url") or "").strip()
if not url:
raise ValueError(f"Playlist config #{idx + 1} missing url")
pid = str(raw.get("id") or raw.get("slug") or playlist_slug(url, f"playlist-{idx + 1}"))
return {
"id": pid,
"url": url,
"title": str(raw.get("title") or pid),
"category": str(raw.get("category") or "ai-tools"),
"tags": [str(t).strip() for t in raw.get("tags", []) if str(t).strip()],
"notes": str(raw.get("notes") or ""),
}
def load_playlists(args: argparse.Namespace) -> list[dict]:
if args.playlist:
return [
normalize_playlist(
{
"id": args.playlist_id or playlist_slug(args.playlist),
"url": args.playlist,
"title": args.playlist_title or args.playlist_id or playlist_slug(args.playlist),
"category": args.category,
"tags": args.tags or [],
}
)
]
if os.getenv("YOUTUBE_PLAYLISTS_JSON"):
data = json.loads(os.environ["YOUTUBE_PLAYLISTS_JSON"])
if isinstance(data, dict) and "playlists" in data:
data = data["playlists"]
if not isinstance(data, list):
raise ValueError("YOUTUBE_PLAYLISTS_JSON must be a list or {playlists: [...]}")
return [normalize_playlist(item, idx) for idx, item in enumerate(data)]
playlists_file = Path(os.getenv("YOUTUBE_PLAYLISTS_FILE", str(DEFAULT_PLAYLISTS_FILE)))
if playlists_file.exists():
data = json.loads(playlists_file.read_text(encoding="utf-8"))
if isinstance(data, dict):
data = data.get("playlists", [])
if not isinstance(data, list):
raise ValueError(f"{playlists_file} must contain a list or {{playlists: [...]}}")
return [normalize_playlist(item, idx) for idx, item in enumerate(data)]
if os.getenv("YOUTUBE_PLAYLIST_URL"):
return [normalize_playlist(os.environ["YOUTUBE_PLAYLIST_URL"])]
return []
def playlist_videos(playlist_url: str) -> list[dict]:
cmd = yt_cmd([
"--flat-playlist",
"--print",
"%(id)s\t%(title)s\t%(duration)s\t%(channel)s",
"--quiet",
playlist_url,
])
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or "yt-dlp playlist listing failed")
videos = []
for line in result.stdout.strip().splitlines():
parts = line.split("\t")
if len(parts) < 2:
continue
vid = parts[0].strip()
title = parts[1].strip()
duration = int(parts[2]) if len(parts) > 2 and parts[2].strip().isdigit() else 0
channel = parts[3].strip() if len(parts) > 3 else ""
if vid:
videos.append({"id": vid, "title": title, "duration": duration, "channel": channel})
return videos
def parse_vtt(vtt_path: Path) -> str:
content = vtt_path.read_text(encoding="utf-8")
seen: set[str] = set()
texts: list[str] = []
for line in content.splitlines():
line = line.strip()
if (
not line
or "-->" in line
or line.startswith(("WEBVTT", "Kind:", "Language:"))
or re.match(r"^\d+$", line)
):
continue
line = re.sub(r"<[^>]+>", "", line)
line = re.sub(r"\s+", " ", line).strip()
if line and line not in seen:
seen.add(line)
texts.append(line)
return " ".join(texts)
def download_subtitle(video_id: str, tmp_dir: Path) -> tuple[str, str]:
cmd = yt_cmd([
"--write-auto-subs",
"--sub-langs",
"zh-Hans,zh-TW,en",
"--skip-download",
"--sub-format",
"vtt",
"--no-progress",
"--quiet",
"-o",
str(tmp_dir / "%(id)s"),
f"https://www.youtube.com/watch?v={video_id}",
])
subprocess.run(cmd, capture_output=True, text=True, timeout=180)
for lang in ["zh-Hans", "zh-TW", "en"]:
vtt = tmp_dir / f"{video_id}.{lang}.vtt"
if vtt.exists():
return parse_vtt(vtt), lang
return "", ""
def existing_card_ids() -> set[str]:
if not YOUTUBE_CARD_DIR.exists():
return set()
return {p.stem for p in YOUTUBE_CARD_DIR.glob("*.md")}
def existing_raw_ids() -> set[str]:
if not RAW_DIR.exists():
return set()
return {p.stem for p in RAW_DIR.glob("*.json")}
def package_video(video: dict, playlist: dict, tmp_dir: Path) -> tuple[dict | None, dict | None]:
transcript, lang = download_subtitle(video["id"], tmp_dir)
base = {
"id": video["id"],
"title": video["title"],
"channel": video.get("channel") or "YouTube",
"duration": video.get("duration", 0),
"source_url": f"https://www.youtube.com/watch?v={video['id']}",
"playlist_id": playlist["id"],
"playlist_title": playlist["title"],
"playlist_url": playlist["url"],
"category": playlist["category"],
"playlist_tags": playlist["tags"],
"playlist_notes": playlist.get("notes", ""),
"created_at": datetime.now(timezone.utc).isoformat(),
"generator": "apan2-cron-agent",
}
out = RAW_DIR / f"{video['id']}.json"
if not transcript:
marker = {**base, "status": "no_subtitles"}
out.write_text(json.dumps(marker, ensure_ascii=False, indent=2), encoding="utf-8")
return None, {**video, "playlist_id": playlist["id"], "reason": "no_subtitles", "raw_path": str(out)}
package = {
**base,
"lang": lang,
"transcript_chars": len(transcript),
"transcript": transcript,
"status": "pending_card_generation",
}
out.write_text(json.dumps(package, ensure_ascii=False, indent=2), encoding="utf-8")
return {**video, "playlist_id": playlist["id"], "lang": lang, "transcript_chars": len(transcript), "raw_path": str(out)}, None
def main() -> int:
parser = argparse.ArgumentParser(description="Monitor YouTube playlists and prepare transcript packages")
parser.add_argument("--playlist", default="", help="Single YouTube playlist URL override")
parser.add_argument("--playlist-id", default="", help="id/slug for single --playlist")
parser.add_argument("--playlist-title", default="", help="title for single --playlist")
parser.add_argument("--category", default="ai-tools", help="default category for single --playlist")
parser.add_argument("--tag", dest="tags", action="append", default=[], help="default tag for single --playlist; repeatable")
parser.add_argument("--limit", type=int, default=0, help="max new videos to package across all playlists; 0 = all")
parser.add_argument("--json", action="store_true", help="print machine-readable JSON summary")
args = parser.parse_args()
playlists = load_playlists(args)
if not playlists:
print("No YouTube playlists configured. Set config/youtube-playlists.json, YOUTUBE_PLAYLISTS_JSON, YOUTUBE_PLAYLISTS_FILE, YOUTUBE_PLAYLIST_URL, or --playlist.", file=sys.stderr)
return 2
RAW_DIR.mkdir(parents=True, exist_ok=True)
YOUTUBE_CARD_DIR.mkdir(parents=True, exist_ok=True)
done = existing_card_ids()
already_raw = existing_raw_ids()
packaged: list[dict] = []
skipped: list[dict] = []
playlist_summaries: list[dict] = []
remaining = args.limit if args.limit else None
with tempfile.TemporaryDirectory() as tmp:
tmp_dir = Path(tmp)
for playlist in playlists:
videos = playlist_videos(playlist["url"])
candidates = [v for v in videos if v["id"] not in done and v["id"] not in already_raw]
if remaining is not None:
candidates = candidates[:remaining]
p_packaged = 0
p_skipped = 0
for video in candidates:
item, skip = package_video(video, playlist, tmp_dir)
already_raw.add(video["id"])
if item:
packaged.append(item)
p_packaged += 1
if skip:
skipped.append(skip)
p_skipped += 1
if remaining is not None:
remaining -= 1
if remaining <= 0:
break
playlist_summaries.append({
"id": playlist["id"],
"title": playlist["title"],
"url": playlist["url"],
"playlist_total": len(videos),
"new_candidates": len(candidates),
"packaged": p_packaged,
"skipped": p_skipped,
"category": playlist["category"],
"tags": playlist["tags"],
})
if remaining is not None and remaining <= 0:
break
summary = {
"playlist_count": len(playlists),
"playlists": playlist_summaries,
"existing_cards": len(done),
"existing_raw": len(already_raw),
"packaged": packaged,
"skipped": skipped,
"raw_dir": str(RAW_DIR),
}
if args.json:
print(json.dumps(summary, ensure_ascii=False, indent=2))
else:
print(f"Playlists: {len(playlists)}")
print(f"Existing YouTube cards: {len(done)}")
print(f"Packaged new transcripts: {len(packaged)}")
for ps in playlist_summaries:
print(f"[{ps['id']}] total={ps['playlist_total']} candidates={ps['new_candidates']} packaged={ps['packaged']} skipped={ps['skipped']}")
for item in packaged:
print(f"+ [{item['playlist_id']}] {item['id']} {item['title']} ({item['lang']}, {item['transcript_chars']} chars)")
for item in skipped:
print(f"- [{item['playlist_id']}] {item['id']} {item['title']} skipped: {item['reason']}")
return 0
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
raise SystemExit(main())