432 lines
19 KiB
Python
432 lines
19 KiB
Python
import os
|
||
import asyncio
|
||
import threading
|
||
import tkinter as tk
|
||
from tkinter import ttk, messagebox, filedialog
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
import aiohttp
|
||
import aiofiles
|
||
import binascii
|
||
import json
|
||
import logging
|
||
import subprocess
|
||
import platform
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||
)
|
||
logger = logging.getLogger("minimax_tts_gui")
|
||
|
||
class MiniMaxTTSGUI:
|
||
def __init__(self, root):
|
||
self.root = root
|
||
self.root.title("MiniMax TTS 工具")
|
||
self.root.geometry("750x700")
|
||
|
||
self.current_audio_file = None
|
||
self.synthesis_in_progress = False
|
||
self.text_entries = []
|
||
self.config_collapsed = False
|
||
|
||
self.load_env()
|
||
self.setup_ui()
|
||
|
||
def load_env(self):
|
||
env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "minimax.env")
|
||
if os.path.exists(env_path):
|
||
load_dotenv(env_path)
|
||
logger.info(f"已加载环境变量文件: {env_path}")
|
||
else:
|
||
logger.warning(f"环境变量文件不存在: {env_path}")
|
||
|
||
def setup_ui(self):
|
||
main_frame = ttk.Frame(self.root, padding="10")
|
||
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
|
||
|
||
self.root.columnconfigure(0, weight=1)
|
||
self.root.rowconfigure(0, weight=1)
|
||
main_frame.columnconfigure(1, weight=1)
|
||
|
||
row = 0
|
||
|
||
config_header_frame = ttk.Frame(main_frame)
|
||
config_header_frame.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 5))
|
||
|
||
ttk.Label(config_header_frame, text="配置信息", font=("Arial", 12, "bold")).pack(side=tk.LEFT)
|
||
self.toggle_btn = ttk.Button(config_header_frame, text="▼", width=3, command=self.toggle_config)
|
||
self.toggle_btn.pack(side=tk.RIGHT)
|
||
|
||
row += 1
|
||
|
||
self.config_frame = ttk.Frame(main_frame)
|
||
self.config_frame.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))
|
||
|
||
config_row = 0
|
||
|
||
ttk.Label(self.config_frame, text="API Key:").grid(row=config_row, column=0, sticky=tk.W, pady=5)
|
||
self.api_key_var = tk.StringVar(value=os.getenv("MINIMAX_API_KEY", ""))
|
||
ttk.Entry(self.config_frame, textvariable=self.api_key_var, width=60).grid(row=config_row, column=1, sticky=(tk.W, tk.E), pady=5)
|
||
config_row += 1
|
||
|
||
ttk.Label(self.config_frame, text="Group ID:").grid(row=config_row, column=0, sticky=tk.W, pady=5)
|
||
self.group_id_var = tk.StringVar(value=os.getenv("MINIMAX_GROUP_ID", "1915292410024300630"))
|
||
ttk.Entry(self.config_frame, textvariable=self.group_id_var, width=60).grid(row=config_row, column=1, sticky=(tk.W, tk.E), pady=5)
|
||
config_row += 1
|
||
|
||
ttk.Label(self.config_frame, text="Base URL:").grid(row=config_row, column=0, sticky=tk.W, pady=5)
|
||
self.base_url_var = tk.StringVar(value=os.getenv("MINIMAX_BASE_URL", "https://api.minimax.chat/v1/t2a_v2"))
|
||
ttk.Entry(self.config_frame, textvariable=self.base_url_var, width=60).grid(row=config_row, column=1, sticky=(tk.W, tk.E), pady=5)
|
||
config_row += 1
|
||
|
||
ttk.Label(self.config_frame, text="音色 ID:").grid(row=config_row, column=0, sticky=tk.W, pady=5)
|
||
self.voice_id_var = tk.StringVar(value=os.getenv("MINIMAX_VOICE_ID", "cartoon-boy-01"))
|
||
ttk.Entry(self.config_frame, textvariable=self.voice_id_var, width=60).grid(row=config_row, column=1, sticky=(tk.W, tk.E), pady=5)
|
||
config_row += 1
|
||
|
||
self.config_frame.columnconfigure(1, weight=1)
|
||
|
||
row += 1
|
||
|
||
ttk.Separator(main_frame, orient='horizontal').grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=10)
|
||
row += 1
|
||
|
||
text_list_header = ttk.Frame(main_frame)
|
||
text_list_header.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 5))
|
||
|
||
ttk.Label(text_list_header, text="文本列表", font=("Arial", 12, "bold")).pack(side=tk.LEFT)
|
||
add_btn = ttk.Button(text_list_header, text="➕", width=3, command=self.add_text_entry)
|
||
add_btn.pack(side=tk.RIGHT)
|
||
row += 1
|
||
|
||
self.text_list_container = ttk.Frame(main_frame)
|
||
self.text_list_container.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=5)
|
||
self.text_list_container.columnconfigure(1, weight=1)
|
||
|
||
self.add_text_entry()
|
||
|
||
row += 1
|
||
|
||
ttk.Separator(main_frame, orient='horizontal').grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=10)
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="合成参数", font=("Arial", 12, "bold")).grid(row=row, column=0, columnspan=2, sticky=tk.W, pady=(0, 10))
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="输出目录:").grid(row=row, column=0, sticky=tk.W, pady=5)
|
||
self.output_dir_var = tk.StringVar(value="assets")
|
||
ttk.Entry(main_frame, textvariable=self.output_dir_var, width=50).grid(row=row, column=1, sticky=(tk.W, tk.E), pady=5)
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="语速:").grid(row=row, column=0, sticky=tk.W, pady=5)
|
||
self.speed_var = tk.DoubleVar(value=1.0)
|
||
ttk.Scale(main_frame, from_=0.5, to=2.0, variable=self.speed_var, orient=tk.HORIZONTAL, length=300).grid(row=row, column=1, sticky=tk.W, pady=5)
|
||
ttk.Label(main_frame, textvariable=self.speed_var).grid(row=row, column=1, sticky=tk.E, pady=5)
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="音量:").grid(row=row, column=0, sticky=tk.W, pady=5)
|
||
self.vol_var = tk.DoubleVar(value=1.0)
|
||
ttk.Scale(main_frame, from_=0.5, to=2.0, variable=self.vol_var, orient=tk.HORIZONTAL, length=300).grid(row=row, column=1, sticky=tk.W, pady=5)
|
||
ttk.Label(main_frame, textvariable=self.vol_var).grid(row=row, column=1, sticky=tk.E, pady=5)
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="音调:").grid(row=row, column=0, sticky=tk.W, pady=5)
|
||
self.pitch_var = tk.IntVar(value=0)
|
||
ttk.Scale(main_frame, from_=-10, to=10, variable=self.pitch_var, orient=tk.HORIZONTAL, length=300).grid(row=row, column=1, sticky=tk.W, pady=5)
|
||
ttk.Label(main_frame, textvariable=self.pitch_var).grid(row=row, column=1, sticky=tk.E, pady=5)
|
||
row += 1
|
||
|
||
button_frame = ttk.Frame(main_frame)
|
||
button_frame.grid(row=row, column=0, columnspan=2, pady=15)
|
||
|
||
self.synthesize_btn = ttk.Button(button_frame, text="开始合成", command=self.start_synthesis)
|
||
self.synthesize_btn.pack(side=tk.LEFT, padx=5)
|
||
|
||
self.play_btn = ttk.Button(button_frame, text="播放最后一个", command=self.play_audio, state=tk.DISABLED)
|
||
self.play_btn.pack(side=tk.LEFT, padx=5)
|
||
|
||
self.open_folder_btn = ttk.Button(button_frame, text="打开输出目录", command=self.open_output_folder)
|
||
self.open_folder_btn.pack(side=tk.LEFT, padx=5)
|
||
row += 1
|
||
|
||
ttk.Separator(main_frame, orient='horizontal').grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=10)
|
||
row += 1
|
||
|
||
ttk.Label(main_frame, text="日志", font=("Arial", 12, "bold")).grid(row=row, column=0, columnspan=2, sticky=tk.W, pady=(0, 10))
|
||
row += 1
|
||
|
||
self.log_text = tk.Text(main_frame, height=8, width=70, state=tk.DISABLED)
|
||
self.log_text.grid(row=row, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=5)
|
||
|
||
log_scrollbar = ttk.Scrollbar(main_frame, orient=tk.VERTICAL, command=self.log_text.yview)
|
||
log_scrollbar.grid(row=row, column=2, sticky=(tk.N, tk.S))
|
||
self.log_text['yscrollcommand'] = log_scrollbar.set
|
||
|
||
main_frame.rowconfigure(row, weight=1)
|
||
|
||
def toggle_config(self):
|
||
self.config_collapsed = not self.config_collapsed
|
||
if self.config_collapsed:
|
||
self.config_frame.grid_remove()
|
||
self.toggle_btn.config(text="▶")
|
||
else:
|
||
self.config_frame.grid()
|
||
self.toggle_btn.config(text="▼")
|
||
|
||
def add_text_entry(self):
|
||
entry_frame = ttk.Frame(self.text_list_container)
|
||
entry_row = len(self.text_entries)
|
||
|
||
ttk.Label(entry_frame, text=f"#{entry_row + 1}").grid(row=0, column=0, sticky=tk.W, padx=2)
|
||
|
||
ttk.Label(entry_frame, text="文件名:").grid(row=0, column=1, sticky=tk.W, padx=2)
|
||
filename_var = tk.StringVar(value=f"text_{entry_row + 1}")
|
||
filename_entry = ttk.Entry(entry_frame, textvariable=filename_var, width=15)
|
||
filename_entry.grid(row=0, column=2, sticky=tk.W, padx=2)
|
||
|
||
ttk.Label(entry_frame, text="文本:").grid(row=0, column=3, sticky=tk.W, padx=2)
|
||
text_var = tk.StringVar(value="")
|
||
text_entry = ttk.Entry(entry_frame, textvariable=text_var, width=35)
|
||
text_entry.grid(row=0, column=4, sticky=(tk.W, tk.E), padx=2)
|
||
|
||
remove_btn = ttk.Button(entry_frame, text="✕", width=3, command=lambda: self.remove_text_entry(entry_frame))
|
||
remove_btn.grid(row=0, column=5, sticky=tk.W, padx=2)
|
||
|
||
entry_frame.columnconfigure(4, weight=1)
|
||
entry_frame.grid(row=entry_row, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=3)
|
||
|
||
self.text_entries.append({
|
||
'frame': entry_frame,
|
||
'filename_var': filename_var,
|
||
'text_var': text_var
|
||
})
|
||
|
||
def remove_text_entry(self, frame):
|
||
if len(self.text_entries) <= 1:
|
||
messagebox.showwarning("提示", "至少保留一个文本条目")
|
||
return
|
||
|
||
for i, entry in enumerate(self.text_entries):
|
||
if entry['frame'] == frame:
|
||
frame.destroy()
|
||
self.text_entries.pop(i)
|
||
self.reorder_entries()
|
||
break
|
||
|
||
def reorder_entries(self):
|
||
for i, entry in enumerate(self.text_entries):
|
||
entry['frame'].grid(row=i, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=3)
|
||
for widget in entry['frame'].winfo_children():
|
||
if isinstance(widget, ttk.Label) and widget.cget("text").startswith("#"):
|
||
widget.config(text=f"#{i + 1}")
|
||
break
|
||
|
||
def log_message(self, message):
|
||
self.log_text.config(state=tk.NORMAL)
|
||
self.log_text.insert(tk.END, message + "\n")
|
||
self.log_text.see(tk.END)
|
||
self.log_text.config(state=tk.DISABLED)
|
||
logger.info(message)
|
||
|
||
def start_synthesis(self):
|
||
if self.synthesis_in_progress:
|
||
messagebox.showwarning("提示", "合成正在进行中，请稍候...")
|
||
return
|
||
|
||
text_list = []
|
||
for entry in self.text_entries:
|
||
text = entry['text_var'].get().strip()
|
||
filename = entry['filename_var'].get().strip()
|
||
if text:
|
||
text_list.append({
|
||
'text': text,
|
||
'filename': filename or "output"
|
||
})
|
||
|
||
if not text_list:
|
||
messagebox.showwarning("提示", "请输入要合成的文本")
|
||
return
|
||
|
||
api_key = self.api_key_var.get().strip()
|
||
group_id = self.group_id_var.get().strip()
|
||
if not api_key or not group_id:
|
||
messagebox.showwarning("提示", "请填写 API Key 和 Group ID")
|
||
return
|
||
|
||
self.synthesis_in_progress = True
|
||
self.synthesize_btn.config(state=tk.DISABLED)
|
||
self.play_btn.config(state=tk.DISABLED)
|
||
|
||
thread = threading.Thread(target=self.run_batch_synthesis, args=(text_list,))
|
||
thread.daemon = True
|
||
thread.start()
|
||
|
||
def run_batch_synthesis(self, text_list):
|
||
try:
|
||
output_dir = Path(self.output_dir_var.get())
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
self.log_message(f"开始批量合成 {len(text_list)} 个文本...")
|
||
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
|
||
success_count = 0
|
||
last_file = None
|
||
|
||
for i, item in enumerate(text_list):
|
||
output_file = output_dir / f"{item['filename']}.mp3"
|
||
self.log_message(f"[{i+1}/{len(text_list)}] 正在合成: {item['text'][:30]}...")
|
||
|
||
result = loop.run_until_complete(
|
||
self.synthesize_speech(
|
||
text=item['text'],
|
||
output_file=str(output_file),
|
||
api_key=self.api_key_var.get(),
|
||
group_id=self.group_id_var.get(),
|
||
base_url=self.base_url_var.get(),
|
||
voice_id=self.voice_id_var.get(),
|
||
speed=self.speed_var.get(),
|
||
vol=self.vol_var.get(),
|
||
pitch=self.pitch_var.get()
|
||
)
|
||
)
|
||
|
||
if result:
|
||
success_count += 1
|
||
last_file = result
|
||
self.log_message(f" ✓ 成功: {result}")
|
||
else:
|
||
self.log_message(f" ✗ 失败")
|
||
|
||
loop.close()
|
||
|
||
if success_count > 0:
|
||
self.current_audio_file = last_file
|
||
self.log_message(f"\n批量合成完成！成功 {success_count}/{len(text_list)} 个")
|
||
self.root.after(0, lambda: self.play_btn.config(state=tk.NORMAL))
|
||
self.root.after(0, lambda: messagebox.showinfo(
|
||
"合成完成",
|
||
f"批量合成完成！\n成功 {success_count}/{len(text_list)} 个\n输出目录: {output_dir.resolve()}"
|
||
))
|
||
else:
|
||
self.log_message("所有合成都失败了")
|
||
self.root.after(0, lambda: messagebox.showerror("错误", "所有合成都失败了，请查看日志"))
|
||
|
||
except Exception as e:
|
||
self.log_message(f"批量合成出错: {str(e)}")
|
||
self.root.after(0, lambda: messagebox.showerror("错误", f"批量合成出错: {str(e)}"))
|
||
finally:
|
||
self.synthesis_in_progress = False
|
||
self.root.after(0, lambda: self.synthesize_btn.config(state=tk.NORMAL))
|
||
|
||
async def synthesize_speech(self, text, output_file, api_key, group_id, base_url, voice_id, speed, vol, pitch):
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
payload = {
|
||
"model": "speech-02-turbo",
|
||
"text": text,
|
||
"stream": True,
|
||
"voice_setting": {
|
||
"voice_id": voice_id,
|
||
"speed": speed,
|
||
"vol": vol,
|
||
"pitch": pitch
|
||
},
|
||
"audio_setting": {
|
||
"sample_rate": 16000,
|
||
"bitrate": 32000,
|
||
"format": "mp3",
|
||
"channel": 1
|
||
}
|
||
}
|
||
|
||
url = f"{base_url}?GroupId={group_id}"
|
||
|
||
try:
|
||
audio_buffer = bytearray()
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(url, json=payload, headers=headers) as response:
|
||
if response.status != 200:
|
||
error_text = await response.text()
|
||
self.log_message(f" 请求失败: {response.status}, {error_text}")
|
||
return None
|
||
|
||
async for line in response.content:
|
||
line = line.strip()
|
||
if line.startswith(b'data:'):
|
||
try:
|
||
json_str = line[5:].decode('utf-8').strip()
|
||
if not json_str:
|
||
continue
|
||
|
||
data_json = json.loads(json_str)
|
||
|
||
if "data" in data_json and "audio" in data_json["data"]:
|
||
status = data_json["data"].get("status", 1)
|
||
audio_hex = data_json["data"]["audio"]
|
||
|
||
if status == 1:
|
||
if audio_hex:
|
||
audio_binary = binascii.unhexlify(audio_hex)
|
||
audio_buffer.extend(audio_binary)
|
||
elif status == 2:
|
||
break
|
||
except json.JSONDecodeError as e:
|
||
self.log_message(f" JSON解析失败: {e}")
|
||
except Exception as e:
|
||
self.log_message(f" 处理响应出错: {str(e)}")
|
||
|
||
if len(audio_buffer) == 0:
|
||
self.log_message(" 音频缓冲区为空")
|
||
return None
|
||
|
||
async with aiofiles.open(output_file, 'wb') as f:
|
||
await f.write(audio_buffer)
|
||
|
||
return output_file
|
||
|
||
except Exception as e:
|
||
self.log_message(f" 合成失败: {str(e)}")
|
||
return None
|
||
|
||
def play_audio(self):
|
||
if not self.current_audio_file or not os.path.exists(self.current_audio_file):
|
||
messagebox.showwarning("提示", "请先合成音频")
|
||
return
|
||
|
||
try:
|
||
if platform.system() == "Darwin":
|
||
subprocess.run(["open", self.current_audio_file])
|
||
elif platform.system() == "Windows":
|
||
os.startfile(self.current_audio_file)
|
||
else:
|
||
subprocess.run(["xdg-open", self.current_audio_file])
|
||
self.log_message(f"正在播放: {self.current_audio_file}")
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"播放失败: {str(e)}")
|
||
|
||
def open_output_folder(self):
|
||
output_dir = self.output_dir_var.get()
|
||
if not os.path.exists(output_dir):
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
try:
|
||
if platform.system() == "Darwin":
|
||
subprocess.run(["open", output_dir])
|
||
elif platform.system() == "Windows":
|
||
os.startfile(output_dir)
|
||
else:
|
||
subprocess.run(["xdg-open", output_dir])
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"打开目录失败: {str(e)}")
|
||
|
||
if __name__ == "__main__":
|
||
root = tk.Tk()
|
||
app = MiniMaxTTSGUI(root)
|
||
root.mainloop()
|