Spaces:
Sleeping
Sleeping
| import io | |
| import tempfile | |
| import os | |
| from typing import List, Dict, Any, Optional | |
| from pathlib import Path | |
| try: | |
| from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer | |
| _vader = SentimentIntensityAnalyzer() | |
| except ImportError: | |
| _vader = None | |
| from app.services.scoring import detect_churn_keywords, CHURN_KEYWORDS | |
| def analyze_sentiment(text: str) -> dict: | |
| """Run VADER sentiment analysis on transcript text.""" | |
| if not text.strip(): | |
| return {"compound": 0.0, "pos": 0.0, "neg": 0.0, "neu": 1.0, "label": "neutral"} | |
| if _vader is None: | |
| # fallback: simple heuristic | |
| positive = sum(1 for w in ["good", "great", "excellent", "happy", "love", "thanks", "amazing", "helpful"] if w in text.lower()) | |
| negative = sum(1 for w in ["bad", "terrible", "awful", "hate", "frustrated", "angry", "broken", "useless"] if w in text.lower()) | |
| compound = (positive - negative) / max(positive + negative + 1, 1) | |
| return { | |
| "compound": round(compound, 4), | |
| "pos": 0.0, "neg": 0.0, "neu": 0.0, | |
| "label": "positive" if compound > 0.1 else ("negative" if compound < -0.1 else "neutral") | |
| } | |
| scores = _vader.polarity_scores(text) | |
| compound = scores["compound"] | |
| label = "positive" if compound >= 0.05 else ("negative" if compound <= -0.05 else "neutral") | |
| return { | |
| "compound": round(compound, 4), | |
| "pos": round(scores["pos"], 4), | |
| "neg": round(scores["neg"], 4), | |
| "neu": round(scores["neu"], 4), | |
| "label": label, | |
| } | |
| def transcribe_with_whisper(audio_bytes: bytes, filename: str, api_key: str) -> str: | |
| """Transcribe audio using OpenAI Whisper API.""" | |
| import httpx | |
| # Determine content type from extension | |
| ext = Path(filename).suffix.lower() | |
| content_type_map = { | |
| ".mp3": "audio/mpeg", | |
| ".wav": "audio/wav", | |
| ".m4a": "audio/mp4", | |
| ".mp4": "video/mp4", | |
| ".ogg": "audio/ogg", | |
| ".webm": "audio/webm", | |
| } | |
| content_type = content_type_map.get(ext, "audio/mpeg") | |
| # Write bytes to temp file | |
| with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp: | |
| tmp.write(audio_bytes) | |
| tmp_path = tmp.name | |
| try: | |
| with open(tmp_path, "rb") as f: | |
| response = httpx.post( | |
| "https://api.openai.com/v1/audio/transcriptions", | |
| headers={"Authorization": f"Bearer {api_key}"}, | |
| data={"model": "whisper-1", "response_format": "text"}, | |
| files={"file": (filename, f, content_type)}, | |
| timeout=60.0, | |
| ) | |
| if response.status_code != 200: | |
| raise Exception(f"Whisper API error ({response.status_code}): {response.text[:200]}") | |
| return response.text | |
| finally: | |
| os.unlink(tmp_path) | |
| def analyze_call(audio_bytes: bytes, filename: str, api_key: str) -> dict: | |
| """Full call analysis: transcript → sentiment → churn keywords.""" | |
| transcript = transcribe_with_whisper(audio_bytes, filename, api_key) | |
| sentiment = analyze_sentiment(transcript) | |
| keywords = detect_churn_keywords(transcript) | |
| # Combined churn intent from transcript | |
| churn_score = keywords["churn_intent_score"] | |
| if sentiment["label"] == "negative": | |
| churn_score = min(churn_score + 20, 100) | |
| elif sentiment["label"] == "positive": | |
| churn_score = max(churn_score - 15, 0) | |
| return { | |
| "filename": filename, | |
| "transcript_snippet": transcript[:300] + ("..." if len(transcript) > 300 else ""), | |
| "transcript_full": transcript, | |
| "sentiment_score": sentiment["compound"], | |
| "sentiment_label": sentiment["label"], | |
| "churn_intent_score": churn_score, | |
| "flagged_keywords": keywords["flagged_keywords"], | |
| "flagged_keyword_count": keywords["flagged_keyword_count"], | |
| } | |