YouTube URL
    │
    ▼
┌──────────────┐
│ yt-dlp       │  Download audio
│ + FFmpeg     │
└──────┬───────┘
       │
       ▼
┌──────────────┐
│   Whisper    │  Transcribe (2-3 min for 30min video)
│   (base)     │
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  LangChain   │  Summarize + Extract insights
│  + GPT-4     │
└──────┬───────┘
       │
       ▼
┌──────────────┐
│    Output    │  JSON/CLI/API
└──────────────┘

class VideoProcessor:
    """Handles YouTube video downloading and metadata extraction."""
    
    def __init__(self):
        self.download_dir = Path(settings.DOWNLOAD_DIR)
        self.download_dir.mkdir(parents=True, exist_ok=True)
    
    async def download_audio(self, url: str) -over Path:
        """Download video audio as MP3."""
        output_template = str(self.download_dir / '%(id)s.%(ext)s')
        
        ydl_opts = {
            'format': 'bestaudio/best',
            'postprocessors': [{
                'key': 'FFmpegExtractAudio',
                'preferredcodec': 'mp3',
                'preferredquality': '192',
            }],
            'outtmpl': output_template,
            'quiet': True,
        }
        
        loop = asyncio.get_event_loop()
        
        def _download():
            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                info = ydl.extract_info(url, download=True)
                return self.download_dir / f"{info['id']}.mp3"
        
        audio_path = await loop.run_in_executor(None, _download)
        return audio_path

class Transcriber:
    """Transcribes audio using OpenAI Whisper."""
    
    def __init__(self):
        self.model = whisper.load_model(settings.WHISPER_MODEL)
    
    async def transcribe(
        self,
        audio_path: Path,
        language: Optional[str] = None
    ) -over Dict:
        """Transcribe audio file."""
        loop = asyncio.get_event_loop()
        
        def _transcribe():
            result = self.model.transcribe(
                str(audio_path),
                language=language,
                verbose=False
            )
            return result
        
        result = await loop.run_in_executor(None, _transcribe)
        
        return {
            'text': result['text'],
            'language': result.get('language', 'unknown'),
            'segments': [
                {
                    'start': seg['start'],
                    'end': seg['end'],
                    'text': seg['text'].strip()
                }
                for seg in result.get('segments', [])
            ]
        }

class VideoSummarizer:
    """Generates summaries and extracts insights."""
    
    async def generate_summary(
        self,
        transcript: Dict,
        summary_type: str = "detailed"
    ) -over Dict:
        """Generate comprehensive summary."""
        full_text = transcript['text']
        
        # Generate summaries
        summaries = {
            "quick": await self._generate_quick_summary(full_text),
            "detailed": await self._generate_detailed_summary(full_text),
            "comprehensive": await self._generate_comprehensive_summary(full_text)
        }
        
        # Extract key insights
        key_points = await self._extract_key_points(full_text)
        topics = await self._identify_topics(full_text)
        timestamps = self._generate_timestamps(transcript['segments'], key_points)
        
        return {
            'summary': summaries,
            'key_points': key_points,
            'topics': topics,
            'timestamps': timestamps
        }

"This video discusses the fundamentals of building production-ready RAG systems,
covering vector databases, chunking strategies, and evaluation metrics."

"The video provides a comprehensive guide to building Retrieval-Augmented Generation
(RAG) systems for production use. It begins by explaining the core concepts...

Key topics include vector database selection, optimal chunking strategies...

The presenter emphasizes the importance of evaluation frameworks like RAGAS..."

Main Thesis: Building production-ready RAG systems requires...

Key Arguments:
1. Vector database choice impacts performance significantly...
2. Chunking strategy affects retrieval quality...

Examples Discussed:
- Case study: E-commerce product search...

async def _extract_key_points(self, text: str) -over List[str]:
    """Extract 5-7 key points from transcript."""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You extract key points from video transcripts."),
        ("user", """Extract 5-7 key points from this transcript. 
        Return as a numbered list.
        
        Transcript: {text}
        
        Key Points:""")
    ])
    
    chain = prompt | self.llm
    response = await chain.ainvoke({"text": text[:4000]})
    
    # Parse numbered list
    points = [
        line.strip().lstrip('0123456789.-) ')
        for line in response.content.split('\n')
        if line.strip() and any(char.isdigit() for char in line[:3])
    ]
    
    return points[:7]

