app.py

"""
Hugging Face Spaces Demo - Voice Assistant API
Multi-language voice assistant with division matching and contact search
"""

import gradio as gr
import logging
from typing import Optional, Tuple
import numpy as np

# Import existing services
from embedding_service import EmbeddingService
from name_extraction_service import NameExtractor
from voice_processing_service import VoiceProcessor
from contact_search_service import ContactSearchService

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Global services (initialized once)
embedding_service: Optional[EmbeddingService] = None
name_extractor: Optional[NameExtractor] = None
voice_processor: Optional[VoiceProcessor] = None
contact_search_service: Optional[ContactSearchService] = None


def initialize_services():
    """Initialize all AI services (called once on startup)"""
    global embedding_service, name_extractor, voice_processor, contact_search_service

    logger.info("🚀 Initializing services...")

    # Initialize embedding service (fast & lightweight)
    logger.info("Loading embedding model...")
    embedding_service = EmbeddingService(model_name="all-MiniLM-L6-v2")
    logger.info("✓ Embedding service ready!")

    # Initialize name extractor
    logger.info("Loading name extraction model...")
    name_extractor = NameExtractor(model_name="urchade/gliner_small-v2.1")
    logger.info("✓ Name extractor ready!")

    # Initialize voice processor (using small model for better accuracy)
    logger.info("Loading Whisper model...")
    voice_processor = VoiceProcessor(model_size="small")  # Using small for better accuracy
    logger.info("✓ Voice processor ready!")

    # Initialize contact search
    logger.info("Loading contact database...")
    contact_search_service = ContactSearchService(name_extractor, embedding_service)
    stats = contact_search_service.get_contact_stats()
    logger.info(f"✓ Loaded {stats['total_contacts']} contacts across {stats['divisions']} divisions")

    return stats


def format_division_matches(matches, names):
    """Format division matching results for display"""
    if not matches:
        return "No matches found."

    output = []

    if names:
        output.append(f"**Extracted Names:** {', '.join(names)}\n")

    output.append("### 🎯 Division Matches:\n")

    for i, match in enumerate(matches[:3], 1):
        confidence_pct = match.confidence * 100
        confidence_bar = "🟢" * int(confidence_pct / 20) + "⚪" * (5 - int(confidence_pct / 20))

        output.append(f"**{i}. {match.division}**")
        output.append(f"   - Confidence: {confidence_pct:.1f}% {confidence_bar}")
        if match.department:
            output.append(f"   - Department: {match.department}")
        output.append("")

    return "\n".join(output)


def format_contact_results(contacts, extracted_names, matched_divisions):
    """Format contact search results for display"""
    if not contacts:
        return "No contacts found."

    output = []

    if extracted_names:
        output.append(f"**Extracted Names:** {', '.join(extracted_names)}\n")

    if matched_divisions:
        output.append(f"**Matched Divisions:** {', '.join(matched_divisions[:3])}\n")

    output.append(f"### 👥 Found {len(contacts)} Contact(s):\n")

    for i, contact in enumerate(contacts[:10], 1):
        confidence_pct = contact['confidence'] * 100
        confidence_bar = "🟢" * int(confidence_pct / 20) + "⚪" * (5 - int(confidence_pct / 20))

        # Use full_name_en (English name) for display
        output.append(f"**{i}. {contact['full_name_en']}** ({contact['full_name_ar']})")
        output.append(f"   - Title: {contact['title_en']}")
        output.append(f"   - Division: {contact['division']}")
        output.append(f"   - Department: {contact['department']}")
        output.append(f"   - Phone: {contact['phone']}")
        output.append(f"   - Email: {contact['email']}")
        output.append(f"   - Confidence: {confidence_pct:.1f}% {confidence_bar}")
        output.append(f"   - Match Reason: {contact['match_reason']}")
        output.append("")

    return "\n".join(output)


