import os import sys import re import time import subprocess import shutil import tempfile from pathlib import Path import traceback import inspect # --- 必要なライブラリのチェック --- required_libs = ["pydub", "pyperclip", "chardet", "pyttsx3", "openai", "simpleaudio"] missing = [] for lib in required_libs: try: __import__(lib) except ImportError: missing.append(lib) if missing: print(f"Error: Missing libraries:\n{', '.join(missing)}") print("------------------------------------------------------------------") print("インストールが必要です:") print(" pip install pyttsx3 openai pydub chardet pyperclip") print("\n[FFmpeg のインストール]") print(" pydub, simpleaudio の利用に必要です。公式サイトからダウンロードし、PATHを設定してください。") print("------------------------------------------------------------------") input("\nPress ENTER to terminate>>\n") sys.exit(1) import pyperclip import chardet from pydub import AudioSegment from pydub.playback import play import simpleaudio as sa tktts_pyttsx3 = None tktts_voicevox = None tktts_aquestalkplayer = None tktts_openai = None def _safe_import(module_name): """モジュールを安全にインポートし、エラーメッセージを表示する""" try: return __import__(module_name) except Exception as e: print(f"\nWarning in tktts.py: Import error for {module_name}") print("------------------------------------------------------------------") print(f"Error message: {e}") print("Traceback:") traceback.print_exc() print("------------------------------------------------------------------") return None tktts_pyttsx3 = _safe_import('tktts_pyttsx3') tktts_voicevox = _safe_import('tktts_voicevox') tktts_aquestalkplayer = _safe_import('tktts_aquestalkplayer') tktts_openai = _safe_import('tktts_openai') default_pyttsx3_voice = "Zira" default_voicevox_voice = "四国めたん" default_aqt_preset = "れいむ" default_optnai_voice = "alloy" TTS_ENGINES = { "pyttsx3": { "engine": tktts_pyttsx3, "default_voice": default_pyttsx3_voice, "ext": "wav" }, "voicevox": { "engine": tktts_voicevox, "default_voice": default_voicevox_voice, "ext": "wav" }, "aquestalkplayer": { "engine": tktts_aquestalkplayer, "default_voice": default_aqt_preset, "ext": "wav" }, "atp": { "engine": tktts_aquestalkplayer, # aquestalkplayerとエンジンを共有 "default_voice": default_aqt_preset, "ext": "wav" }, "openai": { "engine": tktts_openai, "default_voice": default_optnai_voice, "ext": "mp3" }, } def apply_replacements(text, replacements): """置換辞書を使ってテキストを置換（大文字小文字無視）""" for key, val in replacements.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_aqt_preset = "れいむ", default_optnai_voice = "alloy"): 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(f"\n話者リスト作成...") voices_specified = voices.split(';') if len(voices_specified) > 0: default_voice = voices_specified[0] speakers = [] target_voices = {} ispeaker = 0 for speaker, text in dialogue: if speaker is not None and speaker != "" and speaker not in speakers: speakers.append(speaker) if len(voices_specified) > ispeaker: target_voices[speaker] = voices_specified[ispeaker] else: target_voices[speaker] = default_voice ispeaker += 1 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): # 📌 修正箇所 2: TTSモジュールオブジェクトの取得 if tts_engine in TTS_ENGINES: config = TTS_ENGINES[tts_engine] tts_module = config["engine"] if tts_module is not None: return tts_module else: # エンジンは登録されているが、インポートに失敗した場合 print(f"\nError in tktts.get_tts(): TTS engine module [tktts_{tts_engine}] failed to import.") print(f"Scroll back the terminal console to check specific import error message.") print(f" Run tktts_{tts_engine}.py to check missing libraries in this module") return None else: print(f"\nError in tktts.get_tts(): Invalid tts engine [{tts_engine}]\n") return None def get_available_voices_info(endpoint = None): tts = get_tts(tts_engine) if tts is None: return None return tts.get_available_voices_info(endpoint) def get_available_voices(endpoint = None): tts = get_tts(tts_engine) if tts is None: return None return tts.get_available_voices(endpoint) def list_available_voices(tts_engine, endpoint = None): tts = get_tts(tts_engine) if tts is None: return None config = TTS_ENGINES.get(tts_engine) if config is None: print(f"\nError in tktts.list_available_voices(): Invalid tts engine [{tts_engine}]\n") return None ret = False try: if hasattr(tts, 'list_available_voices'): # VoiceVoxなど、endpoint引数を持つ可能性があるものに対応 if tts_engine == "voicevox" and endpoint is not None: ret = tts.list_available_voices(endpoint=endpoint) else: # 引数なしで呼び出す ret = tts.list_available_voices() else: print(f"Error: TTS module [{tts_engine}] has no list_available_voices function.") except Exception as e: print(f"Error calling list_available_voices for {tts_engine}: {e}") ret = False print("===============================") return ret """ def normalize_speaker(speaker, tts_engine): tts = get_tts(tts_engine) if hasattr(tts, "normalize_speaker"): return tts.normalize_speaker(speaker) return speaker """ def normalize_speaker(speaker, tts_engine): if speaker is None: return None # style付きのspeakerに対応 speaker_list = speaker.split('（') # '(', ')'をreplace.pyで' 'に変換してしまう問題に対応 speaker_list = speaker.split(' ') if len(speaker_list) > 0: return speaker_list[0] return speaker 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 speak_dialogue(args, dialogue, voice_map = None, speakers = {}, replacements = None, default_voicevox_voice = "四国めたん", default_pyttsx3_voice = "Zira", default_aqt_preset = "れいむ", default_optnai_voice = "alloy", endpoint = None, output_format = 'mp3'): """選択されたエンジンで音声生成・結合・保存/再生を行う""" # print() print(f"tktts.speak_dialogue(): Generate audio files:") tts_engine = 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 = args.outfile is_save_mode = bool(outfile) print() print("speak_dialogue:") print(f" tts_engine : {tts_engine} is monologue: {args.monologue}" + f" speak rate : {args.speak_rate} tinterval : {args.tinterval}") print(f" voices : {args.voices}") if is_save_mode: print(f" outfile : {args.outfile}") else: print(f" is_save_mode: {is_save_mode}") if voice_map is None: target_voices = get_speaker_dict(tts_engine, dialogue, args.voices, default_voicevox_voice = default_voicevox_voice, default_pyttsx3_voice = default_pyttsx3_voice, default_aqt_preset = default_aqt_preset, default_optnai_voice = default_optnai_voice) else: target_voices = voice_map # print("target_voices:") # for key, val in target_voices.items(): # print(f" {key}: {val}") # pyttsx3以外のエンジンはファイル保存が必須というチェック # pyttsx3はインメモリ再生が可能なので除外 if tts_engine != "pyttsx3": if args.outfile is None or args.outfile == "": print("❌ ファイル保存が必要です。--outfile を指定してください。") return False if not args.temp_dir: print("❌ 一次ファイル保存が必要です。--temp_dir を指定してください。") return False temp_dir = create_temp_dir(args.temp_dir) tmpfiles = [] success = True # 共通引数 call_kwargs = { "dialogue": dialogue, "replacements": replacements, "target_voices": target_voices, "speakers": speakers, "temp_dir": temp_dir, "outfile": args.outfile, "ext": config["ext"], # 辞書から取得 "cfg": args } # エンジン固有の引数を追加 # inspect.signature()を使って、実際にモジュールがどの引数をサポートしているかチェックするのが理想的だが、 # シンプルに引数名で判断する if tts_engine == "pyttsx3": call_kwargs["speak_rate"] = args.speak_rate elif tts_engine == "voicevox": if endpoint: call_kwargs["endpoint"] = endpoint elif tts_engine == "aquestalkplayer" or tts_engine == "atp": if hasattr(args, 'aquestalk_path'): call_kwargs["aquestalk_path"] = args.aquestalk_path elif tts_engine == "openai": if hasattr(args, 'instruction'): call_kwargs["instruction"] = args.instruction # print() print(f"\n⚙️ {tts_engine.upper()}で音声ファイルを生成中...") try: success, tmpfiles = tts.speak_dialogue(**call_kwargs) except Exception as e: print(f"❌ {tts_engine} の speak_dialogue 呼び出しエラー: {e}") traceback.print_exc() success = False if not success: return False # ファイル結合ロジック (変更なし) print(" ファイルを結合中...") combined_audio = AudioSegment.silent(duration=0) ext = config["ext"] # 辞書から取得したextを使用 # 3. pydubでセグメントを結合 for tmpfile in tmpfiles: # print(f" add {tmpfile}...") segment = AudioSegment.from_file(tmpfile, format = ext) print(f" insert {args.tinterval} sec interval") tinterval_ms = int(args.tinterval * 1000) combined_audio += segment + AudioSegment.silent(duration=tinterval_ms) # 4. クリーンアップと最終出力 # output_format = 'mp3' if is_save_mode: combined_audio.export(outfile, format=output_format) print(f"✅ 出力音声を {outfile} に保存しました (形式: {output_format})。") time.sleep(0.1) # API/CPU負荷軽減のため print("🗑️ 一時ファイルを削除中...") for f in tmpfiles: try: os.remove(f) except: pass shutil.rmtree(args.temp_dir) return outfile class tkTTS: def __init__(self, tts_name = None, config = {}): self.tts_name = tts_name self.tts = None self.config = config self.endpoint = config.endpoint if hasattr(config, "endpoint") else None self.aquestalk_path = config.aquestalk_path if hasattr(config, "aquestalk_path") else None self.set_engine(tts_name) def set_engine(self, tts_name): self.tts_name = tts_name self.tts = get_tts(tts_name) def set_endpoint(self, endpoint): self.config.endpoint = endpoint def set_aquestalk_path(self, aquestalk_path): self.config.aquestalk_path = aquestalk_path def get_tts_name(self, tts_name): 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 list_available_voices(self, tts_name = None, endpoint = None): if tts_name is None: tts_name = self.tts_name if endpoint is None: endpoint = self.endpoint return list_available_voices(tts_name, endpoint = endpoint) 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 = {}, replacements = None, endpoint = None, output_format = 'mp3', config = None): if endpoint is None: endpoint = self.endpoint 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_aqt_preset = self.get_default_voice("aqt"), default_optnai_voice = self.get_default_voice("openai"), endpoint = endpoint, output_format = output_format)