app.py

# app.py - YouTube Video RAG Q&A for Hugging Face Spaces

import gradio as gr
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._errors import TranscriptsDisabled, NoTranscriptFound
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
import pickle
import os
import re
import groq
from typing import List, Dict, Tuple
import tempfile

# ============================================
# Configuration - Optimized for Token Limits
# ============================================

GROQ_API_KEY = os.getenv("GROQ_API_KEY")  # Get from Hugging Face Secrets
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
CHUNK_SIZE = 300
MAX_CONTEXT_TOKENS = 1500
MAX_RETRIEVAL_CHUNKS = 2

# ============================================
# YouTube Transcript Extraction
# ============================================

class YouTubeTranscriptProcessor:
    """Handles YouTube transcript extraction and processing using new API"""
    
    @staticmethod
    def extract_transcript(youtube_url: str) -> Tuple[List[Dict], str]:
        """Extract transcript from YouTube video"""
        try:
            video_id = YouTubeTranscriptProcessor.extract_video_id(youtube_url)
            if not video_id:
                return None, "Invalid YouTube URL"
            
            print(f"Processing video ID: {video_id}")
            
            # Create API instance and fetch transcript
            ytt_api = YouTubeTranscriptApi()
            
            try:
                fetched_transcript = ytt_api.fetch(video_id, languages=['en'])
                print("Found English transcript")
            except:
                print("English transcript not found, trying any available language...")
                fetched_transcript = ytt_api.fetch(video_id)
                print(f"Found transcript in language: {fetched_transcript.language}")
            
            # Convert to formatted transcript
            formatted_transcript = []
            for snippet in fetched_transcript.snippets:
                formatted_transcript.append({
                    'text': snippet.text,
                    'start': snippet.start,
                    'duration': snippet.duration
                })
            
            print(f"Successfully extracted {len(formatted_transcript)} transcript entries")
            return formatted_transcript, None
            
        except Exception as e:
            return None, f"Error extracting transcript: {str(e)}"
    
    @staticmethod
    def extract_video_id(url: str) -> str:
        """Extract video ID from YouTube URL"""
        patterns = [
            r'(?:youtube\.com\/watch\?v=)([\w-]+)',
            r'(?:youtu\.be\/)([\w-]+)',
            r'(?:youtube\.com\/embed\/)([\w-]+)',
            r'(?:youtube\.com\/v\/)([\w-]+)',
            r'(?:youtube\.com\/shorts\/)([\w-]+)'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, url)
            if match:
                return match.group(1)
        return None
    
    @staticmethod
    def get_full_transcript_text(transcript: List[Dict]) -> str:
        """Convert transcript to readable full text without timestamps"""
        # Just join all text entries with spaces
        full_text = " ".join([entry['text'] for entry in transcript])
        
        # Clean up extra spaces
        full_text = re.sub(r'\s+', ' ', full_text).strip()
        
        # Add line breaks every ~100 characters for better readability
        lines = []
        words = full_text.split()
        current_line = []
        current_length = 0
        
        for word in words:
            if current_length + len(word) + 1 <= 100:
                current_line.append(word)
                current_length += len(word) + 1
            else:
                lines.append(" ".join(current_line))
                current_line = [word]
                current_length = len(word)
        
        if current_line:
            lines.append(" ".join(current_line))
        
        return "\n".join(lines)
    
    @staticmethod
    def chunk_transcript(transcript: List[Dict]) -> List[Dict]:
        """Split transcript into smaller overlapping chunks"""
        full_text = " ".join([entry['text'] for entry in transcript])
        sentences = re.split(r'(?<=[.!?])\s+', full_text)
        
        chunks = []
        current_chunk = []
        current_length = 0
        
        for sentence in sentences:
            sentence_length = len(sentence)
            
            if current_length + sentence_length <= CHUNK_SIZE:
                current_chunk.append(sentence)
                current_length += sentence_length
            else:
                if current_chunk:
                    chunk_text = " ".join(current_chunk)
                    chunks.append({
                        'text': chunk_text,
                        'chunk_id': len(chunks)
                    })
                
                overlap_text = " ".join(current_chunk[-2:]) if len(current_chunk) > 2 else " ".join(current_chunk)
                current_chunk = [overlap_text, sentence] if overlap_text else [sentence]
                current_length = len(overlap_text) + sentence_length if overlap_text else sentence_length
        
        if current_chunk:
            chunks.append({
                'text': " ".join(current_chunk),
                'chunk_id': len(chunks)
            })
        
        print(f"Created {len(chunks)} chunks from transcript")
        return chunks

# ============================================
# Vector Database Management
# ============================================

