app.py

import streamlit as st
from urllib.parse import urlparse, parse_qs
from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_groq import ChatGroq
import os
import pycountry
# from dotenv import load_dotenv

# Load environment variables
# load_dotenv()
groq_api_key = os.getenv("GROQ_API_KEY")

# Check for API key
if not groq_api_key:
    st.error("GROQ_API_KEY not found. Please check your .env file.")
    st.stop()

# Initialize model
model = ChatGroq(
    api_key=groq_api_key,
    temperature=0.7,
    model_name="llama3-70b-8192"
)

# Extract video ID
def extract_youtube_video_id(url):
    parsed_url = urlparse(url)
    query_params = parse_qs(parsed_url.query)
    video_id = query_params.get('v')
    return video_id[0] if video_id else None

# Language code to name
def get_language_name(lang_code):
    try:
        return pycountry.languages.get(alpha_2=lang_code).name
    except:
        return lang_code  # fallback

# Fetch transcript
def get_transcript(video_id, lang="en"):
    try:
        transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[lang])
        transcript = " ".join(chunk["text"] for chunk in transcript_list)
        return {"transcript": transcript}
    except TranscriptsDisabled:
        return {"error": "Transcripts are disabled for this video."}
    except Exception:
        try:
            transcript_options = YouTubeTranscriptApi.list_transcripts(video_id)
            return {
                "available_languages": {
                    t.language_code: get_language_name(t.language_code)
                    for t in transcript_options
                }
            }
        except:
            return {"error": "Transcript not available and available languages could not be determined."}

# Embed transcript
def embed_transcript(transcript):
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    chunks = splitter.create_documents([transcript])
    embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5")
    vector_store = FAISS.from_documents(chunks, embeddings)
    return vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 4})

# Prompt
prompt = PromptTemplate(
    template="""
You are a helpful assistant.
Answer ONLY using the context below.
If the context is insufficient, say Sorry and you are helpless.

Context:
{context}

Question: {question}
""",
    input_variables=['context', 'question']
)
parser = StrOutputParser()

# Format documents
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Create chain
def create_chain(retriever):
    parallel_chain = RunnableParallel({
        'context': retriever | RunnableLambda(format_docs),
        'question': RunnablePassthrough()
    })
    return parallel_chain

# Streamlit UI
st.title("YouTube Transcript Q&A")

yt_url = st.text_input("Enter YouTube Video URL")
user_question = st.text_input("Enter your question about the video")

video_id = None
transcript_language = None
available_languages = {}

if yt_url:
    video_id = extract_youtube_video_id(yt_url)
    if video_id:
        lang_check = get_transcript(video_id)
        if "available_languages" in lang_check:
            st.warning("English transcript not found. Please select a language from the dropdown.")
            available_languages = lang_check["available_languages"]
            language_display = st.selectbox("Choose transcript language", list(available_languages.values()))
            # Map back to code
            transcript_language = next(code for code, name in available_languages.items() if name == language_display)
        else:
            transcript_language = "en"
    else:
        st.error("Invalid YouTube URL.")

# Process
if st.button("Get Answer"):
    if not yt_url or not user_question:
        st.error("Please enter both a video URL and a question.")
    elif not video_id or not transcript_language:
        st.error("Please ensure a valid video and transcript language are selected.")
    else:
        transcript_result = get_transcript(video_id, lang=transcript_language)
        if "transcript" not in transcript_result:
            st.error(transcript_result.get("error", "Transcript could not be fetched."))
        else:
            transcript = transcript_result["transcript"]
            retriever = embed_transcript(transcript)
            chain = create_chain(retriever)
            with st.spinner("Processing..."):
                main_chain = chain | prompt | model | parser
                response = main_chain.invoke(user_question)
            st.markdown("### Answer:")
            st.write(response)