app.py

from dotenv import load_dotenv
load_dotenv()

import os

if not os.environ.get("GOOGLE_API_KEY"):
    raise RuntimeError("Please set the GOOGLE_API_KEY environment variable with your Google API key.")

import gradio as gr
from supadata import Supadata
from langchain_core.prompts import PromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores.faiss import FAISS
from langchain_huggingface import HuggingFaceEmbeddings, HuggingFaceEndpoint, ChatHuggingFace
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage
from langchain_google_genai import GoogleGenerativeAIEmbeddings

# Initialize Supadata
supadata = Supadata(api_key=os.environ.get("SUPADATA_API_KEY"))

# Initialize the text splitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=300,
    chunk_overlap=30,
    separators=["\n\n", "\n", " ", ".", ","],
    length_function=len,
    is_separator_regex=False
)

# Initialize the embeddings model
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2"
)

# Initialize the LLM
# llm = HuggingFaceEndpoint(
#     repo_id="microsoft/Phi-3-mini-4k-instruct",
#     task="text-generation",
#     max_new_tokens=512,
#     do_sample=False,
#     repetition_penalty=1.03,
#     huggingfacehub_api_token=os.getenv("HUGGINGFACE_TOKEN")
# )

chat = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    temperature=0.3,
    max_tokens=5000,
    timeout=None,
    max_retries=2
)

# Define the prompt template
prompt = PromptTemplate(
    template="""
You are an intelligent AI assistant specialized in analyzing YouTube video transcripts. Your task is to provide accurate, detailed, and helpful answers based solely on the provided transcript content.

IMPORTANT GUIDELINES:
- Answer ONLY from the provided transcript context
- If the context is insufficient to answer the question, clearly state "I don't have enough information from the transcript to answer this question"
- Provide specific details and examples from the transcript when possible
- Be concise but comprehensive in your responses
- If asked for a summary, organize the information logically
- If asked about specific topics, focus on what was actually discussed in the video
- Maintain a helpful and informative tone

TRANSCRIPT CONTEXT:
{context}

QUESTION: {question}

Please provide a clear and detailed answer based on the transcript above:
""",
    input_variables=['context', 'question']
)

# Global variable to store the current retriever
current_retriever = None
current_video_id = None

def extract_video_id(url):
    """Extract video ID from YouTube URL."""
    if "youtube.com/watch?v=" in url:
        return url.split("watch?v=")[1].split("&")[0]
    elif "youtu.be/" in url:
        return url.split("youtu.be/")[1].split("?")[0]
    return url  # Assume it's already a video ID

def process_video_url(video_url_or_id):
    """Process video URL and create retriever object."""
    global current_retriever, current_video_id
    
    try:
        # Extract video ID if URL is provided
        video_id = extract_video_id(video_url_or_id)
        
        # Check if we already have a retriever for this video
        if current_video_id == video_id and current_retriever is not None:
            return f"✅ Video already processed: {video_id}"
        
        # Get transcript using Supadata
        transcript_response = supadata.youtube.transcript(
            video_id=video_id,
            text=True  # Get plain text transcript
        )
        
        # Extract the transcript text
        full_transcript_text = transcript_response.content
        # print(f"✅ Full transcript text: {full_transcript_text}")
        
        # Create chunks
        chunks = text_splitter.create_documents([full_transcript_text])
        
        # Create vector store
        vector_store = FAISS.from_documents(chunks, embeddings)
        current_retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 8})
        # print(f"✅ Current Retreiver : {current_retriever}")
        current_video_id = video_id
        
        return f"✅ Video processed successfully: {video_id}"
    
    except Exception as e:
        return f"❌ Error processing video: {str(e)}"

def answer_question(question):
    global current_retriever
    
    if current_retriever is None:
        return "❌ Process a video first."
    
    try:
        # Retrieve docs with scores
        docs_and_scores = current_retriever.vectorstore.similarity_search_with_score(
            question, k=8
        )
        for doc, score in docs_and_scores:
            print(f"[DEBUG] score={score:.3f}, text={doc.page_content}")
        
        # Extract docs
        retrieved_docs = [doc for doc, _ in docs_and_scores]
        
        # Build context and reply as before
        context_text = "\n\n".join(doc.page_content for doc in retrieved_docs)
        # print("\nContext text:\n", context_text)
        final_prompt = prompt.invoke({"context": context_text, "question": question})
        answer = chat.invoke(final_prompt)
        return answer.content
    
    except Exception as e:
        print(f"[ERROR] in answer_question: {e}")
        return f"❌ Error: {str(e)}"

def process_video(video_url_or_id, question):
    """Legacy function for backward compatibility."""
    # Process video first
    process_result = process_video_url(video_url_or_id)
    if "❌" in process_result:
        return process_result
    
    # Then answer question
    return answer_question(question)

def main():
    with gr.Blocks(title="YouTube Transcript Q&A", theme=gr.themes.Soft()) as iface:
        gr.Markdown("""
        # YouTube Transcript Q&A
        
        Ask questions about any YouTube video's content! 
        
        ## How to use:
        1. Paste a YouTube video URL (e.g., https://www.youtube.com/watch?v=JaRGJVrJBQ8) or just the video ID
        2. Click "Process Video" to download and process the transcript (this happens once per video)
        3. Type your question about the video content
        4. Click "Ask Question" to get your answer
        """)
        
        with gr.Row():
            with gr.Column():
                video_input = gr.Textbox(
                    label="YouTube Video URL or ID",
                    placeholder="Enter YouTube URL or video ID (e.g., JaRGJVrJBQ8)",
                    lines=1
                )
                process_btn = gr.Button("Process Video", variant="primary")
                process_status = gr.Textbox(
                    label="Processing Status",
                    lines=1,
                    interactive=False
                )
                
                gr.Markdown("---")
                
                question_input = gr.Textbox(
                    label="Your Question",
                    placeholder="What would you like to know about this video?",
                    lines=2
                )
                ask_btn = gr.Button("Ask Question", variant="secondary")
            
            with gr.Column():
                output = gr.Textbox(
                    label="Answer",
                    lines=8,
                    show_copy_button=True
                )
        
        # Example inputs
        gr.Examples(
            examples=[
                ["https://www.youtube.com/watch?v=-moW9jvvMr4&t=1s", "What is this video about?"],
                ["https://www.youtube.com/watch?v=-moW9jvvMr4&t=1s", "What are the main topics discussed in this video?"],
                ["https://www.youtube.com/watch?v=-moW9jvvMr4&t=1s", "Can you summarize the key points from this video?"],
                ["-moW9jvvMr4", "What are the most important takeaways from this content?"]
            ],
            inputs=[video_input, question_input]
        )
        
        # Connect the buttons
        process_btn.click(
            fn=process_video_url,
            inputs=[video_input],
            outputs=[process_status]
        )
        
        ask_btn.click(
            fn=answer_question,
            inputs=[question_input],
            outputs=[output]
        )
        
        iface.launch()

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--test":
        result = process_video("https://www.youtube.com/watch?v=gN-QWM5iY9M", "What is this video about?")
        print(result)
    else:
        main()