Update app.py

app.py

@@ -1,208 +1,245 @@
-# Call Sentiment Analyzer
-# app.py
-# Features:
-# - Record via microphone OR upload an audio file
-# - Transcription via Whisper
-# - 5-star multilingual sentiment analysis
-# - Call Quality Indicator (noise/volume estimation)
-# - Keyword Extraction (top relevant words)
-
 import gradio as gr
-import torch
-import numpy as np
+from dotenv import load_dotenv
+import assemblyai as aai
 import librosa
-from transformers import (
-    AutoProcessor,
-    AutoModelForSpeechSeq2Seq,
-    AutoTokenizer,
-    AutoModelForSequenceClassification,
-    pipeline
-)
+import soundfile as sf
+import torch
+import json
+import csv
+import os
+import tempfile
+from datetime import datetime
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
 import torch.nn.functional as F
+from docx import Document
+from reportlab.platypus import SimpleDocTemplate, Paragraph
+from reportlab.lib.styles import getSampleStyleSheet
+
+# =========================
+# CONFIG
+# =========================
+load_dotenv()
 
-print("Loading models... This may take 1-2 minutes.")
+aai.settings.api_key = os.getenv("ASSEMBLYAI_API_KEY")
+hf_token = os.getenv("HF_TOKEN")
 
 device = "cuda" if torch.cuda.is_available() else "cpu"
 
-# === Whisper for transcription ===
-processor = AutoProcessor.from_pretrained("openai/whisper-base.en")
-whisper_model = AutoModelForSpeechSeq2Seq.from_pretrained("openai/whisper-base.en")
-whisper_model.eval()
-whisper_model.to(device)
+tokenizer = AutoTokenizer.from_pretrained(
+    "nlptown/bert-base-multilingual-uncased-sentiment"
+)
 
-# === Sentiment Model (5-star multilingual) ===
-sentiment_tokenizer = AutoTokenizer.from_pretrained('nlptown/bert-base-multilingual-uncased-sentiment')
-sentiment_model = AutoModelForSequenceClassification.from_pretrained('nlptown/bert-base-multilingual-uncased-sentiment')
-sentiment_model.eval()
-sentiment_model.to(device)
+model = AutoModelForSequenceClassification.from_pretrained(
+    "nlptown/bert-base-multilingual-uncased-sentiment"
+)
 
-# === Keyword Extraction ===
-keyword_extractor = pipeline("ner", aggregation_strategy="simple", device=0 if device == "cuda" else -1)
+model.to(device)
+model.eval()
 
-print("All models ready!")
+# =========================
+# GLOBAL
+# =========================
+global_segments = []
+global_conversation = ""
 
+# =========================
+# HELPERS
+# =========================
+def format_time(ms):
+    s = ms / 1000
+    return f"{int(s//60):02d}:{int(s%60):02d}"
 
-# Helper: Transcribe audio
-def transcribe(audio_path):
-    import soundfile as sf
-    audio, sr = sf.read(audio_path)
 
-    # Convert stereo to mono if needed
-    if audio.ndim > 1:
-        audio = audio.mean(axis=1)
+def build_segments(transcript):
+    speaker_map = {}
+    current_id = 1
+    segments = []
 
-    # Resample to 16kHz if needed
-    if sr != 16000:
-        audio = librosa.resample(audio.astype(np.float32), orig_sr=sr, target_sr=16000)
+    for u in transcript.utterances:
+        raw = str(u.speaker)
 
-    audio = audio.astype(np.float32)
+        if raw not in speaker_map:
+            speaker_map[raw] = current_id
+            current_id += 1
 
-    # Whisper processes in 30s chunks
-    chunk_size = 16000 * 30
-    texts = []
-    for i in range(0, len(audio), chunk_size):
-        chunk = audio[i:i + chunk_size]
-        input_features = processor(chunk, sampling_rate=16000, return_tensors="pt").input_features.to(device)
-        with torch.no_grad():
-            predicted_ids = whisper_model.generate(input_features)
-        text = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0].strip()
-        if text:
-            texts.append(text)
+        segments.append({
+            "speaker": speaker_map[raw],
+            "start": format_time(u.start or 0),
+            "end": format_time(u.end or 0),
+            "text": u.text
+        })
 
-    return " ".join(texts)
+    return segments
 
 
-# Helper: Sentiment analysis
-def get_sentiment(text):
-    if not text.strip():
-        return "—", "—", "—"
+def analyze_text(text):
+    inputs = tokenizer(text[:512], return_tensors="pt", truncation=True).to(device)
 
-    inputs = sentiment_tokenizer(text[:512], return_tensors="pt", truncation=True).to(device)
     with torch.no_grad():
-        logits = sentiment_model(**inputs).logits
+        logits = model(**inputs).logits
+
     probs = F.softmax(logits, dim=-1)[0]
