#!/usr/bin/env python3 """ Fish Speech 命令行语音克隆脚本 无需 Web UI，纯命令行控制 """ import os import sys import subprocess import time import signal from pathlib import Path class FishSpeechCLI: def __init__(self): self.fish_speech_dir = Path("/root/tts/fish-speech") self.model_path = self.fish_speech_dir / "checkpoints/fish-speech-1.5/model.pth" self.decoder_path = self.fish_speech_dir / "checkpoints/fish-speech-1.5/firefly-gan-vq-fsq-8x1024-21hz-generator.pth" self.reference_audio = Path("/root/tts/ben_guanquelou.wav") self.output_dir = Path("/root/tts/audio_files") self.output_dir.mkdir(exist_ok=True) # 默认参数 self.reference_text = "登鹳雀楼，白日依山尽，黄河入海流。欲穷千里目，更上一层楼。" self.target_text = """我们习惯于赞美黄河之水天上来，习惯于歌颂大地的厚德载物。教科书告诉我们，河流是水循环的恩赐，大陆是漂浮在岩浆上的方舟。这是一个完美的、闭环的、温情脉脉的解释。但如果，这一切都是关于"摩擦力"的谎言呢？请试着像挤压一个注满水的海绵球一样，去想象我们脚下的这颗星球。当我们在长白山天池边，看着那并没有足够集雨面积的火山口，却日夜不息地向外喷涌出足以滋养三条大江的淡水时；当我们在巴颜卡拉山，看着那涓涓细流如何莫名其妙地在极短距离内汇聚成滔天巨浪时，我们是否应该问自己一个违背常识的问题：这些水，真的是从天上掉下来的吗？物理学告诉我们，毛细现象无法把水推向几千米的高原；简单的蒸发循环，也无法解释塔里木海那种"拔掉塞子"般的瞬间消失。这背后，一定存在一个"第一推动"。它不是温柔的渗透，它是暴力的"挤压"。""" self.server_process = None def check_files(self): """检查必需文件""" print("📦 检查文件...") files = [ (self.model_path, "主模型"), (self.decoder_path, "解码器"), (self.reference_audio, "参考音频") ] for file_path, name in files: if file_path.exists(): size_mb = file_path.stat().st_size / (1024 * 1024) print(f" ✅ {name}: {file_path.name} ({size_mb:.1f}MB)") else: print(f" ❌ {name}: {file_path.name} (缺失)") return False return True def start_api_server(self): """启动 API 服务器""" print("🚀 启动 Fish Speech API 服务器...") # 清理旧进程 subprocess.run("pkill -f 'api_server'", shell=True) time.sleep(2) # 切换到 Fish Speech 目录 os.chdir(self.fish_speech_dir) # 启动命令 cmd = [ sys.executable, "tools/api_server.py", "--llama-checkpoint-path", str(self.model_path), "--decoder-checkpoint-path", str(self.decoder_path), "--device", "cpu" ] print(f"执行命令: {' '.join(cmd)}") # 启动服务器 self.server_process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 等待服务器启动 print("⏳ 等待服务器启动...") max_wait = 120 # 最多等待2分钟 wait_time = 0 while wait_time < max_wait: if self.server_process.poll() is not None: print("❌ 服务器启动失败") stdout, stderr = self.server_process.communicate() print(f"错误: {stderr}") return False # 检查端口 try: import requests for port in [8080, 7860, 5000]: try: response = requests.get(f"http://127.0.0.1:{port}/health", timeout=2) if response.status_code == 200: print(f"✅ 服务器已启动: http://127.0.0.1:{port}") self.server_url = f"http://127.0.0.1:{port}" return True except: continue except ImportError: pass time.sleep(2) wait_time += 2 print(f" 等待中... ({wait_time}s)") print("⏰ 服务器启动超时") return False def synthesize_speech(self, output_filename="fish_speech_cli_output"): """进行语音合成""" print("🎙️ 开始语音合成...") print(f"📝 参考文本: {self.reference_text}") print(f"📝 目标文本长度: {len(self.target_text)} 字符") # 准备客户端命令 client_cmd = [ sys.executable, "tools/api_client.py", "--text", self.target_text, "--reference_audio", str(self.reference_audio), "--reference_text", self.reference_text, "--output", str(self.output_dir / output_filename), "--no-play", "--max_new_tokens", "2048", "--chunk_length", "300", "--top_p", "0.8", "--temperature", "0.8", "--repetition_penalty", "1.1", "--url", f"{self.server_url}/v1/tts", "--format", "wav" ] print(f"执行命令: {' '.join(client_cmd)}") # 运行客户端 result = subprocess.run( client_cmd, capture_output=True, text=True, timeout=600 # 10分钟超时 ) print("🎙️ 合成结果:") if result.stdout: print("输出:", result.stdout.strip()) if result.stderr: print("错误:", result.stderr.strip()) return result.returncode == 0 def check_output(self, output_filename): """检查输出文件""" output_files = [ self.output_dir / f"{output_filename}.wav", self.output_dir / f"{output_filename}.mp3", self.output_dir / f"{output_filename}.flac" ] for output_file in output_files: if output_file.exists(): try: import torchaudio waveform, sample_rate = torchaudio.load(str(output_file)) duration = waveform.shape[1] / sample_rate print(f"\n✅ 音频生成成功!") print(f"📁 文件: {output_file}") print(f"📊 大小: {output_file.stat().st_size:,} bytes") print(f"🎵 时长: {duration:.2f} 秒") print(f"🎵 采样率: {sample_rate:,} Hz") if duration >= 25: print("🎉 时长符合30秒要求!") else: print(f"⚠️ 时长为 {duration:.2f} 秒") return True, str(output_file) except Exception as e: print(f"⚠️ 读取音频失败: {e}") return True, str(output_file) print("❌ 未找到生成的音频文件") return False, None def cleanup(self): """清理资源""" if self.server_process: print("🧹 停止服务器...") self.server_process.terminate() time.sleep(2) def run(self, output_filename="fish_speech_cli_output"): """运行完整的命令行语音合成流程""" print("🎊 Fish Speech 命令行语音克隆") print("=" * 60) try: # 1. 检查文件 if not self.check_files(): print("❌ 文件检查失败") return False # 2. 启动服务器 if not self.start_api_server(): print("❌ 服务器启动失败") return False # 3. 语音合成 if not self.synthesize_speech(output_filename): print("❌ 语音合成失败") return False # 4. 检查结果 success, output_file = self.check_output(output_filename) if success: print(f"\n🎉 命令行语音合成完成!") print(f"📁 输出文件: {output_file}") return True else: print("❌ 未找到输出文件") return False except KeyboardInterrupt: print("\n🛑 用户中断操作") return False except Exception as e: print(f"❌ 执行失败: {e}") return False finally: # 清理 self.cleanup() def main(): """主函数""" if len(sys.argv) > 1: output_filename = sys.argv[1] else: output_filename = "fish_speech_cli_output" cli = FishSpeechCLI() success = cli.run(output_filename) if success: print(f"\n🎊 成功! 使用命令播放音频:") print(f" aplay {cli.output_dir}/{output_filename}.wav") print(f" 或使用文件管理器打开: {cli.output_dir}/") else: print("\n💔 失败，请检查错误信息") if __name__ == "__main__": main()