feat: Implement core pronunciation analysis API

.gitignore

@@ -1,16 +1,21 @@
-# 忽略 Python 虛擬環境
+# Python
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.env
 venv/
+env/
 
-# 忽略 VS Code 的設定
+# IDE / Editor
 .vscode/
+.idea/
 
-# 忽略 Python 的快取檔案
-__pycache__/
-*.pyc
-
-# 忽略下載的本地模型 (非常重要，因為它太大了！)
+# ASR Models (非常重要，模型檔案通常很大)
 ASRs/
 
-# 忽略音訊檔案 (如果它們只是測試用的話)
-TestAudio/
-*.wav
+# Temporary files
+temp_audio/
+
+# macOS
+.DS_Store

ASR.py → analyzer/ASR_en_us.py

@@ -1,3 +1,5 @@
+# analyzer/ASR_en_us.py
+
 import torch
 import soundfile as sf
 import librosa
@@ -5,70 +7,94 @@ from transformers import Wav2Vec2Processor, Wav2Vec2ForCTC
 import os
 from phonemizer import phonemize
 import numpy as np
-from datetime import datetime
-from colorama import init, Fore, Style
-
-# 初始化 colorama
-init(autoreset=True)
+from datetime import datetime, timezone
 
-# --- 1. 全域設定 ---
-TARGET_SENTENCE = "how was your day"
-AUDIO_FILE_PATH = "./TestAudio/hello.wav"
+# --- 1. 全域設定與模型載入函數 ---
+# 模型名稱和路徑保持不變
 MODEL_NAME = "MultiBridge/wav2vec-LnNor-IPA-ft"
 MODEL_SAVE_PATH = "./ASRs/MultiBridge-wav2vec-LnNor-IPA-ft-local"
 
