Panniantong
74c3df5c3d
BREAKING: Complete architectural rewrite.
Before: Copied x-reader's fetcher code into readers/ (1205 lines of borrowed code)
After: Pluggable channel system where each channel is a thin wrapper (~50 lines)
around the best external tool for that platform. Zero copied code.
Architecture:
- channels/base.py — Universal Channel interface (read, search, check)
- channels/web.py — Jina Reader API (swappable)
- channels/github.py — GitHub API (swappable)
- channels/twitter.py — birdx + Jina fallback (swappable)
- channels/youtube.py — yt-dlp (swappable)
- channels/reddit.py — Reddit JSON API + proxy (swappable)
- channels/rss.py — feedparser (swappable)
- channels/bilibili.py — Bilibili API (swappable)
- channels/exa_search.py — Exa semantic search (swappable)
Key design: every backend can be swapped by changing ONE file.
YouTube dies? Change youtube.py. Exa sucks? Swap exa_search.py for Tavily.
Nothing else changes.
Removed: reader.py, schema.py, readers/, search/, utils/ (all x-reader code)
Tests: 36/36 passing
95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""YouTube — via yt-dlp (free, pip install yt-dlp).
|
|
|
|
Backend: yt-dlp (https://github.com/yt-dlp/yt-dlp)
|
|
Swap to: any YouTube subtitle extractor
|
|
"""
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse, parse_qs
|
|
from .base import Channel, ReadResult
|
|
|
|
|
|
class YouTubeChannel(Channel):
|
|
name = "youtube"
|
|
description = "YouTube video transcripts"
|
|
backends = ["yt-dlp"]
|
|
requires_tools = ["yt-dlp"]
|
|
tier = 0
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
domain = urlparse(url).netloc.lower()
|
|
return "youtube.com" in domain or "youtu.be" in domain
|
|
|
|
async def read(self, url: str, config=None) -> ReadResult:
|
|
if not shutil.which("yt-dlp"):
|
|
raise RuntimeError("yt-dlp not installed. Install: pip install yt-dlp")
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Get video info
|
|
info = self._get_info(url)
|
|
title = info.get("title", url)
|
|
author = info.get("uploader", "")
|
|
|
|
# Try to get subtitles
|
|
transcript = self._get_subtitles(url, tmpdir)
|
|
|
|
if not transcript:
|
|
transcript = f"[Video: {title}]\n[No subtitles available. Use Groq Whisper for transcription.]"
|
|
|
|
return ReadResult(
|
|
title=title,
|
|
content=transcript,
|
|
url=url,
|
|
author=author,
|
|
platform="youtube",
|
|
extra={
|
|
"duration": info.get("duration"),
|
|
"view_count": info.get("view_count"),
|
|
"upload_date": info.get("upload_date"),
|
|
},
|
|
)
|
|
|
|
def _get_info(self, url: str) -> dict:
|
|
try:
|
|
result = subprocess.run(
|
|
["yt-dlp", "--dump-json", "--no-download", url],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
if result.returncode == 0:
|
|
return json.loads(result.stdout)
|
|
except (subprocess.TimeoutExpired, json.JSONDecodeError):
|
|
pass
|
|
return {}
|
|
|
|
def _get_subtitles(self, url: str, tmpdir: str) -> str:
|
|
"""Extract subtitles using yt-dlp."""
|
|
try:
|
|
subprocess.run(
|
|
["yt-dlp", "--write-auto-sub", "--write-sub",
|
|
"--sub-lang", "en,zh-Hans,zh",
|
|
"--skip-download", "--sub-format", "vtt",
|
|
"-o", f"{tmpdir}/%(id)s.%(ext)s", url],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
|
|
# Find and read subtitle file
|
|
for f in Path(tmpdir).glob("*.vtt"):
|
|
text = f.read_text(errors="replace")
|
|
# Strip VTT headers and timestamps
|
|
lines = []
|
|
for line in text.split("\n"):
|
|
line = line.strip()
|
|
if not line or line.startswith("WEBVTT") or "-->" in line or line.isdigit():
|
|
continue
|
|
if line not in lines[-1:]: # deduplicate
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
return ""
|