MFA集成

.gitignore

@@ -7,11 +7,6 @@ __pycache__/
 *$py.class
 *.so
 
-# 分发/打包
-dist/
-build/
-*.egg-info/
-
 # pip-tools
 *.egg
 
@@ -29,5 +24,10 @@ build/
 temp/
 *.tmp
 
-# 数据目录（根据需要调整）
+# 数据（根据需要调整）
+config.json
 bank/
+
+# AI 模型相关
+tools/mfa_engine
+models

requirements.in

@@ -10,3 +10,7 @@ customtkinter
 transformers>=4.25.0
 torch
 accelerate
+
+# Silero VAD 语音活动检测
+silero-vad>=5.1
+onnxruntime

requirements.txt

@@ -2,7 +2,7 @@
 # This file is autogenerated by pip-compile with Python 3.13
 # by the following command:
 #
-#    pip-compile requirements.in
+#    pip-compile --output-file=requirements.txt requirements.in
 #
 --index-url https://pypi.tuna.tsinghua.edu.cn/simple
 
@@ -28,6 +28,8 @@ colorama==0.4.6
     # via
     #   click
     #   tqdm
+coloredlogs==15.0.1
+    # via onnxruntime
 customtkinter==5.2.2
     # via -r requirements.in
 darkdetect==0.8.0
@@ -37,6 +39,8 @@ filelock==3.20.3
     #   huggingface-hub
     #   torch
     #   transformers
+flatbuffers==25.12.19
+    # via onnxruntime
 fsspec==2026.1.0
     # via
     #   huggingface-hub
@@ -54,6 +58,8 @@ huggingface-hub==1.3.5
     #   accelerate
     #   tokenizers
     #   transformers
+humanfriendly==10.0
+    # via coloredlogs
 idna==3.11
     # via
     #   anyio
@@ -71,18 +77,29 @@ numpy==2.4.1
     #   accelerate
     #   audiofile
     #   audmath
+    #   onnxruntime
     #   soundfile
     #   transformers
+onnxruntime==1.23.2
+    # via
+    #   -r requirements.in
+    #   silero-vad
 packaging==26.0
     # via
     #   accelerate
     #   customtkinter
     #   huggingface-hub
+    #   onnxruntime
+    #   silero-vad
     #   transformers
+protobuf==6.33.5
+    # via onnxruntime
 psutil==7.2.2
     # via accelerate
 pycparser==3.0
     # via cffi
+pyreadline3==3.5.4
+    # via humanfriendly
 pyyaml==6.0.3
     # via
     #   accelerate
@@ -96,10 +113,14 @@ safetensors==0.7.0
     #   transformers
 shellingham==1.5.4
     # via huggingface-hub
+silero-vad==6.2.0
+    # via -r requirements.in
 soundfile==0.13.1
     # via audiofile
 sympy==1.14.0
-    # via torch
+    # via
+    #   onnxruntime
+    #   torch
 textgrid==1.6.1
     # via -r requirements.in
 tokenizers==0.22.2
@@ -108,6 +129,10 @@ torch==2.10.0
     # via
     #   -r requirements.in
     #   accelerate
+    #   silero-vad
+    #   torchaudio
+torchaudio==2.10.0
+    # via silero-vad
 tqdm==4.67.1
     # via
     #   -r requirements.in

src/gui.py

@@ -353,17 +353,43 @@ class ModelDownloadFrame(ctk.CTkFrame):
         ctk.CTkEntry(self, textvariable=self.mfa_dir_var, width=320).grid(row=7, column=1, padx=5, pady=5, sticky="w")
         ctk.CTkButton(self, text="浏览", width=60, command=self._browse_mfa_dir).grid(row=7, column=2, padx=5, pady=5)
         
-        # MFA 状态
-        ctk.CTkLabel(self, text="状态:").grid(row=8, column=0, padx=10, pady=5, sticky="w")
-        self.mfa_status = ctk.CTkLabel(self, text="🚧 TODO: 自动下载功能开发中", text_color="orange")
-        self.mfa_status.grid(row=8, column=1, columnspan=2, padx=5, pady=5, sticky="w")
+        # MFA 语言选择
+        ctk.CTkLabel(self, text="选择语言:").grid(row=8, column=0, padx=10, pady=5, sticky="w")
+        self.mfa_lang_var = ctk.StringVar(value="mandarin")
+        self.mfa_lang_dropdown = ctk.CTkComboBox(
+            self,
+            values=["mandarin", "japanese"],
+            variable=self.mfa_lang_var,
+            width=200,
+            command=self._on_mfa_lang_change
+        )
+        self.mfa_lang_dropdown.grid(row=8, column=1, padx=5, pady=5, sticky="w")
+        
+        self.mfa_lang_desc = ctk.CTkLabel(self, text="中文 (普通话)", text_color="gray")
+        self.mfa_lang_desc.grid(row=8, column=2, padx=5, pady=5, sticky="w")
+        
+        # MFA 下载按钮和状态
+        ctk.CTkLabel(self, text="状态:").grid(row=9, column=0, padx=10, pady=5, sticky="w")
+        self.mfa_status = ctk.CTkLabel(self, text="⏳ 未下载", text_color="gray")
+        self.mfa_status.grid(row=9, column=1, padx=5, pady=5, sticky="w")
+        
+        self.mfa_download_btn = ctk.CTkButton(
+            self,
+            text="下载模型",
+            command=self._download_mfa_models,
+            width=140
+        )
+        self.mfa_download_btn.grid(row=9, column=2, padx=5, pady=5, sticky="w")
         
         # MFA 文件列表
-        ctk.CTkLabel(self, text="已有文件:").grid(row=9, column=0, padx=10, pady=(10, 5), sticky="nw")
+        ctk.CTkLabel(self, text="已有文件:").grid(row=10, column=0, padx=10, pady=(10, 5), sticky="nw")
         self.mfa_files_text = ctk.CTkTextbox(self, height=70, width=400)
-        self.mfa_files_text.grid(row=9, column=1, columnspan=2, padx=5, pady=(10, 5), sticky="w")
+        self.mfa_files_text.grid(row=10, column=1, columnspan=2, padx=5, pady=(10, 5), sticky="w")
         self.mfa_files_text.insert("end", "选择目录后显示文件列表")
         self.mfa_files_text.configure(state="disabled")
+        
+        # 初始扫描
+        self._scan_mfa_dir()
     
     def _get_model_desc(self):
         """获取当前选中模型的描述"""
@@ -397,6 +423,51 @@ class ModelDownloadFrame(ctk.CTkFrame):
             self._save_config()
             self._scan_mfa_dir()
     
