t-youtube/backend/summarizer.py

import os
import asyncio
import logging
import httpx
import aiosqlite
import requests
from youtube_transcript_api import YouTubeTranscriptApi
from db import DB_PATH

log = logging.getLogger(__name__)

# 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 download_transcript(video_id: str) -> str:
    try:
        session = requests.Session()
        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:
            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"Failed to fetch transcript for {video_id}: {e}")
        return None

async def summarize_video(video_id: str) -> str:
    # 1. Try transcript, fall back to description
    transcript = 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(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}")
            summary = f"Error generating summary: {e}"

    # 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():
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        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())