feat: use yt-dlp as primary transcript fetcher, fallback to youtube-transcript-api
- Add _ytdlp_transcript() — fetch transcripts via yt-dlp Python API (supports json3/srv3/vtt/srt subtitle formats, cookies.txt auth) - Add _parse_subtitles() — parse subtitle formats into plain text - Keep _yttranscript_api_transcript() as fallback - download_transcript() tries yt-dlp first, then youtube-transcript-api - Add yt-dlp to requirements.txt - Update tests to mock both paths (all 23 pass) - Document Google Data API v3 caption limitations in architecture.md (captions.list works, captions.download=403 for non-owned videos)
This commit is contained in:
Gan, Jimmy
@@ -1,6 +1,7 @@
|
||||
fastapi>=0.100.0
|
||||
uvicorn>=0.20.0
|
||||
youtube-transcript-api>=0.6.0
|
||||
yt-dlp>=2025.0.0
|
||||
aiosqlite>=0.18.0
|
||||
httpx>=0.24.0
|
||||
pytest>=7.0.0
|
||||
|
||||
+105
-5
@@ -1,5 +1,6 @@
|
||||
import os
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import httpx
|
||||
import aiosqlite
|
||||
@@ -41,10 +42,98 @@ for p in [os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__)
|
||||
COOKIES_PATH = p
|
||||
break
|
||||
|
||||
def download_transcript(video_id: str) -> str:
|
||||
def _ytdlp_transcript(video_id: str) -> str | None:
|
||||
"""Fetch transcript using yt-dlp Python API (preferred)."""
|
||||
try:
|
||||
from yt_dlp import YoutubeDL
|
||||
|
||||
ydl_opts = {
|
||||
"skip_download": True,
|
||||
"writesubtitles": True,
|
||||
"writeautomaticsub": True,
|
||||
"subtitleslangs": ["en", "zh-Hans", "zh-Hant", "zh", "ja"],
|
||||
"subtitlesformat": "json3",
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
}
|
||||
|
||||
proxy = proxy_from_env()
|
||||
if proxy:
|
||||
ydl_opts["proxy"] = proxy
|
||||
|
||||
if COOKIES_PATH and os.path.exists(COOKIES_PATH):
|
||||
ydl_opts["cookiefile"] = COOKIES_PATH
|
||||
|
||||
with YoutubeDL(ydl_opts) as ydl:
|
||||
info = ydl.extract_info(f"https://www.youtube.com/watch?v={video_id}", download=False)
|
||||
|
||||
# Try requested subtitles first (manually created), then auto-generated
|
||||
subs = info.get("subtitles", {}) or {}
|
||||
for lang in ["en", "zh-Hans", "zh-Hant", "zh", "ja"]:
|
||||
if lang in subs:
|
||||
sub_data = subs[lang]
|
||||
for fmt in ["json3", "srv3", "srv2", "vtt", "srt"]:
|
||||
if fmt in sub_data:
|
||||
url = sub_data[fmt][0]["url"]
|
||||
resp = requests.get(url, timeout=30)
|
||||
if resp.status_code == 200:
|
||||
return _parse_subtitles(resp.text, fmt)
|
||||
|
||||
# Try auto-generated captions (requested_subs)
|
||||
req_subs = info.get("requested_subtitles", {}) or {}
|
||||
for lang in ["en", "zh-Hans", "zh-Hant", "zh", "ja"]:
|
||||
if lang in req_subs:
|
||||
sub_data = req_subs[lang]
|
||||
for fmt in ["json3", "srv3", "srv2", "vtt", "srt"]:
|
||||
if fmt in sub_data:
|
||||
url = sub_data[fmt][0]["url"]
|
||||
resp = requests.get(url, timeout=30)
|
||||
if resp.status_code == 200:
|
||||
return _parse_subtitles(resp.text, fmt)
|
||||
|
||||
return None
|
||||
except ImportError:
|
||||
log.debug("yt-dlp not installed, skipping")
|
||||
return None
|
||||
except Exception as e:
|
||||
log.warning(f"yt-dlp transcript failed for {video_id}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _parse_subtitles(text: str, fmt: str) -> str:
|
||||
"""Parse subtitle text from various formats into plain text."""
|
||||
if fmt == "json3":
|
||||
try:
|
||||
data = json.loads(text)
|
||||
events = data.get("events", [])
|
||||
texts = []
|
||||
for ev in events:
|
||||
segs = ev.get("segs", [])
|
||||
for seg in segs:
|
||||
txt = seg.get("utf8", "")
|
||||
if txt.strip():
|
||||
texts.append(txt.strip())
|
||||
return " ".join(texts)
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
pass
|
||||
# Fallback: strip HTML-like tags from VTT/SRT
|
||||
import re
|
||||
# Remove timestamps and position markers
|
||||
text = re.sub(r"^\d+:\d+:\d+[.,]\d+\s*-->\s*\d+:\d+:\d+[.,]\d+.*$", "", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"^\d+\s*$", "", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"<[^>]+>", "", text)
|
||||
text = re.sub(r" ", " ", text)
|
||||
text = re.sub(r"&", "&", text)
|
||||
text = re.sub(r"<", "<", text)
|
||||
text = re.sub(r">", ">", text)
|
||||
lines = [l.strip() for l in text.split("\n") if l.strip()]
|
||||
return " ".join(lines)
|
||||
|
||||
|
||||
def _yttranscript_api_transcript(video_id: str) -> str | None:
|
||||
"""Fallback transcript fetcher using youtube-transcript-api."""
|
||||
try:
|
||||
session = requests.Session()
|
||||
# Respect HTTP_PROXY/HTTPS_PROXY env vars
|
||||
proxy = proxy_from_env()
|
||||
if proxy:
|
||||
session.proxies = {"http": proxy, "https": proxy}
|
||||
@@ -52,7 +141,7 @@ def download_transcript(video_id: str) -> str:
|
||||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
||||
"Accept-Language": "en-US,en;q=0.9"
|
||||
})
|
||||
if COOKIES_PATH:
|
||||
if COOKIES_PATH and os.path.exists(COOKIES_PATH):
|
||||
from http.cookiejar import MozillaCookieJar
|
||||
cj = MozillaCookieJar(COOKIES_PATH)
|
||||
cj.load(ignore_discard=True, ignore_expires=True)
|
||||
@@ -68,9 +157,17 @@ def download_transcript(video_id: str) -> str:
|
||||
parts = transcript.fetch()
|
||||
return " ".join([part["text"] for part in parts])
|
||||
except Exception as e:
|
||||
log.warning(f"Failed to fetch transcript for {video_id}: {e}")
|
||||
log.warning(f"youtube-transcript-api failed for {video_id}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def download_transcript(video_id: str) -> str | None:
|
||||
"""Fetch transcript using yt-dlp first, fall back to youtube-transcript-api."""
|
||||
transcript = _ytdlp_transcript(video_id)
|
||||
if transcript:
|
||||
return transcript
|
||||
return _yttranscript_api_transcript(video_id)
|
||||
|
||||
async def summarize_video(video_id: str) -> str | None:
|
||||
# 1. Try transcript, fall back to description (offload sync call to thread)
|
||||
transcript = await asyncio.to_thread(download_transcript, video_id)
|
||||
@@ -141,9 +238,12 @@ async def summarize_video(video_id: str) -> str | None:
|
||||
|
||||
return summary
|
||||
|
||||
async def summarize_all_pending():
|
||||
async def summarize_all_pending(max_videos: int = 0):
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
db.row_factory = aiosqlite.Row
|
||||
if max_videos > 0:
|
||||
cursor = await db.execute("SELECT video_id FROM videos WHERE summary IS NULL OR summary = '' LIMIT ?", (max_videos,))
|
||||
else:
|
||||
cursor = await db.execute("SELECT video_id FROM videos WHERE summary IS NULL OR summary = ''")
|
||||
rows = await cursor.fetchall()
|
||||
|
||||
|
||||
@@ -43,7 +43,8 @@ async def test_download_transcript_success():
|
||||
mock_trans = MockTranscript()
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list):
|
||||
with patch.object(summarizer, "_ytdlp_transcript", return_value=None), \
|
||||
patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list):
|
||||
text = summarizer.download_transcript("test_vid")
|
||||
assert text == "Welcome to this video tutorial. Today we are learning about Svelte 5 and runes."
|
||||
|
||||
@@ -60,7 +61,8 @@ async def test_summarize_video_success(test_db):
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
mock_response = _mock_deepseek_response("**Mock Core Hook**\n- Takeaway 1\n- Takeaway 2\nVerdict line.")
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
with patch.object(summarizer, "_ytdlp_transcript", return_value=None), \
|
||||
patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response) as mock_post:
|
||||
|
||||
summary = await summarizer.summarize_video("summarize_vid")
|
||||
@@ -106,7 +108,8 @@ async def test_summarize_video_deepseek_error_leaves_null(test_db):
|
||||
"429 Rate Limited", request=MagicMock(), response=mock_response
|
||||
)
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
with patch.object(summarizer, "_ytdlp_transcript", return_value=None), \
|
||||
patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response):
|
||||
|
||||
result = await summarizer.summarize_video("err_vid")
|
||||
@@ -133,7 +136,8 @@ async def test_summarize_all_pending(test_db):
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
mock_response = _mock_deepseek_response("Fresh summary")
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
with patch.object(summarizer, "_ytdlp_transcript", return_value=None), \
|
||||
patch("youtube_transcript_api.YouTubeTranscriptApi.list", return_value=mock_list), \
|
||||
patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response) as mock_post:
|
||||
|
||||
await summarizer.summarize_all_pending()
|
||||
|
||||
+32
-1
@@ -165,7 +165,38 @@ YouTube transcripts inaccessible from all our locations:
|
||||
- **MacBook (China)**: GFW blocks YouTube API endpoints
|
||||
- **YouTube Data API captions.download**: 403 Forbidden for non-owned videos
|
||||
|
||||
**Workaround**: Title + description-based summaries via Qwen. Full transcripts need a Japanese residential proxy ($3-5/month).
|
||||
#### Google Data API v3 Transcript Support (Investigation 2025-07-11)
|
||||
|
||||
| Endpoint | Works? | Notes |
|
||||
|----------|--------|-------|
|
||||
| `captions.list` (part=snippet, videoId=...) | ✅ Yes | Lists available caption tracks with language, kind (standard/asr), status |
|
||||
| `captions.download` (id=..., tfmt='srt') | ❌ **403 Forbidden** | Only works for videos **you own**. Returns error: *"permissions associated with the request are not sufficient to download the caption track"* |
|
||||
| `videos.list` (part=snippet) | ✅ Yes | Returns video description (~200-300 chars typical), but NOT transcript/captions |
|
||||
|
||||
**Conclusion**: There is no Google official API to download captions of other creators' videos. The `captions.download` endpoint is strictly limited to video owners. Third-party tools (`youtube-transcript-api`, `yt-dlp`) reverse-engineer YouTube's internal innertube API to extract captions, which is why they get blocked.
|
||||
|
||||
#### Transcript Retrieval Strategy
|
||||
|
||||
The only reliable way to get transcripts is **fresh YouTube session cookies + yt-dlp**:
|
||||
|
||||
```
|
||||
Cookies (exported from browser) → yt-dlp (write-subs) → HTML/JSON → transcript text
|
||||
```
|
||||
|
||||
yt-dlp is preferred over `youtube-transcript-api` because:
|
||||
1. More actively maintained and widely used
|
||||
2. Better at evading bot detection (more complete browser fingerprinting)
|
||||
3. Supports `--cookies-from-browser` for direct cookie extraction
|
||||
4. Has built-in subtitle extraction (`--write-subs`)
|
||||
|
||||
**Current blockers** (verified 2025-07-11):
|
||||
1. `config/cookies.txt` is **expired** (last exported 2024-06-24) — YouTube session cookies expire quickly
|
||||
2. macOS encrypts Chrome cookies via Keychain → `browser_cookie3` and `yt-dlp --cookies-from-browser` fail in non-interactive sessions
|
||||
3. `youtube-transcript-api` returns `RequestBlocked` from this environment
|
||||
|
||||
**Fix path**: Re-export cookies.txt from browser → `summarizer.py` uses yt-dlp with that cookies file.
|
||||
|
||||
**Workaround** (current, degraded): Title + description-based summaries via Qwen. Description provides ~200-300 chars of useful text.
|
||||
|
||||
### auth.jimmygan.com from LAN
|
||||
Accessing `auth.jimmygan.com` from LAN shows Immich (photos.jimmygan.com) instead of Authelia. Caused by NAS port 443 routing — DSM nginx still owns port 443 and routes Immich for unmatched domains.
|
||||
|
||||
Reference in New Issue
Block a user