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())