Update app.py

app.py

@@ -6,7 +6,6 @@ import traceback
 import numpy as np
 import librosa
 import soundfile as sf
-import noisereduce as nr
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Depends, Security
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from fastapi.responses import JSONResponse, PlainTextResponse
@@ -29,13 +28,13 @@ def verify_token(credentials: HTTPAuthorizationCredentials = Security(security))
         )
     return True
 
-# 亲昵口语热词配置（完整保留原有全部人物与称呼）
+# 亲昵口语热词配置
 CUSTOM_HOTWORDS = os.getenv(
     "HOTWORDS", 
-    "爹地 妈咪 学长 学姐 欧巴 欧尼 宝贝 亲爱的 哥哥 姐姐 弟弟 妹妹 师傅 老公 老婆 爹咪 小路 小路哥哥 因因 岚岚 甜甜 芝芝瓜 小选 小蛇 宝宝 叶瑄 艾因 路辰 罗夏 司岚 星星嘟嘟 牧首 教宗 虎先生 小叶哥哥"
+    "爹地 妈咪 Daddy 学长 学姐 欧巴 欧尼 宝贝 亲爱的 哥哥 姐姐 弟弟 妹妹 师傅 老公 老婆 爹咪 小路 小路哥哥 因因 岚岚 甜甜 芝芝瓜 小选 小蛇 宝宝 叶瑄 艾因 路辰 罗夏 司岚 星星嘟嘟 牧首 教宗 虎先生 小叶哥哥"
 )
 
-# 口语错别字纠错映射
+# 口语错别字规则纠错映射
 COLLOQUIAL_CORRECTIONS = {
     r"跌地|跌弟|爹底|迭地|嗲地": "爹地",
     r"学掌|鞋长|薛长|雪长": "学长",
@@ -47,41 +46,32 @@ COLLOQUIAL_CORRECTIONS = {
 }
 
 def correct_colloquial_words(text: str) -> str:
-    """自动修复口语常见同音错别字"""
     for pattern, target in COLLOQUIAL_CORRECTIONS.items():
         text = re.sub(pattern, target, text)
     return text
 
 # ================= 2. 加载模型 =================
-# ASR 文字与标点转录模型
 asr_model = AutoModel(
     model="iic/SenseVoiceSmall",
     device="cpu",
     disable_update=True
 )
 
-# 独立声学情绪与语调表征模型
 emo_model = AutoModel(
     model="iic/emotion2vec_plus_seed",
     device="cpu",
     disable_update=True
 )
 
-# ================= 3. 音频前处理流水线 =================
+# ================= 3. 无损音频标准化 =================
 def preprocess_audio(input_path: str, target_sr: int = 16000):
-    """音频标准化：转 16kHz 单声道 -> 降噪 -> 峰值归一化，并返回总时长"""
+    """仅做重采样与响度归一化，严禁有损频域降噪以保护 ASR 精度"""
     y, sr = librosa.load(input_path, sr=target_sr, mono=True)
     if len(y) == 0:
         return input_path, 0.0
 
     duration = float(len(y) / sr)
 
-    # 适度频谱降噪 (强度 0.6)
-    try:
-        y = nr.reduce_noise(y=y, sr=sr, stationary=True, prop_decrease=0.6)
-    except Exception as e:
-        print(f"[Warn] 降噪跳过: {e}")
-
     # 峰值归一化至 -0.5 dB
     max_val = float(np.max(np.abs(y)))
     if max_val > 0:
@@ -91,89 +81,90 @@ 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_frame_res, total_duration: float, fallback_emotion: str, window_size: float = 0.5):
-    """
-    将 Emotion2Vec+ 的帧级时序概率矩阵安全聚合为易读的时间轴
-    强制转为 Python 原生 float/str 类型，杜绝 JSON 序列化崩溃
-    """
+# ================= 4. 情绪与时序解析 =================
+def parse_emotion_and_timeline(emo_res, total_duration: float, window_size: float = 0.5):
+    """统一提取全局情绪与时序走势"""
     labels = ["angry", "disgusted", "fearful", "happy", "neutral", "sad", "surprised"]
+    
+    top_emotion = "neutral"
+    emotion_scores = {}
     timeline = []
+
+    if not emo_res or len(emo_res) == 0:
+        return top_emotion, emotion_scores, timeline
+
+    item = emo_res[0]
+    top_emotion = str(item.get("key", item.get("emotion", "neutral")))
     
-    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 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
+    # 提取置信度
+    raw_scores = item.get("scores", {})
+    if isinstance(raw_scores, dict):
+        emotion_scores = {str(k): round(float(v), 4) for k, v in raw_scores.items()}
+
+    # 提取时间轴
+    raw_feats = item.get("feats") if "feats" in item else None
+    if raw_feats is not None:
+        try:
+            feats = np.array(raw_feats, dtype=np.float32)
+            if feats.ndim == 1:
+                feats = np.expand_dims(feats, axis=0)
             
