import os import sys import re import time import subprocess import shutil import tempfile from pathlib import Path import traceback import inspect # --- コアライブラリとオプションTTSバックエンド --- # 各TTSエンジン固有の依存関係は必須にしない。 # 使えるバックエンドだけを安全に登録する。 import importlib import importlib.util from types import SimpleNamespace CORE_LIBS = ["pydub", "pyperclip", "chardet"] missing_core = [name for name in CORE_LIBS if importlib.util.find_spec(name) is None] if missing_core: print(f"Error: Missing core libraries: {', '.join(missing_core)}") print(" install: pip install pydub pyperclip chardet") print(" MP3などを扱う場合は FFmpeg もPATHに設定してください。") raise SystemExit(1) import pyperclip import chardet from pydub import AudioSegment def check_ffmpeg_available() -> bool: ffmpeg = shutil.which("ffmpeg") ffprobe = shutil.which("ffprobe") if ffmpeg is None: print("❌ ffmpeg が見つかりません。") return False if ffprobe is None: print("⚠️ ffprobe が見つかりません。読み込み時に問題が出る可能性があります。") return False print(f"✅ ffmpeg: {ffmpeg}") print(f"✅ ffprobe: {ffprobe}") return True ffmpeg_path = shutil.which("ffmpeg") if ffmpeg_path is None: print("tktts.py: ffmpeg が見つかりません") raise SystemExit(1) else: # print("tktts.py: ffmpeg が使えます:", ffmpeg_path) pass def _safe_import(module_name, dependency=None): """オプションのTTSバックエンドを安全に読み込む。 dependency が未導入の場合は、バックエンド側の ``input()`` や ``sys.exit()`` を実行させず、利用不可として登録する。 """ if dependency and importlib.util.find_spec(dependency) is None: print( f"Warning in tktts.py: {module_name} is disabled " f"because optional dependency [{dependency}] is not installed." ) return None try: return importlib.import_module(module_name) except SystemExit as e: print(f"Warning in tktts.py: {module_name} terminated during import: {e}") return None except Exception as e: print(f"\nWarning in tktts.py: Import error for {module_name}") print("------------------------------------------------------------------") print(f"Error message: {e}") traceback.print_exc() print("------------------------------------------------------------------") return None tktts_pyttsx3 = _safe_import("tktts_pyttsx3", "pyttsx3") tktts_winrt = _safe_import("tktts_winrt", "winsdk") tktts_voicevox = _safe_import("tktts_voicevox") tktts_aquestalkplayer = _safe_import("tktts_aquestalkplayer") tktts_openai = _safe_import("tktts_openai", "openai") tktts_elevenlabs = _safe_import("tktts_elevenlabs", "elevenlabs") default_pyttsx3_voice = "Zira" default_winrt_voice = "Ayumi" default_voicevox_voice = "四国めたん" default_aqt_preset = "れいむ" default_openai_voice = "alloy" # 旧コードとの互換性のため、綴り間違いの変数名も残す。 default_optnai_voice = default_openai_voice default_elevenlabs_voice = "Rachel" TTS_ENGINES = { "pyttsx3": { "engine": tktts_pyttsx3, "default_voice": default_pyttsx3_voice, "ext": "wav", "direct_playback": True, }, "winrt": { "engine": tktts_winrt, "default_voice": default_winrt_voice, "ext": "wav", "direct_playback": True, }, "onecore": { "engine": tktts_winrt, "default_voice": default_winrt_voice, "ext": "wav", "direct_playback": True, }, "voicevox": { "engine": tktts_voicevox, "default_voice": default_voicevox_voice, "ext": "wav", "direct_playback": False, }, "aquestalkplayer": { "engine": tktts_aquestalkplayer, "default_voice": default_aqt_preset, "ext": "wav", "direct_playback": False, }, "atp": { "engine": tktts_aquestalkplayer, "default_voice": default_aqt_preset, "ext": "wav", "direct_playback": False, }, "openai": { "engine": tktts_openai, "default_voice": default_openai_voice, "ext": "mp3", "direct_playback": False, }, "elevenlabs": { "engine": tktts_elevenlabs, "default_voice": default_elevenlabs_voice, "ext": "mp3", "direct_playback": False, }, "eleven": { "engine": tktts_elevenlabs, "default_voice": default_elevenlabs_voice, "ext": "mp3", "direct_playback": False, }, } def get_available_engines(available_only=True): """登録済みエンジン名を返す。GUIのプルダウン生成にも利用できる。""" if not available_only: return list(TTS_ENGINES.keys()) return [name for name, cfg in TTS_ENGINES.items() if cfg["engine"] is not None] def _supported_kwargs(func, kwargs): """関数が受け取れるキーワード引数だけを残す。""" try: signature = inspect.signature(func) except (TypeError, ValueError): return {k: v for k, v in kwargs.items() if v is not None} if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in signature.parameters.values()): return {k: v for k, v in kwargs.items() if v is not None} return { k: v for k, v in kwargs.items() if v is not None and k in signature.parameters } def _call_backend(func, **kwargs): return func(**_supported_kwargs(func, kwargs)) def apply_replacements(text, replacements): """置換辞書を使ってテキストを置換（大文字小文字無視）。""" for key, val in (replacements or {}).items(): text = re.sub(key, val, text, flags=re.IGNORECASE) return text def load_text(input_path, monologue = False, wait_for_clipboard = True): """入力元('clip'またはファイルパス)からテキストを取得し、(speaker, text)リストを返す""" print() print(f"load_text from {input_path}") if input_path.lower() == "clip" and wait_for_clipboard: print() input("読み上げるテキストをクリップボードにコピーしてください:\n") print("📥 クリップボードからテキストを取得中...") text = pyperclip.paste() text_lines = text.splitlines() else: file_name = input_path if not os.path.isfile(file_name): print(f"Error: ファイル [{file_name}] が見つかりません。") return None print(f"📖 ファイル [{file_name}] を読み込み中...") try: with open(file_name, "rb") as f: raw_data = f.read() result = chardet.detect(raw_data) encoding = result["encoding"] if encoding is None: encoding = 'utf-8' text = raw_data.decode(encoding, errors='ignore') print(f"　 検出されたエンコード: {encoding}") text_lines = text.splitlines() except Exception as e: print(f"❌ ファイル読み込み/エンコード判別エラー: {e}") return None # dialogueリストへの変換 (既存ロジックを流用) dialogue = [] for line in text_lines: line = line.strip() if not line: continue if line.startswith("#"): continue if monologue: dialogue.append((None, line.strip())) else: if "," in line: try: speaker, text = line.split(",", 1) dialogue.append((speaker.strip(), text.strip())) except ValueError: continue # 対話モードでカンマがない行は無視 # 修正前のコードではカンマがない行は無視されているため、ここでは continue continue return dialogue def create_temp_dir(temp_dir): if not os.path.exists(temp_dir): os.makedirs(temp_dir, exist_ok=True) print(f"一時ディレクトリを作成: {temp_dir}") return temp_dir def get_speaker_dict(tts_engine, dialogue, voices, default_voicevox_voice="四国めたん", default_pyttsx3_voice="Zira", default_winrt_voice="Ayumi", default_aqt_preset="れいむ", default_optnai_voice="alloy", default_elevenlabs_voice="Rachel"): tts_engine = str(tts_engine).lower() if tts_engine in TTS_ENGINES: default_voice = TTS_ENGINES[tts_engine]["default_voice"] else: print(f"\nError in tktts.get_speaker_dict(): Invalid tts engine [{tts_engine}]\n") return None print("\n話者リスト作成...") voices_specified = [v.strip() for v in str(voices or "").split(";") if v.strip()] if voices_specified: default_voice = voices_specified[0] speaker_names = [] target_voices = {} for speaker, _text in dialogue or []: if speaker is None or speaker == "" or speaker in speaker_names: continue speaker_names.append(speaker) index = len(speaker_names) - 1 target_voices[speaker] = ( voices_specified[index] if index < len(voices_specified) else default_voice ) target_voices[""] = default_voice target_voices[None] = default_voice for i, (speaker, voice) in enumerate(target_voices.items()): print(f" {i:02d}: {speaker} => {voice}") return target_voices def get_tts(tts_engine): tts_engine = str(tts_engine or "").lower() config = TTS_ENGINES.get(tts_engine) if config is None: print(f"\nError in tktts.get_tts(): Invalid tts engine [{tts_engine}]\n") return None tts_module = config["engine"] if tts_module is not None: return tts_module print(f"\nError in tktts.get_tts(): TTS engine [{tts_engine}] is unavailable.") print("対応バックエンドファイルと、そのオプション依存ライブラリを確認してください。") return None def get_available_voices_info(tts_engine, endpoint=None, api_key=None): tts_engine = str(tts_engine or "").lower() tts = get_tts(tts_engine) if tts is None or not hasattr(tts, "get_available_voices_info"): return None return _call_backend( tts.get_available_voices_info, endpoint=endpoint, api_key=api_key, ) def get_available_voices(tts_engine, endpoint=None, api_key=None): tts_engine = str(tts_engine or "").lower() tts = get_tts(tts_engine) if tts is None or not hasattr(tts, "get_available_voices"): return None return _call_backend( tts.get_available_voices, endpoint=endpoint, api_key=api_key, ) def list_available_voices(tts_engine, endpoint=None, api_key=None): tts_engine = str(tts_engine or "").lower() tts = get_tts(tts_engine) if tts is None: return False if not hasattr(tts, "list_available_voices"): print(f"Error: TTS module [{tts_engine}] has no list_available_voices function.") return False try: ret = _call_backend( tts.list_available_voices, endpoint=endpoint, api_key=api_key, ) except Exception as e: print(f"Error calling list_available_voices for {tts_engine}: {e}") traceback.print_exc() ret = False print("===============================") return ret def normalize_speaker(speaker, tts_engine=None): """話者ラベルを正規化する。 半角スペースでは分割しない。WinRTなどの音声表示名には ``Microsoft Ayumi ...`` のように空白が含まれるためである。 明示的なスタイル表記 ``話者（style）`` / ``話者(style)`` のみ除く。 """ if speaker is None: return None normalized = str(speaker).strip() for opening in ("（", "("): if opening in normalized: normalized = normalized.split(opening, 1)[0].rstrip() return normalized def parse_kv_string(kv_string, keys = [], allow_no_key_kv_string = False): """key=val;key=val 形式を dict に変換""" if not kv_string: return {} nkeys = len(keys) d = {} idx = 0 for item in kv_string.split(";"): if "=" in item: k, v = item.split("=", 1) d[k.strip()] = v.strip() elif allow_no_key_kv_string: if "*" in item or item.strip() == "": continue if idx < nkeys: d[keys[idx]] = item d[idx] = item idx += 1 for speaker in keys: if speaker in d.keys(): continue d[speaker] = speaker return d def _get_attr(args, name, default=None): return getattr(args, name, default) if args is not None else default def _effective_pyttsx3_rate(args): base_rate = float(_get_attr(args, "speak_rate", 150) or 150) multiplier = _get_attr(args, "fspeak_rate", None) if multiplier is not None: try: base_rate *= float(multiplier) except (TypeError, ValueError): pass return max(1, int(round(base_rate))) def _effective_winrt_rate(args): # GUIの fspeak_rate は相対倍率なので、WinRTにはそのまま渡せる。 relative_rate = _get_attr(args, "fspeak_rate", None) if relative_rate is not None: return relative_rate return _get_attr(args, "speak_rate", 150) def _elevenlabs_voice_settings(args): settings = _get_attr(args, "elevenlabs_voice_settings", None) if settings is None: settings = _get_attr(args, "voice_settings", None) if isinstance(settings, dict): return dict(settings) values = {} for key in ("stability", "similarity_boost", "style", "use_speaker_boost"): value = _get_attr(args, f"elevenlabs_{key}", None) if value is not None: values[key] = value return values or None def _infer_output_format(outfile, requested_format, default_format): if requested_format: return str(requested_format).lower().lstrip(".") suffix = Path(str(outfile)).suffix.lower().lstrip(".") if suffix: return "wav" if suffix == "wave" else suffix return default_format def speak_dialogue(args, dialogue, voice_map=None, speakers=None, replacements=None, default_voicevox_voice="四国めたん", default_pyttsx3_voice="Zira", default_winrt_voice="Ayumi", default_aqt_preset="れいむ", default_optnai_voice="alloy", default_elevenlabs_voice="Rachel", endpoint=None, api_key=None, output_format=None): """選択されたエンジンで音声生成・結合・保存または直接再生を行う。""" print("tktts.speak_dialogue(): Generate audio files:") tts_engine = str(_get_attr(args, "tts", "")).lower() tts = get_tts(tts_engine) if tts is None: return False config = TTS_ENGINES.get(tts_engine) if config is None: return False outfile = _get_attr(args, "outfile", "") or "" is_save_mode = bool(outfile) monologue = bool(_get_attr(args, "monologue", False)) speak_rate = _get_attr(args, "speak_rate", 150) tinterval = float(_get_attr(args, "tinterval", 0.5) or 0.0) voices = _get_attr(args, "voices", "") or "" speakers = speakers or {} replacements = replacements or {} print("\nspeak_dialogue:") print( f" tts_engine : {tts_engine} is monologue: {monologue} " f"speak rate: {speak_rate} tinterval: {tinterval}" ) print(f" voices : {voices}") print(f" outfile : {outfile}" if is_save_mode else " is_save_mode: False") if voice_map is None or isinstance(voice_map, str): # GUIなどから単一の音声名が文字列で渡された場合も、話者マップへ # 変換してバックエンドへ渡す。WinRTバックエンド側で音声名を # speakerとして正規化すると、空白を含む表示名が切れる実装との # 互換問題を避けられる。 selected_voices = voice_map if isinstance(voice_map, str) else voices target_voices = get_speaker_dict( tts_engine, dialogue, selected_voices, default_voicevox_voice=default_voicevox_voice, default_pyttsx3_voice=default_pyttsx3_voice, default_winrt_voice=default_winrt_voice, default_aqt_preset=default_aqt_preset, default_optnai_voice=default_optnai_voice, default_elevenlabs_voice=default_elevenlabs_voice, ) else: target_voices = voice_map if target_voices is None: return False if not is_save_mode and not config.get("direct_playback", False): print("❌ このエンジンではファイル保存が必要です。--outfile を指定してください。") return False temp_dir_name = _get_attr(args, "temp_dir", None) if not temp_dir_name: print("❌ 一時ファイル保存先が必要です。--temp_dir を指定してください。") return False temp_dir = create_temp_dir(temp_dir_name) call_kwargs = { "dialogue": dialogue or [], "replacements": replacements, "target_voices": target_voices, "speakers": speakers, "temp_dir": temp_dir, "outfile": outfile, "ext": config["ext"], "cfg": args, } if tts_engine == "pyttsx3": call_kwargs["speak_rate"] = _effective_pyttsx3_rate(args) elif tts_engine in ("winrt", "onecore"): call_kwargs["speak_rate"] = _effective_winrt_rate(args) elif tts_engine == "voicevox": call_kwargs["endpoint"] = endpoint or _get_attr(args, "endpoint", None) elif tts_engine in ("aquestalkplayer", "atp"): call_kwargs["aquestalk_path"] = _get_attr(args, "aquestalk_path", None) elif tts_engine == "openai": call_kwargs["instruction"] = _get_attr(args, "instruction", None) elif tts_engine in ("elevenlabs", "eleven"): call_kwargs.update({ "api_key": ( api_key or _get_attr(args, "elevenlabs_api_key", None) or _get_attr(args, "api_key", None) ), "model_id": ( _get_attr(args, "elevenlabs_model_id", None) or _get_attr(args, "model_id", None) ), "voice_settings": _elevenlabs_voice_settings(args), "output_format": _get_attr(args, "elevenlabs_output_format", None), "optimize_streaming_latency": _get_attr( args, "elevenlabs_optimize_streaming_latency", None ), "language_code": _get_attr(args, "elevenlabs_language_code", None), }) print(f"\n⚙️ {tts_engine.upper()}で音声ファイルを生成中...") try: result = _call_backend(tts.speak_dialogue, **call_kwargs) success, tmpfiles = result except Exception as e: print(f"❌ {tts_engine} の speak_dialogue 呼び出しエラー: {e}") traceback.print_exc() return False if not success: return False # pyttsx3 / WinRT の直接再生はバックエンド側で完了する。 if not is_save_mode: return True tmpfiles = list(tmpfiles or []) if not tmpfiles: if os.path.isfile(outfile) and os.path.getsize(outfile) > 0: return outfile print("❌ 結合対象の一時音声ファイルが生成されませんでした。") return False final_format = _infer_output_format(outfile, output_format, config["ext"]) output_parent = os.path.dirname(os.path.abspath(outfile)) if output_parent: os.makedirs(output_parent, exist_ok=True) print(" ファイルを結合中...") combined_audio = AudioSegment.silent(duration=0) try: for tmpfile in tmpfiles: segment = AudioSegment.from_file(tmpfile, format=config["ext"]) combined_audio += segment if tinterval > 0: print(f" insert {tinterval} sec interval") combined_audio += AudioSegment.silent(duration=int(tinterval * 1000)) combined_audio.export(outfile, format=final_format) print(f"✅ 出力音声を {outfile} に保存しました (形式: {final_format})。") except Exception as e: print(f"❌ 音声ファイルの結合または保存に失敗しました: {e}") traceback.print_exc() return False finally: time.sleep(0.1) print("🗑️ 一時ファイルを削除中...") for filename in tmpfiles: try: if os.path.isfile(filename): os.remove(filename) except OSError as e: print(f" [warn] 一時ファイルを削除できません: {filename}: {e}") try: if os.path.isdir(temp_dir) and not os.listdir(temp_dir): os.rmdir(temp_dir) except OSError: pass return outfile class tkTTS: def __init__(self, tts_name=None, config=None): self.tts_name = tts_name self.tts = None self.config = config if config is not None else SimpleNamespace() self.endpoint = getattr(self.config, "endpoint", None) self.aquestalk_path = getattr(self.config, "aquestalk_path", None) self.api_key = ( getattr(self.config, "elevenlabs_api_key", None) or getattr(self.config, "api_key", None) ) if tts_name is not None: self.set_engine(tts_name) def set_engine(self, tts_name): self.tts_name = str(tts_name).lower() self.tts = get_tts(self.tts_name) def set_endpoint(self, endpoint): self.endpoint = endpoint setattr(self.config, "endpoint", endpoint) def set_aquestalk_path(self, aquestalk_path): self.aquestalk_path = aquestalk_path setattr(self.config, "aquestalk_path", aquestalk_path) def set_api_key(self, api_key): self.api_key = api_key setattr(self.config, "elevenlabs_api_key", api_key) def get_tts_name(self, tts_name=None): if tts_name is None: return self.tts_name return tts_name def get_default_voice(self, tts_name = None): tts_def = TTS_ENGINES.get(self.get_tts_name(tts_name), None) if tts_def:return tts_def["default_voice"] return None def normalize_speaker(self, speaker, tts_name = None): if tts_name is None: tts_name = self.tts_name return normalize_speaker(speaker, tts_name) def get_available_voices_info(self, tts_name=None, endpoint=None, api_key=None): if tts_name is None: tts_name = self.tts_name if endpoint is None: endpoint = self.endpoint if api_key is None: api_key = self.api_key return get_available_voices_info(tts_name, endpoint=endpoint, api_key=api_key) def get_available_voices(self, tts_name=None, endpoint=None, api_key=None): if tts_name is None: tts_name = self.tts_name if endpoint is None: endpoint = self.endpoint if api_key is None: api_key = self.api_key return get_available_voices(tts_name, endpoint=endpoint, api_key=api_key) def list_available_voices(self, tts_name=None, endpoint=None, api_key=None): if tts_name is None: tts_name = self.tts_name if endpoint is None: endpoint = self.endpoint if api_key is None: api_key = self.api_key return list_available_voices(tts_name, endpoint=endpoint, api_key=api_key) def show_voice_map(self, infile, voices, VOICE_MAPS, is_monologue, tts_name = None, endpoint = None): print() print(f"[{infile}]を解析します:") dialogue = self.load_text(infile, is_monologue, wait_for_clipboard = False) if not dialogue: print("エラー: 有効なテキストデータが取得できませんでした。") if not is_monologue: print(" 対話形式でない場合は --monologue=1 オプションをつけてください。") return False speakers_in_file = self.get_speakers_from_dialogue(dialogue) print(f" Speakers in [{infile}]") for idx, sp in enumerate(speakers_in_file): print(f" {idx:02d}: {sp}") current_voice_map = self.update_voice_map(voice_map = VOICE_MAPS, voices = voices, speakers = speakers_in_file) print() print(f"Voice map:") print(f" {'Speaker':<20} => Voice") for key, val in current_voice_map.items(): if type(key) is str: print(f" {key:<20} => {val}") for key, val in current_voice_map.items(): if type(key) is not str and type(key) is not int: if key is None: key = 'None' print(f" {key:<20} => {val}") for key, val in current_voice_map.items(): if type(key) is int: print(f" {key:<20} => {val}") print() print(f"[{infile}] から検出された話者:") for s in sorted(speakers_in_file): if s is None or s == "": voice = current_voice_map.get(s, None) if voice is None: voice = current_voice_map.get(0, None) print(f" (独話): {voice}") else: ns = self.normalize_speaker(s) voice = current_voice_map.get(ns, '未設定') print(f" (speaker) {s}: {ns}: (voice) {voice}") def load_text(self, input_path, monologue = False, wait_for_clipboard = True): return load_text(input_path, monologue = monologue, wait_for_clipboard = wait_for_clipboard) def get_speakers_from_dialogue(self, dialogue): return list(set(s for s, _ in dialogue)) def parse_kv_string(self, kv_string, keys = [], allow_no_key_kv_string = False): """key=val;key=val 形式を dict に変換""" print() print("parse_kv_string:") print(" kv_string:", kv_string) print(" keys:", keys) if not kv_string: return {} nkeys = len(keys) speakers_kv = [] d = {} idx = 0 # kv_stringのspeaker for item in kv_string.split(";"): if "=" in item: k, voice = item.split("=", 1) speaker = self.normalize_speaker(k.strip()) d[speaker] = voice if voice not in d.keys(): d[voice] = voice if speaker not in speakers_kv: speakers_kv.append(speaker) # print("478 add:", speaker) elif allow_no_key_kv_string: # = が無い場合は整数idxの辞書をつくる if "*" in item or item.strip() == "": continue # ユーザ指定辞書keysで与えられたspeaker（読み上げファイルのspeaker）をkv_stringにマップ if idx < nkeys: voice = item # speaker = self.normalize_speaker(voice) # voice = self.normalize_speaker(keys[idx]) speaker = self.normalize_speaker(keys[idx]) d[speaker] = voice if voice not in d.keys(): d[voice] = voice if speaker not in speakers_kv: speakers_kv.append(speaker) d[idx] = item idx += 1 # ユーザ指定辞書keysで与えられたspeaker（読み上げファイルのspeaker）をkv_stringにマップ for idx, speaker_keys in enumerate(keys): speaker_keys = self.normalize_speaker(speaker_keys) if speaker_keys in d.keys(): continue d[speaker_keys] = speaker_keys return d def update_voice_map(self, voice_map = {}, voices = "", speakers = {}): current_voice_map = voice_map.get(self.tts_name, {}).copy() print(f" Voice maps for [{self.tts_name}]];", current_voice_map) voices = voices.strip() if voices == "": print(f"voices is blank. Use given speakers") for idx, sp in enumerate(speakers): sp = self.normalize_speaker(sp) current_voice_map[sp] = sp current_voice_map[idx] = sp else: print(f"Read voice maps from voices:", voices) # 読み上げファイルから抽出したspeakersは使わない # voices_override = self.parse_kv_string(voices, {}, allow_no_key_kv_string = True) voices_override = self.parse_kv_string(voices, speakers, allow_no_key_kv_string = True) current_voice_map.update(voices_override) current_voice_map[None] = current_voice_map.get(0, None) current_voice_map[""] = current_voice_map.get(0, None) return current_voice_map def speak_dialogue(self, dialogue=None, voice_map=None, speakers=None, replacements=None, endpoint=None, api_key=None, output_format=None, config=None): if endpoint is None: endpoint = self.endpoint if api_key is None: api_key = self.api_key if config is None: config = self.config return speak_dialogue( config, dialogue, voice_map, speakers=speakers, replacements=replacements, default_voicevox_voice=self.get_default_voice("voicevox"), default_pyttsx3_voice=self.get_default_voice("pyttsx3"), default_winrt_voice=self.get_default_voice("winrt"), default_aqt_preset=self.get_default_voice("aqt"), default_optnai_voice=self.get_default_voice("openai"), default_elevenlabs_voice=self.get_default_voice("elevenlabs"), endpoint=endpoint, api_key=api_key, output_format=output_format, )