def search_divisions_text(query: str) -> str:
    """Search for divisions based on text query"""
    if not query or not query.strip():
        return "Please enter a query."

    try:
        # Extract names
        names = name_extractor.extract_names(query)

        # Find matching divisions
        matches = embedding_service.find_division(query, top_k=3)

        return format_division_matches(matches, names)

    except Exception as e:
        logger.error(f"Error in division search: {e}")
        return f"Error: {str(e)}"


def search_divisions_voice(audio: Optional[Tuple[int, np.ndarray]]) -> str:
    """Search for divisions based on voice query"""
    if audio is None:
        return "Please record audio first."

    try:
        # Save audio to temporary file
        sample_rate, audio_data = audio
        temp_path = voice_processor.save_audio_array(audio_data, sample_rate)

        # Process voice query
        voice_result = voice_processor.process_voice_query(temp_path)
        query = voice_result['query']

        # Extract names
        names = name_extractor.extract_names(query)

        # Find matching divisions
        matches = embedding_service.find_division(query, top_k=3)

        # Format output
        output = []
        output.append(f"**🎤 Transcribed Text:** {query}")
        output.append(f"**🌍 Language:** {voice_result['language_name']}")
        if voice_result['was_translated']:
            output.append(f"**📝 Original:** {voice_result['original_text']}")
        output.append("")
        output.append(format_division_matches(matches, names))

        # Cleanup
        voice_processor.cleanup_temp_file(temp_path)

        return "\n".join(output)

    except Exception as e:
        logger.error(f"Error in voice division search: {e}")
        return f"Error: {str(e)}"


def search_contacts_text(query: str) -> str:
    """Search for contacts based on text query"""
    if not query or not query.strip():
        return "Please enter a query."

    try:
        # Search contacts
        contacts = contact_search_service.search_contacts(query, top_k=10, min_confidence=0.3)

        # Extract names and divisions
        names = name_extractor.extract_names(query)
        division_matches = embedding_service.find_division(query, top_k=3)
        matched_divisions = [m.division for m in division_matches]

        return format_contact_results(contacts, names, matched_divisions)

    except Exception as e:
        logger.error(f"Error in contact search: {e}")
        return f"Error: {str(e)}"


def search_contacts_voice(audio: Optional[Tuple[int, np.ndarray]]) -> str:
    """Search for contacts based on voice query"""
    if audio is None:
        return "Please record audio first."

    try:
        # Save audio to temporary file
        sample_rate, audio_data = audio
        temp_path = voice_processor.save_audio_array(audio_data, sample_rate)

        # Process voice query
        voice_result = voice_processor.process_voice_query(temp_path)
        query = voice_result['query']

        # Search contacts (this already extracts names and divisions internally)
        contacts = contact_search_service.search_contacts(query, top_k=10, min_confidence=0.3)

        # Format output - names and divisions are already extracted by search_contacts
        output = []
        output.append(f"**🎤 Transcribed Text:** {query}")
        output.append(f"**🌍 Language:** {voice_result['language_name']}")
        if voice_result['was_translated']:
            output.append(f"**📝 Original:** {voice_result['original_text']}")
        output.append("")
        # Pass empty arrays since contact_search already did the matching
        output.append(format_contact_results(contacts, [], []))

        # Cleanup
        voice_processor.cleanup_temp_file(temp_path)

        return "\n".join(output)

    except Exception as e:
        logger.error(f"Error in voice contact search: {e}")
        return f"Error: {str(e)}"