-            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
-            })
-
-    except Exception as err:
-        print(f"[Warn] Timeline extraction fallback: {err}")
-        timeline = [{
+            total_frames = feats.shape[0]
+            if total_frames > 0:
+                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):
+                        t_idx = int(np.argmax(mean_vec))
+                        t_label = labels[t_idx]
+                        conf = float(mean_vec[t_idx])
+                    else:
+                        t_label = top_emotion
+                        conf = 0.8
+
+                    state = "moderate / flat"
+                    if t_label in ["angry", "surprised"]:
+                        state = "high / rising"
+                    elif t_label in ["sad"]:
+                        state = "low / falling"
+                    elif t_label == "happy":
+                        state = "cheerful / dynamic"
+
+                    timeline.append({
+                        "start": start_t,
+                        "end": end_t,
+                        "tone": str(t_label),
+                        "confidence": round(float(conf), 4),
+                        "energy_state": state
+                    })
+        except Exception as e:
+            print(f"[Warn] Timeline parse failed: {e}")
+
+    # 若无有效帧数据则生成单段保底
+    if not timeline:
+        timeline.append({
             "start": 0.0,
             "end": round(float(total_duration), 2),
-            "tone": str(fallback_emotion),
-            "confidence": 0.8,
+            "tone": top_emotion,
+            "confidence": round(float(max(emotion_scores.values())), 4) if emotion_scores else 1.0,
             "energy_state": "moderate / flat"
-        }]
+        })
 
-    return timeline
+    return top_emotion, emotion_scores, timeline
 
 def evaluate_tone_and_emotion(text: str, emotion_scores: dict, raw_top_emotion: str):
-    """全局综合语气判定：结合声学分数与文本关键词"""
+    """多模态语气判断"""
     top_score = float(emotion_scores.get(raw_top_emotion, 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))
@@ -181,7 +172,6 @@ def evaluate_tone_and_emotion(text: str, emotion_scores: dict, raw_top_emotion:
     if (happy_score + surprised_score >= 0.55) and has_soft_words:
         return "coquetry", top_score
 
-    # 低沉嗓音中性保底
     if top_score < 0.60:
         return "neutral", top_score
 
@@ -192,7 +182,7 @@ def evaluate_tone_and_emotion(text: str, emotion_scores: dict, raw_top_emotion:
 async def root():
     return {
         "status": "online",
-        "service": "SenseVoice + Emotion2Vec+ Dynamic Prosody API",
+        "service": "SenseVoice + Emotion2Vec+ STT API",
         "docs": "/docs",
         "endpoint": "/v1/audio/transcriptions"
     }
@@ -212,72 +202,49 @@ async def transcribe_audio(
 
     processed_path = None
     try:
-        # 1. 音频标准化前处理
+        # 1. 无损标准化前处理
         processed_path, duration = preprocess_audio(raw_path)
 
-        # 2. SenseVoice 识别文本（带热词增强）
+        # 2. SenseVoice 原生调用（标准 ITN，不硬传干扰参数）
         asr_res = asr_model.generate(
             input=processed_path,
             language=language if language in ["zh", "en", "yue", "ja", "ko"] else "auto",
             use_itn=True,
-            emotion=False,
-            event=False,
             hotword=CUSTOM_HOTWORDS
         )
         raw_text = asr_res[0]["text"] if asr_res and len(asr_res) > 0 else ""
         clean_text = re.sub(r"<\|.*?\|>", "", raw_text).strip()
         clean_text = correct_colloquial_words(clean_text)
 
-        # 3. Emotion2Vec+ 提取全局情绪 (Utterance-level)
-        emo_res_global = emo_model.generate(
+        # 3. Emotion2Vec+ 单次推理获取情绪与特征
+        emo_res = emo_model.generate(
             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 = str(item.get("key", item.get("emotion", "neutral")))
-            raw_scores = item.get("scores", {})
-            if isinstance(raw_scores, dict):
-                emotion_scores = {str(k): round(float(v), 4) for k, v in raw_scores.items()}
-
-        # 4. Emotion2Vec+ 提取帧级动态时序 (Frame-level)
-        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, 
+        raw_top_emotion, emotion_scores, prosody_timeline = parse_emotion_and_timeline(
+            emo_res, 
             duration, 
-            raw_top_emotion, 
             window_size=0.5
         )
 
-        # 5. 全局多模态语气裁决
+        # 4. 多模态综合语气裁决
         final_emotion, confidence = evaluate_tone_and_emotion(
             clean_text, 
             emotion_scores, 
             raw_top_emotion
         )
 
-        # 6. 构造纯 Python 基础类型的安全字典
+        # 5. 构造双重兼容的返回体（同时保留 emotion 和 overall_emotion）
         response_payload = {
             "text": str(clean_text),
             "duration": round(float(duration), 2),
-            "overall_emotion": str(final_emotion),
+            "emotion": str(final_emotion),                 # 兼容旧版及常规调用
+            "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":
@@ -291,11 +258,9 @@ async def transcribe_audio(
             return JSONResponse(response_payload)
 
     except Exception as e:
-        # 将完整错误堆栈打入 Hugging Face 日志，方便排查
         traceback.print_exc()
         raise HTTPException(status_code=500, detail=str(e))
     finally:
-        # 清理临时文件
         if os.path.exists(raw_path):
             os.remove(raw_path)
         if processed_path and os.path.exists(processed_path):