+    def _on_mfa_lang_change(self, choice):
+        """MFA 语言选择变更"""
+        from src.mfa_model_downloader import get_available_languages
+        langs = get_available_languages()
+        self.mfa_lang_desc.configure(text=langs.get(choice, ""))
+    
+    def _download_mfa_models(self):
+        """下载 MFA 模型"""
+        if self._download_thread and self._download_thread.is_alive():
+            return
+        
+        self.mfa_download_btn.configure(state="disabled")
+        self.mfa_status.configure(text="⏳ 下载中...", text_color="gray")
+        self._download_thread = threading.Thread(target=self._do_download_mfa, daemon=True)
+        self._download_thread.start()
+    
+    def _do_download_mfa(self):
+        """执行 MFA 模型下载（后台线程）"""
+        from src.mfa_model_downloader import download_language_models
+        
+        language = self.mfa_lang_var.get()
+        output_dir = self.mfa_dir_var.get()
+        
+        # 确保目录存在
+        if not os.path.exists(output_dir):
+            os.makedirs(output_dir)
+        
+        self.log_callback(f"开始下载 MFA 模型: {language}")
+        
+        success, acoustic_path, dict_path = download_language_models(
+            language=language,
+            output_dir=output_dir,
+            progress_callback=self.log_callback
+        )
+        
+        if success:
+            self.after(0, lambda: self.mfa_status.configure(text="✅ 已下载", text_color="green"))
+            self.log_callback(f"声学模型: {acoustic_path}")
+            self.log_callback(f"字典文件: {dict_path}")
+        else:
+            self.after(0, lambda: self.mfa_status.configure(text="❌ 下载失败", text_color="red"))
+        
+        self.after(0, lambda: self.mfa_download_btn.configure(state="normal"))
+        self.after(0, self._scan_mfa_dir)
+    
     def _scan_mfa_dir(self):
         """扫描 MFA 模型目录"""
         mfa_dir = self.mfa_dir_var.get()
@@ -498,82 +569,156 @@ class MakeDatasetFrame(ctk.CTkFrame):
     def __init__(self, master, log_callback):
         super().__init__(master)
         self.log_callback = log_callback
+        self._is_running = False
         self._setup_ui()
+        self._check_mfa_status()
     
     def _setup_ui(self):
+        # MFA 状态提示
+        self.mfa_status_label = ctk.CTkLabel(
+            self, 
+            text="⏳ 检查 MFA 环境...",
+            font=ctk.CTkFont(size=12)
+        )
+        self.mfa_status_label.grid(row=0, column=0, columnspan=3, padx=10, pady=(10, 5), sticky="w")
+        
         # 数据集原始目录
-        ctk.CTkLabel(self, text="① 切片及LAB目录:").grid(row=0, column=0, padx=10, pady=5, sticky="w")
+        ctk.CTkLabel(self, text="① 切片及LAB目录:").grid(row=1, column=0, padx=10, pady=5, sticky="w")
         self.raw_dir_var = ctk.StringVar()
-        ctk.CTkEntry(self, textvariable=self.raw_dir_var, width=400).grid(row=0, column=1, padx=5, pady=5)
-        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_raw_dir).grid(row=0, column=2, padx=5, pady=5)
+        ctk.CTkEntry(self, textvariable=self.raw_dir_var, width=400).grid(row=1, column=1, padx=5, pady=5)
+        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_raw_dir).grid(row=1, column=2, padx=5, pady=5)
+        
+        # 输出目录
+        ctk.CTkLabel(self, text="② TextGrid输出目录:").grid(row=2, column=0, padx=10, pady=5, sticky="w")
+        self.output_dir_var = ctk.StringVar()
+        ctk.CTkEntry(self, textvariable=self.output_dir_var, width=400).grid(row=2, column=1, padx=5, pady=5)
+        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_output_dir).grid(row=2, column=2, padx=5, pady=5)
         
         # 字典路径
-        ctk.CTkLabel(self, text="② 字典文件:").grid(row=1, column=0, padx=10, pady=5, sticky="w")
-        self.dict_path_var = ctk.StringVar()
-        ctk.CTkEntry(self, textvariable=self.dict_path_var, width=400).grid(row=1, column=1, padx=5, pady=5)
-        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_dict).grid(row=1, column=2, padx=5, pady=5)
+        ctk.CTkLabel(self, text="③ 字典文件:").grid(row=3, column=0, padx=10, pady=5, sticky="w")
+        self.dict_path_var = ctk.StringVar(value="models/mfa/mandarin_china_mfa.dict")
+        ctk.CTkEntry(self, textvariable=self.dict_path_var, width=400).grid(row=3, column=1, padx=5, pady=5)
+        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_dict).grid(row=3, column=2, padx=5, pady=5)
         
         # MFA模型路径
-        ctk.CTkLabel(self, text="③ MFA模型文件:").grid(row=2, column=0, padx=10, pady=5, sticky="w")
-        self.mfa_model_var = ctk.StringVar()
-        ctk.CTkEntry(self, textvariable=self.mfa_model_var, width=400).grid(row=2, column=1, padx=5, pady=5)
-        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_mfa).grid(row=2, column=2, padx=5, pady=5)
-        
-        # 临时目录
-        ctk.CTkLabel(self, text="④ 临时目录:").grid(row=3, column=0, padx=10, pady=5, sticky="w")
-        self.temp_dir_var = ctk.StringVar(value="temp")
-        ctk.CTkEntry(self, textvariable=self.temp_dir_var, width=400).grid(row=3, column=1, padx=5, pady=5)
-        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_temp).grid(row=3, column=2, padx=5, pady=5)
-        
-        # 数据集名称
-        ctk.CTkLabel(self, text="⑤ 数据集名称:").grid(row=4, column=0, padx=10, pady=5, sticky="w")
-        self.dataset_name_var = ctk.StringVar()
-        ctk.CTkEntry(self, textvariable=self.dataset_name_var, width=400).grid(row=4, column=1, padx=5, pady=5)
+        ctk.CTkLabel(self, text="④ MFA模型文件:").grid(row=4, column=0, padx=10, pady=5, sticky="w")
+        self.mfa_model_var = ctk.StringVar(value="models/mfa/mandarin_mfa.zip")
+        ctk.CTkEntry(self, textvariable=self.mfa_model_var, width=400).grid(row=4, column=1, padx=5, pady=5)
+        ctk.CTkButton(self, text="浏览", width=60, command=self._browse_mfa).grid(row=4, column=2, padx=5, pady=5)
+        
+        # 选项
+        options_frame = ctk.CTkFrame(self)
+        options_frame.grid(row=5, column=0, columnspan=3, padx=10, pady=10, sticky="w")
+        
+        self.single_speaker_var = ctk.BooleanVar(value=True)
+        ctk.CTkCheckBox(
+            options_frame, 
+            text="单说话人模式", 
+            variable=self.single_speaker_var
+        ).pack(side="left", padx=10)
+        
+        self.clean_var = ctk.BooleanVar(value=True)
+        ctk.CTkCheckBox(
+            options_frame, 
+            text="清理旧缓存", 
+            variable=self.clean_var
+        ).pack(side="left", padx=10)
         
         # 执行按钮