class VectorDatabase:
    """Manages FAISS vector database and embeddings"""
    
    def __init__(self):
        print("Loading embedding model...")
        self.embedding_model = SentenceTransformer(EMBEDDING_MODEL)
        self.index = None
        self.chunks = []
        # Use temporary files for Hugging Face Spaces
        self.index_path = tempfile.NamedTemporaryFile(delete=False, suffix='.bin').name
        self.chunks_path = tempfile.NamedTemporaryFile(delete=False, suffix='.pkl').name
    
    def create_embeddings(self, texts: List[str]) -> np.ndarray:
        """Create embeddings for texts"""
        print(f"Creating embeddings for {len(texts)} chunks...")
        batch_size = 32
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            batch_embeddings = self.embedding_model.encode(batch, show_progress_bar=True)
            all_embeddings.append(batch_embeddings)
        
        return np.vstack(all_embeddings)
    
    def build_index(self, chunks: List[Dict]):
        """Build FAISS index from chunks"""
        self.chunks = chunks
        texts = [chunk['text'] for chunk in chunks]
        embeddings = self.create_embeddings(texts)
        
        dimension = embeddings.shape[1]
        self.index = faiss.IndexFlatL2(dimension)
        self.index.add(embeddings.astype('float32'))
        
        self.save()
        return True
    
    def search(self, query: str, k: int = MAX_RETRIEVAL_CHUNKS) -> List[Tuple[str, float]]:
        """Search for similar chunks"""
        if self.index is None or not self.chunks:
            return []
        
        query_embedding = self.embedding_model.encode([query])
        distances, indices = self.index.search(query_embedding.astype('float32'), k)
        
        results = []
        for i, idx in enumerate(indices[0]):
            if idx != -1 and idx < len(self.chunks):
                results.append((self.chunks[idx]['text'], float(distances[0][i])))
        
        return results
    
    def save(self):
        if self.index:
            faiss.write_index(self.index, self.index_path)
        with open(self.chunks_path, 'wb') as f:
            pickle.dump(self.chunks, f)
        print("Database saved successfully")
    
    def load(self):
        if os.path.exists(self.index_path) and os.path.exists(self.chunks_path):
            self.index = faiss.read_index(self.index_path)
            with open(self.chunks_path, 'rb') as f:
                self.chunks = pickle.load(f)
            print(f"Loaded database with {len(self.chunks)} chunks")
            return True
        return False
    
    def clear(self):
        self.index = None
        self.chunks = []
        if os.path.exists(self.index_path):
            os.remove(self.index_path)
        if os.path.exists(self.chunks_path):
            os.remove(self.chunks_path)
        print("Database cleared")

# ============================================
# RAG Question Answering
# ============================================

class RAGQA:
    """Handles RAG-based question answering using Groq directly"""
    
    def __init__(self):
        self.vector_db = VectorDatabase()
        self.client = groq.Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None
        self.current_transcript_text = ""
        self.vector_db.load()
    
    def truncate_context(self, context: str, max_tokens: int = MAX_CONTEXT_TOKENS) -> str:
        max_chars = max_tokens * 4
        if len(context) > max_chars:
            return context[:max_chars] + "..."
        return context
    
    def process_video(self, youtube_url: str) -> Tuple[str, str, bool]:
        """Process YouTube video and build vector database, return full transcript"""
        # Extract transcript
        transcript, error = YouTubeTranscriptProcessor.extract_transcript(youtube_url)
        if error:
            return error, "", False
        
        if not transcript:
            return "No transcript data found", "", False
        
        # Get full transcript text without timestamps
        self.current_transcript_text = YouTubeTranscriptProcessor.get_full_transcript_text(transcript)
        
        # Chunk transcript for RAG
        chunks = YouTubeTranscriptProcessor.chunk_transcript(transcript)
        
        if not chunks:
            return "No content to process", self.current_transcript_text, False
        
        # Build vector database
        self.vector_db.build_index(chunks)
        
        return f"✅ Successfully processed {len(chunks)} chunks from video!", self.current_transcript_text, True
    
    def ask_question(self, question: str) -> str:
        """Answer question using RAG with Groq"""
        if not GROQ_API_KEY:
            return "⚠️ Please set your Groq API key in Hugging Face Secrets."
        
        if self.vector_db.index is None or not self.vector_db.chunks:
            return "⚠️ Please load a video transcript first (click 'Get Transcript') before asking questions."
        
        relevant_chunks = self.vector_db.search(question, k=MAX_RETRIEVAL_CHUNKS)
        
        if not relevant_chunks:
            return "❓ No relevant information found in the transcript. Please try a different question."
        
        context = "\n\n---\n\n".join([chunk[0] for chunk in relevant_chunks])
        context = self.truncate_context(context, MAX_CONTEXT_TOKENS)
        
        system_prompt = """Answer questions based ONLY on the provided transcript context. Be brief (2-3 sentences max). If the answer isn't in the context, say so."""
        user_prompt = f"""Context: {context}\n\nQuestion: {question}\n\nAnswer:"""
        
        try:
            chat_completion = self.client.chat.completions.create(
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                model="llama-3.1-8b-instant",
                temperature=0.3,
                max_tokens=150
            )
            
            return chat_completion.choices[0].message.content
            
        except Exception as e:
            if "rate_limit_exceeded" in str(e) or "too large" in str(e):
                return "⚠️ Context too large. Please ask a more specific question."
            return f"❌ Error: {str(e)}"
    
    def clear_database(self) -> str:
        self.vector_db.clear()
        self.current_transcript_text = ""
        return "🗑️ Database cleared successfully!"