def create_demo():
    """Create the Gradio demo interface"""

    # Initialize services on startup
    stats = initialize_services()

    # Create the interface
    with gr.Blocks(title="Voice Assistant Demo", theme=gr.themes.Soft()) as demo:

        gr.Markdown(f"""
        # 🎙️ Voice Assistant Demo
        ### Multi-language voice assistant with division matching and contact search

        **Database:** {stats['total_contacts']} contacts • {stats['departments']} departments • {stats['divisions']} divisions

        **Features:**
        - 🗣️ Speech-to-text in 99+ languages
        - 🔍 Smart division matching
        - 👤 Name extraction (English & Arabic)
        - 📞 Contact search with confidence scoring
        """)

        with gr.Tabs():

            # Tab 1: Division Matching (Text)
            with gr.Tab("📝 Division Matching (Text)"):
                gr.Markdown("""
                ### Search for divisions by text query
                Try queries like:
                - "I need help from IT Security"
                - "Find someone in Finance"
                - "Connect me to Human Resources"
                - "Find Ahmed in App Dev"
                """)

                with gr.Row():
                    with gr.Column():
                        div_text_input = gr.Textbox(
                            label="Enter your query",
                            placeholder="e.g., I need help from IT Security",
                            lines=2
                        )
                        div_text_btn = gr.Button("🔍 Search Divisions", variant="primary")

                    with gr.Column():
                        div_text_output = gr.Markdown(label="Results")

                div_text_btn.click(
                    fn=search_divisions_text,
                    inputs=[div_text_input],
                    outputs=[div_text_output]
                )

            # Tab 2: Division Matching (Voice)
            with gr.Tab("🎤 Division Matching (Voice)"):
                gr.Markdown("""
                ### Search for divisions by voice
                Speak your query in any language - it will be automatically transcribed and translated.
                """)

                with gr.Row():
                    with gr.Column():
                        div_voice_input = gr.Audio(
                            sources=["microphone"],
                            type="numpy",
                            label="Record your voice query"
                        )
                        div_voice_btn = gr.Button("🔍 Search Divisions", variant="primary")

                    with gr.Column():
                        div_voice_output = gr.Markdown(label="Results")

                div_voice_btn.click(
                    fn=search_divisions_voice,
                    inputs=[div_voice_input],
                    outputs=[div_voice_output]
                )

            # Tab 3: Contact Search (Text)
            with gr.Tab("👥 Contact Search (Text)"):
                gr.Markdown("""
                ### Search for contacts by text query
                Try queries like:
                - "Find Dima in Information Technology"
                - "Ahmed Al-Malek"
                - "I need to talk to someone in Legal"
                - "Find Rashed in Finance"
                """)

                with gr.Row():
                    with gr.Column():
                        contact_text_input = gr.Textbox(
                            label="Enter your query",
                            placeholder="e.g., Find Dima in Information Technology",
                            lines=2
                        )
                        contact_text_btn = gr.Button("🔍 Search Contacts", variant="primary")

                    with gr.Column():
                        contact_text_output = gr.Markdown(label="Results")

                contact_text_btn.click(
                    fn=search_contacts_text,
                    inputs=[contact_text_input],
                    outputs=[contact_text_output]
                )

            # Tab 4: Contact Search (Voice)
            with gr.Tab("🎙️ Contact Search (Voice)"):
                gr.Markdown("""
                ### Search for contacts by voice
                Speak your query in any language to find contacts.
                """)

                with gr.Row():
                    with gr.Column():
                        contact_voice_input = gr.Audio(
                            sources=["microphone"],
                            type="numpy",
                            label="Record your voice query"
                        )
                        contact_voice_btn = gr.Button("🔍 Search Contacts", variant="primary")

                    with gr.Column():
                        contact_voice_output = gr.Markdown(label="Results")

                contact_voice_btn.click(
                    fn=search_contacts_voice,
                    inputs=[contact_voice_input],
                    outputs=[contact_voice_output]
                )

        gr.Markdown("""
        ---
        **Models:**
        - Embeddings: `sentence-transformers/all-MiniLM-L6-v2`
        - Name Extraction: `urchade/gliner_small-v2.1`
        - Speech-to-Text: `openai/whisper-tiny`

        **Supported Languages:** 99+ languages (auto-detected)
        """)

    return demo


if __name__ == "__main__":
    demo = create_demo()
    demo.launch()