video_chat / app.py
JustusI's picture
Update app.py
99bcf2e verified
import streamlit as st
import tempfile
import os
import yt_dlp
from moviepy.editor import VideoFileClip
from openai import OpenAI
client = OpenAI()
# Set your OpenAI API key (make sure it's set in Hugging Face Spaces secrets)
#openai.api_key = os.getenv("OPENAI_API_KEY")
# ---------------------------
# Helper Functions
# ---------------------------
def download_video(youtube_url: str, output_path: str) -> str:
"""Download a YouTube video using yt-dlp and save it to the given output path."""
try:
ydl_opts = {
'format': 'best',
'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
'noplaylist': True,
'quiet': True,
'cookiesfrombrowser': ('chrome',), # Extract cookies from Chrome
'verbose': True # Optional: for more detailed output
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(youtube_url, download=True)
title = info.get('title', 'video')
video_path = os.path.join(output_path, f"{title}.mp4")
return video_path
except Exception as e:
st.error(f"Error downloading video: {e}")
return None
def extract_audio(video_path: str) -> str:
"""Extract audio from the video file and save as MP3."""
try:
# Sanitize filename if needed (e.g., replace problematic characters)
safe_video_path = video_path.replace("|", "_").replace(":", "_")
clip = VideoFileClip(safe_video_path)
audio_path = safe_video_path.replace(".mp4", ".mp3")
clip.audio.write_audiofile(audio_path, codec='mp3')
clip.close()
return audio_path
except Exception as e:
st.error(f"Error extracting audio: {e}")
return None
# audio_file = open("/path/to/file/speech.mp3", "rb")
# transcription = client.audio.transcriptions.create(
# model="whisper-1",
# file=audio_file,
# response_format="text"
# )
# print(transcription.text)
def transcribe_audio(audio_path: str) -> str:
"""Transcribe the audio to text using OpenAI's Whisper API."""
try:
with open(audio_path, "rb") as audio_file:
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
#response_format="text"
)
return transcript.text
except Exception as e:
st.error(f"Error transcribing audio: {e}")
return ""
def generate_summary(transcript_text: str) -> str:
"""Generate a concise summary of the transcript using OpenAI."""
prompt = f"Summarize the following video transcript in a concise manner, highlighting the key points that the user should know:\n\n{transcript_text}. Feel free to use bullet points, bold, italics and headers to empasize key points where necessary"
messages = [
{"role": "system", "content": "You are an helpful assistant."},
{"role": "user", "content": prompt}
]
completion = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
return completion.choices[0].message.content.strip()
def get_chat_response(transcript_text: str, conversation_history: list, user_query: str) -> str:
"""Generate a chat response using the transcript as context."""
messages = conversation_history + [
{"role": "user", "content": f"Based on the video transcript:\n\n{transcript_text}\n\nQuestion: {user_query}"}
]
completion = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
return completion.choices[0].message.content.strip()
# ---------------------------
# Sidebar: Input Options
# ---------------------------
#st.sidebar.title("Video Input Options")
st.sidebar.title("Upload a video")
input_mode = st.sidebar.radio("Select Input Type", ("Upload Video")) #, "YouTube URL"))
transcript_text = ""
if input_mode == "Upload Video":
uploaded_video = st.sidebar.file_uploader("Upload a video file (MP4)", type="mp4")
# elif input_mode == "YouTube URL":
# youtube_url = st.sidebar.text_input("Enter YouTube video URL:")
# ---------------------------
# Process Video Input
# ---------------------------
if input_mode == "Upload Video" and uploaded_video is not None:
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp:
tmp.write(uploaded_video.read())
video_path = tmp.name
st.sidebar.success("Video uploaded successfully!")
audio_path = extract_audio(video_path)
if audio_path:
transcript_text = transcribe_audio(audio_path)
if transcript_text:
st.sidebar.success("Audio transcribed successfully!")
# elif input_mode == "YouTube URL" and youtube_url:
# with tempfile.TemporaryDirectory() as temp_dir:
# video_path = download_video(youtube_url, temp_dir)
# if video_path:
# st.sidebar.success("Video downloaded successfully!")
# audio_path = extract_audio(video_path)
# if audio_path:
# transcript_text = transcribe_audio(audio_path)
# if transcript_text:
# st.sidebar.success("Audio transcribed successfully!")
if transcript_text:
st.session_state.transcript_text = transcript_text
# ---------------------------
# Sidebar: Action Selection
# ---------------------------
st.sidebar.title("Select Action")
action_mode = st.sidebar.radio("Choose Action", ("Summary", "Chat"))
# ---------------------------
# Session State Initialization for Chat
# ---------------------------
if "chat_history" not in st.session_state:
st.session_state.chat_history = [{"role": "assistant", "content": "Hi, how can I help you with the video content?"}]
# ---------------------------
# Main Display Area
# ---------------------------
st.title("Video Chat")
st.write('Tired of watching long boring videos? Summarize your videos in seconds or just chat!')
if "transcript_text" not in st.session_state or not st.session_state.transcript_text:
st.info("Please provide a video input from the sidebar to begin.")
else:
transcript_text = st.session_state.transcript_text
if action_mode == "Summary":
#st.header("Summary & Key Points")
with st.spinner("Generating summary..."):
summary = generate_summary(transcript_text)
st.write(summary)
elif action_mode == "Chat":
st.header("Chat with Your Study Companion")
for msg in st.session_state.chat_history:
st.chat_message(msg["role"]).write(msg["content"])
user_query = st.chat_input("Ask a question about the video content:")
if user_query:
st.session_state.chat_history.append({"role": "user", "content": user_query})
st.chat_message("user").write(user_query)
with st.spinner("Processing your question..."):
response = get_chat_response(transcript_text, st.session_state.chat_history, user_query)
st.session_state.chat_history.append({"role": "assistant", "content": response})
st.chat_message("assistant").write(response)