app.py

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'<a href="data:file/json;base64,{b64}" download="search_history.json">Download JSON</a>'
            st.markdown(href, unsafe_allow_html=True)

if __name__ == "__main__":
    main()