Update app.py

app.py

@@ -1,102 +1,141 @@
 import gradio as gr
 import torch
+import torchaudio
 import os
-import soundfile as sf
-import librosa
-import logging
 import tempfile
+import logging
 import traceback
 from datetime import datetime
-from DPTNet_eval.DPTNet_quant_sep import load_dpt_model, dpt_sep_process
 
-# 配置日志系统
+# 設定日誌系統
 logging.basicConfig(
-    filename='app.log',
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s'
 )
 logger = logging.getLogger(__name__)
 
-# 全局模型加载（避免重复加载）
+# 檢查 Hugging Face 環境變數
+if not os.getenv("SpeechSeparation"):
+    logger.warning("⚠️ 環境變數 SpeechSeparation 未設定！請在 Hugging Face Space 的 Secrets 中設定 HF_TOKEN")
+
+# 載入模型模組
 try:
-    logger.info("開始加載語音分離模型...")
+    logger.info("🔧 開始載入語音分離模型...")
+    from DPTNet_eval.DPTNet_quant_sep import load_dpt_model, dpt_sep_process
+    logger.info("✅ 模型模組載入成功")
+except ImportError as e:
+    logger.error(f"❌ 模組載入失敗: {str(e)}")
+    raise RuntimeError("本地模組路徑配置錯誤") from e
+
+# 全域模型初始化
+try:
+    logger.info("🔄 初始化模型中...")
     model = load_dpt_model()
-    logger.info("模型加載成功")
+    logger.info(f"🧠 模型載入完成，運行設備: {'GPU' if torch.cuda.is_available() else 'CPU'}")
 except Exception as e:
-    logger.error(f"模型加載失敗: {str(e)}")
-    raise RuntimeError("模型初始化失敗") from e
+    logger.error(f"💣 模型初始化失敗: {str(e)}")
+    raise RuntimeError("模型載入異常終止") from e
 
-def separate_audio(input_wav):
-    """處理音訊分離的主函數"""
+def validate_audio(path):
+    """驗證音檔格式與內容有效性"""
+    try:
+        info = torchaudio.info(path)
+        logger.info(f"🔊 音檔資訊: 采樣率={info.sample_rate}Hz, 通道數={info.num_channels}")
+        
+        if info.num_channels not in [1, 2]:
+            raise gr.Error("❌ 不支援的音檔通道數（僅支援單聲道或立體聲）")
+            
+        if info.sample_rate < 8000 or info.sample_rate > 48000:
+            raise gr.Error("❌ 不支援的采樣率（需介於 8kHz~48kHz）")
+            
+        return info.sample_rate
+    except Exception as e:
+        logger.error(f"⚠️ 音檔驗證失敗: {str(e)}")
+        raise gr.Error("❌ 無效的音訊檔案格式")
+
+def convert_to_wav(input_path):
+    """統一轉換為 16kHz WAV 格式"""
+    try:
+        # 使用 torchaudio 保持一致性
+        waveform, sample_rate = torchaudio.load(input_path)
+        
+        # 單聲道轉換
+        if waveform.shape[0] > 1:
+            waveform = torch.mean(waveform, dim=0, keepdim=True)
+        
+        # 重采樣至 16kHz
+        if sample_rate != 16000:
+            resampler = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)
+            waveform = resampler(waveform)
+        
+        # 建立臨時 WAV 檔案
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpfile:
+            torchaudio.save(tmpfile.name, waveform, 16000, bits_per_sample=16)
+            logger.info(f"📝 已生成標準 WAV 檔案: {tmpfile.name}")
+            return tmpfile.name
+            
+    except Exception as e:
+        logger.error(f"⚠️ 音檔轉換失敗: {str(e)}")
+        raise gr.Error("❌ 音訊格式轉換失敗")
+
+def separate_audio(input_audio):
+    """主處理函式"""
     process_id = datetime.now().strftime("%Y%m%d%H%M%S%f")
     temp_wav = None
     
     try:
-        logger.info(f"[{process_id}] 開始處理檔案: {input_wav}")
+        logger.info(f"[{process_id}] 🚀 收到新請求: {input_audio}")
         
