app.py

# other imports
import streamlit as st, os
import spaces
import tempfile
from pytube import YouTube
import whisper
from dotenv import load_dotenv

# langchain imports
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_openai import OpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough


# initialize session state variables if not already initialized
if 'chain' not in st.session_state:
    st.session_state.chain = None
if 'transcription' not in st.session_state:
    st.session_state.transcription = None
if 'thumbnail_url' not in st.session_state:
    st.session_state.thumbnail_url = None



# function for getting transcribing YouTube video
def get_yt_trans(yt_obj):
    video_audio = yt_obj.streams.filter(only_audio=True).first()
    # define model for audio to text conversion
    audio_to_text_model = whisper.load_model('base')
    # store transcription in local file
    with tempfile.TemporaryDirectory() as tmpdir:
        audio_file = video_audio.download(output_path=tmpdir)
        st.session_state.transcription = audio_to_text_model.transcribe(audio_file, fp16=False)["text"].strip()
        # with open("transcription.txt", "w") as file:
        #     file.write(transcription)
    


# function for chunking transcription text
def get_vid_text_chunks(transcription_text):
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=20)
    chunked_text = text_splitter.split_text(transcription_text)
    return chunked_text


# function for creating vector db
def create_vecdb(chunks):
    vecdb = FAISS.from_texts(chunks, OpenAIEmbeddings())
    # vecdb.save_local('vec_store') ## if needed to store and retrieve locally
    return vecdb


# define api keys
load_dotenv()
os.environ['OPEN_API_KEY'] = os.getenv('OPENAI_API_KEY')

# define LLM
llm = OpenAI(temperature=0)

# Set page configs
st.set_page_config(
    page_title="YouTube Talks",
    page_icon='📽️'
)
st.header('Query YouTube Videos!')
thumbnail_placeholder = st.empty()
st.sidebar.header("URL details:")
video_url = st.sidebar.text_input("Enter video URL")
submit_button = st.sidebar.button("Submit")
user_query = st.text_input("Query the Video!")
video_query_button = st.button("Ask video!")
progress_updates = st.sidebar.empty()
main_ph = st.empty()

# on button press
if submit_button:
    progress_updates.text("Transcribing Video...")
    # get video transcription
    yt_obj = YouTube(video_url)
    # display video thumbnail
    st.session_state.thumbnail_url = yt_obj.thumbnail_url
    thumbnail_placeholder.image(st.session_state.thumbnail_url)
    # get transcription
    get_yt_trans(yt_obj)
    st.sidebar.subheader("Transcription:")
    st.sidebar.write(st.session_state.transcription)
    progress_updates.text("Making Text Chunks...")
    # get text chunks
    chunks = get_vid_text_chunks(st.session_state.transcription)  
    progress_updates.text("Creating Vector DB...")
    # create vector db
    vector_db = create_vecdb(chunks)          
    # define main prompt
    p_template = '''Answer the question based on the context below. If you can't 
answer the question, reply "I don't know".

Context: {context}

Question: {question}'''
    
    prompt = ChatPromptTemplate.from_template(p_template)
    # define output parser
    parser = StrOutputParser()
    # define main chain
    st.session_state.chain = {"context": vector_db.as_retriever(), "question": RunnablePassthrough()} | prompt | llm | parser
    progress_updates.text("Video Transcribed Successfully!!!")


if video_query_button:
    # keep displaying Transcription and Thumbnail in window
    st.sidebar.subheader("Transcription:")
    st.sidebar.write(st.session_state.transcription)
    thumbnail_placeholder.image(st.session_state.thumbnail_url)
    # if video not transcribed display error
    if st.session_state.chain is None:
        st.error("Please transcribe a video first by submitting a URL.")
    else:
        main_ph.text("Fetching Results...")
        # print results
        st.subheader("Result:")
        main_ph.text("Displaying Results...")
        st.write(st.session_state.chain.invoke(user_query))