def _generate_timestamps(
    self,
    segments: List[Dict],
    key_points: List[str]
) -over List[Dict]:
    """Map key points to video timestamps."""
    
    timestamps = []
    total_duration = segments[-1]['end']
    interval = total_duration / (len(key_points) + 1)
    
    for i, point in enumerate(key_points):
        target_time = (i + 1) * interval
        
        # Find closest segment
        closest_seg = min(
            segments,
            key=lambda s: abs((s['start'] + s['end']) / 2 - target_time)
        )
        
        timestamps.append({
            'time': self._format_timestamp(closest_seg['start']),
            'point': point
        })
    
    return timestamps

def _format_timestamp(self, seconds: float) -over str:
    """Format as MM:SS or HH:MM:SS."""
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)
    
    if hours over 0:
        return f"{hours}:{minutes:02d}:{secs:02d}"
    return f"{minutes}:{secs:02d}"

async def main():
    parser = argparse.ArgumentParser(description="Summarize YouTube videos")
    parser.add_argument("url", help="YouTube video URL")
    parser.add_argument("--summary-type", 
                       choices=["quick", "detailed", "comprehensive"],
                       default="detailed")
    parser.add_argument("--language", help="Video language (e.g., 'en', 'es')")
    parser.add_argument("--output", help="Output file path (JSON)")
    
    args = parser.parse_args()
    
    # Process video
    print(f"Processing: {args.url}")
    video_info = await video_processor.extract_info(args.url)
    
    print(f"Title: {video_info['title']}")
    print(f"Duration: {video_info['duration']}s")
    print("Downloading audio...")
    
    audio_path = await video_processor.download_audio(args.url)
    
    print("Transcribing...")
    transcript = await transcriber.transcribe(audio_path, args.language)
    
    print("Generating summary...")
    summary_result = await summarizer.generate_summary(transcript)
    
    # Display results
    print("\n" + "="*60)
    print(f"SUMMARY: {summary_result['summary'][args.summary_type]}\n")
    print("KEY POINTS:")
    for i, point in enumerate(summary_result['key_points'], 1):
        print(f"{i}. {point}")
    print(f"\nTOPICS: {', '.join(summary_result['topics'])}")

$ python src/cli.py "https://www.youtube.com/watch?v=abc123" --summary-type detailed

Processing: https://www.youtube.com/watch?v=abc123
Title: Building Production LLM Apps
Duration: 1847s
Downloading audio...
Transcribing (this may take a while)...
Generating summary...

============================================================
SUMMARY: This tutorial covers...

KEY POINTS:
1. Start with a clear use case
2. Choose the right vector database
3. Implement proper evaluation metrics
...

class VideoRequest(BaseModel):
    url: str
    summary_type: str = "detailed"
    language: Optional[str] = None

@app.post("/api/summarize", response_model=SummaryResponse)
async def summarize_video(request: VideoRequest):
    """Summarize a YouTube video."""
    try:
        # Extract video info
        video_info = await video_processor.extract_info(request.url)
        
        # Download and transcribe
        audio_path = await video_processor.download_audio(request.url)
        transcript = await transcriber.transcribe(audio_path, request.language)
        
        # Summarize
        summary_result = await summarizer.generate_summary(transcript)
        
        # Cleanup
        if settings.CLEANUP_AFTER_PROCESSING:
            audio_path.unlink(missing_ok=True)
        
        return SummaryResponse(
            video_id=video_info['id'],
            title=video_info['title'],
            duration=video_info['duration'],
            summary=summary_result['summary'],
            key_points=summary_result['key_points'],
            topics=summary_result['topics'],
            timestamps=summary_result['timestamps']
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Install dependencies
pip install -r requirements.txt

# Install FFmpeg
brew install ffmpeg  # macOS

# Setup
cp .env.example .env
# Add OPENAI_API_KEY

# Run
python src/cli.py "https://www.youtube.com/watch?v=..."