-    pred = torch.argmax(probs).item() + 1
-    conf = probs[pred - 1].item() * 100
+    return torch.argmax(probs).item() + 1
 
-    stars = "⭐" * pred
-    label = ["Very Negative", "Negative", "Neutral", "Positive", "Very Positive"][pred - 1]
-    return f"{stars} {label}", f"{conf:.1f}%", pred
 
+# =========================
+# MAIN PROCESS
+# =========================
+def process_audio(file, speakers=0, language="auto"):
+    global global_segments, global_conversation
 
-# Helper: Keyword extraction
-def extract_keywords(text):
-    try:
-        entities = keyword_extractor(text)
-        keywords = [ent['word'] for ent in entities if ent['score'] > 0.8]
-        return ", ".join(keywords[:8]) if keywords else "None detected"
-    except:
-        return "None detected"
+    if file is None:
+        return "❌ No audio provided", "", ""
 
+    path = file if isinstance(file, str) else file.name
+
+    temp_path = None
 
-# Helper: Estimate call quality
-def estimate_call_quality(audio_path):
     try:
-        import soundfile as sf
-        audio, sr = sf.read(audio_path)
-        if audio.ndim > 1:
-            audio = audio.mean(axis=1)
-        audio = audio.astype(np.float32)
-
-        rms = np.sqrt(np.mean(audio ** 2))
-        volume_db = 20 * np.log10(rms + 1e-8)
-        volume_level = "Low" if volume_db < -40 else "Good" if volume_db < -20 else "Loud"
-
-        S = np.abs(librosa.stft(audio))
-        flatness = np.mean(librosa.feature.spectral_flatness(S=S))
-        noise_level = "High Noise" if flatness > 0.3 else "Clear"
-
-        quality = (
-            "Good 🟢" if volume_level in ["Good", "Loud"] and noise_level == "Clear"
-            else "Fair 🟡" if volume_level == "Good"
-            else "Poor 🔴"
+        # Load audio
+        audio, sr = librosa.load(path, sr=None, mono=True)
+
+        # 🔥 Create TEMP FILE (not saved permanently)
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
+            sf.write(tmp.name, audio, sr)
+            temp_path = tmp.name
+
+        config = aai.TranscriptionConfig(
+            speaker_labels=True,
+            speakers_expected=speakers if speakers > 0 else None,
+            language_code=None if language == "auto" else language
         )
-        volume_pct = max(0, min(100, round(volume_db + 60)))
-        return quality, volume_pct, noise_level
-    except:
-        return "Unknown", 0, "Unknown"
-
-
-# Main analysis function
-def analyze_audio(audio_path):
-    if audio_path is None:
-        return "No audio provided.", "—", "—", "—", "—", "—", "—"
-
-    # Transcribe
-    transcript = transcribe(audio_path)
-    if not transcript:
-        return "Could not transcribe audio. Please try again.", "—", "—", "—", "—", "—", "—"
-
-    # Sentiment
-    sentiment_label, confidence, stars = get_sentiment(transcript)
-
-    # Keywords
-    keywords = extract_keywords(transcript)
-
-    # Call quality
-    quality, volume_pct, noise = estimate_call_quality(audio_path)
-
-    # Overall health summary
-    if stars == 5 or stars == 4:
-        health = "🟢 Positive Call"
-    elif stars == 3:
-        health = "🟡 Neutral Call"
-    else:
-        health = "🔴 Needs Attention"
-
-    return (
-        transcript,
-        sentiment_label,
-        confidence,
-        keywords,
-        quality,
-        f"{volume_pct}%",
-        f"{health} | Noise: {noise}"
-    )
 
+        transcript = aai.Transcriber().transcribe(temp_path, config)
 
-# Gradio Interface
-with gr.Blocks(title="Call Sentiment Analyzer", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🎤 Call Sentiment Analyzer")
-    gr.Markdown("""
-    **Record or upload an audio file to analyze:**
-    - Transcription via Whisper
-    - 5-star sentiment analysis (multilingual)
-    - Call quality monitoring
-    - Keyword extraction
-    """)
-
-    with gr.Row():
-        audio_input = gr.Audio(
-            sources=["microphone", "upload"],
-            type="filepath",
-            label="🎙️ Record or Upload Audio"
-        )
+        if transcript.error:
+            return f"❌ {transcript.error}", "", ""
+
+        global_segments = build_segments(transcript)
+
+        speaker_count = len(set(s["speaker"] for s in global_segments))
+
+        label_map = {
+            1: ("🔴", "Very Negative"),
+            2: ("🟠", "Negative"),
+            3: ("🟡", "Neutral"),
+            4: ("🟢", "Positive"),
+            5: ("🟢", "Very Positive")
+        }
+
+        conversation = ""
+
+        for i, seg in enumerate(global_segments, start=1):
+            score = analyze_text(seg["text"])
+            emoji, label = label_map.get(score, ("⚪", "Unknown"))
+
+            conversation += (
+                f"Speaker {seg['speaker']} | Utterance {i}\n"
+                f"({seg['start']} - {seg['end']})\n"
+                f"{emoji} {label}: {seg['text']}\n\n"
+            )
+
+        global_conversation = conversation
 
-    analyze_btn = gr.Button("🔍 Analyze", variant="primary")
+        return "✅ Done", conversation, f"Speakers: {speaker_count} | Utterances: {len(global_segments)}"
 
-    with gr.Row():
-        with gr.Column():
-            transcript_box = gr.Textbox(label="📝 Transcript", lines=6, interactive=False)
-        with gr.Column():
-            sentiment_box = gr.Textbox(label="😊 Sentiment", interactive=False)
-            conf_box = gr.Textbox(label="📊 Confidence", interactive=False)
-            keyword_box = gr.Textbox(label="🔑 Detected Keywords", interactive=False)
+    except Exception as e:
+        return f"❌ Error: {str(e)}", "", ""
 
-    with gr.Row():
-        quality_box = gr.Textbox(label="�� Call Quality", interactive=False)
-        volume_box = gr.Textbox(label="🔊 Volume Level", interactive=False)
-        health_box = gr.Textbox(label="❤️ Call Health", interactive=False)
+    finally:
+        # 🔥 DELETE temp file ALWAYS
+        if temp_path and os.path.exists(temp_path):
+            os.remove(temp_path)
+
+
+# =========================
+# EXPORT
+# =========================
+def export_file(format_type):
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+
+    if format_type == "TXT":
+        path = f"conversation_{timestamp}.txt"
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(global_conversation)
+
+    elif format_type == "JSON":
+        path = f"conversation_{timestamp}.json"
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(global_segments, f, indent=4)
+
+    elif format_type == "CSV":
+        path = f"conversation_{timestamp}.csv"
+        with open(path, "w", newline="", encoding="utf-8") as f:
+            writer = csv.DictWriter(f, fieldnames=["speaker", "start", "end", "text"])
+            writer.writeheader()
+            writer.writerows(global_segments)
+
+    elif format_type == "WORD":
+        path = f"conversation_{timestamp}.docx"
+        doc = Document()
+        doc.add_heading("Conversation Transcript", 0)
+        doc.add_paragraph(global_conversation)
+        doc.save(path)
+
+    elif format_type == "PDF":
+        path = f"conversation_{timestamp}.pdf"
+        doc = SimpleDocTemplate(path)
+        styles = getSampleStyleSheet()
+        content = [Paragraph(global_conversation.replace("\n", "<br/>"), styles["Normal"])]
+        doc.build(content)
+
+    return path
+
+
+# =========================
+# UI
+# =========================
+with gr.Blocks(title="AI Conversation Sentiment System") as app:
+
+    gr.Markdown("# 🎙 AI Conversation Sentiment Analyzer")
+
+    with gr.Group():
+        gr.Markdown("### 🎙 Input Audio")
+        audio = gr.Audio(sources=["upload", "microphone"], type="filepath")
+
+    with gr.Group():
+        gr.Markdown("### ⚙ Settings")
+        with gr.Row():
+            speakers = gr.Number(value=0, label="Speakers (0 = auto)")
+            language = gr.Dropdown(["auto", "en", "fr", "es", "de"], value="auto")
+
+    analyze_btn = gr.Button("🚀 Analyze")
+
+    with gr.Group():
+        gr.Markdown("### 💬 Conversation Output")
+        status = gr.Textbox(label="Status")
+        conversation_box = gr.Textbox(lines=18, label="Conversation + Sentiment")
+        info = gr.Textbox(label="Info")
+
+    with gr.Group():
+        gr.Markdown("### 📁 Export")
+        with gr.Row():
+            export_format = gr.Dropdown(
+                ["TXT", "JSON", "CSV", "WORD", "PDF"],
+                value="TXT",
+                label="Select Format"
+            )
+            export_btn = gr.Button("⬇ Export")
+        download = gr.File()
 
     analyze_btn.click(
-        fn=analyze_audio,
-        inputs=audio_input,
-        outputs=[transcript_box, sentiment_box, conf_box, keyword_box, quality_box, volume_box, health_box]
+        process_audio,
+        inputs=[audio, speakers, language],
+        outputs=[status, conversation_box, info]
+    )
+
+    export_btn.click(
+        export_file,
+        inputs=[export_format],
+        outputs=[download]
     )
 
+
 if __name__ == "__main__":
-    demo.launch()
+    app.launch(theme=gr.themes.Soft())