-        ctk.CTkButton(self, text="⑥ 开始制作", command=self._run).grid(row=5, column=1, pady=20)
+        self.run_btn = ctk.CTkButton(self, text="⑤ 开始对齐", command=self._run)
+        self.run_btn.grid(row=6, column=1, pady=20)
+    
+    def _check_mfa_status(self):
+        """检查 MFA 环境状态"""
+        from src.mfa_runner import check_mfa_available
+        
+        if check_mfa_available():
+            self.mfa_status_label.configure(
+                text="✅ MFA 外挂环境已就绪 (tools/mfa_engine)",
+                text_color="green"
+            )
+        else:
+            self.mfa_status_label.configure(
+                text="❌ MFA 外挂环境不可用，请检查 tools/mfa_engine 目录",
+                text_color="red"
+            )
     
     def _browse_raw_dir(self):
         path = filedialog.askdirectory(title="选择切片及LAB目录")
         if path:
             self.raw_dir_var.set(path)
     
+    def _browse_output_dir(self):
+        path = filedialog.askdirectory(title="选择TextGrid输出目录")
+        if path:
+            self.output_dir_var.set(path)
+    
     def _browse_dict(self):
-        path = filedialog.askopenfilename(title="选择字典文件", filetypes=[("文本文件", "*.txt")])
+        path = filedialog.askopenfilename(
+            title="选择字典文件", 
+            filetypes=[("字典文件", "*.dict *.txt"), ("所有文件", "*.*")]
+        )
         if path:
             self.dict_path_var.set(path)
     
     def _browse_mfa(self):
-        path = filedialog.askopenfilename(title="选择MFA模型", filetypes=[("ZIP文件", "*.zip")])
+        path = filedialog.askopenfilename(
+            title="选择MFA模型", 
+            filetypes=[("ZIP文件", "*.zip"), ("所有文件", "*.*")]
+        )
         if path:
             self.mfa_model_var.set(path)
     
-    def _browse_temp(self):
-        path = filedialog.askdirectory(title="选择临时目录")
-        if path:
-            self.temp_dir_var.set(path)
-    
     def _run(self):
+        if self._is_running:
+            return
+        
         raw_dir = self.raw_dir_var.get()
+        output_dir = self.output_dir_var.get()
         dict_path = self.dict_path_var.get()
         mfa_model = self.mfa_model_var.get()
-        temp_dir = self.temp_dir_var.get()
-        dataset_name = self.dataset_name_var.get()
         
-        if not all([raw_dir, dict_path, mfa_model, temp_dir, dataset_name]):
-            messagebox.showerror("错误", "请填写所有必要字段")
+        if not raw_dir or not output_dir:
+            messagebox.showerror("错误", "请填写输入目录和输出目录")
             return
         
-        self.log_callback("批量制作数据集功能需要MFA环境支持")
-        self.log_callback("请确保已安装Montreal Forced Aligner")
-        self.log_callback(f"配置信息:")
-        self.log_callback(f"  - 原始目录: {raw_dir}")
-        self.log_callback(f"  - 字典: {dict_path}")
-        self.log_callback(f"  - MFA模型: {mfa_model}")
-        self.log_callback(f"  - 临时目录: {temp_dir}")
-        self.log_callback(f"  - 数据集名称: {dataset_name}")
-        self.log_callback("此功能涉及多个外部脚本调用，建议在命令行中执行")
-        logger.info("批量制作数据集配置已记录")
+        self._is_running = True
+        self.run_btn.configure(state="disabled", text="对齐中...")
+        
+        threading.Thread(
+            target=self._process,
+            args=(raw_dir, output_dir, dict_path, mfa_model),
+            daemon=True
+        ).start()
+    
+    def _process(self, raw_dir, output_dir, dict_path, mfa_model):
+        """执行 MFA 对齐（后台线程）"""
+        from src.mfa_runner import run_mfa_alignment
+        
+        self.log_callback("=" * 50)
+        self.log_callback("开始 MFA 对齐任务")
+        
+        success, message = run_mfa_alignment(
+            corpus_dir=raw_dir,
+            output_dir=output_dir,
+            dict_path=dict_path if dict_path else None,
+            model_path=mfa_model if mfa_model else None,
+            single_speaker=self.single_speaker_var.get(),
+            clean=self.clean_var.get(),
+            progress_callback=self.log_callback
+        )
+        
+        if success:
+            self.log_callback("✅ MFA 对齐任务完成!")
+            self.log_callback(f"TextGrid 文件已输出到: {output_dir}")
+        else:
+            self.log_callback(f"❌ MFA 对齐失败: {message}")
+        
+        self.log_callback("=" * 50)
+        
+        # 恢复按钮状态
+        self.after(0, lambda: self.run_btn.configure(state="normal", text="⑤ 开始对齐"))
+        self._is_running = False
 
 
 class App(ctk.CTk):

src/mfa_model_downloader.py

