Spaces:
Sleeping
Sleeping
""" | |
Video Keyword Finder with Gradio Interface | |
======================================== | |
A Python script that finds and timestamps specific keywords within video files. | |
It transcribes the audio using Whisper AI and finds all occurrences of specified keywords | |
with their timestamps and surrounding context. Supports both local video files and YouTube URLs. | |
Requirements | |
----------- | |
- Python 3.8 or higher | |
- ffmpeg (must be installed and accessible in system PATH) | |
- GPU recommended but not required | |
""" | |
import whisper_timestamped as whisper | |
import datetime | |
import os | |
import tempfile | |
import yt_dlp | |
import re | |
import logging | |
import math | |
import subprocess | |
import glob | |
import gradio as gr | |
# Set up logging | |
logging.basicConfig( | |
level=logging.INFO, | |
format='%(asctime)s - %(levelname)s - %(message)s', | |
handlers=[ | |
logging.StreamHandler() | |
] | |
) | |
def parse_time(time_str): | |
"""Convert time string (HH:MM:SS) to seconds""" | |
if not time_str or time_str.strip() == "": | |
return None | |
try: | |
time_parts = list(map(int, time_str.split(':'))) | |
if len(time_parts) == 3: # HH:MM:SS | |
return time_parts[0] * 3600 + time_parts[1] * 60 + time_parts[2] | |
elif len(time_parts) == 2: # MM:SS | |
return time_parts[0] * 60 + time_parts[1] | |
else: | |
return int(time_str) # Just seconds | |
except: | |
raise ValueError("Time must be in HH:MM:SS, MM:SS, or seconds format") | |
def format_time(seconds): | |
"""Convert seconds to HH:MM:SS format""" | |
return str(datetime.timedelta(seconds=int(seconds))) | |
def is_youtube_url(url): | |
"""Check if the provided string is a YouTube URL""" | |
if not url or url.strip() == "": | |
return False | |
youtube_regex = ( | |
r'(https?://)?(www\.)?' | |
r'(youtube|youtu|youtube-nocookie)\.(com|be)/' | |
r'(watch\?v=|embed/|v/|.+\?v=)?([^&=%\?]{11})' | |
) | |
return bool(re.match(youtube_regex, url)) | |
def download_youtube_video(url): | |
"""Download YouTube video and return path to temporary file""" | |
temp_dir = tempfile.gettempdir() | |
temp_file = os.path.join(temp_dir, 'youtube_video.mp4') | |
ydl_opts = { | |
'format': 'best[ext=mp4]', | |
'outtmpl': temp_file, | |
'quiet': False, | |
'progress_hooks': [lambda d: logging.info(f"Download progress: {d.get('status', 'unknown')}")], | |
'socket_timeout': 30, | |
} | |
try: | |
logging.info(f"Starting download of YouTube video: {url}") | |
with yt_dlp.YoutubeDL(ydl_opts) as ydl: | |
info = ydl.extract_info(url, download=True) | |
logging.info(f"Video info extracted: {info.get('title', 'Unknown title')}") | |
if os.path.exists(temp_file): | |
file_size = os.path.getsize(temp_file) | |
logging.info(f"Download complete. File size: {file_size / (1024*1024):.2f} MB") | |
return temp_file | |
else: | |
raise Exception("Download completed but file not found") | |
except Exception as e: | |
logging.error(f"Error downloading YouTube video: {str(e)}") | |
raise | |
def get_video_duration(video_path): | |
"""Get video duration using ffprobe""" | |
cmd = [ | |
'ffprobe', | |
'-v', 'error', | |
'-show_entries', 'format=duration', | |
'-of', 'default=noprint_wrappers=1:nokey=1', | |
video_path | |
] | |
try: | |
output = subprocess.check_output(cmd).decode().strip() | |
return float(output) | |
except subprocess.CalledProcessError as e: | |
logging.error(f"Error getting video duration: {str(e)}") | |
raise | |
def split_video(video_path, segment_duration=120): | |
"""Split video into segments using ffmpeg""" | |
try: | |
temp_dir = tempfile.gettempdir() | |
segment_pattern = os.path.join(temp_dir, 'segment_%03d.mp4') | |
# Remove any existing segments | |
for old_segment in glob.glob(os.path.join(temp_dir, 'segment_*.mp4')): | |
try: | |
os.remove(old_segment) | |
except: | |
pass | |
# Split video into segments | |
cmd = [ | |
'ffmpeg', | |
'-i', video_path, | |
'-f', 'segment', | |
'-segment_time', str(segment_duration), | |
'-c', 'copy', | |
'-reset_timestamps', '1', | |
segment_pattern | |
] | |
logging.info("Splitting video into segments...") | |
subprocess.run(cmd, check=True, capture_output=True) | |
# Get list of generated segments | |
segments = sorted(glob.glob(os.path.join(temp_dir, 'segment_*.mp4'))) | |
logging.info(f"Created {len(segments)} segments") | |
return segments | |
except Exception as e: | |
logging.error(f"Error splitting video: {str(e)}") | |
raise | |
def process_segments(segments, keywords): | |
"""Process each segment sequentially""" | |
results = {keyword: [] for keyword in keywords} | |
# Load whisper model once | |
logging.info("Loading Whisper model...") | |
model = whisper.load_model("base") | |
for i, segment_path in enumerate(segments): | |
try: | |
segment_num = int(re.search(r'segment_(\d+)', segment_path).group(1)) | |
start_time = segment_num * 120 # Each segment is 120 seconds | |
logging.info(f"Processing segment {i+1}/{len(segments)} (starting at {format_time(start_time)})") | |
# Transcribe segment | |
audio = whisper.load_audio(segment_path) | |
transcription = whisper.transcribe(model, audio) | |
# Process results | |
for segment in transcription['segments']: | |
text = segment['text'].lower() | |
timestamp = segment['start'] + start_time # Adjust timestamp relative to full video | |
for keyword in keywords: | |
if keyword.lower() in text: | |
results[keyword].append(( | |
timestamp, | |
segment['text'] | |
)) | |
logging.info(f"Found keyword '{keyword}' at {format_time(timestamp)}: {segment['text']}") | |
except Exception as e: | |
logging.error(f"Error processing segment {segment_path}: {str(e)}") | |
continue | |
finally: | |
# Clean up segment file | |
try: | |
os.remove(segment_path) | |
except: | |
pass | |
return results | |
def find_keywords_in_video(video_path, keywords, begin_time=None, end_time=None): | |
"""Find timestamps for keywords in video transcription""" | |
try: | |
logging.info(f"Processing video: {video_path}") | |
logging.info(f"Searching for keywords: {keywords}") | |
# Convert keywords string to list and clean up | |
if isinstance(keywords, str): | |
keywords = [k.strip() for k in keywords.split(',')] | |
# Get video duration | |
duration = get_video_duration(video_path) | |
logging.info(f"Video duration: {duration:.2f} seconds") | |
# Set time bounds | |
start = parse_time(begin_time) if begin_time else 0 | |
end = min(parse_time(end_time) if end_time else duration, duration) | |
if start is not None and end is not None and start >= end: | |
raise ValueError("End time must be greater than start time") | |
# Split video into segments | |
segment_duration = 120 # 2 minutes per segment | |
segments = split_video(video_path, segment_duration) | |
# Process segments sequentially | |
results = process_segments(segments, keywords) | |
return results | |
except Exception as e: | |
logging.error(f"Error processing video: {str(e)}") | |
raise | |
def process_video(video_file, youtube_url, keywords_input, start_time="0:00", end_time=None): | |
""" | |
Process video file or YouTube URL and find keywords | |
""" | |
try: | |
# Convert keywords string to list | |
keywords = [k.strip() for k in keywords_input.split(',') if k.strip()] | |
if not keywords: | |
return "Please enter at least one keyword (separated by commas)" | |
# Handle input source | |
if youtube_url and youtube_url.strip(): | |
if not is_youtube_url(youtube_url): | |
return "Invalid YouTube URL. Please provide a valid URL." | |
video_path = download_youtube_video(youtube_url) | |
cleanup_needed = True | |
elif video_file is not None: | |
video_path = video_file | |
cleanup_needed = False | |
else: | |
return "Please provide either a video file or a YouTube URL" | |
try: | |
# Find keywords | |
results = find_keywords_in_video( | |
video_path=video_path, | |
keywords=keywords, | |
begin_time=start_time if start_time else None, | |
end_time=end_time if end_time else None | |
) | |
# Format results | |
output = [] | |
total_matches = sum(len(matches) for matches in results.values()) | |
output.append(f"Total matches found: {total_matches}\n") | |
if total_matches == 0: | |
output.append("No matches found for any keywords.") | |
else: | |
for keyword, matches in results.items(): | |
if matches: | |
output.append(f"\nResults for '{keyword}':") | |
for timestamp, context in matches: | |
output.append(f"[{format_time(timestamp)}] {context}") | |
else: | |
output.append(f"\nNo occurrences found for '{keyword}'") | |
return "\n".join(output) | |
finally: | |
# Cleanup temporary files | |
if cleanup_needed and os.path.exists(video_path): | |
try: | |
os.remove(video_path) | |
except: | |
pass | |
except Exception as e: | |
logging.error(f"Error processing video: {str(e)}") | |
return f"Error processing video: {str(e)}" | |
# Create Gradio interface | |
with gr.Blocks(title="Video Keyword Finder", theme=gr.themes.Soft()) as demo: | |
gr.Markdown(""" | |
# 🎥 Video Keyword Finder | |
Find timestamps for specific keywords in your videos using AI transcription. | |
Upload a video file or provide a YouTube URL, then enter the keywords you want to find. | |
""") | |
with gr.Row(): | |
with gr.Column(): | |
video_input = gr.File( | |
label="Upload Video File", | |
file_types=["video"], | |
type="filepath" | |
) | |
youtube_url = gr.Textbox( | |
label="Or Enter YouTube URL", | |
placeholder="https://www.youtube.com/watch?v=..." | |
) | |
keywords = gr.Textbox( | |
label="Keywords (comma-separated)", | |
placeholder="enter, your, keywords, here" | |
) | |
with gr.Row(): | |
start_time = gr.Textbox( | |
label="Start Time (HH:MM:SS)", | |
placeholder="0:00", | |
value="0:00" | |
) | |
end_time = gr.Textbox( | |
label="End Time (HH:MM:SS)", | |
placeholder="Optional" | |
) | |
submit_btn = gr.Button("Find Keywords", variant="primary") | |
with gr.Column(): | |
output = gr.Textbox( | |
label="Results", | |
placeholder="Keywords and timestamps will appear here...", | |
lines=20 | |
) | |
gr.Markdown(""" | |
### Instructions: | |
1. Upload a video file or paste a YouTube URL | |
2. Enter keywords separated by commas (e.g., "hello, world, python") | |
3. Optionally set start and end times (format: HH:MM:SS) | |
4. Click "Find Keywords" and wait for results | |
Note: Processing time depends on video length. A 1-hour video typically takes 15-30 minutes. | |
""") | |
submit_btn.click( | |
fn=process_video, | |
inputs=[video_input, youtube_url, keywords, start_time, end_time], | |
outputs=output | |
) | |
if __name__ == "__main__": | |
demo.launch() | |