-# --- 2. 載入模型和處理器 (保持不變) ---
-print(f"正在準備模型 '{MODEL_NAME}'...")
-try:
-    if not os.path.exists(MODEL_SAVE_PATH):
-        print(f"本地找不到模型，正在從 Hugging Face 下載並儲存...")
-        processor_to_save = Wav2Vec2Processor.from_pretrained(MODEL_NAME)
-        model_to_save = Wav2Vec2ForCTC.from_pretrained(MODEL_NAME)
-        processor_to_save.save_pretrained(MODEL_SAVE_PATH)
-        model_to_save.save_pretrained(MODEL_SAVE_PATH)
-        print("模型已成功下載並儲存。")
-    else:
-        print(f"在 '{MODEL_SAVE_PATH}' 中找到本地模型。")
-    processor = Wav2Vec2Processor.from_pretrained(MODEL_SAVE_PATH)
-    model = Wav2Vec2ForCTC.from_pretrained(MODEL_SAVE_PATH)
-    print("模型和處理器載入成功！")
-except Exception as e:
-    print(f"處理或載入模型時發生錯誤: {e}")
-    exit()
-
-# --- 3. 準備目標音標 (Target) - (已修改) ---
-print("正在準備目標音標...")
-# 在這一步就徹底移除重音符號，得到最乾淨的目標音標列表
-target_ipa_by_word = [
-    word.replace('ˌ', '').replace('ˈ', '').replace('ː', '')
-    for word in phonemize(TARGET_SENTENCE, language='en-us', backend='espeak', with_stress=True).split()
-]
-
-# --- 4. 讀取音訊並進行辨識 (保持不變) ---
-print(f"正在讀取音訊檔案: {AUDIO_FILE_PATH}...")
-try:
-    speech, sample_rate = sf.read(AUDIO_FILE_PATH)
-    if sample_rate != 16000:
-        speech = librosa.resample(y=speech, orig_sr=sample_rate, target_sr=16000)
-except Exception as e:
-    print(f"讀取或處理音訊時發生錯誤: {e}")
-    exit()
-print("正在辨識用戶的實際發音...")
-input_values = processor(speech, sampling_rate=16000, return_tensors="pt").input_values
-with torch.no_grad():
-  logits = model(input_values).logits
-predicted_ids = torch.argmax(logits, dim=-1)
-user_ipa_full = processor.decode(predicted_ids[0])
-
-
-# --- 5. 核心函式：現在處理的都是乾淨的音標，邏輯保持不變 ---
-def get_phoneme_alignments_by_word(user_phoneme_str, target_words_ipa):
+# 將 processor 和 model 設為全域變數，以便快取
+processor = None
+model = None
+
+def load_model():
+    """
+    在應用程式啟動時載入模型和處理器。
+    如果模型已載入，則跳過。
+    """
+    global processor, model
+    if processor and model:
+        print("英文模型已載入，跳過。")
+        return True
+
+    print(f"正在準備英文 (en-us) ASR 模型 '{MODEL_NAME}'...")
+    try:
+        if not os.path.exists(MODEL_SAVE_PATH):
+            print(f"本地找不到模型，正在從 Hugging Face 下載並儲存...")
+            processor_to_save = Wav2Vec2Processor.from_pretrained(MODEL_NAME)
+            model_to_save = Wav2Vec2ForCTC.from_pretrained(MODEL_NAME)
+            processor_to_save.save_pretrained(MODEL_SAVE_PATH)
+            model_to_save.save_pretrained(MODEL_SAVE_PATH)
+            print("模型已成功下載並儲存。")
+        else:
+            print(f"在 '{MODEL_SAVE_PATH}' 中找到本地模型。")
+        
+        processor = Wav2Vec2Processor.from_pretrained(MODEL_SAVE_PATH)
+        model = Wav2Vec2ForCTC.from_pretrained(MODEL_SAVE_PATH)
+        print("英文 (en-us) 模型和處理器載入成功！")
+        return True
+    except Exception as e:
+        print(f"處理或載入 en-us 模型時發生錯誤: {e}")
+        # 將錯誤向上拋出，讓主應用知道失敗
+        raise RuntimeError(f"Failed to load en-us model: {e}")
+
+# --- 2. 核心分析函數 (主入口) ---
+def analyze(audio_file_path: str, target_sentence: str) -> dict:
+    """
+    接收音訊檔案路徑和目標句子，回傳詳細的發音分析字典。
+    這是此模組的主要進入點。
+    """
+    if not processor or not model:
+        raise RuntimeError("模型尚未載入。請確保在呼叫 analyze 之前已成功執行 load_model()。")
+
+    # --- 準備目標音標 (您的原始邏輯) ---
+    target_ipa_by_word = [
+        word.replace('ˌ', '').replace('ˈ', '').replace('ː', '')
+        for word in phonemize(target_sentence, language='en-us', backend='espeak', with_stress=True).split()
+    ]
+    target_words_original = target_sentence.split()
+
+    # --- 讀取音訊並進行辨識 (您的原始邏輯) ---
+    try:
+        speech, sample_rate = sf.read(audio_file_path)
+        if sample_rate != 16000:
+            speech = librosa.resample(y=speech, orig_sr=sample_rate, target_sr=16000)
+    except Exception as e:
+        raise IOError(f"讀取或處理音訊時發生錯誤: {e}")
+    
+    input_values = processor(speech, sampling_rate=16000, return_tensors="pt").input_values
+    with torch.no_grad():
+        logits = model(input_values).logits
+    predicted_ids = torch.argmax(logits, dim=-1)
+    user_ipa_full = processor.decode(predicted_ids[0])
+
+    # --- 音素級對齊 (您的原始邏輯) ---
+    word_alignments = _get_phoneme_alignments_by_word(user_ipa_full, target_ipa_by_word)
+
+    # --- 格式化為指定的 JSON 結構 ---
+    return _format_to_json_structure(word_alignments, target_sentence, target_words_original)
+
+
+# --- 3. 您的原始對齊函數 (設為內部函數，未修改邏輯) ---
+def _get_phoneme_alignments_by_word(user_phoneme_str, target_words_ipa):
+    # ... 您的程式碼完全不變 ...
     user_phonemes = list(user_phoneme_str.replace(' ', ''))
     target_phonemes_flat = []
     word_boundaries = []
     current_idx = 0
     for word_ipa in target_words_ipa:
-        phonemes = list(word_ipa) # 已經是乾淨的音標
+        phonemes = list(word_ipa)
         target_phonemes_flat.extend(phonemes)
         current_idx += len(phonemes)
         word_boundaries.append(current_idx)
@@ -111,79 +137,60 @@ def get_phoneme_alignments_by_word(user_phoneme_str, target_words_ipa):
             
     return alignments_by_word
 
-# --- 6. 格式化輸出函式 (已簡化) ---
-def format_and_print_final_report(alignments):
+# --- 4. 新增的格式化函數 (設為內部函數) ---
+def _format_to_json_structure(alignments, sentence, original_words) -> dict:
+    # ... 與上一版相同，用於生成您指定的 JSON 結構 ...
     total_phonemes = 0
     total_errors = 0
-    correct_words = 0
-    
-    target_line_parts = []
-    user_line_parts = []
+    correct_words_count = 0
+    words_data = []
 
-    for alignment in alignments:
+    for i, alignment in enumerate(alignments):
         word_is_correct = True
