app.py

import gradio as gr
import json
import os
import ssl
import base64
import urllib.request
import tempfile
import soundfile
from datetime import datetime
from typing import Dict, List, Optional, Tuple
import edge_tts
from langdetect import detect


# Custom CSS for better styling
CUSTOM_CSS = """
/* General styling */
.container {
    max-width: 900px;
    margin: auto;
}

/* Header styling */
#header {
    text-align: center;
    padding: 20px;
    margin-bottom: 30px;
}

/* Component styling */
.input-section {
    background: var(--background-fill-primary);
    padding: 20px;
    border-radius: 10px;
    margin-bottom: 20px;
}

.output-section {
    background: var(--background-fill-primary);
    padding: 20px;
    border-radius: 10px;
    box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}

/* Note styling */
.note-card {
    background: var(--background-fill-primary);
    padding: 15px;
    margin: 10px 0;
    border-radius: 8px;
    box-shadow: 0 2px 4px rgba(0,0,0,0.05);
}

/* Adaptive text colors for both themes */
.output-section .prose {
    color: var(--body-text-color) !important;
}

.output-section p, 
.output-section h1, 
.output-section h2, 
.output-section h3, 
.output-section h4, 
.output-section h5, 
.output-section h6 {
    color: var(--body-text-color) !important;
}

/* Ensure Markdown content uses theme colors */
.output-section .markdown-body {
    color: var(--body-text-color) !important;
    background-color: var(--background-fill-primary) !important;
}

/* Style blockquotes and code blocks */
.output-section blockquote {
    border-left: 3px solid var(--border-color-primary);
    color: var(--body-text-color);
    background: var(--background-fill-secondary);
}

.output-section code {
    color: var(--body-text-color);
    background: var(--background-fill-secondary);
}
"""

# Create a light theme
LIGHT_THEME = gr.themes.Default(
    primary_hue="blue",
    secondary_hue="blue",
    neutral_hue="slate",
    font=["Source Sans Pro", "ui-sans-serif", "system-ui", "sans-serif"],
    font_mono=["JetBrains Mono", "ui-monospace", "Consolas", "monospace"],
)

