Gan, Jimmy
f1ada7d277
- 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)
275 lines
11 KiB
Python
275 lines
11 KiB
Python
import os
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import httpx
|
|
import aiosqlite
|
|
import requests
|
|
from youtube_transcript_api import YouTubeTranscriptApi
|
|
from db import DB_PATH
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
def proxy_from_env():
|
|
"""Return proxy URL from HTTP_PROXY/HTTPS_PROXY env vars, or None."""
|
|
return os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") or \
|
|
os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or None
|
|
|
|
# Fallback pattern to retrieve the correct bytecatcode credentials
|
|
API_KEY = os.environ.get("DEEPSEEK_API_KEY", os.environ.get("OPENAI_API_KEY", ""))
|
|
BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
|
|
MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")
|
|
|
|
SYSTEM_PROMPT = "You are a professional video summarizing assistant. Create highly concise, structured summaries from video transcripts. Your output must be in Chinese if the transcript is Chinese, otherwise in English."
|
|
|
|
PROMPT = """Read the following YouTube video information and write a super-concise, structured summary.
|
|
|
|
Guidelines:
|
|
1. **Core Hook (1 sentence in bold)**: Explain exactly what this video is about and its main conclusion or thesis.
|
|
2. **Key Takeaways (max 4-5 bullet points)**: Summarize the most important points, arguments, or steps. Focus on information density. Skip fluff, intros, sponsor segments, and generic remarks.
|
|
3. **Actionable Idea / Final Verdict**: A single line explaining who this is for or a key action item.
|
|
|
|
The input may be a full transcript or just a title+description. Extract maximum value from whatever is provided.
|
|
Keep the entire summary under 150 words. Use the same language as the source (Chinese or English).
|
|
|
|
Content:
|
|
{transcript}"""
|
|
|
|
COOKIES_PATH = None
|
|
for p in [os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config/cookies.txt"),
|
|
"/app/config/cookies.txt"]:
|
|
if os.path.exists(p):
|
|
COOKIES_PATH = p
|
|
break
|
|
|
|
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()
|
|
proxy = proxy_from_env()
|
|
if proxy:
|
|
session.proxies = {"http": proxy, "https": proxy}
|
|
session.headers.update({
|
|
"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 and os.path.exists(COOKIES_PATH):
|
|
from http.cookiejar import MozillaCookieJar
|
|
cj = MozillaCookieJar(COOKIES_PATH)
|
|
cj.load(ignore_discard=True, ignore_expires=True)
|
|
session.cookies.update({c.name: c.value for c in cj})
|
|
transcript_list = YouTubeTranscriptApi(http_client=session).list(video_id)
|
|
try:
|
|
transcript = transcript_list.find_manually_created_transcript(['en', 'zh-Hans', 'zh-Hant', 'zh', 'ja'])
|
|
except Exception:
|
|
try:
|
|
transcript = transcript_list.find_generated_transcript(['en', 'zh-Hans', 'zh-Hant', 'zh', 'ja'])
|
|
except Exception:
|
|
transcript = next(iter(transcript_list))
|
|
parts = transcript.fetch()
|
|
return " ".join([part["text"] for part in parts])
|
|
except Exception as 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)
|
|
|
|
if not transcript:
|
|
# Fallback: use video description from YouTube API
|
|
try:
|
|
from auth import get_credentials
|
|
from googleapiclient.discovery import build
|
|
creds = get_credentials()
|
|
youtube = build("youtube", "v3", credentials=creds)
|
|
resp = youtube.videos().list(part="snippet", id=video_id).execute()
|
|
items = resp.get("items", [])
|
|
if items:
|
|
desc = items[0]["snippet"].get("description", "")
|
|
title = items[0]["snippet"].get("title", "")
|
|
if desc:
|
|
transcript = f"Title: {title}\n\nDescription: {desc[:3000]}"
|
|
except Exception as e:
|
|
log.warning(f"Description fallback failed for {video_id}: {e}")
|
|
|
|
if not transcript:
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
await db.execute(
|
|
"UPDATE videos SET transcript = ?, summary = ? WHERE video_id = ?",
|
|
("No transcript available", "📺 Short clip — title only, no description available. Watch to preview.", video_id)
|
|
)
|
|
await db.commit()
|
|
return None
|
|
|
|
# 2. Call DeepSeek API (OpenAI-compatible)
|
|
headers = {
|
|
"Authorization": f"Bearer {API_KEY}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "t-youtube/0.1"
|
|
}
|
|
|
|
payload = {
|
|
"model": MODEL,
|
|
"max_tokens": 1000,
|
|
"messages": [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": PROMPT.format(transcript=transcript[:50000])}
|
|
]
|
|
}
|
|
|
|
async with httpx.AsyncClient(proxy=proxy_from_env(), timeout=120) as client:
|
|
try:
|
|
resp = await client.post(
|
|
f"{BASE_URL}/v1/chat/completions",
|
|
headers=headers,
|
|
json=payload
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
summary = data["choices"][0]["message"]["content"]
|
|
except Exception as e:
|
|
log.error(f"DeepSeek API request failed for {video_id}: {e}")
|
|
return None # Leave summary NULL for retry on next sync
|
|
|
|
# 3. Save transcript & summary to db
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
await db.execute(
|
|
"UPDATE videos SET transcript = ?, summary = ? WHERE video_id = ?",
|
|
(transcript, summary, video_id)
|
|
)
|
|
await db.commit()
|
|
|
|
return summary
|
|
|
|
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()
|
|
|
|
if not rows:
|
|
log.info("No pending videos to summarize.")
|
|
return
|
|
|
|
log.info(f"Found {len(rows)} pending videos to summarize.")
|
|
for row in rows:
|
|
video_id = row["video_id"]
|
|
log.info(f"Summarizing video: {video_id}")
|
|
await summarize_video(video_id)
|
|
await asyncio.sleep(1)
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
async def main():
|
|
if len(sys.argv) > 1:
|
|
video_id = sys.argv[1]
|
|
print(f"Summarizing single video: {video_id}")
|
|
summary = await summarize_video(video_id)
|
|
print(f"Summary:\n{summary}")
|
|
else:
|
|
await summarize_all_pending()
|
|
|
|
asyncio.run(main())
|