-        # 1. 驗證輸入檔案
-        if not os.path.exists(input_wav):
-            raise gr.Error("檔案不存在，請重新上傳")
-        if os.path.getsize(input_wav) > 50 * 1024 * 1024:  # 50MB限制
-            raise gr.Error("檔案大小超過50MB限制")
-
-        # 2. 讀取並標準化音訊
-        logger.info(f"[{process_id}] 讀取音訊檔案...")
-        data, sr = librosa.load(input_wav, sr=None, mono=True)
+        # 1️⃣ 檔案驗證與轉換
+        if not os.path.exists(input_audio):
+            raise gr.Error("❌ 檔案不存在，請重新上傳")
         
-        # 3. 重採樣處理
-        if sr != 16000:
-            logger.info(f"[{process_id}] 重採樣從 {sr}Hz 到 16000Hz...")
-            data = librosa.resample(data, orig_sr=sr, target_sr=16000)
-            sr = 16000
-
-        # 4. 創建臨時檔案
-        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file:
-            temp_wav = tmp_file.name
-            logger.info(f"[{process_id}] 寫入臨時檔案: {temp_wav}")
-            sf.write(temp_wav, data, sr, subtype='PCM_16')
-
-        # 5. 執行語音分離
-        logger.info(f"[{process_id}] 開始語音分離...")
-        out_dir = tempfile.mkdtemp()  # 使用��時目錄存放輸出
+        if os.path.getsize(input_audio) > 50 * 1024 * 1024:  # 50MB 限制
+            raise gr.Error("❌ 檔案大小超過 50MB 限制")
+            
+        logger.info(f"[{process_id}] 🔁 轉換標準音檔格式...")
+        temp_wav = convert_to_wav(input_audio)
+        validate_audio(temp_wav)
+        
+        # 2️⃣ 建立輸出目錄
+        out_dir = tempfile.mkdtemp()
         outfilename = os.path.join(out_dir, "output.wav")
+        logger.info(f"[{process_id}] 📁 建立臨時輸出目錄: {out_dir}")
         
-        dpt_sep_process(temp_wav, model=model, outfilename=outfilename)
+        # 3️⃣ 執行語音分離
+        logger.info(f"[{process_id}] 🧠 開始執行語音分離...")
+        sep_files = dpt_sep_process(temp_wav, model=model, outfilename=outfilename)
         
-        # 6. 獲取輸出檔案
-        output_files = [
-            outfilename.replace('.wav', '_sep1.wav'),
-            outfilename.replace('.wav', '_sep2.wav')
-        ]
-        logger.info(f"[{process_id}] 預期輸出檔案: {output_files}")
-
-        # 7. 驗證輸出
-        if not all(os.path.exists(f) for f in output_files):
-            missing = [f for f in output_files if not os.path.exists(f)]
-            raise gr.Error(f"分離失敗，缺失檔案: {missing}")
-
-        logger.info(f"[{process_id}] 處理完成")
-        return output_files
-
+        # 4️⃣ 驗證輸出結果
+        for f in sep_files:
+            if not os.path.exists(f):
+                raise gr.Error(f"❌ 缺少輸出檔案: {f}")
+            validate_audio(f)
+        
+        logger.info(f"[{process_id}] ✅ 處理成功完成")
+        return sep_files
+        
+    except RuntimeError as e:
+        if "CUDA out of memory" in str(e):
+            logger.error(f"[{process_id}] 💥 CUDA 記憶體不足")
+            raise gr.Error("⚠️ 記憶體不足，請上傳較短的音檔") from e
+        else:
+            raise
     except Exception as e:
-        error_msg = f"[{process_id}] 處理錯誤: {str(e)}\n{traceback.format_exc()}"
-        logger.error(error_msg)
-        raise gr.Error(f"處理失敗: {str(e)}") from e
-
+        logger.error(f"[{process_id}] ❌ 處理錯誤: {str(e)}\n{traceback.format_exc()}")
+        raise gr.Error(f"⚠️ 處理失敗: {str(e)}") from e
     finally:
         # 清理臨時檔案
         if temp_wav and os.path.exists(temp_wav):
             try:
-                os.remove(temp_wav)
-                logger.info(f"[{process_id}] 已清理臨時檔案")
+                os.unlink(temp_wav)
+                logger.info(f"[{process_id}] 🧹 臨時檔案已清理")
             except Exception as clean_err:
-                logger.warning(f"[{process_id}] 清理失敗: {str(clean_err)}")
+                logger.warning(f"[{process_id}] ⚠️ 清理失敗: {str(clean_err)}")
 
