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("ANTHROPIC_API_KEY")
if not API_KEY:
    API_KEY = os.environ.get("OPENAI_API_KEY", "sk-Kq5JbgmnEEkdzTqNAQ6bV6nztQ0ngeHTHF9Vg8nUglQvrOlx")

BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://www.bytecatcode.org")
MODEL = "claude-haiku-4-5-20251001"

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 transcript 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.

Keep the entire summary under 150 words. Be extremely direct so I can decide in 10 seconds if I should watch the full video or skip it. Use the same language as the transcript (likely Chinese or English).

Transcript:
{transcript}"""

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 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept-Language": "en-US,en;q=0.9"
        })
        transcript_list = YouTubeTranscriptApi(http_client=session).list(video_id)

        # Try to find manual transcripts in preferred languages
        try:
            transcript = transcript_list.find_manually_created_transcript(['en', 'zh-Hans', 'zh-Hant', 'zh', 'ja'])
        except Exception:
            # Fallback to generated (auto-captions) in preferred languages
            try:
                transcript = transcript_list.find_generated_transcript(['en', 'zh-Hans', 'zh-Hant', 'zh', 'ja'])
            except Exception:
                # Fallback to any transcript
                transcript = next(iter(transcript_list))

        parts = transcript.fetch()
        text = " ".join([part["text"] for part in parts])
        return text
    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. Fetch transcript
    transcript = download_transcript(video_id)
    if not transcript:
        # Save placeholder transcript / status
        async with aiosqlite.connect(DB_PATH) as db:
            await db.execute(
                "UPDATE videos SET transcript = ?, summary = ? WHERE video_id = ?",
                ("No transcript available", "Could not retrieve video transcript.", video_id)
            )
            await db.commit()
        return None

    # 2. Call Claude API
    headers = {
        "x-api-key": API_KEY,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }

    # Standard Anthropic payload format
    payload = {
        "model": MODEL,
        "max_tokens": 1000,
        "system": SYSTEM_PROMPT,
        "messages": [
            {"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/messages",
                headers=headers,
                json=payload
            )
            resp.raise_for_status()
            data = resp.json()
            summary = data["content"][0]["text"]
        except Exception as e:
            log.error(f"Claude 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())