MULTILINGUAL_SYSYTEM_PROMPTS = {
    'english': {
        'system': """You are a world class AI assistant. From users' collection of random thoughts, you need to organize into a clear, concise, and actionable format. Please structure them into the following categories that STRICTLY follows this format:
🧠 Main Idea/Theme – Summarize the central topic.

✅ Actionable Steps – Break down what can be done next.

❓ Open Questions – Highlight any uncertainties or areas that need further exploration.

Here are thoughts:""",
        'summarize': "Summarize the text above."
    },
    'chinese': {
        'system': """你是一位世界级的AI助手。你需要将用户的随想整理成清晰、简洁和可行的格式。请严格按照以下格式将内容分类：
🧠 主要想法/主题 – 总结核心话题。

✅ 可行步骤 – 分解下一步可以做什么。

❓ 开放性问题 – 突出任何不确定性或需要进一步探索的领域。

以下是想法：""",
        'summarize': "总结上述文本。"
    },
    'german': {
        'system': """Sie sind ein erstklassiger KI-Assistent. Sie müssen die zufälligen Gedanken der Benutzer in ein klares, prägnantes und umsetzbares Format bringen. Bitte strukturieren Sie die Inhalte streng nach folgendem Format:
🧠 Hauptidee/Thema – Fassen Sie das zentrale Thema zusammen.

✅ Umsetzbare Schritte – Schlüsseln Sie auf, was als nächstes getan werden kann.

❓ Offene Fragen – Heben Sie Unklarheiten oder Bereiche hervor, die weiterer Erforschung bedürfen.

Hier sind die Gedanken:""",
        'summarize': "Fassen Sie den obigen Text zusammen."
    },
    'french': {
        'system': """Vous êtes un assistant IA de classe mondiale. Vous devez organiser les pensées aléatoires des utilisateurs dans un format clair, concis et exploitable. Veuillez structurer le contenu en respectant strictement le format suivant:
🧠 Idée/Thème principal – Résumez le sujet central.

✅ Étapes exploitables – Décomposez ce qui peut être fait ensuite.

❓ Questions ouvertes – Mettez en évidence les incertitudes ou les domaines nécessitant une exploration plus approfondie.

Voici les pensées :""",
        'summarize': "Résumez le texte ci-dessus."
    },
    'italian': {
        'system': """Sei un assistente IA di classe mondiale. Devi organizzare i pensieri casuali degli utenti in un formato chiaro, conciso e pratico. Si prega di strutturare il contenuto seguendo rigorosamente questo formato:
🧠 Idea/Tema principale – Riassumi l'argomento centrale.

✅ Passi attuabili – Scomponi cosa si può fare dopo.

❓ Domande aperte – Evidenzia eventuali incertezze o aree che necessitano di ulteriore esplorazione.

Ecco i pensieri:""",
        'summarize': "Riassumi il testo sopra."
    },
    'japanese': {
        'system': """あなたは世界クラスのAIアシスタントです。ユーザーのランダムな考えを、明確で簡潔で実行可能な形式に整理する必要があります。以下の形式に厳密に従って内容を構造化してください：
🧠 メインアイデア/テーマ – 中心的なトピックを要約します。

✅ 実行可能なステップ – 次に何ができるかを分解します。

❓ オープンな質問 – 不確実性や更なる探求が必要な領域を強調します。

以下が考えです：""",
        'summarize': "上記のテキストを要約してください。"
    },
    'spanish': {
        'system': """Eres un asistente de IA de clase mundial. Necesitas organizar los pensamientos aleatorios de los usuarios en un formato claro, conciso y procesable. Por favor, estructure el contenido siguiendo estrictamente este formato:
🧠 Idea/Tema principal – Resume el tema central.

✅ Pasos procesables – Desglose lo que se puede hacer a continuación.

❓ Preguntas abiertas – Destaca cualquier incertidumbre o áreas que necesiten más exploración.

Aquí están los pensamientos:""",
        'summarize': "Resume el texto anterior."
    },
    'portuguese': {
        'system': """Você é um assistente de IA de classe mundial. Você precisa organizar os pensamentos aleatórios dos usuários em um formato claro, conciso e acionável. Por favor, estruture o conteúdo seguindo rigorosamente este formato:
🧠 Ideia/Tema principal – Resuma o tópico central.

✅ Passos acionáveis – Detalhe o que pode ser feito a seguir.

❓ Questões em aberto – Destaque quaisquer incertezas ou áreas que precisam de mais exploração.

Aqui estão os pensamentos:""",
        'summarize': "Resuma o texto acima."
    }
}