+        phonemes_data = []
         
-        max_lens = [max(len(t), len(u)) for t, u in zip(alignment['target'], alignment['user'])]
-        
-        target_word_parts = [p.ljust(max_lens[j]) for j, p in enumerate(alignment['target'])]
-        target_line_parts.append(f"[ {' '.join(target_word_parts)} ]")
-        
-        user_word_parts = []
-        for j, user_phoneme in enumerate(alignment['user']):
+        for j in range(len(alignment['target'])):
             target_phoneme = alignment['target'][j]
+            user_phoneme = alignment['user'][j]
             is_match = (user_phoneme == target_phoneme)
             
+            phonemes_data.append({
+                "target": target_phoneme,
+                "user": user_phoneme,
+                "isMatch": is_match
+            })
+            
             if not is_match:
                 word_is_correct = False
-                if user_phoneme != '-' and target_phoneme != '-': # 替換
-                    total_errors += 1
-                elif user_phoneme == '-': # 省略
-                    total_errors += 1
-                else: # 插入
-                    total_errors += 1
-            
-            color = Fore.GREEN if is_match else Fore.RED
-            user_word_parts.append(f"{color}{user_phoneme.ljust(max_lens[j])}{Style.RESET_ALL}")
-        
-        user_line_parts.append(f"[ {' '.join(user_word_parts)} ]")
+                if user_phoneme != '-' and target_phoneme != '-': total_errors += 1
+                elif user_phoneme == '-': total_errors += 1
+                else: total_errors += 1
         
         if word_is_correct:
-            correct_words += 1
+            correct_words_count += 1
+            
+        words_data.append({
+            "word": original_words[i] if i < len(original_words) else "N/A",
+            "isCorrect": word_is_correct,
+            "phonemes": phonemes_data
+        })
         
         total_phonemes += sum(1 for p in alignment['target'] if p != '-')
 
-    # --- 計算統計資料 ---
     total_words = len(alignments)
-    incorrect_words = total_words - correct_words
-    overall_score = (correct_words / total_words) * 100 if total_words > 0 else 0
+    overall_score = (correct_words_count / total_words) * 100 if total_words > 0 else 0
     phoneme_error_rate = (total_errors / total_phonemes) * 100 if total_phonemes > 0 else 0
 
-    # --- 列印報告 ---
-    separator = "="*70
-    print("\n" + separator)
-    print("Pronunciation Analysis".center(70))
-    print(separator + "\n")
-
-    print(f"Sentence: {TARGET_SENTENCE}\n")
-    print(f"Target  : {' '.join(target_line_parts)}")
-    print(f"User    : {' '.join(user_line_parts)}")
-
-    print("\n" + "-" * 70)
-    print("[ Summary ]")
-    print("-" * 70)
-    print(f"- Overall Score:         {overall_score:.1f}%")
-    print(f"- Total Words:           {total_words}")
-    print(f"- Correct Words:         {correct_words}")
-    print(f"- Incorrect Words:       {incorrect_words}")
-    print(f"- Phoneme Error Rate:    {phoneme_error_rate:.2f}% ({total_errors} errors in {total_phonemes} target phonemes)")
-    # (已修改) 使用 UTC 時間
-    print(f"- Analysis Timestamp:    {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} (UTC)")
+    final_result = {
+        "sentence": sentence,
+        "analysisTimestampUTC": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'),
+        "summary": {
+            "overallScore": round(overall_score, 1),
+            "totalWords": total_words,
+            "correctWords": correct_words_count,
+            "phonemeErrorRate": round(phoneme_error_rate, 2)
+        },
+        "words": words_data
+    }
     
-    print("\n" + separator)
-
-
-# --- 主流程 ---
-print("正在進行音素級對齊...")
-word_alignments = get_phoneme_alignments_by_word(user_ipa_full, target_ipa_by_word)
-
-format_and_print_final_report(word_alignments)
+    return final_result

main.py