@@ -0,0 +1,236 @@
+# -*- coding: utf-8 -*-
+"""
+MFA 模型下载模块
+支持下载中文和日文的声学模型及字典
+"""
+
+import os
+import logging
+import urllib.request
+import urllib.error
+from pathlib import Path
+from typing import Optional, Callable
+
+logger = logging.getLogger(__name__)
+
+# 模型下载基础 URL
+GITHUB_RELEASE_BASE = "https://github.com/MontrealCorpusTools/mfa-models/releases/download"
+GITHUB_RAW_BASE = "https://raw.githubusercontent.com/MontrealCorpusTools/mfa-models/main"
+
+# 支持的语言配置
+# 格式: {语言代码: {名称, 声学模型信息, 字典信息}}
+LANGUAGE_MODELS = {
+    "mandarin": {
+        "name": "中文 (普通话)",
+        "acoustic": {
+            "tag": "acoustic-mandarin_mfa-v3.0.0",
+            "filename": "mandarin_mfa.zip",
+            "description": "Mandarin MFA acoustic model v3.0.0"
+        },
+        "dictionary": {
+            # 字典从 releases 下载，tag 格式: dictionary-{name}-v{version}
+            "tag": "dictionary-mandarin_china_mfa-v3.0.0",
+            "filename": "mandarin_china_mfa.dict",
+            "description": "Mandarin (China) MFA dictionary v3.0.0"
+        }
+    },
+    "japanese": {
+        "name": "日文",
+        "acoustic": {
+            "tag": "acoustic-japanese_mfa-v3.0.0",
+            "filename": "japanese_mfa.zip",
+            "description": "Japanese MFA acoustic model v3.0.0"
+        },
+        "dictionary": {
+            "tag": "dictionary-japanese_mfa-v3.0.0",
+            "filename": "japanese_mfa.dict",
+            "description": "Japanese MFA dictionary v3.0.0"
+        }
+    }
+}
+
+
+def get_available_languages() -> dict:
+    """获取可用的语言列表"""
+    return {k: v["name"] for k, v in LANGUAGE_MODELS.items()}
+
+
+def _download_file(
+    url: str,
+    dest_path: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> bool:
+    """
+    下载文件
+    
+    参数:
+        url: 下载地址
+        dest_path: 保存路径
+        progress_callback: 进度回调
+    
+    返回:
+        是否成功
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    try:
+        log(f"正在下载: {url}")
+        
+        # 创建目录
+        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
+        
+        # 下载文件
+        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+        
+        with urllib.request.urlopen(req, timeout=60) as response:
+            total_size = response.headers.get("Content-Length")
+            if total_size:
+                total_size = int(total_size)
+                log(f"文件大小: {total_size / 1024 / 1024:.1f} MB")
+            
+            # 分块下载
+            block_size = 8192
+            downloaded = 0
+            
+            with open(dest_path, "wb") as f:
+                while True:
+                    chunk = response.read(block_size)
+                    if not chunk:
+                        break
+                    f.write(chunk)
+                    downloaded += len(chunk)
+                    
+                    if total_size and downloaded % (block_size * 100) == 0:
+                        percent = downloaded / total_size * 100
+                        log(f"下载进度: {percent:.1f}%")
+        
+        log(f"下载完成: {dest_path}")
+        return True
+        
+    except urllib.error.HTTPError as e:
+        log(f"HTTP 错误: {e.code} - {e.reason}")
+        return False
+    except urllib.error.URLError as e:
+        log(f"网络错误: {e.reason}")
+        return False
+    except Exception as e:
+        log(f"下载失败: {e}")
+        return False
+
+
+def download_acoustic_model(
+    language: str,
+    output_dir: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> tuple[bool, str]:
+    """
+    下载声学模型
+    
+    参数:
+        language: 语言代码 (mandarin/japanese)
+        output_dir: 输出目录
+        progress_callback: 进度回调
+    
+    返回:
+        (成功标志, 文件路径或错误信息)
+    """
+    if language not in LANGUAGE_MODELS:
+        return False, f"不支持的语言: {language}"
+    
+    config = LANGUAGE_MODELS[language]["acoustic"]
+    url = f"{GITHUB_RELEASE_BASE}/{config['tag']}/{config['filename']}"
+    dest_path = os.path.join(output_dir, config["filename"])
+    
+    if os.path.exists(dest_path):
+        if progress_callback:
+            progress_callback(f"声学模型已存在: {dest_path}")
+        return True, dest_path
+    
+    if _download_file(url, dest_path, progress_callback):
+        return True, dest_path
+    else:
+        return False, "声学模型下载失败"
+
+
+def download_dictionary(
+    language: str,
+    output_dir: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> tuple[bool, str]:
+    """
+    下载字典文件
+    
+    参数:
+        language: 语言代码 (mandarin/japanese)
+        output_dir: 输出目录
+        progress_callback: 进度回调
+    
+    返回:
+        (成功标志, 文件路径或错误信息)
+    """
+    if language not in LANGUAGE_MODELS:
+        return False, f"不支持的语言: {language}"
+    
+    config = LANGUAGE_MODELS[language]["dictionary"]
+    # 字典文件从 releases 下载
+    url = f"{GITHUB_RELEASE_BASE}/{config['tag']}/{config['filename']}"
+    dest_path = os.path.join(output_dir, config["filename"])
+    
+    if os.path.exists(dest_path):
+        if progress_callback:
+            progress_callback(f"字典文件已存在: {dest_path}")
+        return True, dest_path
+    
+    if _download_file(url, dest_path, progress_callback):
+        return True, dest_path
+    else:
+        return False, "字典文件下载失败"
+
+
+def download_language_models(
+    language: str,
+    output_dir: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> tuple[bool, str, str]:
+    """
+    下载指定语言的声学模型和字典
+    
+    参数:
+        language: 语言代码 (mandarin/japanese)
+        output_dir: 输出目录
+        progress_callback: 进度回调
+    
+    返回:
+        (成功标志, 声学模型路径, 字典路径)
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    if language not in LANGUAGE_MODELS:
+        return False, "", f"不支持的语言: {language}"
+    
+    lang_name = LANGUAGE_MODELS[language]["name"]
+    log(f"开始下载 {lang_name} 模型...")
+    
+    # 下载声学模型
+    log("=" * 40)
+    log("下载声学模型...")
+    success, acoustic_path = download_acoustic_model(language, output_dir, progress_callback)
+    if not success:
+        return False, "", acoustic_path
+    
+    # 下载字典
+    log("=" * 40)
+    log("下载字典文件...")
+    success, dict_path = download_dictionary(language, output_dir, progress_callback)
+    if not success:
+        return False, acoustic_path, dict_path
+    
+    log("=" * 40)
+    log(f"{lang_name} 模型下载完成!")
+    return True, acoustic_path, dict_path

src/mfa_runner.py

@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+"""
+MFA 外挂调用模块
+采用 Sidecar Pattern，通过 subprocess 调用独立的 MFA 环境
+"""
+
+import os
+import subprocess
+import logging
+from pathlib import Path
+from typing import Optional, Callable
+
+logger = logging.getLogger(__name__)
+
+# 定位路径
+BASE_DIR = Path(__file__).parent.parent.absolute()
+MFA_ENGINE_DIR = BASE_DIR / "tools" / "mfa_engine"
+MFA_PYTHON = MFA_ENGINE_DIR / "python.exe"
+
+# 默认模型路径
+DEFAULT_DICT_PATH = BASE_DIR / "models" / "mandarin.dict"
+DEFAULT_MODEL_PATH = BASE_DIR / "models" / "mandarin.zip"
+DEFAULT_TEMP_DIR = BASE_DIR / "mfa_temp"
+
+
+def check_mfa_available() -> bool:
+    """检查 MFA 外挂环境是否可用"""
+    if not MFA_ENGINE_DIR.exists():
+        logger.warning(f"MFA 引擎目录不存在: {MFA_ENGINE_DIR}")
+        return False
+    if not MFA_PYTHON.exists():
+        logger.warning(f"MFA Python 不存在: {MFA_PYTHON}")
+        return False
+    return True
+
+
+def _build_mfa_env() -> dict:
+    """构造 MFA 专用环境变量"""
+    env = os.environ.copy()
+    
+    # 必须把 Library\bin 加入 PATH，否则 Kaldi DLL 找不到
+    mfa_paths = [
+        str(MFA_ENGINE_DIR),
+        str(MFA_ENGINE_DIR / "Library" / "bin"),
+        str(MFA_ENGINE_DIR / "Scripts"),
+        str(MFA_ENGINE_DIR / "bin"),
+    ]
+    env["PATH"] = ";".join(mfa_paths) + ";" + env.get("PATH", "")
+    
+    return env
+
+
+def run_mfa_alignment(
+    corpus_dir: str,
+    output_dir: str,
+    dict_path: Optional[str] = None,
+    model_path: Optional[str] = None,
+    temp_dir: Optional[str] = None,
+    single_speaker: bool = True,
+    clean: bool = True,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> tuple[bool, str]:
+    """
+    执行 MFA 对齐
+    
+    参数:
+        corpus_dir: 包含 wav 和 lab/txt 的输入目录
+        output_dir: TextGrid 输出目录
+        dict_path: 字典文件路径，默认使用 models/mandarin.dict
+        model_path: 声学模型路径，默认使用 models/mandarin.zip
+        temp_dir: 临时目录，默认使用 mfa_temp
+        single_speaker: 是否为单说话人模式
+        clean: 是否清理旧缓存
+        progress_callback: 进度回调函数
+    
+    返回:
+        (成功标志, 输出信息或错误信息)
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    # 检查环境
+    if not check_mfa_available():
+        return False, "MFA 外挂环境不可用，请检查 tools/mfa_engine 目录"
+    
+    # 设置默认路径
+    dict_path = dict_path or str(DEFAULT_DICT_PATH)
+    model_path = model_path or str(DEFAULT_MODEL_PATH)
+    temp_dir = temp_dir or str(DEFAULT_TEMP_DIR)
+    
+    # 验证路径
+    if not os.path.isdir(corpus_dir):
+        return False, f"输入目录不存在: {corpus_dir}"
+    if not os.path.isfile(dict_path):
+        return False, f"字典文件不存在: {dict_path}"
+    if not os.path.isfile(model_path):
+        return False, f"声学模型不存在: {model_path}"
+    
+    # 创建输出和临时目录
+    os.makedirs(output_dir, exist_ok=True)
+    os.makedirs(temp_dir, exist_ok=True)
+    
+    # 构造命令
+    cmd = [
+        str(MFA_PYTHON),
+        "-m", "montreal_forced_aligner",
+        "align",
+        str(corpus_dir),
+        str(dict_path),
+        str(model_path),
+        str(output_dir),
+        "--temp_directory", str(temp_dir),
+    ]
+    
+    if clean:
+        cmd.append("--clean")
+    if single_speaker:
+        cmd.append("--single_speaker")
+    
+    log(f"正在启动 MFA 对齐引擎...")
+    log(f"输入目录: {corpus_dir}")
+    log(f"输出目录: {output_dir}")
+    
+    try:
+        env = _build_mfa_env()
+        
+        result = subprocess.run(
+            cmd,
+            env=env,
+            capture_output=True,
+            text=True,
+            encoding='utf-8',
+            errors='replace'
+        )
+        
+        if result.returncode == 0:
+            log("MFA 对齐完成!")
+            return True, result.stdout
+        else:
+            error_msg = result.stderr or result.stdout or "未知错误"
+            log(f"MFA 运行出错: {error_msg}")
+            return False, error_msg
+            
+    except FileNotFoundError as e:
+        msg = f"找不到 MFA Python: {e}"
+        log(msg)
+        return False, msg
+    except Exception as e:
+        msg = f"MFA 执行异常: {e}"
+        log(msg)
+        return False, msg
+
+
+def run_mfa_validate(
+    corpus_dir: str,
+    dict_path: Optional[str] = None,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> tuple[bool, str]:
+    """
+    验证语料库格式是否正确
+    
+    参数:
+        corpus_dir: 语料库目录
+        dict_path: 字典文件路径
+        progress_callback: 进度回调函数
+    
+    返回:
+        (成功标志, 输出信息)
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    if not check_mfa_available():
+        return False, "MFA 外挂环境不可用"
+    
+    dict_path = dict_path or str(DEFAULT_DICT_PATH)
+    
+    cmd = [
+        str(MFA_PYTHON),
+        "-m", "montreal_forced_aligner",
+        "validate",
+        str(corpus_dir),
+        str(dict_path),
+    ]
+    
+    log("正在验证语料库...")
+    
+    try:
+        env = _build_mfa_env()
+        result = subprocess.run(
+            cmd,
+            env=env,
+            capture_output=True,
+            text=True,
+            encoding='utf-8',
+            errors='replace'
+        )
+        
+        output = result.stdout + "\n" + result.stderr
+        log("验证完成")
+        return result.returncode == 0, output
+        
+    except Exception as e:
+        return False, str(e)

src/silero_vad_downloader.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+"""
+Silero VAD 模型下载模块
+支持自动下载 Silero VAD 模型到指定目录
+"""
+
+import os
+import logging
+import urllib.request
+import urllib.error
+from pathlib import Path
+from typing import Optional, Callable
+
+logger = logging.getLogger(__name__)
+
+# Silero VAD 模型配置
+SILERO_VAD_CONFIG = {
+    "repo": "snakers4/silero-vad",
+    "model_name": "silero_vad",
+    "version": "v5.1",
+    "onnx_filename": "silero_vad.onnx",
+    "jit_filename": "silero_vad.jit",
+    "download_base": "https://github.com/snakers4/silero-vad/raw/master/src/silero_vad/data"
+}
+
+
+def _download_file(
+    url: str,
+    dest_path: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> bool:
+    """
+    下载文件
+    
+    参数:
+        url: 下载地址
+        dest_path: 保存路径
+        progress_callback: 进度回调
+    
+    返回:
+        是否成功
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    try:
+        log(f"正在下载: {url}")
+        
+        # 创建目录
+        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
+        
+        # 下载文件
+        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+        
+        with urllib.request.urlopen(req, timeout=60) as response:
+            total_size = response.headers.get("Content-Length")
+            if total_size:
+                total_size = int(total_size)
+                log(f"文件大小: {total_size / 1024 / 1024:.2f} MB")
+            
+            # 分块下载
+            block_size = 8192
+            downloaded = 0
+            
+            with open(dest_path, "wb") as f:
+                while True:
+                    chunk = response.read(block_size)
+                    if not chunk:
+                        break
+                    f.write(chunk)
+                    downloaded += len(chunk)
+                    
+                    if total_size and downloaded % (block_size * 100) == 0:
+                        percent = downloaded / total_size * 100
+                        log(f"下载进度: {percent:.1f}%")
+        
+        log(f"下载完成: {dest_path}")
+        return True
+        
+    except urllib.error.HTTPError as e:
+        log(f"HTTP 错误: {e.code} - {e.reason}")
+        return False
+    except urllib.error.URLError as e:
+        log(f"网络错误: {e.reason}")
+        return False
+    except Exception as e:
+        log(f"下载失败: {e}")
+        return False
+
+
+def get_vad_model_path(models_dir: str) -> str:
+    """
+    获取 VAD 模型文件路径
+    
+    参数:
+        models_dir: 模型根目录
+    
+    返回:
+        ONNX 模型文件路径
+    """
+    return os.path.join(models_dir, "silero_vad", SILERO_VAD_CONFIG["onnx_filename"])
+
+
+def is_vad_model_downloaded(models_dir: str) -> bool:
+    """
+    检查 VAD 模型是否已下载
+    
+    参数:
+        models_dir: 模型根目录
+    
+    返回:
+        是否已下载
+    """
+    model_path = get_vad_model_path(models_dir)
+    return os.path.exists(model_path)
+
+
+def download_silero_vad(
+    output_dir: str,
+    progress_callback: Optional[Callable[[str], None]] = None,
+    use_onnx: bool = True
+) -> tuple[bool, str]:
+    """
+    下载 Silero VAD 模型
+    
+    参数:
+        output_dir: 输出目录 (模型根目录)
+        progress_callback: 进度回调
+        use_onnx: 是否下载 ONNX 格式 (默认 True，否则下载 JIT 格式)
+    
+    返回:
+        (成功标志, 文件路径或错误信息)
+    """
+    def log(msg: str):
+        logger.info(msg)
+        if progress_callback:
+            progress_callback(msg)
+    
+    # 确定文件名和 URL
+    if use_onnx:
+        filename = SILERO_VAD_CONFIG["onnx_filename"]
+    else:
+        filename = SILERO_VAD_CONFIG["jit_filename"]
+    
+    url = f"{SILERO_VAD_CONFIG['download_base']}/{filename}"
+    vad_dir = os.path.join(output_dir, "silero_vad")
+    dest_path = os.path.join(vad_dir, filename)
+    
+    # 检查是否已存在
+    if os.path.exists(dest_path):
+        log(f"Silero VAD 模型已存在: {dest_path}")
+        return True, dest_path
+    
+    log("开始下载 Silero VAD 模型...")
+    log(f"版本: {SILERO_VAD_CONFIG['version']}")
+    log(f"格式: {'ONNX' if use_onnx else 'JIT'}")
+    
+    if _download_file(url, dest_path, progress_callback):
+        log("Silero VAD 模型下载完成!")
+        return True, dest_path
+    else:
+        return False, "Silero VAD 模型下载失败"
+
+
+def ensure_vad_model(
+    models_dir: str,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> str:
+    """
+    确保 VAD 模型已下载，如未下载则自动下载
+    
+    参数:
+        models_dir: 模型根目录
+        progress_callback: 进度回调
+    
+    返回:
+        模型文件路径
+    
+    异常:
+        RuntimeError: 下载失败时抛出
+    """
+    model_path = get_vad_model_path(models_dir)
+    
+    if os.path.exists(model_path):
+        logger.info(f"Silero VAD 模型已就绪: {model_path}")
+        return model_path
+    
+    success, result = download_silero_vad(models_dir, progress_callback)
+    if success:
+        return result
+    else:
+        raise RuntimeError(f"Silero VAD 模型下载失败: {result}")

tests/test_mfa_model_downloader.py

@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+"""
+MFA 模型下载模块单元测试
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch, MagicMock
+from pathlib import Path
+
+# 添加项目根目录到路径
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.mfa_model_downloader import (
+    get_available_languages,
+    LANGUAGE_MODELS,
+    GITHUB_RELEASE_BASE,
+    download_acoustic_model,
+    download_dictionary,
+    download_language_models,
+)
+
+
+class TestGetAvailableLanguages(unittest.TestCase):
+    """测试获取可用语言列表"""
+    
+    def test_returns_dict(self):
+        """返回值应为字典"""
+        result = get_available_languages()
+        self.assertIsInstance(result, dict)
+    
+    def test_contains_mandarin(self):
+        """应包含中文"""
+        result = get_available_languages()
+        self.assertIn("mandarin", result)
+        self.assertEqual(result["mandarin"], "中文 (普通话)")
+    
+    def test_contains_japanese(self):
+        """应包含日文"""
+        result = get_available_languages()
+        self.assertIn("japanese", result)
+        self.assertEqual(result["japanese"], "日文")
+
+
+class TestLanguageModelsConfig(unittest.TestCase):
+    """测试语言模型配置"""
+    
+    def test_mandarin_config_complete(self):
+        """中文配置应完整"""
+        config = LANGUAGE_MODELS["mandarin"]
+        self.assertIn("name", config)
+        self.assertIn("acoustic", config)
+        self.assertIn("dictionary", config)
+        
+        # 声学模型配置
+        acoustic = config["acoustic"]
+        self.assertIn("tag", acoustic)
+        self.assertIn("filename", acoustic)
+        self.assertTrue(acoustic["filename"].endswith(".zip"))
+        
+        # 字典配置
+        dictionary = config["dictionary"]
+        self.assertIn("tag", dictionary)
+        self.assertIn("filename", dictionary)
+        self.assertTrue(dictionary["filename"].endswith(".dict"))
+    
+    def test_japanese_config_complete(self):
+        """日文配置应完整"""
+        config = LANGUAGE_MODELS["japanese"]
+        self.assertIn("name", config)
+        self.assertIn("acoustic", config)
+        self.assertIn("dictionary", config)
+    
+    def test_acoustic_url_format(self):
+        """声学模型 URL 格式应正确"""
+        for lang, config in LANGUAGE_MODELS.items():
+            acoustic = config["acoustic"]
+            url = f"{GITHUB_RELEASE_BASE}/{acoustic['tag']}/{acoustic['filename']}"
+            self.assertTrue(url.startswith("https://github.com/"))
+            self.assertIn("mfa-models", url)
+    
+    def test_dictionary_url_format(self):
+        """字典 URL 格式应正确"""
+        for lang, config in LANGUAGE_MODELS.items():
+            dictionary = config["dictionary"]
+            url = f"{GITHUB_RELEASE_BASE}/{dictionary['tag']}/{dictionary['filename']}"
+            self.assertTrue(url.startswith("https://github.com/"))
+            self.assertIn("dictionary-", url)
+
+
+class TestDownloadAcousticModel(unittest.TestCase):
+    """测试声学模型下载"""
+    
+    def test_invalid_language(self):
+        """不支持的语言应返回失败"""
+        success, result = download_acoustic_model("invalid_lang", "/tmp")
+        self.assertFalse(success)
+        self.assertIn("不支持的语言", result)
+    
+    @patch('src.mfa_model_downloader._download_file')
+    def test_download_called_with_correct_url(self, mock_download):
+        """应使用正确的 URL 下载"""
+        mock_download.return_value = True
+        
+        with patch('os.path.exists', return_value=False):
+            download_acoustic_model("mandarin", "/tmp/models")
+        
+        # 验证调用参数
+        call_args = mock_download.call_args
+        url = call_args[0][0]
+        self.assertIn("mandarin_mfa.zip", url)
+        self.assertIn("acoustic-mandarin_mfa", url)
+    
+    @patch('os.path.exists')
+    def test_skip_if_exists(self, mock_exists):
+        """文件已存在时应跳过下载"""
+        mock_exists.return_value = True
+        
+        success, result = download_acoustic_model("mandarin", "/tmp/models")
+        self.assertTrue(success)
+        self.assertIn("mandarin_mfa.zip", result)
+
+
+class TestDownloadDictionary(unittest.TestCase):
+    """测试字典下载"""
+    
+    def test_invalid_language(self):
+        """不支持的语言应返回失败"""
+        success, result = download_dictionary("invalid_lang", "/tmp")
+        self.assertFalse(success)
+        self.assertIn("不支持的语言", result)
+    
+    @patch('src.mfa_model_downloader._download_file')
+    def test_download_called_with_correct_url(self, mock_download):
+        """应使用正确的 URL 下载"""
+        mock_download.return_value = True
+        
+        with patch('os.path.exists', return_value=False):
+            download_dictionary("japanese", "/tmp/models")
+        
+        call_args = mock_download.call_args
+        url = call_args[0][0]
+        self.assertIn("github.com", url)
+        self.assertIn("dictionary-japanese", url)
+
+
+class TestDownloadLanguageModels(unittest.TestCase):
+    """测试完整语言模型下载"""
+    
+    def test_invalid_language(self):
+        """不支持的语言应返回失败"""
+        success, acoustic, dict_path = download_language_models("invalid", "/tmp")
+        self.assertFalse(success)
+    
+    @patch('src.mfa_model_downloader.download_dictionary')
+    @patch('src.mfa_model_downloader.download_acoustic_model')
+    def test_downloads_both_models(self, mock_acoustic, mock_dict):
+        """应同时下载声学模型和字典"""
+        mock_acoustic.return_value = (True, "/tmp/acoustic.zip")
+        mock_dict.return_value = (True, "/tmp/dict.dict")
+        
+        success, acoustic, dict_path = download_language_models("mandarin", "/tmp")
+        
+        self.assertTrue(success)
+        mock_acoustic.assert_called_once()
+        mock_dict.assert_called_once()
+    
+    @patch('src.mfa_model_downloader.download_dictionary')
+    @patch('src.mfa_model_downloader.download_acoustic_model')
+    def test_stops_on_acoustic_failure(self, mock_acoustic, mock_dict):
+        """声学模型下载失败时应停止"""
+        mock_acoustic.return_value = (False, "下载失败")
+        
+        success, _, _ = download_language_models("mandarin", "/tmp")
+        
+        self.assertFalse(success)
+        mock_dict.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mfa_runner.py

@@ -0,0 +1,243 @@
+# -*- coding: utf-8 -*-
+"""
+MFA 运行模块单元测试
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch, MagicMock
+from pathlib import Path
+
+# 添加项目根目录到路径
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.mfa_runner import (
+    check_mfa_available,
+    _build_mfa_env,
+    run_mfa_alignment,
+    run_mfa_validate,
+    BASE_DIR,
+    MFA_ENGINE_DIR,
+    MFA_PYTHON,
+)
+
+
+class TestCheckMfaAvailable(unittest.TestCase):
+    """测试 MFA 环境检查"""
+    
+    @patch('src.mfa_runner.MFA_ENGINE_DIR')
+    def test_returns_false_when_dir_not_exists(self, mock_dir):
+        """目录不存在时应返回 False"""
+        mock_path = MagicMock()
+        mock_path.exists.return_value = False
+        
+        with patch.object(Path, 'exists', return_value=False):
+            # 由于模块级变量，需要重新导入或直接测试逻辑
+            pass
+    
+    def test_path_constants_defined(self):
+        """路径常量应正确定义"""
+        self.assertIsInstance(BASE_DIR, Path)
+        self.assertIsInstance(MFA_ENGINE_DIR, Path)
+        self.assertIsInstance(MFA_PYTHON, Path)
+        
+        # 验证路径结构
+        self.assertTrue(str(MFA_ENGINE_DIR).endswith("mfa_engine"))
+        self.assertTrue(str(MFA_PYTHON).endswith("python.exe"))
+
+
+class TestBuildMfaEnv(unittest.TestCase):
+    """测试 MFA 环境变量构建"""
+    
+    def test_returns_dict(self):
+        """应返回字典"""
+        env = _build_mfa_env()
+        self.assertIsInstance(env, dict)
+    
+    def test_path_contains_mfa_dirs(self):
+        """PATH 应包含 MFA 相关目录"""
+        env = _build_mfa_env()
+        path = env.get("PATH", "")
+        
+        self.assertIn("mfa_engine", path)
+        self.assertIn("Library", path)
+    
+    def test_preserves_original_path(self):
+        """应保留原始 PATH"""
+        original_path = os.environ.get("PATH", "")
+        env = _build_mfa_env()
+        
+        # 原始 PATH 应在新 PATH 中
+        self.assertIn(original_path.split(";")[0], env["PATH"])
+
+
+class TestRunMfaAlignment(unittest.TestCase):
+    """测试 MFA 对齐功能"""
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    def test_fails_when_mfa_unavailable(self, mock_check):
+        """MFA 不可用时应返回失败"""
+        mock_check.return_value = False
+        
+        success, msg = run_mfa_alignment("/input", "/output")
+        
+        self.assertFalse(success)
+        self.assertIn("不可用", msg)
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    def test_fails_when_corpus_not_exists(self, mock_isdir, mock_check):
+        """输入目录不存在时应返回失败"""
+        mock_check.return_value = True
+        mock_isdir.return_value = False
+        
+        success, msg = run_mfa_alignment("/nonexistent", "/output")
+        
+        self.assertFalse(success)
+        self.assertIn("不存在", msg)
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    @patch('os.path.isfile')
+    def test_fails_when_dict_not_exists(self, mock_isfile, mock_isdir, mock_check):
+        """字典文件不存在时应返回失败"""
+        mock_check.return_value = True
+        mock_isdir.return_value = True
+        mock_isfile.return_value = False
+        
+        success, msg = run_mfa_alignment(
+            "/input", "/output",
+            dict_path="/nonexistent.dict"
+        )
+        
+        self.assertFalse(success)
+        self.assertIn("不存在", msg)
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    @patch('os.path.isfile')
+    @patch('os.makedirs')
+    @patch('subprocess.run')
+    def test_calls_subprocess_with_correct_args(
+        self, mock_run, mock_makedirs, mock_isfile, mock_isdir, mock_check
+    ):
+        """应使用正确的参数调用 subprocess"""
+        mock_check.return_value = True
+        mock_isdir.return_value = True
+        mock_isfile.return_value = True
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        
+        run_mfa_alignment(
+            "/input", "/output",
+            dict_path="/dict.dict",
+            model_path="/model.zip",
+            single_speaker=True,
+            clean=True
+        )
+        
+        # 验证 subprocess.run 被调用
+        mock_run.assert_called_once()
+        
+        # 验证命令参数
+        call_args = mock_run.call_args
+        cmd = call_args[0][0]
+        
+        self.assertIn("align", cmd)
+        self.assertIn("/input", cmd)
+        self.assertIn("/dict.dict", cmd)
+        self.assertIn("/model.zip", cmd)
+        self.assertIn("/output", cmd)
+        self.assertIn("--single_speaker", cmd)
+        self.assertIn("--clean", cmd)
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    @patch('os.path.isfile')
+    @patch('os.makedirs')
+    @patch('subprocess.run')
+    def test_returns_success_on_zero_returncode(
+        self, mock_run, mock_makedirs, mock_isfile, mock_isdir, mock_check
+    ):
+        """返回码为 0 时应返回成功"""
+        mock_check.return_value = True
+        mock_isdir.return_value = True
+        mock_isfile.return_value = True
+        mock_run.return_value = MagicMock(returncode=0, stdout="完成", stderr="")
+        
+        success, msg = run_mfa_alignment(
+            "/input", "/output",
+            dict_path="/dict.dict",
+            model_path="/model.zip"
+        )
+        
+        self.assertTrue(success)
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    @patch('os.path.isfile')
+    @patch('os.makedirs')
+    @patch('subprocess.run')
+    def test_returns_failure_on_nonzero_returncode(
+        self, mock_run, mock_makedirs, mock_isfile, mock_isdir, mock_check
+    ):
+        """返回码非 0 时应返回失败"""
+        mock_check.return_value = True
+        mock_isdir.return_value = True
+        mock_isfile.return_value = True
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="错误")
+        
+        success, msg = run_mfa_alignment(
+            "/input", "/output",
+            dict_path="/dict.dict",
+            model_path="/model.zip"
+        )
+        
+        self.assertFalse(success)
+
+
+class TestRunMfaValidate(unittest.TestCase):
+    """测试 MFA 验证功能"""
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    def test_fails_when_mfa_unavailable(self, mock_check):
+        """MFA 不可用时应返回失败"""
+        mock_check.return_value = False
+        
+        success, msg = run_mfa_validate("/corpus")
+        
+        self.assertFalse(success)
+        self.assertIn("不可用", msg)
+
+
+class TestProgressCallback(unittest.TestCase):
+    """测试进度回调"""
+    
+    @patch('src.mfa_runner.check_mfa_available')
+    @patch('os.path.isdir')
+    @patch('os.path.isfile')
+    @patch('os.makedirs')
+    @patch('subprocess.run')
+    def test_callback_called_on_success(
+        self, mock_run, mock_makedirs, mock_isfile, mock_isdir, mock_check
+    ):
+        """成功时应调用回调"""
+        mock_check.return_value = True
+        mock_isdir.return_value = True
+        mock_isfile.return_value = True
+        mock_run.return_value = MagicMock(returncode=0, stdout="完成", stderr="")
+        callback = MagicMock()
+        
+        run_mfa_alignment(
+            "/input", "/output",
+            dict_path="/dict.dict",
+            model_path="/model.zip",
+            progress_callback=callback
+        )
+        
+        # 回调应被调用（至少一次）
+        self.assertTrue(callback.called)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_silero_vad_downloader.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+"""
+Silero VAD 下载模块测试
+"""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+
+from src.silero_vad_downloader import (
+    get_vad_model_path,
+    is_vad_model_downloaded,
+    download_silero_vad,
+    ensure_vad_model,
+    SILERO_VAD_CONFIG
+)
+
+
+class TestSileroVadDownloader(unittest.TestCase):
+    """Silero VAD 下载器测试类"""
+    
+    def test_get_vad_model_path(self):
+        """测试获取模型路径"""
+        models_dir = "/test/models"
+        expected = os.path.join(models_dir, "silero_vad", "silero_vad.onnx")
+        self.assertEqual(get_vad_model_path(models_dir), expected)
+    
+    def test_is_vad_model_downloaded_false(self):
+        """测试模型未下载时返回 False"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self.assertFalse(is_vad_model_downloaded(tmpdir))
+    
+    def test_is_vad_model_downloaded_true(self):
+        """测试模型已下载时返回 True"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            vad_dir = os.path.join(tmpdir, "silero_vad")
+            os.makedirs(vad_dir)
+            model_path = os.path.join(vad_dir, "silero_vad.onnx")
+            with open(model_path, "w") as f:
+                f.write("dummy")
+            self.assertTrue(is_vad_model_downloaded(tmpdir))
+    
+    def test_download_silero_vad_already_exists(self):
+        """测试模型已存在时跳过下载"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            vad_dir = os.path.join(tmpdir, "silero_vad")
+            os.makedirs(vad_dir)
+            model_path = os.path.join(vad_dir, "silero_vad.onnx")
+            with open(model_path, "w") as f:
+                f.write("dummy")
+            
+            success, result = download_silero_vad(tmpdir)
+            self.assertTrue(success)
+            self.assertEqual(result, model_path)
+    
+    def test_config_values(self):
+        """测试配置值正确性"""
+        self.assertEqual(SILERO_VAD_CONFIG["onnx_filename"], "silero_vad.onnx")
+        self.assertEqual(SILERO_VAD_CONFIG["jit_filename"], "silero_vad.jit")
+        self.assertIn("snakers4/silero-vad", SILERO_VAD_CONFIG["repo"])
+
+
+if __name__ == "__main__":
+    unittest.main()