-# 🎯 你提供的 description 內容（已轉為 HTML）
+# 🎯 description 內容（轉為 HTML）
 description_html = """
 <h1 align='center'><a href='https://www.twman.org/AI/ASR/SpeechSeparation' target='_blank'>中文語者分離(分割)</a></h1>
-<p align='center'><b>上傳一段混音音檔 （支援 `.mp3`, `.wav`），自動分離出兩個人的聲音</b></p>
-
 <div align='center'>
   <a href='https://www.twman.org' target='_blank'>TonTon Huang Ph.D.</a> | 
   <a href='https://www.twman.org/AI' target='_blank'> AI </a> |
@@ -105,10 +144,7 @@ description_html = """
   <a href='http://deeplearning101.twman.org' target='_blank'>Deep Learning 101</a> |
   <a href='https://www.youtube.com/c/DeepLearning101' target='_blank'>YouTube</a>  
 </div>
-
 <br>
-
-### 📘 相關技術文章：
 <ul>
   <li><a href='https://blog.twman.org/2025/03/AIAgent.html' target='_blank'>避開 AI Agent 開發陷阱：常見問題、挑戰與解決方案 (那些 AI Agent 實戰踩過的坑)</a>：探討多種 AI Agent 工具的應用經驗與挑戰</li>
   <li><a href='https://blog.twman.org/2024/08/LLM.html' target='_blank'>白話文手把手帶你科普 GenAI</a>：淺顯介紹生成式人工智慧核心概念</li>
@@ -123,44 +159,50 @@ description_html = """
   <li><a href='https://blog.twman.org/2023/07/wsl.html' target='_blank'>用PPOCRLabel來幫PaddleOCR做OCR的微調和標註</a></li>
   <li><a href='https://blog.twman.org/2023/07/HugIE.html' target='_blank'>基於機器閱讀理解和指令微調的統一信息抽取框架之診斷書醫囑資訊擷取分析</a></li>
 </ul>
-
 <br>
 """
 
+EXAMPLES = [
+    ["examples/sample1.wav"],
+    ["examples/sample2.mp3"]
+]
+
+AUDIO_INPUT = gr.Audio(
+    label="🔊 上傳混合音檔",
+    type="filepath",
+    sources=["upload", "microphone"],
+    show_label=True,
+    max_length=180  # 最大 3 分鐘
+)
+
+AUDIO_OUTPUTS = [
+    gr.Audio(label="🗣️ 語音軌道 1", type="filepath"),
+    gr.Audio(label="🗣️ 語音軌道 2", type="filepath")
+]
+
+# 🚀 啟動應用程式
+interface = gr.Interface(
+    fn=separate_audio,
+    inputs=AUDIO_INPUT,
+    outputs=AUDIO_OUTPUTS,
+    title="🎙️ 語音分離��上傳一段混音音檔（支援.mp3, .wav），自動分離出兩個人的聲音；Deep Learning 101",
+    description=description_html,
+    examples=EXAMPLES,
+    allow_flagging="never",
+    cache_examples=False,
+    theme="default"
+)
+
+LAUNCH_CONFIG = {
+    "server_name": "0.0.0.0",
+    "server_port": 7860,
+    "share": False,
+    "debug": True,
+    "auth": None,
+    "inbrowser": True,
+    "quiet": False
+}
+
 if __name__ == "__main__":
-    # 完整配置 Gradio 接口
-    interface = gr.Interface(
-        fn=separate_audio,
-        inputs=gr.Audio(
-            type="filepath",
-            label="請上傳混音音檔 (支援格式: mp3/wav/ogg)",
-            sources=["upload", "microphone"],
-            max_length=180
-        ),
-        outputs=[
-            gr.Audio(label="語音軌道 1", format="wav"),
-            gr.Audio(label="語音軌道 2", format="wav")
-        ],
-        title="🎙️ 語音分離 Demo - Deep Learning 101",
-        description=description_html,  # 直接使用HTML描述
-        allow_flagging="never",
-        live=True,
-        examples=[
-            ["examples/sample1.wav"],
-            ["examples/sample2.mp3"]
-        ],
-        theme="default"
-    )
-  
-    launch_kwargs = {
-        "server_name": "0.0.0.0",
-        "server_port": 7860,
-        "share": False,
-        "debug": False,
-        "auth": None,
-        "inbrowser": True,
-        "quiet": False,
-        "prevent_thread_lock": True
-    }
- 
-    interface.launch(**launch_kwargs)
+    logger.info("🚀 啟動 Gradio 服務...")
+    interface.launch(**LAUNCH_CONFIG)