@@ -0,0 +1,127 @@
+# main.py (Final Corrected Version)
+
+import uvicorn
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException
+from fastapi.responses import JSONResponse
+import os
+import shutil
+from contextlib import asynccontextmanager
+import asyncio
+import importlib.util
+import sys
+from datetime import datetime  # The required import statement
+
+# Ngrok is optional, so we handle its potential absence
+try:
+    from pyngrok import ngrok, conf
+    PYNGROK_INSTALLED = True
+except ImportError:
+    PYNGROK_INSTALLED = False
+
+# --- Analyzer Loading Logic ---
+ANALYZER_MODULES = {}
+SUPPORTED_LANGUAGES = ["en_us"]
+
+async def load_analyzers():
+    print("正在預載入所有支援的分析器模型...")
+    for lang in SUPPORTED_LANGUAGES:
+        try:
+            module_name = f"analyzer.ASR_{lang}"
+            spec = importlib.util.find_spec(module_name)
+            if spec is None:
+                print(f"警告：找不到 {lang} 的分析器模組: {module_name}")
+                continue
+            
+            analyzer_module = importlib.util.module_from_spec(spec)
+            sys.modules[module_name] = analyzer_module
+            spec.loader.exec_module(analyzer_module)
+            
+            if hasattr(analyzer_module, 'load_model'):
+                await asyncio.to_thread(analyzer_module.load_model)
+                ANALYZER_MODULES[lang] = analyzer_module
+                print(f"'{lang}' 分析器載入成功。")
+            else:
+                print(f"警告：'{lang}' 模組中沒有找到 load_model 函數。")
+        except Exception as e:
+            print(f"錯誤：載入 '{lang}' 分析器時失敗: {e}")
+
+# --- FastAPI Lifespan ---
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print("應用程式啟動中...")
+    await load_analyzers()
+    
+    if PYNGROK_INSTALLED:
+        NGROK_AUTHTOKEN = os.environ.get("NGROK_AUTHTOKEN")
+        if NGROK_AUTHTOKEN:
+            conf.get_default().auth_token = NGROK_AUTHTOKEN
+            print("正在啟動 ngrok 通道...")
+            public_url = await asyncio.to_thread(ngrok.connect, 8000, name="pronunciation-api")
+            print(f"Ngrok 通道已建立，公開 URL: {public_url}")
+        else:
+            print("警告：未設定 NGROK_AUTHTOKEN，Ngrok 將不會啟動。")
+    else:
+        print("警告: pyngrok 套件未安裝，Ngrok 將不會啟動。")
+
+    yield
+    
+    print("應用程式關閉中...")
+    if PYNGROK_INSTALLED and ngrok.get_tunnels():
+        ngrok.disconnect()
+        print("Ngrok 通道已關閉。")
+
+# --- FastAPI App Initialization ---
+app = FastAPI(lifespan=lifespan)
+TEMP_DIR = "temp_audio"
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+# --- API Endpoint ---
+@app.post("/api/v1/recognize")
+async def recognize_speech_api(
+    language: str = Form(...),
+    target_sentence: str = Form(...),
+    file: UploadFile = File(...)
+):
+    if language not in ANALYZER_MODULES:
+        raise HTTPException(status_code=400, detail=f"不支援的語言: '{language}'。支援的語言: {list(ANALYZER_MODULES.keys())}")
+
+    if not file.filename or not file.filename.lower().endswith('.wav'):
+        raise HTTPException(status_code=400, detail="檔案格式錯誤或檔名無效，請上傳 .wav 檔案。")
+
+    safe_filename = os.path.basename(file.filename)
+    temp_file_path = os.path.join(TEMP_DIR, f"{datetime.now().strftime('%Y%m%d%H%M%S')}-{safe_filename}")
+    
+    try:
+        with open(temp_file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+
+        analyzer_module = ANALYZER_MODULES[language]
+        print(f"使用 '{language}' 分析器處理檔案: {file.filename}")
+        
+        analysis_result = await asyncio.to_thread(
+            analyzer_module.analyze, temp_file_path, target_sentence
+        )
+
+        return JSONResponse(content=analysis_result)
+    except Exception as e:
+        print(f"處理請求時發生未預期的錯誤: {e}")
+        raise HTTPException(status_code=500, detail=f"伺服器內部錯誤: {str(e)}")
+    finally:
+        if os.path.exists(temp_file_path):
+            os.remove(temp_file_path)
+        if file:
+            await file.close()
+
+@app.get("/")
+def read_root():
+    return {"message": "發音分析 API 已啟動。請使用 POST /api/v1/recognize 端點。"}
+
+# --- Server Execution ---
+if __name__ == "__main__":
+    print("="*60)
+    if PYNGROK_INSTALLED:
+        print("請確保已設定 NGROK_AUTHTOKEN 環境變數以便 ngrok 正常運作。")
+    else:
+        print("pyngrok 未安裝，服務僅在本地運行。")
+    print("="*60)
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

requirements.txt

@@ -1,8 +1,10 @@
+fastapi
+uvicorn[standard]
+pyngrok
+python-multipart
 torch
 soundfile
 librosa
 transformers
-phonemizer
-fastapi
-uvicorn[standard]
-colorama
+phonemizer[espeak]
+numpy