"""tktts.py: 複数のTTS（Text-to-Speech）エンジンを統合し、テキストからの音声合成機能を提供します。 さまざまなTTSバックエンド（pyttsx3, WinRT, VOICEVOXなど）を動的にロードし、 テキストを音声に変換して再生またはファイルに保存する機能を提供します。 話者ごとの音声設定、一時ファイルの管理、音声の結合なども行います。 :doc:`tktts_usage` """ 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がインストールされているかを確認します。 FFmpegとFFprobeの両方の実行可能ファイルがシステムのPATHで利用可能かをチェックし、 その結果をコンソールに出力します。 :returns: bool - FFmpegとFFprobeの両方が利用可能であればTrue、そうでなければFalse。 """ 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: str, dependency: str | None = None): """オプションのTTSバックエンドモジュールを安全にインポートします。 依存ライブラリがインストールされていない場合や、インポート中にエラーが発生した場合は、 そのモジュールをスキップし、Noneを返します。 これにより、一部のTTSバックエンドが利用できなくても、モジュール全体がクラッシュすることなく 動作を継続できます。 :param module_name: str: インポートするモジュールの名前。 :param dependency: str or None: ``module_name`` が依存する外部ライブラリの名前。 この依存ライブラリが未導入の場合はインポートを試みません。 :returns: module or None - 正常にインポートされたモジュール、またはインポートに失敗した場合はNone。 """ 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_qwen3 = _safe_import("tktts_qwen3", "qwen_tts") tktts_irodori = _safe_import("tktts_irodori", "irodori_tts") tktts_aquestalkplayer = _safe_import("tktts_aquestalkplayer") tktts_openai = _safe_import("tktts_openai", "openai") tktts_elevenlabs = _safe_import("tktts_elevenlabs", "elevenlabs") tktts_gemini = _safe_import("tktts_gemini", "google.genai") default_pyttsx3_voice = "Zira" default_winrt_voice = "Ayumi" default_voicevox_voice = "四国めたん" default_qwen3_voice = "Ono_Anna" default_irodori_voice = "default" default_aqt_preset = "れいむ" default_openai_voice = "alloy" # 旧コードとの互換性のため、綴り間違いの変数名も残す。 default_optnai_voice = default_openai_voice default_elevenlabs_voice = "Rachel" default_gemini_voice = "Kore" 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, }, "qwen3": { "engine": tktts_qwen3, "default_voice": default_qwen3_voice, "ext": "wav", "direct_playback": False, }, "qwen": { "engine": tktts_qwen3, "default_voice": default_qwen3_voice, "ext": "wav", "direct_playback": False, }, "irodori": { "engine": tktts_irodori, "default_voice": default_irodori_voice, "ext": "wav", "direct_playback": False, }, "irodori-tts": { "engine": tktts_irodori, "default_voice": default_irodori_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, }, "gemini": { "engine": tktts_gemini, "default_voice": default_gemini_voice, "ext": "wav", "direct_playback": False, }, } def get_available_engines(available_only: bool = True) -> list[str]: """利用可能なTTSエンジンまたは登録されている全てのTTSエンジン名をリストとして返します。 GUIのプルダウンメニュー生成などにも利用できます。 :param available_only: bool: Trueの場合、実際にロードされて利用可能なエンジンのみを返します。 Falseの場合、設定ファイルに登録されている全てのエンジン名を返します。 :returns: list[str] - TTSエンジンの名前のリスト。 """ 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: dict) -> dict: """関数が受け取れるキーワード引数だけを残します。 関数のシグネチャを検査し、関数が定義している引数、または ``**kwargs`` を受け入れる場合にのみ、 辞書内の引数を保持します。これにより、バックエンド関数が予期しない引数を受け取ってエラーになるのを防ぎます。 :param func: Callable: キーワード引数を適用する対象の関数。 :param kwargs: dict: フィルタリング対象のキーワード引数の辞書。 :returns: dict - 関数が受け取れるキーワード引数のみを含む辞書。 """ try: signature = inspect.signature(func) except (TypeError, ValueError): # シグネチャが取得できない場合は、Noneでない全ての引数を返す return {k: v for k, v in kwargs.items() if v is not None} # 関数が可変キーワード引数 (**kwargs) を受け入れる場合は、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): """内部的にバックエンド関数の呼び出しをラップし、サポートされている引数のみを渡します。 :param func: Callable: 呼び出すバックエンド関数。 :param kwargs: Any: 関数に渡すキーワード引数。 :returns: Any - バックエンド関数の実行結果。 """ return func(**_supported_kwargs(func, kwargs)) def apply_replacements(text: str, replacements: dict[str, str] | None) -> str: """置換辞書を使ってテキストを置換（大文字小文字無視）。 :param text: str: 置換処理を行う元のテキスト。 :param replacements: dict[str, str] or None: キーを検索パターン、値を置換後の文字列とする辞書。 Noneの場合、置換は行われません。 :returns: str - 置換処理が適用されたテキスト。 """ for key, val in (replacements or {}).items(): text = re.sub(key, val, text, flags=re.IGNORECASE) return text def load_text(input_path: str, monologue: bool = False, wait_for_clipboard: bool = True) -> list[tuple[str | None, str]] | None: """入力元('clip'またはファイルパス)からテキストを取得し、(speaker, text)リストを返します。 入力パスが"clip"の場合、クリップボードからテキストを取得します。 ファイルパスの場合、ファイルを読み込み、chardetでエンコーディングを検出します。 テキストは行ごとに解析され、カンマで区切られた話者とテキストのペア、 またはモノローグ形式で返されます。 :param input_path: str: テキストの入力元。"clip"またはファイルパス。 :param monologue: bool: Trueの場合、全てを単一の話者のテキストとして扱います。 Falseの場合、行を「話者,テキスト」形式で解析します。 :param wait_for_clipboard: bool: ``input_path`` が "clip" の場合に、 ユーザーがクリップボードにコピーするのを待つか。 :returns: list[tuple[str or None, str]] or None - (話者名, テキスト)のタプルのリスト、 またはエラー発生時はNone。 """ 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: str) -> str: """指定されたパスに一時ディレクトリを作成します。 ディレクトリが存在しない場合は新しく作成します。 :param temp_dir: str: 作成する一時ディレクトリのパス。 :returns: str - 作成された一時ディレクトリのパス。 """ 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: str, dialogue: list[tuple[str | None, str]], voices: str | None, default_voicevox_voice: str = "四国めたん", default_pyttsx3_voice: str = "Zira", default_winrt_voice: str = "Ayumi", default_aqt_preset: str = "れいむ", default_optnai_voice: str = "alloy", default_elevenlabs_voice: str = "Rachel", default_gemini_voice: str = "Kore") -> dict[str | None, str] | None: """対話データと指定された音声設定に基づいて、話者と対応する音声名をマッピングする辞書を生成します。 `voices` 引数で指定された優先順位、またはデフォルトの音声設定に基づいて話者と音声の マッピングを決定します。 :param tts_engine: str: 使用するTTSエンジンの名前。 :param dialogue: list[tuple[str or None, str]]: (話者名, テキスト)のタプルのリスト。 :param voices: str or None: セミコロンで区切られた音声名の文字列。 :param default_voicevox_voice: str: VOICEVOXのデフォルト音声名。 :param default_pyttsx3_voice: str: pyttsx3のデフォルト音声名。 :param default_winrt_voice: str: WinRTのデフォルト音声名。 :param default_aqt_preset: str: AquesTalkPlayerのデフォルトプリセット名。 :param default_optnai_voice: str: OpenAIのデフォルト音声名。 :param default_elevenlabs_voice: str: ElevenLabsのデフォルト音声名。 :param default_gemini_voice: str: Geminiのデフォルト音声名。 :returns: dict[str or None, str] or None - 話者名をキー、音声名を値とする辞書、 または無効なエンジン名の場合はNone。 """ 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: str | None): """指定されたTTSエンジンのモジュールを返します。 エンジン名が ``TTS_ENGINES`` に登録されていない場合や、エンジンのモジュールが ロードされていない場合はNoneを返します。 :param tts_engine: str or None: 取得するTTSエンジンの名前。 :returns: module or None - TTSエンジンのモジュール、または利用できない場合はNone。 """ 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: str | None, endpoint: str | None = None, api_key: str | None = None): """指定されたTTSエンジンで利用可能な音声の詳細情報を取得します。 バックエンドモジュールに ``get_available_voices_info`` 関数が存在する場合にそれを呼び出します。 :param tts_engine: str or None: 情報を取得するTTSエンジンの名前。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 :returns: Any or None - 利用可能な音声の詳細情報、または機能が存在しない場合は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: str | None, endpoint: str | None = None, api_key: str | None = None): """指定されたTTSエンジンで利用可能な音声のリストを取得します。 バックエンドモジュールに ``get_available_voices`` 関数が存在する場合にそれを呼び出します。 :param tts_engine: str or None: リストを取得するTTSエンジンの名前。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 :returns: list[str] or None - 利用可能な音声のリスト、または機能が存在しない場合は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: str | None, endpoint: str | None = None, api_key: str | None = None) -> bool: """指定されたTTSエンジンで利用可能な音声をコンソールに出力します。 バックエンドモジュールに ``list_available_voices`` 関数が存在する場合にそれを呼び出します。 :param tts_engine: str or None: リストを出力するTTSエンジンの名前。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 :returns: bool - 処理が成功した場合はTrue、失敗した場合はFalse。 """ 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: str | None, tts_engine: str | None = None) -> str | None: """話者ラベルを正規化します。 半角スペースでは分割しない。WinRTなどの音声表示名には ``Microsoft Ayumi ...`` のように空白が含まれるためである。 明示的なスタイル表記 ``話者（style）`` / ``話者(style)`` のみ除く。 :param speaker: str or None: 正規化する話者名。 :param tts_engine: str or None: 現在は使用されていないが将来的な拡張のために残されているTTSエンジン名。 :returns: str or None - 正規化された話者名、または入力がNoneの場合はNone。 """ 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: str | None, keys: list[str] = [], allow_no_key_kv_string: bool = False) -> dict[str | int, str]: """key=val;key=val 形式を dict に変換します。 `keys` 引数で指定されたキーのリストが与えられた場合、キーが指定されていない値に対しては、 そのリストのインデックスに基づいてキーを割り当てます。 :param kv_string: str or None: "key=val;key=val" 形式の文字列。 :param keys: list[str]: キーがない値に割り当てるためのキーのリスト。 :param allow_no_key_kv_string: bool: キーが指定されていないアイテム（例: "value1;value2"）を許可するかどうか。 :returns: dict[str or int, str] - 変換された辞書。キーは文字列またはインデックス。 """ 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: object | None, name: str, default: Any | None = None) -> Any | None: """オブジェクトから属性を安全に取得します。 `args` がNoneの場合や、指定された属性が存在しない場合はデフォルト値を返します。 :param args: object or None: 属性を取得する対象のオブジェクト（通常は ``argparse.Namespace``）。 :param name: str: 取得する属性の名前。 :param default: Any or None: 属性が存在しない場合に返すデフォルト値。 :returns: Any or None - 取得した属性の値、またはデフォルト値。 """ return getattr(args, name, default) if args is not None else default def _effective_pyttsx3_rate(args: object) -> int: """pyttsx3の音声速度を計算します。 ``--speak_rate`` と ``--fspeak_rate`` オプションを考慮して 最終的な音声速度（WPM: Words Per Minute）を決定します。 :param args: object: ``speak_rate`` と ``fspeak_rate`` 属性を持つオブジェクト。 :returns: int - pyttsx3に設定する有効な音声速度。 """ 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: object) -> float | int: """WinRTの音声速度を計算します。 ``--fspeak_rate`` または ``--speak_rate`` オプションから音声速度を決定します。 GUIの `fspeak_rate` は相対倍率なので、WinRTにはそのまま渡すことができます。 :param args: object: ``fspeak_rate`` と ``speak_rate`` 属性を持つオブジェクト。 :returns: float or int - WinRTに設定する有効な音声速度。 """ # 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: object) -> dict | None: """ElevenLabsの音声設定 (``VoiceSettings``) を ``args`` オブジェクトから構築します。 ``elevenlabs_voice_settings`` または ``voice_settings`` 属性から設定辞書を取得するか、 個別の ``elevenlabs_stability``、``elevenlabs_similarity_boost`` などの属性から設定を組み立てます。 :param args: object: ElevenLabs関連の設定属性を持つオブジェクト。 :returns: dict or None - ElevenLabsの音声設定辞書、または設定がない場合はNone。 """ 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: str | Path, requested_format: str | None, default_format: str) -> str: """出力ファイル名、要求された形式、デフォルト形式から最終的な出力形式を推測します。 ``requested_format`` があればそれを優先し、そうでなければ ``outfile`` の拡張子から推測します。 どちらもなければ ``default_format`` を使用します。 :param outfile: str or Path: 出力ファイルのパス。 :param requested_format: str or None: 要求された出力形式（例: "mp3", "wav"）。 :param default_format: str: デフォルトの出力形式。 :returns: str - 推測された出力形式（拡張子のみ、例: "mp3"）。 """ 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: object, dialogue: list[tuple[str | None, str]], voice_map: dict[str | None, str] | str | None = None, speakers: dict | None = None, replacements: dict[str, str] | None = None, default_voicevox_voice: str = "四国めたん", default_pyttsx3_voice: str = "Zira", default_winrt_voice: str = "Ayumi", default_aqt_preset: str = "れいむ", default_optnai_voice: str = "alloy", default_elevenlabs_voice: str = "Rachel", default_gemini_voice: str = "Kore", endpoint: str | None = None, api_key: str | None = None, output_format: str | None = None) -> str | bool: """選択されたエンジンで音声生成・結合・保存または直接再生を行います。 TTSエンジンを選択し、設定を読み込みます。 話者ごとの音声マッピングを決定し、一時ディレクトリを作成して各セリフの音声ファイルを生成します。 生成された音声ファイルを結合し、指定された出力形式で保存します。 pyttsx3やWinRTのように直接再生に対応している場合は、保存せずに再生を試みます。 :param args: object: コマンドライン引数などの設定オブジェクト。 :param dialogue: list[tuple[str or None, str]]: (話者名, テキスト)のタプルのリスト。 :param voice_map: dict[str or None, str] or str or None: 話者と音声名をマッピングする辞書、 または単一の音声名文字列。 Noneの場合、引数 ``voices`` から生成されます。 :param speakers: dict or None: 以前に解析された話者情報。 :param replacements: dict[str, str] or None: テキスト置換ルール。 :param default_voicevox_voice: str: VOICEVOXのデフォルト音声名。 :param default_pyttsx3_voice: str: pyttsx3のデフォルト音声名。 :param default_winrt_voice: str: WinRTのデフォルト音声名。 :param default_aqt_preset: str: AquesTalkPlayerのデフォルトプリセット名。 :param default_optnai_voice: str: OpenAIのデフォルト音声名。 :param default_elevenlabs_voice: str: ElevenLabsのデフォルト音声名。 :param default_gemini_voice: str: Geminiのデフォルト音声名。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 :param output_format: str or None: 出力音声ファイルの形式（例: "mp3", "wav"）。 :returns: str or bool - 保存されたファイルのパス（str）、 または成功の場合はTrue (直接再生時)、失敗の場合はFalse。 """ 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, default_gemini_voice=default_gemini_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 ("qwen3", "qwen"): call_kwargs.update({ "language": _get_attr(args, "qwen3_language", None), "model_id": _get_attr(args, "qwen3_model_id", None), "device": _get_attr(args, "qwen3_device", None), "dtype": _get_attr(args, "qwen3_dtype", None), }) elif tts_engine in ("irodori", "irodori-tts"): call_kwargs.update({ "caption": _get_attr(args, "irodori_caption", None), "ref_wav": _get_attr(args, "irodori_ref_wav", None), "ref_wavs": _get_attr(args, "irodori_ref_wavs", None), "model_id": _get_attr(args, "irodori_model_id", None), "device": _get_attr(args, "irodori_device", None), "precision": _get_attr(args, "irodori_precision", 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 == "gemini": 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: """複数のTTS（Text-to-Speech）エンジンを統一的に扱うためのラッパークラスです。 選択されたTTSエンジンの設定を管理し、音声合成、利用可能な音声の取得、 話者マッピングなどの機能を提供します。 """ def __init__(self, tts_name: str | None = None, config: SimpleNamespace | None = None): """``tkTTS`` クラスのコンストラクタです。 :param tts_name: str or None: 初期化時に設定するTTSエンジンの名前。 :param config: SimpleNamespace or None: 設定情報を含むオブジェクト。 (例: ``argparse.Namespace`` のインスタンス)。 """ 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: str): """使用するTTSエンジンを設定します。 :param tts_name: str: 設定するTTSエンジンの名前。 """ self.tts_name = str(tts_name).lower() self.tts = get_tts(self.tts_name) def set_endpoint(self, endpoint: str): """APIベースのTTSエンジンのエンドポイントURLを設定します。 :param endpoint: str: 設定するエンドポイントURL。 """ self.endpoint = endpoint setattr(self.config, "endpoint", endpoint) def set_aquestalk_path(self, aquestalk_path: str): """AquesTalkPlayerの実行パスを設定します。 :param aquestalk_path: str: AquesTalkPlayerの実行可能ファイルのパス。 """ self.aquestalk_path = aquestalk_path setattr(self.config, "aquestalk_path", aquestalk_path) def set_api_key(self, api_key: str): """APIベースのTTSエンジンのAPIキーを設定します。 :param api_key: str: 設定するAPIキー。 """ self.api_key = api_key setattr(self.config, "elevenlabs_api_key", api_key) def get_tts_name(self, tts_name: str | None = None) -> str: """現在設定されているTTSエンジン名、または引数で指定されたTTSエンジン名を返します。 :param tts_name: str or None: 取得したいTTSエンジンの名前。 指定しない場合はインスタンスの設定を使用します。 :returns: str - TTSエンジンの名前。 """ if tts_name is None: return self.tts_name return tts_name def get_default_voice(self, tts_name: str | None = None) -> str | None: """指定されたTTSエンジンのデフォルト音声名を返します。 :param tts_name: str or None: デフォルト音声名を取得するTTSエンジンの名前。 指定しない場合はインスタンスの設定を使用します。 :returns: str or None - デフォルト音声名、または設定がない場合は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: str | None, tts_name: str | None = None) -> str | None: """話者ラベルを正規化します。 このメソッドは、グローバル関数 ``normalize_speaker`` のラッパーです。 :param speaker: str or None: 正規化する話者名。 :param tts_name: str or None: 現在は使用されていないが将来的な拡張のために残されているTTSエンジン名。 :returns: str or None - 正規化された話者名、または入力がNoneの場合は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: str | None = None, endpoint: str | None = None, api_key: str | None = None): """現在設定されている、または指定されたTTSエンジンで利用可能な音声の詳細情報を取得します。 このメソッドは、グローバル関数 ``get_available_voices_info`` のラッパーです。 インスタンスに設定されているエンドポイントやAPIキーを自動的に渡します。 :param tts_name: str or None: 情報を取得するTTSエンジンの名前。 指定しない場合はインスタンスの設定を使用します。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 指定しない場合はインスタンスの設定を使用します。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 指定しない場合はインスタンスの設定を使用します。 :returns: Any or None - 利用可能な音声の詳細情報、または機能が存在しない場合は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: str | None = None, endpoint: str | None = None, api_key: str | None = None): """現在設定されている、または指定されたTTSエンジンで利用可能な音声のリストを取得します。 このメソッドは、グローバル関数 ``get_available_voices`` のラッパーです。 インスタンスに設定されているエンドポイントやAPIキーを自動的に渡します。 :param tts_name: str or None: リストを取得するTTSエンジンの名前。 指定しない場合はインスタンスの設定を使用します。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 指定しない場合はインスタンスの設定を使用します。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 指定しない場合はインスタンスの設定を使用します。 :returns: list[str] or None - 利用可能な音声のリスト、または機能が存在しない場合は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: str | None = None, endpoint: str | None = None, api_key: str | None = None) -> bool: """現在設定されている、または指定されたTTSエンジンで利用可能な音声をコンソールに出力します。 このメソッドは、グローバル関数 ``list_available_voices`` のラッパーです。 インスタンスに設定されているエンドポイントやAPIキーを自動的に渡します。 :param tts_name: str or None: リストを出力するTTSエンジンの名前。 指定しない場合はインスタンスの設定を使用します。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 指定しない場合はインスタンスの設定を使用します。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 指定しない場合はインスタンスの設定を使用します。 :returns: bool - 処理が成功した場合はTrue、失敗した場合はFalse。 """ 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: str, voices: str, VOICE_MAPS: dict, is_monologue: bool, tts_name: str | None = None, endpoint: str | None = None) -> bool: """入力ファイルから話者を抽出し、既存の音声マップと指定された音声設定に基づいて、 最終的な話者と音声のマッピングを表示します。 このメソッドは主にデバッグや設定確認のために使用され、読み上げは行いません。 :param infile: str: 解析する入力ファイル（テキストまたはクリップボード）。 :param voices: str: セミコロンで区切られた音声指定文字列（例: "speaker1=voiceA;speaker2=voiceB;voiceC"）。 :param VOICE_MAPS: dict: 全てのTTSエンジンに対する既存の音声マッピング辞書。 :param is_monologue: bool: 入力ファイルが独話形式か。 :param tts_name: str or None: 使用するTTSエンジンの名前。指定しない場合はインスタンスの設定を使用します。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 :returns: bool - 処理が成功した場合はTrue、失敗した場合はFalse。 """ 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}") return True def load_text(self, input_path: str, monologue: bool = False, wait_for_clipboard: bool = True) -> list[tuple[str | None, str]] | None: """指定された入力元（クリップボードまたはファイル）からテキストを読み込み、 (speaker, text)のリストを生成します。 このメソッドは、グローバル関数 ``load_text`` のラッパーです。 :param input_path: str: テキストの入力元。"clip"またはファイルパス。 :param monologue: bool: Trueの場合、全てを単一の話者のテキストとして扱います。 Falseの場合、行を「話者,テキスト」形式で解析します。 :param wait_for_clipboard: bool: ``input_path`` が "clip" の場合に、 ユーザーがクリップボードにコピーするのを待つか。 :returns: list[tuple[str or None, str]] or None - (話者名, テキスト)のタプルのリスト、 またはエラー発生時はNone。 """ return load_text(input_path, monologue = monologue, wait_for_clipboard = wait_for_clipboard) def get_speakers_from_dialogue(self, dialogue: list[tuple[str | None, str]]) -> list[str | None]: """対話データからユニークな話者名のリストを抽出します。 :param dialogue: list[tuple[str or None, str]]: (話者名, テキスト)のタプルのリスト。 :returns: list[str or None] - 対話に含まれるユニークな話者名のリスト。 """ return list(set(s for s, _ in dialogue)) def parse_kv_string(self, kv_string: str | None, keys: list[str] = [], allow_no_key_kv_string: bool = False) -> dict[str | int, str]: """key=val;key=val 形式の文字列を dict に変換します。 このメソッドは、グローバル関数 ``parse_kv_string`` をラップし、 話者名の正規化を適用します。 :param kv_string: str or None: "key=val;key=val" 形式の文字列。 :param keys: list[str]: キーがない値に割り当てるためのキーのリスト。 :param allow_no_key_kv_string: bool: キーが指定されていないアイテム（例: "value1;value2"）を許可するかどうか。 :returns: dict[str or int, str] - 変換された辞書。キーは正規化された話者名またはインデックス。 """ 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: dict = {}, voices: str = "", speakers: list[str | None] = []) -> dict[str | None | int, str]: """既存の音声マップと新しい音声指定、話者リストに基づいて、 現在のTTSエンジンの音声マッピングを更新します。 ``voices`` 引数が空の場合、``speakers`` リストから話者名を取得し、 それをデフォルトの音声名としてマップします。 ``voices`` が指定されている場合は、それを解析して既存のマップを上書きまたは追加します。 :param voice_map: dict: 全てのTTSエンジンに対する既存の音声マッピング辞書。 :param voices: str: セミコロンで区切られた音声指定文字列 （例: "speaker1=voiceA;speaker2=voiceB;voiceC"）。 :param speakers: list[str or None]: 読み上げファイルから検出された話者のリスト。 :returns: dict[str or None or int, str] - 更新された現在のTTSエンジンの音声マッピング辞書。 """ 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: list[tuple[str | None, str]] | None = None, voice_map: dict[str | None, str] | str | None = None, speakers: dict | None = None, replacements: dict[str, str] | None = None, endpoint: str | None = None, api_key: str | None = None, output_format: str | None = None, config: SimpleNamespace | None = None) -> str | bool: """現在設定されているTTSエンジンを使用して、対話データから音声を生成し、 結合してファイルに保存または直接再生します。 このメソッドは、グローバル関数 ``speak_dialogue`` を呼び出すラッパーメソッドです。 インスタンスに設定されているエンドポイントやAPIキー、設定オブジェクトを自動的に渡します。 :param dialogue: list[tuple[str or None, str]] or None: (話者名, テキスト)のタプルのリスト。 :param voice_map: dict[str or None, str] or str or None: 話者と音声名をマッピングする辞書、 または単一の音声名文字列。 :param speakers: dict or None: 以前に解析された話者情報。 :param replacements: dict[str, str] or None: テキスト置換ルール。 :param endpoint: str or None: APIを使用するTTSエンジンのエンドポイントURL。 指定しない場合はインスタンスの設定を使用します。 :param api_key: str or None: APIを使用するTTSエンジンのAPIキー。 指定しない場合はインスタンスの設定を使用します。 :param output_format: str or None: 出力音声ファイルの形式（例: "mp3", "wav"）。 :param config: SimpleNamespace or None: 設定情報を含むオブジェクト。 指定しない場合はインスタンスの設定を使用します。 :returns: str or bool - 保存されたファイルのパス（str）、 または成功の場合はTrue (直接再生時)、失敗の場合はFalse。 """ 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("atp"), # 辞書キーは "atp" or "aquestalkplayer" default_optnai_voice=self.get_default_voice("openai"), default_elevenlabs_voice=self.get_default_voice("elevenlabs"), default_gemini_voice=self.get_default_voice("gemini"), endpoint=endpoint, api_key=api_key, output_format=output_format, )