Update app.py

app.py

@@ -2,6 +2,7 @@ import os
 import re
 import shutil
 import tempfile
+import traceback
 import numpy as np
 import librosa
 import soundfile as sf
@@ -28,7 +29,7 @@ def verify_token(credentials: HTTPAuthorizationCredentials = Security(security))
         )
     return True
 
-# 亲昵口语热词配置
+# 亲昵口语热词配置（完整保留原有全部人物与称呼）
 CUSTOM_HOTWORDS = os.getenv(
     "HOTWORDS", 
     "爹地 妈咪 学长 学姐 欧巴 欧尼 宝贝 亲爱的 哥哥 姐姐 弟弟 妹妹 师傅 老公 老婆 爹咪 小路 小路哥哥 因因 岚岚 甜甜 芝芝瓜 小选 小蛇 宝宝 叶瑄 艾因 路辰 罗夏 司岚 星星嘟嘟 牧首 教宗 虎先生 小叶哥哥"
@@ -46,6 +47,7 @@ COLLOQUIAL_CORRECTIONS = {
 }
 
 def correct_colloquial_words(text: str) -> str:
+    """自动修复口语常见同音错别字"""
     for pattern, target in COLLOQUIAL_CORRECTIONS.items():
         text = re.sub(pattern, target, text)
     return text
@@ -72,16 +74,16 @@ def preprocess_audio(input_path: str, target_sr: int = 16000):
     if len(y) == 0:
         return input_path, 0.0
 
-    duration = len(y) / sr
+    duration = float(len(y) / sr)
 
     # 适度频谱降噪 (强度 0.6)
     try:
         y = nr.reduce_noise(y=y, sr=sr, stationary=True, prop_decrease=0.6)
-    except Exception:
-        pass
+    except Exception as e:
+        print(f"[Warn] 降噪跳过: {e}")
 
     # 峰值归一化至 -0.5 dB
-    max_val = np.max(np.abs(y))
+    max_val = float(np.max(np.abs(y)))
     if max_val > 0:
         y = (y / max_val) * 0.95
 
@@ -89,87 +91,93 @@ def preprocess_audio(input_path: str, target_sr: int = 16000):
     sf.write(processed_path, y, sr, subtype="PCM_16")
     return processed_path, duration
 
-# ================= 4. 语调时序与情绪动态解析 =================
-def extract_prosody_timeline(emo_res, total_duration: float, window_size: float = 0.5):
+# ================= 4. 语调时序与情绪动态解析 (防崩安全版) =================
+def extract_prosody_timeline(emo_frame_res, total_duration: float, fallback_emotion: str, window_size: float = 0.5):
     """
-    将 Emotion2Vec+ 的帧级时序概率矩阵聚合为易读的时间轴
+    将 Emotion2Vec+ 的帧级时序概率矩阵安全聚合为易读的时间轴
+    强制转为 Python 原生 float/str 类型，杜绝 JSON 序列化崩溃
     """
-    if not emo_res or len(emo_res) == 0:
-        return []
-
-    item = emo_res[0]
-    
-    # 尝试提取帧级时序矩阵 (Shape: [T, Num_Classes] 或 scores 列表)
-    feats = item.get("feats") if "feats" in item else item.get("scores")
-    
-    # 若模型仅返回整句级结果时的安全降级
-    if feats is None or not isinstance(feats, (list, np.ndarray)) or len(feats) == 0:
-        return [{
-            "start": 0.0,
-            "end": round(total_duration, 2),
-            "tone": item.get("key", item.get("emotion", "neutral")),
-            "confidence": round(float(max(item.get("scores", {}).values())), 4) if item.get("scores") else 1.0,
-            "energy_state": "moderate"
-        }]
-
-    feats = np.array(feats)
-    total_frames = len(feats)
-    if total_frames == 0:
-        return []
-
-    # 标签名称映射列表 (Emotion2Vec 典型标签索引)
     labels = ["angry", "disgusted", "fearful", "happy", "neutral", "sad", "surprised"]
-    
     timeline = []
-    # 按 window_size (秒) 划分时间窗并计算均值
-    frames_per_window = max(1, int(total_frames * (window_size / max(total_duration, 0.1))))
-    
-    num_windows = int(np.ceil(total_frames / frames_per_window))
     
-    for i in range(num_windows):
-        start_frame = i * frames_per_window
-        end_frame = min(total_frames, (i + 1) * frames_per_window)
-        
-        window_feat = feats[start_frame:end_frame]
-        mean_scores = np.mean(window_feat, axis=0) if window_feat.ndim > 1 else window_feat
-        
-        start_time = round(i * window_size, 2)
-        end_time = round(min(total_duration, (i + 1) * window_size), 2)
-        
-        # 找出当前时间段最突出的情绪/语调
-        top_idx = int(np.argmax(mean_scores)) if len(mean_scores) == len(labels) else 0
-        top_label = labels[top_idx] if top_idx < len(labels) else "neutral"
-        conf = float(mean_scores[top_idx]) if top_idx < len(mean_scores) else 0.5
+    try:
+        if not emo_frame_res or len(emo_frame_res) == 0:
+            raise ValueError("Empty emo_frame_res")
+
+        item = emo_frame_res[0]
+        raw_feats = item.get("feats") if "feats" in item else item.get("scores")
         
-        # 根据情绪类型判定当前声学唤醒度/语调状态
-        if top_label in ["angry", "surprised"]:
-            energy_state = "high / rising"      # 激昂 / 语调上扬
-        elif top_label in ["sad"]:
-            energy_state = "low / falling"       # 低沉 / 语调下沉
-        elif top_label == "happy":
-            energy_state = "cheerful / dynamic"  # 轻快 / 活跃
-        else:
-            energy_state = "moderate / flat"     # 平稳 / ���常
+        if raw_feats is None:
+            raise ValueError("No feats or scores found in frame result")
+
+        feats = np.array(raw_feats, dtype=np.float32)
+        if feats.ndim == 1:
+            feats = np.expand_dims(feats, axis=0)
+
+        total_frames = feats.shape[0]
+        if total_frames == 0:
+            raise ValueError("Zero frames")
+
+        frames_per_window = max(1, int(total_frames * (window_size / max(total_duration, 0.1))))
+        num_windows = int(np.ceil(total_frames / frames_per_window))
+
+        for i in range(num_windows):
+            start_f = i * frames_per_window
+            end_f = min(total_frames, (i + 1) * frames_per_window)
+            chunk = feats[start_f:end_f]
+            
+            mean_vec = np.mean(chunk, axis=0) if chunk.ndim > 1 else chunk
+            
+            start_t = round(float(i * window_size), 2)
+            end_t = round(float(min(total_duration, (i + 1) * window_size)), 2)
+
+            if len(mean_vec) == len(labels):
+                top_idx = int(np.argmax(mean_vec))
+                top_tone = labels[top_idx]
+                conf = float(mean_vec[top_idx])
+            else:
+                top_tone = fallback_emotion
+                conf = 0.8
+
+            # 根据情绪类型判定当前声学唤醒度/语调状态
+            if top_tone in ["angry", "surprised"]:
+                energy_state = "high / rising"      # 激昂 / 语调上扬
+            elif top_tone in ["sad"]:
+                energy_state = "low / falling"       # 低沉 / 语调下沉
+            elif top_tone == "happy":
+                energy_state = "cheerful / dynamic"  # 轻快 / 活跃
+            else:
+                energy_state = "moderate / flat"     # 平稳 / 正常
+
+            timeline.append({
+                "start": start_t,
+                "end": end_t,
+                "tone": str(top_tone),
+                "confidence": round(float(conf), 4),
+                "energy_state": energy_state
+            })
 
-        timeline.append({
-            "start": start_time,
-            "end": end_time,
-            "tone": top_label,
-            "confidence": round(conf, 4),
-            "energy_state": energy_state
-        })
+    except Exception as err:
+        print(f"[Warn] Timeline extraction fallback: {err}")
+        timeline = [{
+            "start": 0.0,
+            "end": round(float(total_duration), 2),
+            "tone": str(fallback_emotion),
+            "confidence": 0.8,
+            "energy_state": "moderate / flat"
+        }]
 
     return timeline
 
 def evaluate_tone_and_emotion(text: str, emotion_scores: dict, raw_top_emotion: str):
     """全局综合语气判定：结合声学分数与文本关键词"""
-    top_score = emotion_scores.get(raw_top_emotion, 0.0)
+    top_score = float(emotion_scores.get(raw_top_emotion, 0.0))
 
     # 撒娇 / 卖萌判定
-    happy_score = emotion_scores.get("happy", 0.0)
-    surprised_score = emotion_scores.get("surprised", 0.0)
+    happy_score = float(emotion_scores.get("happy", 0.0))
+    surprised_score = float(emotion_scores.get("surprised", 0.0))
     has_soft_words = bool(re.search(r"[嘛啦呀吧啵呗嘤~～…]", text))
-    
+
     if (happy_score + surprised_score >= 0.55) and has_soft_words:
         return "coquetry", top_score
 
@@ -225,23 +233,32 @@ async def transcribe_audio(
             input=processed_path,
             granularity="utterance"
         )
-        
+
         raw_top_emotion = "neutral"
         emotion_scores = {}
         if emo_res_global and len(emo_res_global) > 0:
             item = emo_res_global[0]
-            raw_top_emotion = item.get("key", item.get("emotion", "neutral"))
+            raw_top_emotion = str(item.get("key", item.get("emotion", "neutral")))
             raw_scores = item.get("scores", {})
-            emotion_scores = {k: round(float(v), 4) for k, v in raw_scores.items()}
+            if isinstance(raw_scores, dict):
+                emotion_scores = {str(k): round(float(v), 4) for k, v in raw_scores.items()}
 
         # 4. Emotion2Vec+ 提取帧级动态时序 (Frame-level)
-        emo_res_frame = emo_model.generate(
-            input=processed_path,
-            granularity="frame"
-        )
-        
+        try:
+            emo_res_frame = emo_model.generate(
+                input=processed_path,
+                granularity="frame"
+            )
+        except Exception:
+            emo_res_frame = []
+
         # 生成语调变化时间轴 (按 0.5s 窗口聚合)
-        prosody_timeline = extract_prosody_timeline(emo_res_frame, duration, window_size=0.5)
+        prosody_timeline = extract_prosody_timeline(
+            emo_res_frame, 
+            duration, 
+            raw_top_emotion, 
+            window_size=0.5
+        )
 
         # 5. 全局多模态语气裁决
         final_emotion, confidence = evaluate_tone_and_emotion(
@@ -250,33 +267,32 @@ async def transcribe_audio(
             raw_top_emotion
         )
 
-        # 6. 按客户端格式输出
+        # 6. 构造纯 Python 基础类型的安全字典
+        response_payload = {
+            "text": str(clean_text),
+            "duration": round(float(duration), 2),
+            "overall_emotion": str(final_emotion),
+            "confidence": round(float(confidence), 4),
+            "emotion_scores": emotion_scores,
+            "prosody_timeline": prosody_timeline
+        }
+
+        # 7. 按客户端格式输出
         if response_format == "text":
             return PlainTextResponse(clean_text)
         elif response_format == "verbose_json":
-            return JSONResponse({
+            response_payload.update({
                 "task": "transcribe",
-                "language": language,
-                "duration": round(duration, 2),
-                "text": clean_text,
-                "overall_emotion": final_emotion,
-                "confidence": round(float(confidence), 4),
-                "emotion_scores": emotion_scores,
-                "prosody_timeline": prosody_timeline,
+                "language": str(language),
                 "segments": []
             })
+            return JSONResponse(response_payload)
         else:
-            # 默认 json 格式
-            return JSONResponse({
-                "text": clean_text,
-                "duration": round(duration, 2),
-                "overall_emotion": final_emotion,
-                "confidence": round(float(confidence), 4),
-                "emotion_scores": emotion_scores,
-                "prosody_timeline": prosody_timeline
-            })
+            return JSONResponse(response_payload)
 
     except Exception as e:
+        # 将完整错误堆栈打入 Hugging Face 日志，方便排查
+        traceback.print_exc()
         raise HTTPException(status_code=500, detail=str(e))
     finally:
         # 清理临时文件