import os import sys # 在 windowed exe 中 sys.stdout/stderr 為 None， # 關閉 HuggingFace Hub 的內建 tqdm 進度條以防止 AttributeError。 # 下載進度改由自訂 GUI 進度條顯示。 os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "1" # 若 stdout/stderr 為 None（windowed exe），替換為 /dev/null， # 防止其他套件試圖寫入時崩潰。 if sys.stdout is None: import io sys.stdout = io.StringIO() if sys.stderr is None: import io sys.stderr = io.StringIO() def _fix_nagisa_for_frozen_exe(): """ 在 frozen exe 模式下，nagisa 使用的 DyNet C++ 函式庫（_dynet.pyd）的 fopen() 無法開啟含 CJK 字元或特殊 Unicode 字元的路徑。 解法：將 nagisa/data/ 複製到純 ASCII 的 %TEMP% 路徑， 並用 sys.meta_path hook 讓 nagisa 的 __file__ 指向乾淨路徑。 若 %TEMP% 也含 Unicode，則改用 stub 模組（停用 SRT 時間戳）。 """ if not getattr(sys, 'frozen', False): return # 開發模式不需要 import shutil, types, importlib.abc, importlib.util, tempfile nagisa_src = os.path.join(sys._MEIPASS, 'nagisa') if not os.path.isdir(nagisa_src): return # nagisa 未打包進來 # ── 複製 nagisa 整包到 temp 目錄 ── tmp_root = os.path.join(tempfile.gettempdir(), 'qwenasr_pkgs') nagisa_dst = os.path.join(tmp_root, 'nagisa') model_dst = os.path.join(nagisa_dst, 'data', 'nagisa_v001.model') try: _ascii_ok = tmp_root.encode('ascii') # 確認 temp 路徑是純 ASCII except UnicodeEncodeError: _ascii_ok = None if _ascii_ok: if not os.path.exists(model_dst): if os.path.exists(nagisa_dst): shutil.rmtree(nagisa_dst, ignore_errors=True) try: shutil.copytree(nagisa_src, nagisa_dst) except Exception: _ascii_ok = None # 複製失敗，退到 stub if not _ascii_ok: # ── 退到 stub 模式：停用 nagisa 避免崩潰 ── stub = types.ModuleType('nagisa') class _Stub: words = []; postags = [] class _StubTagger: def wakati(self, t): return t.split() def tagging(self, t): return _Stub() stub.tagger = _StubTagger() for _n in ['nagisa', 'nagisa.tagger', 'nagisa.model', 'nagisa.hp', 'nagisa.prepro', 'prepro', 'nagisa.utils', 'nagisa.train']: sys.modules[_n] = stub return # ── ASCII temp 路徑可用：用 meta_path hook 讓 nagisa 從 temp 載入 ── # 把 __file__ 指向 tmp_root 下的 nagisa，DyNet 就能讀到模型 class _NagisaLoader(importlib.abc.MetaPathFinder, importlib.abc.Loader): def find_spec(self, name, path=None, target=None): if name in ('nagisa',) + tuple( f'nagisa.{s}' for s in ('tagger','model','hp','prepro','utils','train')): # 這個模組已在 sys.modules（stub / real）就跳過 if name in sys.modules: return None fpath = os.path.join(nagisa_dst, name.replace('nagisa.', '').replace('nagisa', '__init__') .replace('__init__', '__init__') + '.py') if not os.path.exists(fpath): fpath = os.path.join(nagisa_dst, '__init__.py') \ if name == 'nagisa' else None if fpath and os.path.exists(fpath): return importlib.util.spec_from_file_location( name, fpath, submodule_search_locations=[nagisa_dst] if name == 'nagisa' else None) return None def create_module(self, spec): return None def exec_module(self, module): pass sys.meta_path.insert(0, _NagisaLoader()) # 同時確保 prepro 可被找到（nagisa 套件內以頂層 import 使用） if tmp_root not in sys.path: sys.path.insert(0, tmp_root) sys.path.insert(0, nagisa_dst) _fix_nagisa_for_frozen_exe() # 取得此腳本所在目錄，用於絕對路徑計算 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # 取得系統下載資料夾路徑（輸出檔案預設存放位置） def _get_downloads_dir() -> str: try: import winreg key = winreg.OpenKey( winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders", ) path, _ = winreg.QueryValueEx(key, "{374DE290-123F-4565-9164-39C4925E467B}") winreg.CloseKey(key) if path and os.path.isdir(path): return path except Exception: pass # fallback：~/Downloads fallback = os.path.join(os.path.expanduser("~"), "Downloads") os.makedirs(fallback, exist_ok=True) return fallback DOWNLOADS_DIR = _get_downloads_dir() import wave import threading import queue import time import datetime import traceback import numpy as np import customtkinter as ctk from tkinter import filedialog, messagebox try: import sounddevice as sd import soundfile as sf HAS_SOUNDDEVICE = True except ImportError: HAS_SOUNDDEVICE = False try: from huggingface_hub import snapshot_download HAS_HF_HUB = True except ImportError: HAS_HF_HUB = False class QwenASRApp(ctk.CTk): # 語系顯示名稱 → API 用名稱 對照表 LANG_DISPLAY = { "自動偵測": None, "繁體中文": "Chinese", "簡體中文": "Chinese", # Qwen3-ASR 的 Chinese 包含繁簡，就由輸出文字判斷 "英語": "English", "日語": "Japanese", "韓語": "Korean", "粤語": "Cantonese", "法語": "French", "德語": "German", "西班牙語": "Spanish", "葡萄牙語": "Portuguese", "日語": "Japanese", "阿拉伯語": "Arabic", "印尼語": "Indonesian", "泰語": "Thai", "越南語": "Vietnamese", "土耳其語": "Turkish", "印地語": "Hindi", "馬來語": "Malay", "俄語": "Russian", "義大利語": "Italian", } def __init__(self): super().__init__() self.title("Qwen3-ASR 本地語音辨識工具") self.geometry("900x720") ctk.set_appearance_mode("dark") ctk.set_default_color_theme("blue") self.is_recording = False self.is_streaming = False # 即時串流辨識狀態 self.audio_queue = queue.Queue() self.recorded_frames = [] self.sample_rate = 16000 # Qwen3-ASR 建議取樣率 self.model = None self.model_loaded = False self.active_model_path = None self.selected_file = None self._opencc_warned = False # 避免重複提示安裝 opencc self._last_output_path = None # 最後一次輸出的檔案完整路徑 # 模型清單（名稱 → HuggingFace Model ID） self.available_models = { "Qwen3-ASR-0.6B（速度快，約 1.2GB）": "Qwen/Qwen3-ASR-0.6B", "Qwen3-ASR-1.7B（準確高，約 3.5GB）": "Qwen/Qwen3-ASR-1.7B", } self.setup_ui() self.check_local_models() # ───────────────────────────────────────────────── # UI 建立 # ───────────────────────────────────────────────── def setup_ui(self): # 標題列（含使用說明按鈕） title_row = ctk.CTkFrame(self, fg_color="transparent") title_row.pack(fill="x", padx=20, pady=(16, 4)) ctk.CTkLabel( title_row, text="Qwen3-ASR 本地語音辨識（100% 本機運行）", font=ctk.CTkFont(size=22, weight="bold"), ).pack(side="left", expand=True) ctk.CTkButton( title_row, text="❓ 使用說明", width=100, command=self.show_help, ).pack(side="right", padx=4) # 主框架 self.main_frame = ctk.CTkFrame(self) self.main_frame.pack(fill="both", expand=True, padx=20, pady=8) # ── 模型管理區塊 ── mf = ctk.CTkFrame(self.main_frame) mf.pack(fill="x", padx=10, pady=8) ctk.CTkLabel(mf, text="選擇模型:", font=ctk.CTkFont(size=14)).grid( row=0, column=0, padx=10, pady=8, sticky="w" ) self.model_combobox = ctk.CTkComboBox( mf, values=list(self.available_models.keys()), width=270, command=self.on_model_change, ) self.model_combobox.set(list(self.available_models.keys())[0]) self.model_combobox.grid(row=0, column=1, padx=10, pady=8) self.btn_load_model = ctk.CTkButton( mf, text="載入 / 下載模型", command=self.download_or_load_model ) self.btn_load_model.grid(row=1, column=0, columnspan=2, padx=10, pady=8) # ForcedAligner 勾選框（啟用 SRT 時間戳需要） self.use_aligner_var = ctk.BooleanVar(value=False) self.aligner_checkbox = ctk.CTkCheckBox( mf, text="啟用時間戳（額外下載 ForcedAligner-0.6B，約 1.2GB）", variable=self.use_aligner_var, command=self.on_model_change, ) self.aligner_checkbox.grid(row=1, column=2, columnspan=2, padx=10, pady=8, sticky="w") self.model_status_label = ctk.CTkLabel( mf, text="等待檢查...", text_color="orange" ) self.model_status_label.grid( row=2, column=0, columnspan=4, padx=10, pady=4, sticky="w" ) # ── 設定區塊 ── sf_frame = ctk.CTkFrame(self.main_frame) sf_frame.pack(fill="x", padx=10, pady=8) ctk.CTkLabel(sf_frame, text="來源語系:", font=ctk.CTkFont(size=14)).grid( row=0, column=0, padx=10, pady=8, sticky="w" ) self.lang_combobox = ctk.CTkComboBox( sf_frame, values=list(self.LANG_DISPLAY.keys()), width=180, ) self.lang_combobox.set("自動偵測") self.lang_combobox.grid(row=0, column=1, padx=10, pady=8) ctk.CTkLabel(sf_frame, text="輸出格式:", font=ctk.CTkFont(size=14)).grid( row=0, column=2, padx=10, pady=8, sticky="w" ) self.format_combobox = ctk.CTkComboBox( sf_frame, values=["純文字逐字稿（含標點）", "SRT 字幕檔（需勾選時間戳）"], width=230, ) self.format_combobox.set("純文字逐字稿（含標點）") self.format_combobox.grid(row=0, column=3, padx=10, pady=8) ctk.CTkLabel(sf_frame, text="最大Token數:", font=ctk.CTkFont(size=14)).grid( row=0, column=4, padx=10, pady=8, sticky="w" ) self.max_tokens_entry = ctk.CTkEntry(sf_frame, width=80) self.max_tokens_entry.insert(0, "1024") self.max_tokens_entry.grid(row=0, column=5, padx=10, pady=8) # ── 操作分頁 ── self.tabview = ctk.CTkTabview(self.main_frame, width=800, height=160) self.tabview.pack(padx=10, pady=8, fill="x") self.tab_file = self.tabview.add("📁 上傳音檔") self.tab_mic = self.tabview.add("🎙️ 麥克風錄音") # 上傳音檔頁 self.file_label = ctk.CTkLabel(self.tab_file, text="尚未選擇檔案") self.file_label.pack(pady=8) btn_row = ctk.CTkFrame(self.tab_file, fg_color="transparent") btn_row.pack() ctk.CTkButton(btn_row, text="選擇本地音檔", command=self.select_file).pack( side="left", padx=10, pady=8 ) ctk.CTkButton( btn_row, text="▶ 開始辨識", command=self.transcribe_file, fg_color="green" ).pack(side="left", padx=10, pady=8) # 麥克風頁 mic_top = ctk.CTkFrame(self.tab_mic, fg_color="transparent") mic_top.pack(fill="x", padx=4, pady=(6, 0)) # ── 左半：整段錄音 ── mic_left = ctk.CTkFrame(mic_top, fg_color="transparent") mic_left.pack(side="left", fill="both", expand=True, padx=(0, 6)) ctk.CTkLabel(mic_left, text="💾 整段錄音後轉文", font=ctk.CTkFont(size=12, weight="bold")).pack(anchor="w") self.mic_status = ctk.CTkLabel(mic_left, text="準備就緒", font=ctk.CTkFont(size=12), text_color="gray") self.mic_status.pack(anchor="w", pady=(2, 4)) self.btn_record_mic = ctk.CTkButton( mic_left, text="● 開始麥克風錄音", command=lambda: self.toggle_recording("mic"), width=180, ) self.btn_record_mic.pack(anchor="w") # ── 右半：即時串流 ── mic_right = ctk.CTkFrame(mic_top, fg_color="transparent") mic_right.pack(side="left", fill="both", expand=True) ctk.CTkLabel(mic_right, text="⚡ 即時串流轉文（每區分段推論）", font=ctk.CTkFont(size=12, weight="bold")).pack(anchor="w") self.stream_status = ctk.CTkLabel(mic_right, text="準備就緒", font=ctk.CTkFont(size=12), text_color="gray") self.stream_status.pack(anchor="w", pady=(2, 4)) self.btn_stream = ctk.CTkButton( mic_right, text="● 開始即時串流轉文", command=self.toggle_stream, fg_color="#8B4513", hover_color="#A0522D", width=180, ) self.btn_stream.pack(anchor="w") # 即時串流結果顯示區 ctk.CTkLabel(self.tab_mic, text="即時轉文結果：", font=ctk.CTkFont(size=12)).pack(anchor="w", padx=8, pady=(6, 0)) self.stream_textbox = ctk.CTkTextbox( self.tab_mic, height=70, font=ctk.CTkFont(size=13) ) self.stream_textbox.pack(fill="x", padx=8, pady=(2, 6)) self.stream_textbox.insert("1.0", "（即時辨識結果將顯示於此）") self.stream_textbox.configure(state="disabled") if not HAS_SOUNDDEVICE: ctk.CTkLabel( self.tab_mic, text="⚠ 未安裝 sounddevice，請執行: pip install sounddevice soundfile", text_color="orange", ).pack() # ── 進度條區塊 ── progress_frame = ctk.CTkFrame(self.main_frame, fg_color="transparent") progress_frame.pack(fill="x", padx=10, pady=(4, 0)) self.progress_bar = ctk.CTkProgressBar( progress_frame, mode="indeterminate", height=14, corner_radius=6 ) self.progress_bar.pack(fill="x", side="left", expand=True, padx=(0, 8)) self.progress_bar.set(0) # 初始設為 0（靜止） self.progress_label = ctk.CTkLabel( progress_frame, text="閒置", font=ctk.CTkFont(size=12), text_color="gray", width=120, anchor="w", ) self.progress_label.pack(side="left") self._progress_active = False self._progress_elapsed = 0 self._progress_timer_id = None # ── 輸出快捷按鈕列（辨識完成後才顯示） ── self.output_btn_frame = ctk.CTkFrame(self.main_frame, fg_color="transparent") # 初始不 pack，辨識完成後才顯示 self.btn_open_file = ctk.CTkButton( self.output_btn_frame, text="📄 開啟輸出檔案", command=self._open_output_file, fg_color="#3E6B8F", hover_color="#4C82AE", width=160, ) self.btn_open_file.pack(side="left", padx=(0, 8)) self.btn_open_folder = ctk.CTkButton( self.output_btn_frame, text="📂 開啟輸出目錄", command=self._open_output_folder, fg_color="#5A5A2E", hover_color="#6F6F38", width=160, ) self.btn_open_folder.pack(side="left") # 日誌區 self.log_textbox = ctk.CTkTextbox( self.main_frame, height=200, font=ctk.CTkFont(size=13) ) self.log_textbox.pack(fill="both", expand=True, padx=10, pady=8) self.log("系統初始化完成，等待模型載入...") # ── 頁腳（作者 + CC 授權） ── footer = ctk.CTkFrame(self, fg_color="transparent") footer.pack(fill="x", padx=20, pady=(0, 8)) footer_text = ctk.CTkLabel( footer, text="Made by ", font=ctk.CTkFont(size=11), text_color="gray", ) footer_text.pack(side="left") link = ctk.CTkLabel( footer, text="阿剛老師", font=ctk.CTkFont(size=11, underline=True), text_color="#61AFEF", cursor="hand2", ) link.pack(side="left") link.bind("", lambda e: self._open_url("https://kentxchang.blogspot.tw")) ctk.CTkLabel( footer, text="   本程式以 CC BY-NC-SA 4.0 授權公開，可自由使用與修改，但請註明作者，且不得商業利用。", font=ctk.CTkFont(size=11), text_color="gray", ).pack(side="left") # ───────────────────────────────────────────────── # 工具方法 # ───────────────────────────────────────────────── def log(self, message): """將訊息寫入日誌文字框""" ts = datetime.datetime.now().strftime("%H:%M:%S") self.log_textbox.insert("end", f"[{ts}] {message}\n") self.log_textbox.see("end") @staticmethod def _open_url(url: str): """用系統預設瀏覽器開啟連結""" import webbrowser webbrowser.open(url) def _get_lang_param(self) -> str | None: """將語系選單的中文顯示名轉換為 API 所需的英文名稱""" display = self.lang_combobox.get() return self.LANG_DISPLAY.get(display, None) def show_help(self): """顯示完整使用說明視窗""" win = ctk.CTkToplevel(self) win.title("Qwen3-ASR 使用說明") win.geometry("720x600") win.grab_set() # 設為模態視窗 ctk.CTkLabel( win, text="Qwen3-ASR 本地語音辨識 — 使用說明", font=ctk.CTkFont(size=18, weight="bold"), ).pack(pady=(16, 8)) text = ctk.CTkTextbox(win, font=ctk.CTkFont(size=13), wrap="word") text.pack(fill="both", expand=True, padx=16, pady=(0, 8)) help_content = """\ 【一、軟體簡介】 本工具以阿里巴巴開源的 Qwen3-ASR 語音辨識模型為基礎，所有運算均在本機進行，不上傳任何音訊。 支援辨識語言：繁體中文、簡體中文、英語、日語、韓語、粵語及多種語言（共 20 種）。 ────────────────────────────── 【二、初次設定（模型下載）】 1. 選擇「AI 模型」： • Qwen3-ASR-0.6B（速度快，約 1.2GB）：適合快速試用或低規硬體。 • Qwen3-ASR-1.7B（準確度高，約 3.5GB）：適合要求較高的辨識任務。 2. 若需要 SRT 字幕時間戳，請勾選「啟用時間戳」，系統會額外下載 ForcedAligner-0.6B 模型（約 1.2GB）。 3. 按「載入/下載模型」，首次使用需連線下載，之後可完全離線。 注意：本版本為純 CPU 版，所有運算均在 CPU 上執行，無需 GPU。 ────────────────────────────── 【三、最大 Token 數 的說明】 「最大 Token 數」控制模型一次最多可以生成的文字量。 預設值：1024（約可涵蓋 3～5 分鐘的語音）。 • 音訊較短（< 3 分鐘）：使用預設 1024 即可。 • 音訊較長（5～15 分鐘）：建議設為 2048 或 4096。 • 非常長的音訊（> 30 分鐘）：建議設為 8192（需較多記憶體）。 • 若辨識結果被截斷（文字突然中斷），請增大此數值後重新載入模型。 注意：增大 Token 數會增加推論時間與記憶體用量。 ────────────────────────────── 【四、音檔辨識】 1. 切換至「📁 上傳音檔」分頁。 2. 點「選擇本地音檔」，支援格式：wav / mp3 / flac / m4a / ogg / mp4 / mkv / aac。 3. 選擇「來源語系」與「輸出格式」。 4. 點「▶ 開始辨識」，進度條會顯示辨識中及耗時。 5. 辨識完成後，結果儲存至系統「下載」資料夾，副檔名為 .txt 或 .srt。 6. 完成後可使用「📄 開啟輸出檔案」或「📂 開啟輸出目錄」按鈕直接存取結果。 ────────────────────────────── 【五、輸出格式說明】 • 純文字逐字稿（含標點）：直接輸出連續文字，無時間資訊。 • SRT 字幕檔：輸出帶時間戳的字幕，每句約 10～25 字。 ⚠ 需要在「載入模型前」先勾選「啟用時間戳（ForcedAligner）」才能使用。 ────────────────────────────── 【六、麥克風錄音（整段模式）】 1. 切換至「🎙️ 麥克風錄音」分頁（需安裝 sounddevice）。 2. 點「● 開始麥克風錄音」，對麥克風說話。 3. 說完後點「⏹ 停止錄音」，系統將自動辨識整段音訊並儲存結果。 安裝指令：pip install sounddevice soundfile ────────────────────────────── 【七、麥克風錄音（即時串流模式）】 即時串流模式會在錄音的同時，每累積數秒就立即辨識一段，不必等待錄音完成。 1. 切換至「🎙️ 麥克風錄音」分頁。 2. 點「● 開始即時串流轉文」。 3. 辨識結果會即時顯示在下方文字框。 4. 點「⏹ 停止串流」後，系統會詢問是否儲存完整結果。 注意：即時串流模式準確度略低於整段模式，適合需要即時看到結果的場合。 ────────────────────────────── 【八、常見問題】 Q：辨識結果為空或只有幾個字？ A：請加大「最大 Token 數」後重新載入模型再試。 Q：SRT 格式無法使用？ A：請確認載入模型前已勾選「啟用時間戳（ForcedAligner）」，並重新點擊「載入模型」。 Q：輸出檔案存在哪裡？ A：音檔辨識結果存至系統「下載」資料夾；麥克風錄音結果存至程式目錄。 辨識完成後可直接點擊「📄 開啟輸出檔案」按鈕快速存取。 Q：想要離線使用？ A：首次下載完成後即可完全離線，模型存放於程式目錄的 models 資料夾中。 """ text.insert("end", help_content) text.configure(state="disabled") ctk.CTkButton(win, text="關閉", command=win.destroy).pack(pady=8) def _start_progress(self, label="辨識中"): """啟動不確定式進度條動畫""" self._progress_active = True self._progress_elapsed = 0 self._transcription_start = time.time() # 記錄開始時間 self.progress_bar.configure(mode="indeterminate") self.progress_bar.start() self.progress_label.configure(text=f"{label}… 0s", text_color="#61AFEF") self._tick_progress(label) def _tick_progress(self, label): """每秒更新計時器""" if not self._progress_active: return self._progress_elapsed += 1 self.progress_label.configure(text=f"{label}… {self._progress_elapsed}s") self._progress_timer_id = self.after(1000, lambda: self._tick_progress(label)) def _stop_progress(self, success=True): """停止進度條動畫，並顯示總耗時""" self._progress_active = False if self._progress_timer_id: self.after_cancel(self._progress_timer_id) self._progress_timer_id = None self.progress_bar.stop() self.progress_bar.configure(mode="determinate") self.progress_bar.set(1.0 if success else 0.0) color = "#98C379" if success else "#E06C75" # 計算總耗時 elapsed = time.time() - getattr(self, "_transcription_start", time.time()) m, s = divmod(int(elapsed), 60) elapsed_str = f"{m}m{s:02d}s" if m else f"{s}s" msg = f"✅ 完成 （{elapsed_str}）" if success else f"❌ 失敗 （{elapsed_str}）" self.progress_label.configure(text=msg, text_color=color) if success: self.log(f"⏱ 總辨識耗時：{elapsed_str}") # 5 秒後恢復閒置狀態 self.after(5000, self._reset_progress) def _reset_progress(self): """恢復進度條為閒置狀態""" self.progress_bar.set(0) self.progress_label.configure(text="閒置", text_color="gray") def _show_output_buttons(self, output_path: str): """辨識完成後顯示輸出快捷按鈕，並記錄路徑""" self._last_output_path = output_path # 先 forget 再 pack，確保在日誌區上方正確插入 self.output_btn_frame.pack_forget() self.output_btn_frame.pack( in_=self.main_frame, fill="x", padx=10, pady=(2, 4), before=self.log_textbox, ) def _open_output_file(self): """用系統預設程式開啟最後一次輸出的檔案""" if self._last_output_path and os.path.exists(self._last_output_path): import subprocess subprocess.Popen(["explorer", self._last_output_path]) else: messagebox.showwarning("找不到檔案", "輸出檔案不存在，請先執行辨識。") def _open_output_folder(self): """用檔案總管開啟輸出目錄並選取該檔案""" if self._last_output_path and os.path.exists(self._last_output_path): import subprocess subprocess.Popen(["explorer", "/select,", self._last_output_path]) else: # 找不到檔案時，至少開啟下載資料夾 import subprocess subprocess.Popen(["explorer", DOWNLOADS_DIR]) def _get_model_dir(self): """取得目前選取模型的本地資料夾路徑（絕對路徑）""" selected_name = self.model_combobox.get() model_id = self.available_models[selected_name] model_folder = model_id.split("/")[-1] return os.path.join(SCRIPT_DIR, "models", model_folder), model_id def _get_max_tokens(self): try: return int(self.max_tokens_entry.get()) except ValueError: return 1024 # ───────────────────────────────────────────────── # 模型管理 # ───────────────────────────────────────────────── def check_local_models(self): """檢查目前選擇的模型是否已存在本地""" model_dir, _ = self._get_model_dir() if os.path.exists(os.path.join(model_dir, "config.json")): self.model_status_label.configure( text="✅ 已下載（可離線使用）", text_color="green" ) self.btn_load_model.configure(text="載入此模型") else: self.model_status_label.configure( text="❌ 尚未下載（需網路）", text_color="red" ) self.btn_load_model.configure(text="下載並載入模型") def on_model_change(self, choice=None): self.check_local_models() # 切換選項後需重新載入 if self.model_loaded: self.model_loaded = False self.model_status_label.configure( text="⚠ 模型設定已變更，請重新載入", text_color="orange" ) def _download_repo_with_progress(self, repo_id: str, local_dir: str, label_prefix: str): """ 逐檔下載 HuggingFace repo，並即時更新 GUI 進度條。 改用 list_repo_files + hf_hub_download 逐檔處理， 不依賴 tqdm_class（在 exe 環境下 tqdm_class 不可靠）。 """ from huggingface_hub import list_repo_files, hf_hub_download # ── 取得檔案清單 ── self.log(f"　正在取得 {repo_id} 的檔案清單...") all_files = [f for f in list_repo_files(repo_id) if not f.startswith(".gitattributes")] total = len(all_files) self.log(f"　共 {total} 個檔案") for idx, filename in enumerate(all_files, start=1): dest = os.path.join(local_dir, *filename.split("/")) # 已存在則跳過（支援斷點續傳概念） if os.path.exists(dest): self._update_file_progress(idx, total, filename, label_prefix) continue short = filename.split("/")[-1] self.log(f"　[{idx}/{total}] ⬇ {short}") hf_hub_download( repo_id=repo_id, filename=filename, local_dir=local_dir, ) self._update_file_progress(idx, total, filename, label_prefix) def _update_file_progress(self, current: int, total: int, filename: str, label_prefix: str): """在主執行緒更新進度條（依檔案數計算百分比）""" pct = current / total if total > 0 else 0 short = filename.split("/")[-1] text = f"{label_prefix}… {current}/{total} 個檔案 ({pct*100:.0f}%)" def _gui_update(pv=pct, st=text, fn=short): try: self.progress_bar.configure(mode="determinate") self.progress_bar.set(pv) self.progress_label.configure(text=st, text_color="#61AFEF") self.model_status_label.configure( text=f"⬇ {fn}", text_color="orange" ) except Exception: pass self.after(0, _gui_update) def _start_download_progress(self, label: str): """初始化下載進度條（determinate 模式）""" self.progress_bar.configure(mode="determinate") self.progress_bar.set(0) self.progress_label.configure(text=f"{label}… 0%", text_color="#61AFEF") def _finish_download_progress(self): """下載完成後還原進度條""" self.progress_bar.set(1.0) self.progress_label.configure(text="下載完成 ✅", text_color="#98C379") self.after(2000, lambda: ( self.progress_bar.set(0), self.progress_label.configure(text="閒置", text_color="gray"), )) def download_or_load_model(self): self.btn_load_model.configure(state="disabled") model_dir, model_id = self._get_model_dir() threading.Thread( target=self._model_worker, args=(model_id, model_dir), daemon=True, ).start() def _model_worker(self, model_id, model_dir): """背景執行緒：下載或載入模型（CPU 模式）""" try: import torch # ── Step 1：下載主模型（必要時）── if not os.path.exists(os.path.join(model_dir, "config.json")): if not HAS_HF_HUB: self.log("❌ 找不到 huggingface_hub 庫，請執行: pip install huggingface_hub") return model_short = model_id.split("/")[-1] self.log(f"開始從 HuggingFace 下載 {model_id}...") self.log("　進度條會依每個檔案更新，請耐心等候（首次下載後可離線使用）...") self.after(0, lambda ms=model_short: self._start_download_progress(f"⬇ {ms}")) os.makedirs(model_dir, exist_ok=True) self._download_repo_with_progress( repo_id=model_id, local_dir=model_dir, label_prefix=f"⬇ {model_short}", ) self.log(f"✅ {model_id} 下載完成！") self.after(0, self._finish_download_progress) # ── Step 1b：若勾選時間戳，下載 ForcedAligner（必要時）── use_aligner = self.use_aligner_var.get() aligner_id = "Qwen/Qwen3-ForcedAligner-0.6B" aligner_dir = os.path.join(SCRIPT_DIR, "models", "Qwen3-ForcedAligner-0.6B") if use_aligner and not os.path.exists(os.path.join(aligner_dir, "config.json")): if not HAS_HF_HUB: self.log("❌ 找不到 huggingface_hub 庫，無法下載 ForcedAligner") use_aligner = False else: self.log(f"開始下載 ForcedAligner 模型（{aligner_id}）...") self.after(0, lambda: self._start_download_progress("⬇ ForcedAligner")) os.makedirs(aligner_dir, exist_ok=True) self._download_repo_with_progress( repo_id=aligner_id, local_dir=aligner_dir, label_prefix="⬇ ForcedAligner", ) self.log("✅ ForcedAligner 下載完成！") self.after(0, self._finish_download_progress) # ── Step 2：載入模型（純 CPU）── from qwen_asr import Qwen3ASRModel max_tokens = self._get_max_tokens() loaded = False # 準備 ForcedAligner 參數（若啟用） aligner_kwargs = None if use_aligner and os.path.exists(os.path.join(aligner_dir, "config.json")): aligner_kwargs = dict( dtype=torch.float32, device_map="cpu", ) self.log("✅ ForcedAligner 就緒，SRT 時間戳功能已啟用。") elif use_aligner: self.log("⚠ ForcedAligner 目錄不存在，時間戳功能將停用。") use_aligner = False try: self.log("嘗試以 CPU (float32) 載入模型...") self.model_status_label.configure( text="載入中…（CPU float32）", text_color="orange" ) common_kwargs = dict( dtype=torch.float32, max_inference_batch_size=1, max_new_tokens=max_tokens, ) if use_aligner and aligner_kwargs: common_kwargs["forced_aligner"] = aligner_dir common_kwargs["forced_aligner_kwargs"] = aligner_kwargs self.model = Qwen3ASRModel.from_pretrained( model_dir, device_map="cpu", **common_kwargs, ) self.model_loaded = True self.aligner_enabled = use_aligner self.active_model_path = model_dir ts_note = "含時間戳" if use_aligner else "不含時間戳" self.log(f"✅ 模型載入成功！（CPU float32，{ts_note}）") self.model_status_label.configure( text=f"✅ 就緒（CPU，{ts_note}）", text_color="green" ) loaded = True except Exception as e: self.log(f"❌ 模型載入失敗：{e}") loaded = False if not loaded: self.log("❌ 模型載入失敗，請檢查錯誤訊息。") self.model_status_label.configure(text="❌ 載入失敗", text_color="red") except Exception as e: self.log(f"❌ 模型處理發生錯誤: {e}") self.log(traceback.format_exc()) self.model_status_label.configure(text="❌ 處理失敗", text_color="red") finally: self.btn_load_model.configure(state="normal") self.check_local_models() # ───────────────────────────────────────────────── # 音檔辨識 # ───────────────────────────────────────────────── def select_file(self): path = filedialog.askopenfilename( title="選擇音訊或影片檔案", filetypes=[ ("音訊/影片", "*.wav *.mp3 *.flac *.m4a *.ogg *.mp4 *.mkv *.aac"), ("全部檔案", "*.*"), ], ) if path: self.selected_file = path self.file_label.configure(text=f"已選擇：{os.path.basename(path)}") self.log(f"已選擇音檔：{path}") def transcribe_file(self): if not self.selected_file: messagebox.showwarning("警告", "請先選擇音檔！") return if not self.model_loaded: messagebox.showwarning("警告", "模型尚未載入完成，請先載入模型！") return self.log(f"開始處理：{self.selected_file}") threading.Thread( target=self._process_file_worker, args=(self.selected_file,), daemon=True, ).start() def _process_file_worker(self, file_path): out_format = self.format_combobox.get() # 只有在「模型載入時勾選了時間戳（含 ForcedAligner）」才能要求 SRT want_srt = "SRT" in out_format aligner_ready = getattr(self, "aligner_enabled", False) return_ts = want_srt and aligner_ready if want_srt and not aligner_ready: self.log("⚠ SRT 需要 ForcedAligner，但載入模型時未勾選「啟用時間戳」。") self.log(" 請重新勾選並點擊「載入模型」後再試。改為輸出純文字。") lang_param = self._get_lang_param() # 中文顯示名轉 API 名 lang_display = self.lang_combobox.get() self.log(f"語系：{lang_display} ｜ 輸出格式：{'SRT' if return_ts else '純文字'}") try: self.log("模型推論中，請稍候...") self.after(0, lambda: self._start_progress("辨識中")) results = self.model.transcribe( file_path, language=lang_param, return_time_stamps=return_ts, ) if not results: self.after(0, lambda: self._stop_progress(success=False)) self.log("❌ 辨識結果為空。") return res = results[0] # 輸出檔案存至下載資料夾，保留原始檔名 base_name = os.path.splitext(os.path.basename(file_path))[0] ext = ".srt" if return_ts else ".txt" save_path = os.path.join(DOWNLOADS_DIR, base_name + ext) self._save_result(save_path, res, "SRT" if return_ts else "txt", lang_display=lang_display) self.after(0, lambda: self._stop_progress(success=True)) # 辨識成功後顯示快捷按鈕 self.after(0, lambda p=save_path: self._show_output_buttons(p)) except Exception as e: self.after(0, lambda: self._stop_progress(success=False)) self.log(f"❌ 辨識發生錯誤: {e}") self.log(traceback.format_exc()) def _save_result(self, path, res, out_format, lang_display: str = ""): """儲存辨識結果為 SRT 或 TXT， 依 lang_display 自動做繁簡轉換： 選「繁體中文」或「自動偵測」且偵測為中文 → 轉繁體 選「簡體中文」 → 保留簡體輸出 """ # 判斷是否需要轉繁體 detected = getattr(res, "language", "") or "" need_traditional = ( lang_display == "繁體中文" or (lang_display == "自動偵測" and "chinese" in detected.lower()) ) def cvt(text: str) -> str: return self._convert_chinese(text, to_traditional=need_traditional) try: with open(path, "w", encoding="utf-8") as f: if "SRT" in out_format: ts_data = getattr(res, "time_stamps", None) if ts_data: sentences = self._merge_char_timestamps(ts_data) for i, (start_s, end_s, text) in enumerate(sentences, start=1): f.write( f"{i}\n" f"{self._sec_to_srt(start_s)} --> {self._sec_to_srt(end_s)}\n" f"{cvt(text)}\n\n" ) else: self.log("⚠ 無法取得時間戳，改為輸出純文字。") f.write(cvt(res.text)) else: f.write(cvt(res.text)) final_text = cvt(res.text) self.log(f"✅ 辨識完成！已儲存至：\n {path}") preview = final_text[:80] + ("..." if len(final_text) > 80 else "") self.log(f"📃 預覽：{preview}") except Exception as e: self.log(f"❌ 儲存檔案時發生錯誤: {e}") self.log(traceback.format_exc()) def _convert_chinese(self, text: str, to_traditional: bool = True) -> str: """繁簡轉換輔助方法，使用 OpenCC。套件未安裝時日誌提示。""" if not to_traditional or not text: return text try: import opencc cc = opencc.OpenCC("s2twp") # 簡體 → 繁體（台灣用詞） return cc.convert(text) except ImportError: if not self._opencc_warned: self._opencc_warned = True self.log("⚠️ 繁簡轉換需要 opencc，請先安裝：") self.log(" pip install opencc-python-reimplemented") self.log(" 安裝後重新開啟程式即可自動轉換繁體中文。") return text except Exception as e: self.log(f"⚠️ OpenCC 轉換失敗：{e}") return text @staticmethod def _merge_char_timestamps( ts_data, max_chars: int = 25, gap_threshold: float = 0.8, ) -> list: """ 將字元級 time_stamps 合併為句子級列表。 回傳格式： List[(start_sec, end_sec, text)] 拆句脚虚： 1. 字元為句尾標點 (。！？…!?) 2. 相鄰字元間隔 > gap_threshold 秒 3. 累積字元數 > max_chars """ SENTENCE_END = set("。！？…!?\u2026") sentences = [] buf_text = "" buf_start = None buf_end = None prev_end = None for seg in ts_data: ch = getattr(seg, "text", "").strip() if not ch: continue t_start = getattr(seg, "start_time", 0.0) or 0.0 t_end = getattr(seg, "end_time", t_start) or t_start # 判斷是否要斷句 force_break = False if prev_end is not None and (t_start - prev_end) > gap_threshold: force_break = True # 時間門戳大 if len(buf_text) >= max_chars: force_break = True # 超過最大字數 if force_break and buf_text: sentences.append((buf_start, buf_end, buf_text)) buf_text = "" buf_start = None if buf_start is None: buf_start = t_start buf_text += ch buf_end = t_end prev_end = t_end # 句尾標點斷句 if ch in SENTENCE_END and buf_text: sentences.append((buf_start, buf_end, buf_text)) buf_text = "" buf_start = None # 收尾殘餘 if buf_text: sentences.append((buf_start, buf_end, buf_text)) return sentences @staticmethod def _sec_to_srt(seconds: float) -> str: """將秒數轉換為 SRT 時間格式 HH:MM:SS,mmm""" if seconds is None: seconds = 0.0 ms = int(round(seconds * 1000)) h = ms // 3_600_000 ms %= 3_600_000 m = ms // 60_000 ms %= 60_000 s = ms // 1000 ms %= 1000 return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" # ───────────────────────────────────────────────── # 即時串流辨識 # ───────────────────────────────────────────────── def toggle_stream(self): """切換即時串流辨識的開始 / 停止""" if not self.model_loaded: messagebox.showwarning("警告", "模型尚未載入完成，請先載入模型！") return if self.is_recording: messagebox.showwarning("警告", "整段錄音進行中，請先停止再使用串流辨識。") return if self.is_streaming: # 停止串流 self.is_streaming = False self.btn_stream.configure( text="● 開始即時串流轉文", fg_color="#8B4513", hover_color="#A0522D", ) self.stream_status.configure(text="已停止，處理最後片段...", text_color="gray") else: # 開始串流 self.is_streaming = True # 清空 queue 和結果框 while not self.audio_queue.empty(): self.audio_queue.get_nowait() self.stream_textbox.configure(state="normal") self.stream_textbox.delete("1.0", "end") self.stream_textbox.configure(state="disabled") self.btn_stream.configure( text="⏹ 停止串流辨識", fg_color="red", hover_color="#CC0000" ) self.stream_status.configure(text="● 串流中… 說話即時轉文", text_color="#FF8C00") self.log("⚡ 即時串流辨識啟動（每 4 秒推論一段）...") threading.Thread(target=self._stream_worker, daemon=True).start() def _stream_worker(self): """ 即時串流辨識背景執行緒： - 每累積 CHUNK_SECS 秒音訊就推論一次 - 結果即時追加到 stream_textbox - 停止後推論剩餘音訊，並詢問是否儲存 """ CHUNK_SECS = 4 # 每段推論長度（秒） MIN_SECS = 0.5 # 少於此秒數的尾段忽略不推論 lang_param = self._get_lang_param() lang_display = self.lang_combobox.get() chunk_buf = [] # 當前累積音訊片段 all_text = [] # 全部推論文字（用於最終存檔） seg_n = 0 # 段落計數 def _append(text: str): """在主執行緒安全地更新 stream_textbox""" def _do(): self.stream_textbox.configure(state="normal") if self.stream_textbox.get("1.0", "end").strip(): self.stream_textbox.insert("end", text) else: self.stream_textbox.insert("1.0", text) self.stream_textbox.see("end") self.stream_textbox.configure(state="disabled") self.after(0, _do) def _infer_chunk(audio_np: np.ndarray) -> str: """送出一段音訊推論，回傳文字""" try: res = self.model.transcribe( (audio_np, self.sample_rate), language=lang_param, return_time_stamps=False, ) if res: txt = getattr(res[0], "text", "") or "" if lang_display == "繁體中文": txt = self._convert_chinese(txt, to_traditional=True) return txt.strip() except Exception as e: self.log(f"⚠ 串流推論錯誤: {e}") return "" # ── 錄音 + 分段推論迴圈 ── if not HAS_SOUNDDEVICE: self.log("❌ 請先安裝 sounddevice：pip install sounddevice") self.is_streaming = False return with sd.InputStream( samplerate=self.sample_rate, channels=1, dtype="float32", callback=self._audio_callback, ): while self.is_streaming: time.sleep(0.05) # 把 queue 中收到的片段收進 chunk_buf while not self.audio_queue.empty(): try: chunk_buf.append(self.audio_queue.get_nowait()) except queue.Empty: break # 累積夠了就推論 total_samples = sum(len(c) for c in chunk_buf) if total_samples >= CHUNK_SECS * self.sample_rate: audio_np = np.concatenate(chunk_buf) chunk_buf = [] seg_n += 1 self.after(0, lambda n=seg_n: self.stream_status.configure( text=f"● 串流中… 已推論第 {n} 段", text_color="#FF8C00")) text = _infer_chunk(audio_np) if text: all_text.append(text) _append(text + " ") # ── 停止後處理剩餘音訊 ── while not self.audio_queue.empty(): try: chunk_buf.append(self.audio_queue.get_nowait()) except queue.Empty: break total_remain = sum(len(c) for c in chunk_buf) if total_remain >= MIN_SECS * self.sample_rate: audio_np = np.concatenate(chunk_buf) text = _infer_chunk(audio_np) if text: all_text.append(text) _append(text) # 更新狀態 self.after(0, lambda: self.stream_status.configure( text=f"✅ 完成（共 {seg_n} 段）", text_color="green")) self.log(f"⚡ 串流辨識完成，共 {seg_n} 段。") # 詢問是否儲存 if all_text: full_text = " ".join(all_text) def _ask_save(): if messagebox.askyesno("儲存結果", "串流辨識完成！\n是否將結果儲存為 TXT 至下載資料夾？"): ts_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") save_path = os.path.join(DOWNLOADS_DIR, f"stream_{ts_str}.txt") try: with open(save_path, "w", encoding="utf-8") as f: f.write(full_text) self.log(f"✅ 已儲存：{save_path}") except Exception as e: self.log(f"❌ 儲存失敗：{e}") self.after(0, _ask_save) # ───────────────────────────────────────────────── # 錄音功能 # ───────────────────────────────────────────────── def toggle_recording(self, mode): if not self.model_loaded: messagebox.showwarning("警告", "模型尚未載入完成，請先載入模型！") return if self.is_recording: # 停止錄音 self.is_recording = False self.btn_record_mic.configure( text="開始麥克風錄音", fg_color=["#3B8ED0", "#1F6AA5"] ) self.mic_status.configure(text="已停止，正在處理...") self.log("停止錄音，等待推論...") else: # 開始錄音 self.is_recording = True self.recorded_frames = [] # 清空 queue while not self.audio_queue.empty(): self.audio_queue.get_nowait() self.btn_record_mic.configure(text="⏹ 停止錄音", fg_color="red") self.mic_status.configure(text="● 錄音中... 點擊按鈕停止") self.log("開始麥克風錄音...") threading.Thread( target=self._record_and_transcribe_worker, args=(mode,), daemon=True, ).start() def _record_and_transcribe_worker(self, mode): try: if mode == "mic": if not HAS_SOUNDDEVICE: self.log("❌ 請先安裝 sounddevice：pip install sounddevice soundfile") self.is_recording = False return # 使用 sounddevice InputStream；音訊透過 callback 放入 queue with sd.InputStream( samplerate=self.sample_rate, channels=1, dtype="float32", callback=self._audio_callback, ): while self.is_recording: time.sleep(0.05) except Exception as e: self.log(f"❌ 錄音發生錯誤: {e}") self.log(traceback.format_exc()) self.is_recording = False # ── 錄音結束：收集所有已錄音訊並推論 ── while not self.audio_queue.empty(): try: self.recorded_frames.append(self.audio_queue.get_nowait()) except queue.Empty: break if len(self.recorded_frames) == 0: self.log("❌ 未收集到任何音訊，請確認裝置正常。") return audio_data = np.concatenate(self.recorded_frames) duration = len(audio_data) / self.sample_rate self.log(f"錄音完成！共 {duration:.1f} 秒，正在辨識...") try: lang_param = self._get_lang_param() # 中文顯示名轉 API 名 lang_display = self.lang_combobox.get() out_format = self.format_combobox.get() # 與音檔辨識相同邀輯：只有 aligner_enabled 時才能要求時間戳 want_srt = "SRT" in out_format aligner_ready = getattr(self, "aligner_enabled", False) return_ts = want_srt and aligner_ready if want_srt and not aligner_ready: self.log("⚠ SRT 需要 ForcedAligner，但載入模型時未勾選「啟用時間戳」。改為輸出純文字。") self.after(0, lambda: self._start_progress("辨識錄音中")) # API 接受 (numpy_array, sample_rate) 元組 results = self.model.transcribe( (audio_data, self.sample_rate), language=lang_param, return_time_stamps=return_ts, ) if results: res = results[0] ts_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") ext = ".srt" if return_ts else ".txt" save_path = os.path.join(DOWNLOADS_DIR, f"record_{ts_str}{ext}") self._save_result(save_path, res, "SRT" if return_ts else "txt", lang_display=lang_display) self.after(0, lambda: self._stop_progress(success=True)) else: self.log("❌ 辨識結果為空。") self.after(0, lambda: self._stop_progress(success=False)) except Exception as e: self.after(0, lambda: self._stop_progress(success=False)) self.log(f"❌ 辨識發生錯誤: {e}") self.log(traceback.format_exc()) def _audio_callback(self, indata, frames, time_info, status): """sounddevice 麥克風音訊片段回呼""" if status: self.log(f"⚠ 音訊裝置狀態: {status}") if self.is_recording or self.is_streaming: self.audio_queue.put(indata[:, 0].copy()) if __name__ == "__main__": app = QwenASRApp() app.mainloop()