class VoiceNotesApp:
    def __init__(self):
        # Azure endpoint configuration
        # self.azure_url = os.getenv("AZURE_ENDPOINT")
        # self.api_key = os.getenv("AZURE_API_KEY")
        self.azure_url = "https://nguyenbach-9897-westus3-covns.westus3.inference.ml.azure.com/score"
        self.api_key = "uABZxEGiK6iXlBWsvy7xOEJ1zCmeLkvSHuf6wEykfqtf4I7KbbWiJQQJ99BBAAAAAAAAAAAAINFRAZML2JhK"
        
        # Initialize sample audio files
        self.sample_audios = {
            "English - Weekend Plan": "content/english.weekend.plan.wav",
            "Chinese - Kids & Work": "content/chinese.kid.work.mp3",
            "German - Vacation Planning": "content/german.vacation.work.mp3",
            "French - Random Thoughts": "content/french.random.vacation.mp3",
            "Italian - Daily Life": "content/italian.daily.life.mp3",
            "Japanese - Seattle Trip Report": "content/japanese.seattle.trip.report.mp3",
            "Spanish - Soccer Class": "content/spanish.soccer.class.mp3",
            "Portuguese - Buying House & Friends": "content/portugese.house.friends.mp3"
        }
        
        # Initialize storage
        self.notes_file = "voice_notes.json"
        self.notes = self.load_notes()
    
    def load_notes(self):
        if os.path.exists(self.notes_file):
            with open(self.notes_file, 'r') as f:
                notes = json.load(f)
                # Sort notes by timestamp in descending order (most recent first)
                return sorted(notes, key=lambda x: x['timestamp'], reverse=True)
        return []

    def save_notes(self):
        with open(self.notes_file, 'w') as f:
            json.dump(self.notes, f, indent=2)
    
    def encode_base64_from_file(self, file_path):
        """Encode file content to base64 string and determine MIME type."""
        file_extension = os.path.splitext(file_path)[1].lower()
        
        # Map file extensions to MIME types
        if file_extension in ['.jpg', '.jpeg']:
            mime_type = "image/jpeg"
        elif file_extension == '.png':
            mime_type = "image/png"
        elif file_extension == '.gif':
            mime_type = "image/gif"
        elif file_extension in ['.bmp', '.tiff', '.webp']:
            mime_type = f"image/{file_extension[1:]}"
        elif file_extension == '.flac':
            mime_type = "audio/flac"
        elif file_extension == '.wav':
            mime_type = "audio/wav"
        elif file_extension == '.mp3':
            mime_type = "audio/mpeg"
        elif file_extension in ['.m4a', '.aac']:
            mime_type = "audio/aac"
        elif file_extension == '.ogg':
            mime_type = "audio/ogg"
        else:
            mime_type = "application/octet-stream"
        
        # Read and encode file content
        with open(file_path, "rb") as file:
            encoded_string = base64.b64encode(file.read()).decode('utf-8')
        
        return encoded_string, mime_type

    def call_azure_endpoint(self, data):
        """Call Azure ML endpoint with the given data."""
        parameters = {"temperature": 0.0}
        data["input_data"]["parameters"] = parameters
        
        def allowSelfSignedHttps(allowed):
            # bypass the server certificate verification on client side
            if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
                ssl._create_default_https_context = ssl._create_unverified_context

        allowSelfSignedHttps(True) 
        body = str.encode(json.dumps(data))
        
        if not self.api_key:
            raise Exception("A key should be provided to invoke the endpoint")

        headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}

        req = urllib.request.Request(self.azure_url, body, headers)

        try:
            response = urllib.request.urlopen(req)
            result = response.read().decode('utf-8')
            return result
        except urllib.error.HTTPError as error:
            print("The request failed with status code: " + str(error.code))
            print(error.info())
            print(error.read().decode("utf8", 'ignore'))
            return f"Error: {error.code}"

    def transcribe_audio(self, audio_input):
        """Convert speech to text using Azure endpoint"""
        try:
            # Encode audio file to base64
            encoded_audio, mime_type = self.encode_base64_from_file(audio_input)
            
            # Prepare the request payload
            speech_prompt = "Based on the attached audio, generate a comprehensive text transcription of the spoken content."
            payload = {
                "input_data": {
                    "input_string": [
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": speech_prompt
                                },
                                {
                                    "type": "audio_url",
                                    "audio_url": {
                                        "url": f"data:{mime_type};base64,{encoded_audio}"
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
            
            # Call Azure endpoint
            response_json = self.call_azure_endpoint(payload)
            
            # Parse response
            try:
                response_data = json.loads(response_json)
                # Extract the actual response text
                if isinstance(response_data, dict) and "output" in response_data:
                    transcription = response_data["output"]
                else:
                    transcription = response_json
            except:
                transcription = response_json
            
            print(f"Debug transcription:\n{transcription}")
            
            return transcription
        except Exception as e:
            print(f"Transcription error: {str(e)}")
            return f"Error transcribing: {str(e)}"

    def summarize_text(self, text):
        """Generate a summary in the detected language using Azure endpoint"""
        if not text:
            return "No text to summarize"

        try:
            # First, detect language
            detected_language = detect(text)
            
            # Map detected language to supported languages
            language_mapping = {
                'en': 'english',
                'zh-cn': 'chinese',
                'de': 'german',
                'fr': 'french',
                'it': 'italian',
                'ja': 'japanese',
                'es': 'spanish',
                'pt': 'portuguese'
            }
            
            # Default to English if language not supported
            selected_language = language_mapping.get(detected_language, 'english')
            
            print(f"Detected language: {detected_language} ; Selected language: {selected_language} ; Text: {text}")

            # Generate summary in detected language
            prompts = MULTILINGUAL_SYSYTEM_PROMPTS[selected_language]
            
            # Prepare the request payload for summarization
            payload = {
                "input_data": {
                    "input_string": [
                        {
                            "role": "system",
                            "content": [
                                {
                                    "type": "text",
                                    "text": prompts["system"]
                                }
                            ]
                        },
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": text
                                }
                            ]
                        },
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": prompts["summarize"]
                                }
                            ]
                        }
                    ]
                }
            }
            
            # Call Azure endpoint
            response_json = self.call_azure_endpoint(payload)
            
            # Parse response
            try:
                response_data = json.loads(response_json)
                # Extract the actual response text
                if isinstance(response_data, dict) and "output" in response_data:
                    summary = response_data["output"]
                else:
                    summary = response_json
            except:
                summary = response_json
            
            print(f"Debug summary:\n{summary}")
            
            return summary, selected_language
        except Exception as e:
            print(f"Summarization error: {str(e)}")
            return f"Error summarizing: {str(e)}", "english"

    def format_note_display(self, note):
        """Format note for display in a Gradio interface"""
        return f"""## 📝 Summary  
{note['summary']}

---  

## 🎙️ Transcription  
{note['transcription']}  

---

## 🕒 Timestamp  
**{note['timestamp']}**

---

## 🌐 Detected Language
**{note['language']}**"""    
    
    def search_notes(self, query):
        """Search through notes content"""
        if not query:
            return self.list_all_notes()
            
        matching_notes = []
        query = query.lower()
        
        for note in self.notes:
            if (query in note['transcription'].lower() or 
                query in note['summary'].lower()):
                matching_notes.append(note)
        
        if not matching_notes:
            return "No matching notes found"
            
        return "\n\n---\n\n".join([self.format_note_display(note) for note in matching_notes])

    def list_all_notes(self):
        """Return all notes in formatted string"""
        if not self.notes:
            return "No notes found"
        
        # Notes are already sorted by timestamp in load_notes()
        return "\n\n---\n\n".join([self.format_note_display(note) for note in self.notes])

    def process_note(self, audio, progress=gr.Progress()):
        """Process audio input and generate note with summary"""
        if audio is None:
            return "Please record or upload an audio file.", None, "❌ No audio provided"
        
        try:
            # Start processing
            progress(0, desc="Starting audio processing...")
            
            # Transcribe audio (25% of progress)
            progress(0.25, desc="Transcribing audio...")
            transcription = self.transcribe_audio(audio)
            
            # Generate summary (35% more progress)
            progress(0.60, desc="Generating summary...")
            summary, selected_language = self.summarize_text(transcription)
            
            # Create and save note (25% more progress)
            progress(0.85, desc="Saving note...")
            note = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "transcription": transcription,
                "summary": summary,
                "language": selected_language
            }
            
            # Save note
            self.notes.append(note)
            self.save_notes()
            
            # Complete
            progress(1.0, desc="Complete!")
            return self.format_note_display(note), audio, "✅ Processing complete!"
            
        except Exception as e:
            return str(e), None, f"❌ Error: {str(e)}"
    
    async def text_to_speech(self, text, detected_lang):
        """Convert text to speech using Edge TTS with language-specific voices"""
        if not text.strip():
            return None
        
        # Map of language codes to Edge TTS voices
        voice_mapping = {
            'english': 'en-US-RogerNeural',       # English
            'chinese': 'zh-CN-XiaoxiaoNeural',    # Chinese
            'german': 'de-DE-KatjaNeural',        # German
            'french': 'fr-FR-HenriNeural',        # French
            'italian': 'it-IT-DiegoNeural',       # Italian
            'japanese': 'ja-JP-KeitaNeural',      # Japanese
            'spanish': 'es-ES-XimenaNeural',      # Spanish
            'portuguese': 'pt-BR-AntonioNeural',  # Portuguese
        }
        
        # Default to English if language not supported
        voice = voice_mapping.get(detected_lang, 'en-US-EricNeural')
        
        try:
            communicate = edge_tts.Communicate(text, voice)
            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
                tmp_path = tmp_file.name
                await communicate.save(tmp_path)
                return tmp_path
        except Exception as e:
            print(f"TTS Error: {str(e)}")
            return None
    
    def create_interface(self):
        """Create an enhanced Gradio interface with improved styling and layout"""
         # Preload English sample
        default_audio_path = self.sample_audios["English - Weekend Plan"]
        # Read the audio file for proper initialization
        try:
            audio_data, sample_rate = soundfile.read(default_audio_path)
            default_audio = (sample_rate, audio_data)
        except Exception as e:
            print(f"Error loading default audio: {e}")
            default_audio = None

        # Create the layout with tabs
        with gr.Blocks(
            css=CUSTOM_CSS,
            theme=LIGHT_THEME,
            title="Thoughts Organizer"
        ) as interface:
            # Header section
            with gr.Row(elem_id="header"):
                gr.Markdown(
                    """
                    # 🎙️ Thoughts Organizer with Phi-4-Multimodal
                    ### Transform your spoken thoughts into organized, actionable insights

                    Capture, organize, and act on your spoken thoughts with AI-powered voice notes in multiple languages, including English, Chinese, German, French, Italian, Japanese, Spanish, and Portuguese.
                    """
                )

            # Main content tabs
            with gr.Tabs():
                # Record New Note Tab
                with gr.Tab("📝 New Note", id="new_note"):
                    with gr.Column(elem_classes="input-section"):
                        # Audio input with clear instructions
                        gr.Markdown("### Record or Upload Your Voice Note")
                        audio_input = gr.Audio(
                            sources=["microphone", "upload"],
                            type="filepath",
                            label="",
                            interactive=True,
                            show_download_button=True,
                            value=default_audio
                        )

                        with gr.Row():
                            sample_audio = gr.Dropdown(
                                choices=list(self.sample_audios.keys()),
                                label="Select a sample audio to try",
                                value="English - Weekend Plan"
                            )
                            # Process button with loading state
                            process_btn = gr.Button("🔄 Process Note", variant="primary")
                            clear_btn = gr.Button("🗑️ Clear", variant="secondary")
                            
                        # Add progress bar
                        progress_bar = gr.Progress()

                    with gr.Column(elem_classes="output-section"):
                        # Status message display
                        processing_status = gr.Markdown(label="Status", value="Ready to process...")
                        # Note output display
                        note_display = gr.Markdown(label="")
                        
                        # Add TTS playback controls
                        with gr.Row():
                            play_btn = gr.Button("🔊 Listen Note", variant="secondary")
                            tts_audio = gr.Audio(label="TTS Output", visible=True, interactive=False)

                # Enhanced All Notes Tab
                with gr.Tab("📚 All Notes", id="all_notes"):
                    with gr.Column():
                        # Search box
                        search_box = gr.Textbox(
                            label="🔍 Search Notes",
                            placeholder="Enter keywords to search...",
                            show_label=True
                        )

                        # Refresh button for notes
                        refresh_btn = gr.Button("🔄 Refresh Notes")

                        # All notes display
                        all_notes_display = gr.Markdown()

            # Function to load sample audio
            def load_sample(sample_name):
                if not sample_name:
                    return None
                try:
                    audio_path = self.sample_audios[sample_name]
                    audio_data, sample_rate = soundfile.read(audio_path)
                    return (sample_rate, audio_data)
                except Exception as e:
                    print(f"Error loading sample audio: {e}")
                    return None
            
            # Automatically load sample when selected
            sample_audio.change(
                fn=load_sample,
                inputs=[sample_audio],
                outputs=[audio_input],
                api_name="load_sample"
            )
                
            def process_and_get_note(audio, progress=gr.Progress()):
                try:
                    note_text, _, status = self.process_note(audio, progress)
                    # Return both the note text and updated all notes display
                    all_notes = self.list_all_notes()
                    return note_text, all_notes, status
                except Exception as e:
                    error_msg = f"❌ Error processing note: {str(e)}"
                    return "", "", error_msg

            # Update the process button click event
            process_btn.click(
                fn=process_and_get_note,
                inputs=[audio_input],
                outputs=[note_display, all_notes_display, processing_status],
                api_name="process_note"
            )
            
            # Function to handle TTS playback
            async def play_note(note_text):
                if not note_text:
                    return None
                
                try:
                    # Extract the detected language from the note display
                    lang_section = note_text.split("Detected Language")[-1].strip()
                    detected_lang = lang_section.strip('*').strip()
                    
                    # Extract the summary section (everything before the first ---)
                    summary_section = note_text.split("---")[0].strip()
                    
                    # Remove Markdown headers (#)
                    cleaned_text = summary_section.replace('#', '')
                    
                    # Remove emojis and section labels
                    cleaned_text = cleaned_text.replace('📝 Summary', '').strip()
                    
                    # Remove all special characters except basic punctuation
                    # Keep: letters, numbers, spaces, and basic punctuation
                    clean_chars = []
                    for char in cleaned_text:
                        if (char.isalnum() or 
                            char.isspace() or 
                            char in '.,!?-:;()[]{}"\''):
                            clean_chars.append(char)
                    
                    cleaned_text = ''.join(clean_chars)
                    
                    # Remove multiple spaces
                    cleaned_text = ' '.join(cleaned_text.split())
                    
                    print(f"Debug - Original text: {summary_section}")
                    print(f"Debug - Cleaned text: {cleaned_text}")
                    
                    audio_path = await self.text_to_speech(cleaned_text, detected_lang)
                    return audio_path
                    
                except Exception as e:
                    print(f"Error in play_note: {str(e)}")
                    return None

            # Update event handlers
            play_btn.click(
                fn=play_note,
                inputs=[note_display],
                outputs=[tts_audio],
                api_name="play_note"
            )

            # Clear button functionality
            def clear_all():
                # Return default/empty values for all components
                return None, "", "Ready to process...", ""
            
            clear_btn.click(
                fn=clear_all,
                inputs=[],
                outputs=[
                    audio_input,        # Clear audio input
                    note_display,       # Clear note display
                    processing_status,  # Reset status message
                    all_notes_display   # Clear all notes display
                ]
            )
            
            def refresh_notes():
                # Reload notes from disk
                self.notes = self.load_notes()
                # Return updated notes display
                return self.list_all_notes()
                
            refresh_btn.click(
                fn=refresh_notes,
                inputs=[],
                outputs=[all_notes_display],
                api_name="refresh_notes"
            )
            # Add search functionality
            search_box.change(
                fn=self.search_notes,
                inputs=[search_box],
                outputs=[all_notes_display],
                api_name="search_notes"
            )

            # Instructions and tips
            with gr.Accordion("ℹ️ Tips & Instructions", open=False):
                gr.Markdown(
                    """
                    ### How to Use:
                    1. **Record or Upload**: Use the microphone to record directly or upload an audio file
                    2. **Process**: Click 'Process Note' to convert your voice note into organized text
                    3. **Review**: View your processed note with main ideas, action items, and questions
                    4. **Listen**: Click the '🔊 Play Note' button to hear the summary read aloud
                    5. **Browse**: Switch to 'All Notes' tab to view your note history

                    ### Features:
                    - 🎙️ Record or upload voice notes
                    - 📝 Automatic transcription
                    - 🧠 Smart organization of ideas
                    - 📚 Historical note tracking
                    """
                )
            # Footer
            with gr.Column(elem_classes="output-section"):
                gr.Markdown("Powered by Microsoft [Phi-4 multimodal model](https://aka.ms/phi-4-multimodal/azure) on Azure AI. © 2025")

        return interface

def run_app():
    # Create app instance
    app = VoiceNotesApp()
    
    # Launch Gradio interface
    interface = app.create_interface()
    interface.launch(
        share=True,
        server_name="0.0.0.0",
    )
    
    
run_app()