import gradio as gr
import torch
import yt_dlp
import os
import subprocess
import json
from threading import Thread
from transformers import AutoTokenizer, AutoModelForCausalLM
import spaces
import moviepy.editor as mp
import time
import langdetect
import uuid
HF_TOKEN = os.environ.get("HF_TOKEN")
print("Starting the program...")
model_path = "Qwen/Qwen2.5-7B-Instruct"
print(f"Loading model {model_path}...")
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.float16, trust_remote_code=True).cuda()
model = model.eval()
print("Model successfully loaded.")
def generate_unique_filename(extension):
return f"{uuid.uuid4()}{extension}"
def cleanup_files(*files):
for file in files:
if file and os.path.exists(file):
os.remove(file)
print(f"Removed file: {file}")
def download_youtube_audio(url):
print(f"Downloading audio from YouTube: {url}")
output_path = generate_unique_filename(".wav")
ydl_opts = {
'format': 'bestaudio/best',
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'wav',
}],
'outtmpl': output_path,
'keepvideo': True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([url])
# Check if the file was renamed to .wav.wav
if os.path.exists(output_path + ".wav"):
os.rename(output_path + ".wav", output_path)
if os.path.exists(output_path):
print(f"Audio download completed. File saved at: {output_path}")
print(f"File size: {os.path.getsize(output_path)} bytes")
else:
print(f"Error: File {output_path} not found after download.")
return output_path
@spaces.GPU(duration=90)
def transcribe_audio(file_path):
print(f"Starting transcription of file: {file_path}")
temp_audio = None
if file_path.endswith(('.mp4', '.avi', '.mov', '.flv')):
print("Video file detected. Extracting audio...")
try:
video = mp.VideoFileClip(file_path)
temp_audio = generate_unique_filename(".wav")
video.audio.write_audiofile(temp_audio)
file_path = temp_audio
except Exception as e:
print(f"Error extracting audio from video: {e}")
raise
print(f"Does the file exist? {os.path.exists(file_path)}")
print(f"File size: {os.path.getsize(file_path) if os.path.exists(file_path) else 'N/A'} bytes")
output_file = generate_unique_filename(".json")
command = [
"insanely-fast-whisper",
"--file-name", file_path,
"--device-id", "0",
"--model-name", "openai/whisper-large-v3",
"--task", "transcribe",
"--timestamp", "chunk",
"--transcript-path", output_file
]
print(f"Executing command: {' '.join(command)}")
try:
result = subprocess.run(command, check=True, capture_output=True, text=True)
print(f"Standard output: {result.stdout}")
print(f"Error output: {result.stderr}")
except subprocess.CalledProcessError as e:
print(f"Error running insanely-fast-whisper: {e}")
print(f"Standard output: {e.stdout}")
print(f"Error output: {e.stderr}")
raise
print(f"Reading transcription file: {output_file}")
try:
with open(output_file, "r") as f:
transcription = json.load(f)
except json.JSONDecodeError as e:
print(f"Error decoding JSON: {e}")
print(f"File content: {open(output_file, 'r').read()}")
raise
if "text" in transcription:
result = transcription["text"]
else:
result = " ".join([chunk["text"] for chunk in transcription.get("chunks", [])])
print("Transcription completed.")
# Cleanup
cleanup_files(output_file)
if temp_audio:
cleanup_files(temp_audio)
return result
@spaces.GPU(duration=90)
def generate_summary_stream(transcription):
print("Starting summary generation...")
print(f"Transcription length: {len(transcription)} characters")
detected_language = langdetect.detect(transcription)
prompt = f"""Summarize the following video transcription in 150-300 words.
The summary should be in the same language as the transcription, which is detected as {detected_language}.
Please ensure that the summary captures the main points and key ideas of the transcription:
{transcription[:300000]}..."""
response, history = model.chat(tokenizer, prompt, history=[])
print(f"Final summary generated: {response[:100]}...")
print("Summary generation completed.")
return response
def process_youtube(url):
if not url:
print("YouTube URL not provided.")
return "Please enter a YouTube URL.", None
print(f"Processing YouTube URL: {url}")
audio_file = None
try:
audio_file = download_youtube_audio(url)
if not os.path.exists(audio_file):
raise FileNotFoundError(f"File {audio_file} does not exist after download.")
print(f"Audio file found: {audio_file}")
print("Starting transcription...")
transcription = transcribe_audio(audio_file)
print(f"Transcription completed. Length: {len(transcription)} characters")
return transcription, None
except Exception as e:
print(f"Error processing YouTube: {e}")
return f"Processing error: {str(e)}", None
finally:
if audio_file and os.path.exists(audio_file):
cleanup_files(audio_file)
print(f"Directory content after processing: {os.listdir('.')}")
def process_uploaded_video(video_path):
print(f"Processing uploaded video: {video_path}")
try:
print("Starting transcription...")
transcription = transcribe_audio(video_path)
print(f"Transcription completed. Length: {len(transcription)} characters")
return transcription, None
except Exception as e:
print(f"Error processing video: {e}")
return f"Processing error: {str(e)}", None
print("Setting up Gradio interface...")
with gr.Blocks(theme=gr.themes.Soft()) as demo:
gr.Markdown(
"""
# 🎥 Video Transcription and Smart Summary
Upload a video or provide a YouTube link to get a transcription and AI-generated summary. HF Zero GPU has a usage time limit. So if you want to run longer videos I recommend you clone the space. Remove @Spaces.gpu from the code and run it locally on your GPU!
"""
)
with gr.Tabs():
with gr.TabItem("📤 Video Upload"):
video_input = gr.Video(label="Drag and drop or click to upload")
video_button = gr.Button("🚀 Process Video", variant="primary")
with gr.TabItem("🔗 YouTube Link"):
url_input = gr.Textbox(label="Paste YouTube URL here", placeholder="https://www.youtube.com/watch?v=...")
url_button = gr.Button("🚀 Process URL", variant="primary")
with gr.Row():
with gr.Column():
transcription_output = gr.Textbox(label="📝 Transcription", lines=10, show_copy_button=True)
with gr.Column():
summary_output = gr.Textbox(label="📊 Summary", lines=10, show_copy_button=True)
summary_button = gr.Button("📝 Generate Summary", variant="secondary")
gr.Markdown(
"""
### How to use:
1. Upload a video or paste a YouTube link.
2. Click 'Process' to get the transcription.
3. Click 'Generate Summary' to get a summary of the content.
*Note: Processing may take a few minutes depending on the video length.*
"""
)
def process_video_and_update(video):
if video is None:
return "No video uploaded.", "Please upload a video."
print(f"Video received: {video}")
transcription, _ = process_uploaded_video(video)
print(f"Returned transcription: {transcription[:100] if transcription else 'No transcription generated'}...")
return transcription or "Transcription error", ""
video_button.click(process_video_and_update, inputs=[video_input], outputs=[transcription_output, summary_output])
url_button.click(process_youtube, inputs=[url_input], outputs=[transcription_output, summary_output])
summary_button.click(generate_summary_stream, inputs=[transcription_output], outputs=[summary_output])
print("Launching Gradio interface...")
demo.launch()