import streamlit as st import pandas as pd import numpy as np from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import torch import json import os from pathlib import Path from datetime import datetime import edge_tts import asyncio import base64 from openai import OpenAI import anthropic import streamlit.components.v1 as components # Page configuration st.set_page_config( page_title="Video Search with Speech", page_icon="🎥", layout="wide" ) # Initialize session state if 'messages' not in st.session_state: st.session_state['messages'] = [] if 'search_history' not in st.session_state: st.session_state['search_history'] = [] if 'last_voice_input' not in st.session_state: st.session_state['last_voice_input'] = "" # Load environment variables openai_client = OpenAI() claude_client = anthropic.Anthropic() # Initialize the speech component speech_component = components.declare_component("speech_recognition", path="mycomponent") class VideoSearch: def __init__(self): self.text_model = SentenceTransformer('all-MiniLM-L6-v2') self.load_dataset() def load_dataset(self): """Load the Omega Multimodal dataset""" try: # Load dataset from Hugging Face self.dataset = pd.read_csv("paste.txt") self.prepare_features() except Exception as e: st.error(f"Error loading dataset: {e}") self.create_dummy_data() def prepare_features(self): """Prepare and cache embeddings""" # Convert string representations of embeddings back to numpy arrays self.video_embeds = np.array([json.loads(e) if isinstance(e, str) else e for e in self.dataset.video_embed]) self.text_embeds = np.array([json.loads(e) if isinstance(e, str) else e for e in self.dataset.description_embed]) def create_dummy_data(self): """Create dummy data for testing""" self.dataset = pd.DataFrame({ 'video_id': [f'video_{i}' for i in range(10)], 'youtube_id': ['dQw4w9WgXcQ'] * 10, # Example YouTube ID 'description': ['Sample video description'] * 10, 'views': [1000] * 10, 'start_time': [0] * 10, 'end_time': [60] * 10 }) # Create dummy embeddings self.video_embeds = np.random.randn(10, 384) # Match model dimensions self.text_embeds = np.random.randn(10, 384) def search(self, query, top_k=5): """Search videos using query""" query_embedding = self.text_model.encode([query])[0] # Compute similarities video_sims = cosine_similarity([query_embedding], self.video_embeds)[0] text_sims = cosine_similarity([query_embedding], self.text_embeds)[0] # Combine similarities combined_sims = 0.5 * video_sims + 0.5 * text_sims # Get top results top_indices = np.argsort(combined_sims)[-top_k:][::-1] results = [] for idx in top_indices: results.append({ 'video_id': self.dataset.iloc[idx]['video_id'], 'youtube_id': self.dataset.iloc[idx]['youtube_id'], 'description': self.dataset.iloc[idx]['description'], 'start_time': self.dataset.iloc[idx]['start_time'], 'end_time': self.dataset.iloc[idx]['end_time'], 'relevance_score': float(combined_sims[idx]), 'views': self.dataset.iloc[idx]['views'] }) return results async def generate_speech(text, voice="en-US-AriaNeural"): """Generate speech using Edge TTS""" if not text.strip(): return None communicate = edge_tts.Communicate(text, voice) audio_file = f"speech_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp3" await communicate.save(audio_file) return audio_file def process_with_gpt4(prompt): """Process text with GPT-4""" try: response = openai_client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content except Exception as e: st.error(f"Error with GPT-4: {e}") return None def process_with_claude(prompt): """Process text with Claude""" try: response = claude_client.messages.create( model="claude-3-sonnet-20240229", max_tokens=1000, messages=[{"role": "user", "content": prompt}] ) return response.content[0].text except Exception as e: st.error(f"Error with Claude: {e}") return None def main(): st.title("🎥 Video Search with Speech Recognition") # Initialize video search search = VideoSearch() # Create tabs tab1, tab2, tab3 = st.tabs(["🔍 Search", "🎙️ Voice Input", "💾 History"]) with tab1: st.subheader("Search Videos") # Text search query = st.text_input("Enter your search query:") col1, col2 = st.columns(2) with col1: search_button = st.button("🔍 Search") with col2: num_results = st.slider("Number of results:", 1, 10, 5) if search_button and query: results = search.search(query, num_results) st.session_state['search_history'].append({ 'query': query, 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'results': results }) for i, result in enumerate(results, 1): with st.expander(f"Result {i}: {result['description'][:100]}...", expanded=i==1): cols = st.columns([2, 1]) with cols[0]: st.markdown(f"**Full Description:**") st.write(result['description']) st.markdown(f"**Time Range:** {result['start_time']}s - {result['end_time']}s") st.markdown(f"**Views:** {result['views']:,}") with cols[1]: st.markdown(f"**Relevance Score:** {result['relevance_score']:.2%}") if result['youtube_id']: st.video(f"https://youtube.com/watch?v={result['youtube_id']}&t={result['start_time']}") # Generate audio summary if st.button(f"🔊 Generate Audio Summary", key=f"audio_{i}"): summary = f"Video summary: {result['description'][:200]}" audio_file = asyncio.run(generate_speech(summary)) if audio_file: st.audio(audio_file) # Cleanup audio file if os.path.exists(audio_file): os.remove(audio_file) with tab2: st.subheader("Voice Input") # Speech recognition component voice_input = speech_component() if voice_input and voice_input != st.session_state['last_voice_input']: st.session_state['last_voice_input'] = voice_input st.markdown("**Transcribed Text:**") st.write(voice_input) cols = st.columns(3) with cols[0]: if st.button("🔍 Search Videos"): results = search.search(voice_input, num_results) st.session_state['search_history'].append({ 'query': voice_input, 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'results': results }) for i, result in enumerate(results, 1): with st.expander(f"Result {i}: {result['description'][:100]}...", expanded=i==1): st.write(result['description']) if result['youtube_id']: st.video(f"https://youtube.com/watch?v={result['youtube_id']}&t={result['start_time']}") with cols[1]: if st.button("🤖 Process with GPT-4"): gpt_response = process_with_gpt4(voice_input) if gpt_response: st.markdown("**GPT-4 Response:**") st.write(gpt_response) with cols[2]: if st.button("🧠 Process with Claude"): claude_response = process_with_claude(voice_input) if claude_response: st.markdown("**Claude Response:**") st.write(claude_response) with tab3: st.subheader("Search History") if st.button("🗑️ Clear History"): st.session_state['search_history'] = [] st.experimental_rerun() for i, entry in enumerate(reversed(st.session_state['search_history'])): with st.expander(f"Query: {entry['query']} ({entry['timestamp']})", expanded=False): st.markdown(f"**Original Query:** {entry['query']}") st.markdown(f"**Time:** {entry['timestamp']}") for j, result in enumerate(entry['results'], 1): st.markdown(f"**Result {j}:**") st.write(result['description']) if result['youtube_id']: st.video(f"https://youtube.com/watch?v={result['youtube_id']}&t={result['start_time']}") # Sidebar configuration with st.sidebar: st.subheader("⚙️ Configuration") st.markdown("**Video Search Settings**") st.slider("Default Results:", 1, 10, 5, key="default_results") st.markdown("**Voice Settings**") st.selectbox("TTS Voice:", ["en-US-AriaNeural", "en-US-GuyNeural", "en-GB-SoniaNeural"], key="tts_voice") st.markdown("**Model Settings**") st.selectbox("Text Embedding Model:", ["all-MiniLM-L6-v2", "paraphrase-multilingual-MiniLM-L12-v2"], key="embedding_model") if st.button("📥 Download Search History"): # Convert history to JSON history_json = json.dumps(st.session_state['search_history'], indent=2) b64 = base64.b64encode(history_json.encode()).decode() href = f'Download JSON' st.markdown(href, unsafe_allow_html=True) if __name__ == "__main__": main()