# ============================================
# Gradio UI Application
# ============================================

# Initialize RAG system
rag_system = RAGQA()

def process_youtube_url(youtube_url):
    if not youtube_url or youtube_url.strip() == "":
        return "❌ Please enter a YouTube URL", "⚠️ Waiting for video...", ""
    
    message, transcript_text, success = rag_system.process_video(youtube_url)
    if success:
        return message, "✅ Ready for questions!", transcript_text
    else:
        return message, "❌ Failed to process video", ""

def answer_question(question, history):
    if not question or question.strip() == "":
        return history
    
    answer = rag_system.ask_question(question)
    history = history or []
    history.append((question, answer))
    return history

def clear_everything():
    message = rag_system.clear_database()
    return message, "⚠️ Waiting for video...", "", []

# Create Gradio interface
with gr.Blocks(title="🎥 YouTube Video RAG Q&A", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 📚 YouTube Video Q&A with RAG
    ### Extract transcript and ask questions about any YouTube video!
    
    **How it works:**
    1. Enter a YouTube URL
    2. Click "Get Transcript" to extract and process the video transcript
    3. Ask questions about the video content
    4. Get accurate answers based solely on the transcript
    
    **Note:** Make sure the video has captions/transcripts enabled.
    """)
    
    with gr.Row():
        with gr.Column(scale=3):
            youtube_url = gr.Textbox(
                label="🔗 YouTube URL",
                placeholder="https://www.youtube.com/watch?v=...",
                lines=1
            )
        
        with gr.Column(scale=1):
            process_btn = gr.Button("🎬 Get Transcript", variant="primary", size="lg")
    
    with gr.Row():
        status_text = gr.Textbox(label="📊 Status", interactive=False, lines=2)
        qa_status = gr.Textbox(label="QA Status", interactive=False, lines=1, value="⚠️ Waiting for video...")
    
    gr.Markdown("---")
    
    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### 📝 Complete Transcript")
            transcript_display = gr.Textbox(
                label="",
                interactive=False,
                lines=25,
                max_lines=25,
                placeholder="Transcript will appear here after processing..."
            )
        
        with gr.Column(scale=1):
            gr.Markdown("### 💬 Ask Questions")
            chatbot = gr.Chatbot(
                label="Chat",
                height=400,
                bubble_full_width=False,
                avatar_images=(None, "🤖")
            )
            
            with gr.Row():
                question = gr.Textbox(
                    label="Your Question",
                    placeholder="Ask about the video...",
                    lines=2,
                    scale=4
                )
                submit_btn = gr.Button("Ask", variant="primary", scale=1)
            
            with gr.Row():
                clear_chat_btn = gr.Button("🗑️ Clear Chat", variant="secondary", size="sm")
                clear_all_btn = gr.Button("🔄 Clear All", variant="stop", size="sm")
    
    # Event handlers
    process_btn.click(
        process_youtube_url,
        inputs=[youtube_url],
        outputs=[status_text, qa_status, transcript_display]
    )
    
    submit_btn.click(
        answer_question,
        inputs=[question, chatbot],
        outputs=[chatbot]
    ).then(
        lambda: "", None, [question]
    )
    
    clear_chat_btn.click(
        lambda: [], None, [chatbot]
    )
    
    clear_all_btn.click(
        clear_everything,
        outputs=[status_text, qa_status, transcript_display, chatbot]
    )
    
    question.submit(
        answer_question,
        inputs=[question, chatbot],
        outputs=[chatbot]
    ).then(
        lambda: "", None, [question]
    )

# Launch the app
if __name__ == "__main__":
    demo.launch()