Fish Speech 1.5开发者工具链：CLI命令行工具封装与自动化测试框架
·
Fish Speech 1.5开发者工具链：CLI命令行工具封装与自动化测试框架
如果你已经体验过Fish Speech 1.5的Web界面，觉得它生成语音又快又好，那么接下来我要分享的内容，会让你在开发效率上再上一个台阶。想象一下，你不再需要每次都打开浏览器、输入文字、点击按钮，而是用一行命令就能生成语音，还能批量处理成百上千个文本文件，甚至自动测试模型在不同场景下的表现。
这就是开发者工具链的魅力。今天，我们不聊怎么用界面，而是深入后台，看看如何把Fish Speech 1.5变成一个更强大、更自动化的开发工具。我会带你从零开始，封装一个命令行工具，并搭建一个自动化测试框架，让你真正把TTS能力集成到自己的项目中。
1. 为什么需要开发者工具链？
在开始动手之前，我们先搞清楚一个问题：已经有了好用的Web界面，为什么还要折腾命令行和自动化测试？
场景一：批量处理 你手头有100篇产品说明文档，需要全部转换成语音。用Web界面？你得复制粘贴100次，点击100次按钮，下载100个文件。这显然不现实。
场景二：集成到其他系统 你的聊天机器人需要语音回复功能。你不可能让机器人去打开浏览器操作，它需要一个程序化的接口来调用。
场景三：持续验证 每次模型更新后，你怎么知道它还能正常工作？靠人工一个个测试？太慢了，而且容易遗漏。
这就是开发者工具链要解决的问题：效率、集成、可靠性。通过命令行工具，你可以用脚本批量处理；通过自动化测试，你可以确保每次更新都不会破坏原有功能。
2. 环境准备与基础API调用
在封装自己的工具之前，我们先看看Fish Speech 1.5提供了哪些原生能力。镜像已经内置了完整的API服务，这是我们一切工作的基础。
2.1 检查API服务状态
首先，确保你的Fish Speech实例正在运行。打开终端，连接到你的实例，检查服务是否正常：
# 检查后端API服务（端口7861）
curl -I http://127.0.0.1:7861/v1/tts
# 检查前端WebUI服务（端口7860）
curl -I http://127.0.0.1:7860
如果看到HTTP/1.1 200 OK或者HTTP/1.1 405 Method Not Allowed（对于GET请求到POST端点），说明服务正常。405是正常的，因为/v1/tts只接受POST请求。
2.2 基础API调用示例
让我们先写一个最简单的Python脚本来调用API：
# basic_api_call.py
import requests
import json
import soundfile as sf
import io
def generate_speech(text, output_path="output.wav"):
"""
调用Fish Speech API生成语音
参数:
text: 要合成的文本
output_path: 输出音频文件路径
"""
# API端点
url = "http://127.0.0.1:7861/v1/tts"
# 请求参数
payload = {
"text": text,
"reference_id": None, # 不使用参考音色
"max_new_tokens": 1024 # 最大token数
}
# 设置请求头
headers = {
"Content-Type": "application/json"
}
try:
print(f"正在生成语音: {text[:50]}...")
# 发送POST请求
response = requests.post(url, json=payload, headers=headers)
# 检查响应状态
if response.status_code == 200:
# 保存音频文件
with open(output_path, "wb") as f:
f.write(response.content)
print(f"✅ 语音生成成功，保存到: {output_path}")
return True
else:
print(f"❌ 请求失败，状态码: {response.status_code}")
print(f"错误信息: {response.text}")
return False
except requests.exceptions.ConnectionError:
print("❌ 无法连接到API服务，请检查服务是否启动")
return False
except Exception as e:
print(f"❌ 发生未知错误: {str(e)}")
return False
# 测试调用
if __name__ == "__main__":
# 测试文本
test_text = "你好，这是一个Fish Speech API调用测试。"
# 生成语音
success = generate_speech(test_text, "test_output.wav")
if success:
print("测试完成！")
else:
print("测试失败，请检查问题。")
运行这个脚本，你会在当前目录得到一个test_output.wav文件。这就是最基础的API调用方式，但每次都要写这么多代码，显然不够方便。
3. 封装CLI命令行工具
现在，我们来创建一个更友好的命令行工具。目标是：通过一行命令就能生成语音，支持各种参数配置，还能批量处理。
3.1 设计命令行接口
一个好的CLI工具应该：
- 命令简单易记
- 参数清晰明了
- 有详细的帮助信息
- 支持默认值和快捷方式
我设计了一个叫fish-tts的命令行工具，它支持以下功能：
# 基础用法：生成单个语音
fish-tts generate "你好，世界" -o hello.wav
# 使用参考音频进行音色克隆
fish-tts generate "早上好" -r reference.wav -o morning.wav
# 批量处理文本文件
fish-tts batch input.txt -o output_dir/
# 从标准输入读取文本
echo "这是一个测试" | fish-tts generate -o test.wav
3.2 实现CLI工具
让我们一步步实现这个工具。首先创建项目结构：
fish-speech-cli/
├── fish_tts/
│ ├── __init__.py
│ ├── cli.py # 命令行入口
│ ├── api_client.py # API客户端
│ ├── batch_processor.py # 批量处理器
│ └── utils.py # 工具函数
├── setup.py
├── requirements.txt
└── README.md
3.2.1 API客户端封装
首先，我们封装一个更健壮的API客户端：
# fish_tts/api_client.py
import requests
import json
import time
from typing import Optional, Dict, Any
from pathlib import Path
class FishSpeechClient:
"""Fish Speech API客户端"""
def __init__(self, base_url: str = "http://127.0.0.1:7861"):
"""
初始化客户端
参数:
base_url: API服务地址，默认本地7861端口
"""
self.base_url = base_url.rstrip('/')
self.api_url = f"{self.base_url}/v1/tts"
self.session = requests.Session()
# 设置超时时间
self.timeout = 30 # 30秒超时
def check_health(self) -> bool:
"""检查API服务是否健康"""
try:
response = self.session.get(f"{self.base_url}/docs", timeout=5)
return response.status_code == 200
except:
return False
def generate_speech(
self,
text: str,
output_path: Optional[str] = None,
reference_audio: Optional[str] = None,
max_new_tokens: int = 1024,
temperature: float = 0.7,
retry_times: int = 3
) -> Optional[bytes]:
"""
生成语音
参数:
text: 要合成的文本
output_path: 输出文件路径，如果为None则返回字节数据
reference_audio: 参考音频文件路径，用于音色克隆
max_new_tokens: 最大生成token数
temperature: 采样温度
retry_times: 失败重试次数
返回:
音频字节数据（如果output_path为None）
"""
# 构建请求数据
payload = {
"text": text,
"reference_id": None,
"max_new_tokens": max_new_tokens,
"temperature": temperature
}
# 如果有参考音频，需要特殊处理
# 注意：当前API可能需要通过文件上传方式传递参考音频
# 这里我们先实现基础版本
headers = {"Content-Type": "application/json"}
# 重试机制
for attempt in range(retry_times):
try:
print(f"生成语音中 (尝试 {attempt + 1}/{retry_times})...")
response = self.session.post(
self.api_url,
json=payload,
headers=headers,
timeout=self.timeout
)
if response.status_code == 200:
audio_data = response.content
# 如果指定了输出路径，保存文件
if output_path:
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "wb") as f:
f.write(audio_data)
print(f"✅ 语音已保存到: {output_path}")
return None
else:
return audio_data
elif response.status_code == 422:
print(f"❌ 参数错误: {response.text}")
break
else:
print(f"❌ 请求失败，状态码: {response.status_code}")
except requests.exceptions.Timeout:
print(f"⏱️ 请求超时 (尝试 {attempt + 1})")
if attempt < retry_times - 1:
time.sleep(1) # 等待1秒后重试
except requests.exceptions.ConnectionError:
print(f"🔌 连接失败 (尝试 {attempt + 1})")
if attempt < retry_times - 1:
time.sleep(2)
except Exception as e:
print(f"❌ 未知错误: {str(e)}")
break
print("❌ 所有重试均失败")
return None
def batch_generate(
self,
texts: list,
output_dir: str,
prefix: str = "speech",
**kwargs
) -> Dict[str, bool]:
"""
批量生成语音
参数:
texts: 文本列表
output_dir: 输出目录
prefix: 文件名前缀
**kwargs: 传递给generate_speech的其他参数
返回:
字典，键为文件名，值为是否成功
"""
results = {}
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
total = len(texts)
for i, text in enumerate(texts, 1):
# 清理文本，用于生成文件名
clean_text = text[:30].replace(" ", "_").replace("/", "_")
filename = f"{prefix}_{i:03d}_{clean_text}.wav"
filepath = output_path / filename
print(f"[{i}/{total}] 处理: {text[:50]}...")
success = self.generate_speech(
text=text,
output_path=str(filepath),
**kwargs
) is not None
results[str(filepath)] = success
# 添加短暂延迟，避免服务器压力过大
if i < total:
time.sleep(0.5)
return results
3.2.2 命令行接口实现
接下来实现命令行入口：
# fish_tts/cli.py
import argparse
import sys
import json
from pathlib import Path
from typing import List
from .api_client import FishSpeechClient
from .batch_processor import BatchProcessor
def main():
"""命令行主函数"""
parser = argparse.ArgumentParser(
description="Fish Speech TTS 命令行工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
使用示例:
# 生成单个语音
fish-tts generate "你好，世界" -o hello.wav
# 批量处理文本文件
fish-tts batch input.txt -o outputs/
# 从标准输入读取
echo "测试文本" | fish-tts generate -o test.wav
# 使用自定义API地址
fish-tts generate "Hello" --api-url http://192.168.1.100:7861
"""
)
# 全局参数
parser.add_argument(
"--api-url",
default="http://127.0.0.1:7861",
help="Fish Speech API地址 (默认: http://127.0.0.1:7861)"
)
# 子命令
subparsers = parser.add_subparsers(dest="command", help="可用命令")
# generate 命令
generate_parser = subparsers.add_parser("generate", help="生成单个语音")
generate_parser.add_argument("text", nargs="?", help="要合成的文本")
generate_parser.add_argument("-o", "--output", required=True, help="输出文件路径")
generate_parser.add_argument("-r", "--reference", help="参考音频文件路径")
generate_parser.add_argument("-t", "--tokens", type=int, default=1024,
help="最大token数 (默认: 1024)")
generate_parser.add_argument("--temperature", type=float, default=0.7,
help="采样温度 (默认: 0.7)")
# batch 命令
batch_parser = subparsers.add_parser("batch", help="批量处理")
batch_parser.add_argument("input", help="输入文件或目录")
batch_parser.add_argument("-o", "--output", required=True, help="输出目录")
batch_parser.add_argument("-p", "--prefix", default="speech",
help="文件名前缀 (默认: speech)")
batch_parser.add_argument("--format", choices=["txt", "json", "csv"],
default="txt", help="输入文件格式")
# health 命令
health_parser = subparsers.add_parser("health", help="检查API健康状态")
args = parser.parse_args()
# 创建客户端
client = FishSpeechClient(base_url=args.api_url)
# 处理不同命令
if args.command == "generate":
# 从参数或标准输入获取文本
if args.text:
text = args.text
elif not sys.stdin.isatty(): # 有标准输入
text = sys.stdin.read().strip()
else:
print("❌ 错误: 请提供文本或通过管道输入")
sys.exit(1)
if not text:
print("❌ 错误: 文本不能为空")
sys.exit(1)
print(f"📝 文本: {text}")
print(f"💾 输出: {args.output}")
success = client.generate_speech(
text=text,
output_path=args.output,
reference_audio=args.reference,
max_new_tokens=args.tokens,
temperature=args.temperature
) is not None
if success:
print("✅ 生成完成")
sys.exit(0)
else:
print("❌ 生成失败")
sys.exit(1)
elif args.command == "batch":
processor = BatchProcessor(client)
# 根据输入类型处理
input_path = Path(args.input)
if input_path.is_file():
results = processor.process_file(
str(input_path),
args.output,
args.prefix,
args.format
)
elif input_path.is_dir():
results = processor.process_directory(
str(input_path),
args.output,
args.prefix,
args.format
)
else:
print(f"❌ 错误: 输入路径不存在: {args.input}")
sys.exit(1)
# 统计结果
total = len(results)
success_count = sum(1 for success in results.values() if success)
print(f"\n📊 批量处理完成:")
print(f" 总计: {total} 个文件")
print(f" 成功: {success_count}")
print(f" 失败: {total - success_count}")
# 保存结果报告
report_path = Path(args.output) / "batch_report.json"
with open(report_path, "w", encoding="utf-8") as f:
json.dump({
"summary": {
"total": total,
"success": success_count,
"failed": total - success_count
},
"details": results
}, f, ensure_ascii=False, indent=2)
print(f"📋 详细报告已保存到: {report_path}")
elif args.command == "health":
if client.check_health():
print("✅ API服务运行正常")
sys.exit(0)
else:
print("❌ API服务不可用")
sys.exit(1)
else:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()
3.2.3 批量处理器实现
# fish_tts/batch_processor.py
import json
import csv
from pathlib import Path
from typing import List, Dict, Any
class BatchProcessor:
"""批量处理器"""
def __init__(self, client):
self.client = client
def read_text_file(self, filepath: str) -> List[str]:
"""读取文本文件，每行作为一个文本"""
with open(filepath, "r", encoding="utf-8") as f:
lines = [line.strip() for line in f if line.strip()]
return lines
def read_json_file(self, filepath: str) -> List[str]:
"""读取JSON文件"""
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
# 支持多种JSON格式
if isinstance(data, list):
# 如果是字符串列表
if all(isinstance(item, str) for item in data):
return data
# 如果是对象列表，尝试提取text字段
elif all(isinstance(item, dict) for item in data):
texts = []
for item in data:
if "text" in item:
texts.append(item["text"])
elif "content" in item:
texts.append(item["content"])
return texts
elif isinstance(data, dict):
# 如果是字典，尝试找texts或data字段
if "texts" in data:
return data["texts"]
elif "data" in data and isinstance(data["data"], list):
return [item.get("text", "") for item in data["data"]]
return []
def read_csv_file(self, filepath: str) -> List[str]:
"""读取CSV文件"""
texts = []
with open(filepath, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
# 尝试找text或content列
if "text" in row:
texts.append(row["text"])
elif "content" in row:
texts.append(row["content"])
elif "sentence" in row:
texts.append(row["sentence"])
return texts
def process_file(
self,
input_file: str,
output_dir: str,
prefix: str = "speech",
format: str = "txt"
) -> Dict[str, bool]:
"""处理单个文件"""
filepath = Path(input_file)
# 根据格式读取文本
if format == "txt":
texts = self.read_text_file(input_file)
elif format == "json":
texts = self.read_json_file(input_file)
elif format == "csv":
texts = self.read_csv_file(input_file)
else:
raise ValueError(f"不支持的格式: {format}")
print(f"📖 从 {filepath.name} 读取了 {len(texts)} 个文本")
# 批量生成
return self.client.batch_generate(
texts=texts,
output_dir=output_dir,
prefix=f"{prefix}_{filepath.stem}"
)
def process_directory(
self,
input_dir: str,
output_dir: str,
prefix: str = "speech",
format: str = "txt"
) -> Dict[str, bool]:
"""处理整个目录"""
input_path = Path(input_dir)
all_results = {}
# 根据格式查找文件
if format == "txt":
pattern = "*.txt"
elif format == "json":
pattern = "*.json"
elif format == "csv":
pattern = "*.csv"
else:
raise ValueError(f"不支持的格式: {format}")
files = list(input_path.glob(pattern))
print(f"📁 在 {input_dir} 中找到 {len(files)} 个 {format} 文件")
for i, filepath in enumerate(files, 1):
print(f"\n[{i}/{len(files)}] 处理文件: {filepath.name}")
# 为每个文件创建子目录
file_output_dir = Path(output_dir) / filepath.stem
file_output_dir.mkdir(parents=True, exist_ok=True)
# 处理文件
results = self.process_file(
str(filepath),
str(file_output_dir),
prefix,
format
)
all_results.update(results)
return all_results
3.2.4 安装脚本
最后，创建setup.py让用户可以安装这个工具：
# setup.py
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="fish-speech-cli",
version="1.0.0",
author="Fish Speech Developer",
author_email="developer@example.com",
description="Fish Speech TTS 命令行工具",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/yourusername/fish-speech-cli",
packages=find_packages(),
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
python_requires=">=3.8",
install_requires=[
"requests>=2.28.0",
"soundfile>=0.12.0",
],
entry_points={
"console_scripts": [
"fish-tts=fish_tts.cli:main",
],
},
)
创建requirements.txt：
requests>=2.28.0
soundfile>=0.12.0
argparse>=1.4.0
3.3 使用CLI工具
现在，让我们试试这个工具。首先安装：
# 进入项目目录
cd fish-speech-cli
# 安装到当前环境
pip install -e .
# 或者直接使用（不安装）
python -m fish_tts.cli --help
测试各种功能：
# 1. 检查API健康状态
fish-tts health
# 2. 生成单个语音
fish-tts generate "欢迎使用Fish Speech命令行工具" -o welcome.wav
# 3. 批量处理文本文件
# 先创建一个测试文件
echo -e "第一条测试语音\n这是第二条测试内容\n第三条测试语句" > test.txt
fish-tts batch test.txt -o batch_output/
# 4. 处理JSON文件
echo '[{"text": "JSON测试1"}, {"text": "JSON测试2"}]' > test.json
fish-tts batch test.json -o json_output/ --format json
# 5. 从管道输入
echo "管道输入测试" | fish-tts generate -o pipe.wav
4. 构建自动化测试框架
有了CLI工具，我们还需要确保它稳定可靠。自动化测试框架就是我们的"安全网"，它能自动发现潜在问题。
4.1 测试框架设计
我们的测试框架需要覆盖：
- 单元测试：测试单个函数是否正确
- 集成测试：测试整个流程是否正常
- 性能测试：测试响应时间和资源使用
- 回归测试：确保新版本不破坏旧功能
4.2 实现测试框架
创建测试目录结构：
tests/
├── __init__.py
├── conftest.py # 测试配置
├── test_unit/ # 单元测试
│ ├── __init__.py
│ ├── test_utils.py
│ └── test_validators.py
├── test_integration/ # 集成测试
│ ├── __init__.py
│ ├── test_api.py
│ └── test_cli.py
├── test_performance/ # 性能测试
│ ├── __init__.py
│ └── test_benchmark.py
└── test_regression/ # 回归测试
├── __init__.py
└── test_compatibility.py
4.2.1 测试配置
# tests/conftest.py
import pytest
import time
import sys
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# 测试配置
class TestConfig:
"""测试配置类"""
API_URL = "http://127.0.0.1:7861"
TEST_TEXT_SHORT = "测试文本"
TEST_TEXT_LONG = "这是一个较长的测试文本，用于测试模型处理长文本的能力。" * 5
TEST_OUTPUT_DIR = "test_outputs"
@classmethod
def ensure_output_dir(cls):
"""确保输出目录存在"""
Path(cls.TEST_OUTPUT_DIR).mkdir(exist_ok=True)
# 测试夹具
@pytest.fixture
def config():
"""提供测试配置"""
TestConfig.ensure_output_dir()
return TestConfig
@pytest.fixture
def api_client():
"""提供API客户端实例"""
from fish_tts.api_client import FishSpeechClient
client = FishSpeechClient(base_url=TestConfig.API_URL)
# 等待服务就绪
max_retries = 10
for i in range(max_retries):
if client.check_health():
break
if i < max_retries - 1:
time.sleep(2)
else:
pytest.skip("API服务未就绪，跳过测试")
return client
@pytest.fixture
def cleanup_output():
"""测试后清理输出文件"""
yield
import shutil
output_dir = Path(TestConfig.TEST_OUTPUT_DIR)
if output_dir.exists():
shutil.rmtree(output_dir)
4.2.2 单元测试
# tests/test_unit/test_utils.py
import pytest
from fish_tts.utils import validate_text, sanitize_filename
class TestUtils:
"""工具函数测试"""
def test_validate_text(self):
"""测试文本验证"""
# 正常文本
assert validate_text("正常文本") == "正常文本"
# 空文本
with pytest.raises(ValueError, match="文本不能为空"):
validate_text("")
# 过长文本
long_text = "a" * 10000
with pytest.raises(ValueError, match="文本过长"):
validate_text(long_text)
def test_sanitize_filename(self):
"""测试文件名清理"""
test_cases = [
("正常文件名", "正常文件名"),
("test/file.txt", "test_file.txt"),
("a*b?c:d", "a_b_c_d"),
(" 前后空格 ", "前后空格"),
("中英文混合 English", "中英文混合_English"),
]
for input_text, expected in test_cases:
result = sanitize_filename(input_text)
assert result == expected, f"输入: {input_text}, 期望: {expected}, 实际: {result}"
4.2.3 集成测试
# tests/test_integration/test_api.py
import pytest
from pathlib import Path
class TestAPIIntegration:
"""API集成测试"""
def test_api_health(self, api_client):
"""测试API健康检查"""
assert api_client.check_health() is True
def test_generate_speech_short(self, api_client, config, cleanup_output):
"""测试生成短文本语音"""
output_path = Path(config.TEST_OUTPUT_DIR) / "short_test.wav"
result = api_client.generate_speech(
text=config.TEST_TEXT_SHORT,
output_path=str(output_path)
)
assert result is None # 成功时返回None
assert output_path.exists()
assert output_path.stat().st_size > 1024 # 文件大小应大于1KB
def test_generate_speech_long(self, api_client, config, cleanup_output):
"""测试生成长文本语音"""
output_path = Path(config.TEST_OUTPUT_DIR) / "long_test.wav"
result = api_client.generate_speech(
text=config.TEST_TEXT_LONG,
output_path=str(output_path),
max_new_tokens=2048 # 增加token限制
)
assert result is None
assert output_path.exists()
def test_batch_generate(self, api_client, config, cleanup_output):
"""测试批量生成"""
texts = [
"批量测试第一条",
"批量测试第二条",
"批量测试第三条"
]
results = api_client.batch_generate(
texts=texts,
output_dir=config.TEST_OUTPUT_DIR,
prefix="batch_test"
)
assert len(results) == 3
# 检查文件是否生成
output_dir = Path(config.TEST_OUTPUT_DIR)
wav_files = list(output_dir.glob("*.wav"))
assert len(wav_files) == 3
# 统计成功数量
success_count = sum(1 for success in results.values() if success)
assert success_count >= 2 # 至少成功2个
@pytest.mark.slow
def test_concurrent_requests(self, api_client, config, cleanup_output):
"""测试并发请求（标记为慢速测试）"""
import concurrent.futures
texts = [f"并发测试{i}" for i in range(5)]
output_files = [
Path(config.TEST_OUTPUT_DIR) / f"concurrent_{i}.wav"
for i in range(5)
]
def generate_one(text, output_path):
return api_client.generate_speech(
text=text,
output_path=str(output_path)
)
# 使用线程池并发请求
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = []
for text, output_path in zip(texts, output_files):
future = executor.submit(generate_one, text, output_path)
futures.append(future)
# 等待所有任务完成
results = [future.result() for future in futures]
# 检查结果
success_count = sum(1 for result in results if result is None)
assert success_count >= 3 # 至少成功3个
4.2.4 CLI测试
# tests/test_integration/test_cli.py
import pytest
import subprocess
import json
from pathlib import Path
class TestCLIIntegration:
"""CLI集成测试"""
def test_cli_help(self):
"""测试帮助命令"""
result = subprocess.run(
["fish-tts", "--help"],
capture_output=True,
text=True
)
assert result.returncode == 0
assert "usage:" in result.stdout.lower()
assert "generate" in result.stdout
assert "batch" in result.stdout
def test_cli_generate(self, config, cleanup_output):
"""测试生成命令"""
output_path = Path(config.TEST_OUTPUT_DIR) / "cli_test.wav"
result = subprocess.run(
["fish-tts", "generate", config.TEST_TEXT_SHORT, "-o", str(output_path)],
capture_output=True,
text=True
)
print("STDOUT:", result.stdout)
print("STDERR:", result.stderr)
# 允许非零退出码（如果API未启动）
if result.returncode != 0:
pytest.skip("CLI命令执行失败，可能API未启动")
assert output_path.exists()
def test_cli_batch(self, config, cleanup_output):
"""测试批量命令"""
# 创建测试文件
test_file = Path(config.TEST_OUTPUT_DIR) / "test_input.txt"
test_file.write_text("第一行\n第二行\n第三行")
output_dir = Path(config.TEST_OUTPUT_DIR) / "batch_output"
result = subprocess.run(
["fish-tts", "batch", str(test_file), "-o", str(output_dir)],
capture_output=True,
text=True
)
if result.returncode != 0:
pytest.skip("批量命令执行失败")
# 检查输出
assert output_dir.exists()
wav_files = list(output_dir.glob("*.wav"))
assert len(wav_files) == 3
# 检查报告文件
report_file = output_dir / "batch_report.json"
assert report_file.exists()
with open(report_file, "r", encoding="utf-8") as f:
report = json.load(f)
assert report["summary"]["total"] == 3
assert report["summary"]["success"] >= 2
4.2.5 性能测试
# tests/test_performance/test_benchmark.py
import pytest
import time
import statistics
from pathlib import Path
class TestPerformance:
"""性能测试"""
@pytest.mark.benchmark
def test_response_time(self, api_client, config, cleanup_output):
"""测试响应时间"""
iterations = 5
response_times = []
for i in range(iterations):
output_path = Path(config.TEST_OUTPUT_DIR) / f"perf_test_{i}.wav"
start_time = time.time()
result = api_client.generate_speech(
text=f"性能测试迭代 {i}",
output_path=str(output_path)
)
end_time = time.time()
if result is None: # 成功
response_times.append(end_time - start_time)
# 短暂延迟
if i < iterations - 1:
time.sleep(1)
if response_times:
print(f"\n响应时间统计 (共{len(response_times)}次成功):")
print(f" 平均: {statistics.mean(response_times):.2f}秒")
print(f" 最小: {min(response_times):.2f}秒")
print(f" 最大: {max(response_times):.2f}秒")
print(f" 标准差: {statistics.stdev(response_times):.2f}秒")
# 断言平均响应时间小于10秒
assert statistics.mean(response_times) < 10.0
@pytest.mark.benchmark
def test_memory_usage(self, api_client, config, cleanup_output):
"""测试内存使用（近似）"""
import psutil
import os
# 获取初始内存
process = psutil.Process(os.getpid())
initial_memory = process.memory_info().rss / 1024 / 1024 # MB
# 执行多次生成
for i in range(3):
output_path = Path(config.TEST_OUTPUT_DIR) / f"memory_test_{i}.wav"
api_client.generate_speech(
text=f"内存测试 {i}",
output_path=str(output_path)
)
# 获取最终内存
final_memory = process.memory_info().rss / 1024 / 1024
memory_increase = final_memory - initial_memory
print(f"\n内存使用变化:")
print(f" 初始: {initial_memory:.1f} MB")
print(f" 最终: {final_memory:.1f} MB")
print(f" 增加: {memory_increase:.1f} MB")
# 断言内存增加不超过200MB
assert memory_increase < 200.0
4.2.6 回归测试
# tests/test_regression/test_compatibility.py
import pytest
import sys
from pathlib import Path
class TestRegression:
"""回归测试"""
def test_python_version_compatibility(self):
"""测试Python版本兼容性"""
python_version = sys.version_info
# 确保支持Python 3.8+
assert python_version.major == 3
assert python_version.minor >= 8
print(f"Python版本: {python_version.major}.{python_version.minor}.{python_version.micro}")
def test_import_compatibility(self):
"""测试导入兼容性"""
# 测试所有主要模块都能导入
import fish_tts
import fish_tts.cli
import fish_tts.api_client
import fish_tts.batch_processor
import fish_tts.utils
assert fish_tts.__version__ is not None
@pytest.mark.parametrize("text,expected_valid", [
("正常文本", True),
("", False), # 空文本
("a" * 10000, False), # 过长文本
("Hello 123", True), # 英文数字
("中文测试", True), # 中文
("🎉表情符号", True), # 包含表情符号
])
def test_text_validation_regression(self, text, expected_valid):
"""测试文本验证回归"""
from fish_tts.utils import validate_text
if expected_valid:
result = validate_text(text)
assert result == text.strip()
else:
with pytest.raises(ValueError):
validate_text(text)
4.3 运行测试
创建测试运行脚本：
# run_tests.sh
#!/bin/bash
echo "🐟 开始运行 Fish Speech 测试套件"
echo "=================================="
# 1. 运行单元测试
echo -e "\n📋 运行单元测试..."
pytest tests/test_unit/ -v --tb=short
# 2. 运行集成测试（需要API服务）
echo -e "\n🔗 运行集成测试..."
pytest tests/test_integration/ -v --tb=short
# 3. 运行性能测试（标记为benchmark）
echo -e "\n⚡ 运行性能测试..."
pytest tests/test_performance/ -v -m benchmark --tb=short
# 4. 运行回归测试
echo -e "\n🔄 运行回归测试..."
pytest tests/test_regression/ -v --tb=short
# 5. 生成测试报告
echo -e "\n📊 生成测试报告..."
pytest --junitxml=test-results.xml --cov=fish_tts --cov-report=html
echo -e "\n✅ 所有测试完成！"
echo "查看覆盖率报告: file://$(pwd)/htmlcov/index.html"
或者使用更简单的pytest命令：
# 运行所有测试
pytest tests/ -v
# 只运行快速测试（跳过慢速和性能测试）
pytest tests/ -v -m "not slow and not benchmark"
# 运行特定测试类
pytest tests/test_integration/test_api.py::TestAPIIntegration -v
# 生成HTML覆盖率报告
pytest --cov=fish_tts --cov-report=html
5. 实际应用案例
现在，让我们看看这个工具链在实际项目中怎么用。
5.1 案例一：有声电子书生成
假设你有一本小说，想把它转换成有声书：
# audiobook_generator.py
from fish_tts.api_client import FishSpeechClient
from pathlib import Path
import json
class AudiobookGenerator:
"""有声书生成器"""
def __init__(self, api_url="http://127.0.0.1:7861"):
self.client = FishSpeechClient(api_url)
def generate_from_novel(self, novel_path, output_dir, voice_settings=None):
"""
从小说文件生成有声书
参数:
novel_path: 小说文件路径
output_dir: 输出目录
voice_settings: 语音设置
"""
# 读取小说
with open(novel_path, "r", encoding="utf-8") as f:
content = f.read()
# 分割章节（简单按段落分割）
paragraphs = [p.strip() for p in content.split("\n\n") if p.strip()]
print(f"📚 小说《{Path(novel_path).stem}》")
print(f" 章节数: {len(paragraphs)}")
print(f" 总字数: {sum(len(p) for p in paragraphs)}")
# 生成语音
results = self.client.batch_generate(
texts=paragraphs,
output_dir=output_dir,
prefix="chapter",
max_new_tokens=2048, # 允许更长的文本
temperature=0.8 # 稍微增加随机性，让语音更自然
)
# 生成元数据
metadata = {
"title": Path(novel_path).stem,
"total_chapters": len(paragraphs),
"chapters": []
}
for i, (filepath, success) in enumerate(results.items()):
metadata["chapters"].append({
"index": i + 1,
"text_preview": paragraphs[i][:100] + "..." if len(paragraphs[i]) > 100 else paragraphs[i],
"audio_file": Path(filepath).name,
"success": success,
"text_length": len(paragraphs[i])
})
# 保存元数据
metadata_path = Path(output_dir) / "metadata.json"
with open(metadata_path, "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
print(f"✅ 有声书生成完成！")
print(f" 音频文件保存在: {output_dir}")
print(f" 元数据文件: {metadata_path}")
return metadata
# 使用示例
if __name__ == "__main__":
generator = AudiobookGenerator()
# 生成有声书
metadata = generator.generate_from_novel(
novel_path="novel.txt",
output_dir="audiobook_output"
)
# 统计结果
success_count = sum(1 for chapter in metadata["chapters"] if chapter["success"])
print(f"\n📊 生成统计:")
print(f" 总章节: {metadata['total_chapters']}")
print(f" 成功: {success_count}")
print(f" 失败: {metadata['total_chapters'] - success_count}")
5.2 案例二：多语言产品演示生成
如果你的产品需要支持多语言演示：
# multilingual_demo.py
from fish_tts.api_client import FishSpeechClient
from pathlib import Path
import json
class MultilingualDemoGenerator:
"""多语言演示生成器"""
# 多语言演示文本
DEMO_TEXTS = {
"zh": {
"welcome": "欢迎使用我们的产品，这是一个智能语音合成演示。",
"features": "我们的产品支持高质量语音合成，多种音色选择，以及实时生成功能。",
"closing": "感谢观看演示，如有任何问题，请随时联系我们。"
},
"en": {
"welcome": "Welcome to our product. This is a smart text-to-speech demo.",
"features": "Our product supports high-quality speech synthesis, multiple voice options, and real-time generation.",
"closing": "Thank you for watching the demo. If you have any questions, please feel free to contact us."
},
"ja": {
"welcome": "当社製品へようこそ。これはスマートな音声合成デモです。",
"features": "当社製品は高品質な音声合成、複数の音声オプション、リアルタイム生成をサポートしています。",
"closing": "デモをご覧いただきありがとうございます。ご質問がございましたら、お気軽にお問い合わせください。"
}
}
def __init__(self, api_url="http://127.0.0.1:7861"):
self.client = FishSpeechClient(api_url)
def generate_all_demos(self, output_dir="multilingual_demos"):
"""生成所有语言的演示"""
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
all_results = {}
for lang_code, texts in self.DEMO_TEXTS.items():
print(f"\n🌐 生成 {lang_code} 语言演示...")
lang_dir = output_path / lang_code
lang_dir.mkdir(exist_ok=True)
lang_results = {}
for text_key, text_content in texts.items():
filename = f"{text_key}.wav"
filepath = lang_dir / filename
print(f" 生成: {text_key}...")
success = self.client.generate_speech(
text=text_content,
output_path=str(filepath)
) is not None
lang_results[text_key] = {
"success": success,
"file": filename,
"text": text_content
}
# 短暂延迟
import time
time.sleep(0.5)
all_results[lang_code] = lang_results
# 生成索引文件
self._generate_index(output_path, all_results)
return all_results
def _generate_index(self, output_dir, results):
"""生成索引HTML文件"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>多语言语音演示</title>
<style>
body { font-family: Arial, sans-serif; margin: 40px; }
.language { margin-bottom: 30px; border: 1px solid #ddd; padding: 20px; border-radius: 8px; }
.language h2 { color: #333; border-bottom: 2px solid #4CAF50; padding-bottom: 10px; }
.demo-item { margin: 15px 0; padding: 15px; background: #f9f9f9; border-radius: 5px; }
audio { width: 100%; margin-top: 10px; }
.success { color: #4CAF50; }
.failed { color: #f44336; }
</style>
</head>
<body>
<h1>🎙️ 多语言语音演示</h1>
<p>使用 Fish Speech 1.5 生成的多语言产品演示</p>
"""
for lang_code, demos in results.items():
# 获取语言名称
lang_names = {"zh": "中文", "en": "English", "ja": "日本語"}
lang_name = lang_names.get(lang_code, lang_code)
html_content += f"""
<div class="language">
<h2>{lang_name} ({lang_code})</h2>
"""
for demo_key, demo_info in demos.items():
status_class = "success" if demo_info["success"] else "failed"
status_text = "✅ 成功" if demo_info["success"] else "❌ 失败"
html_content += f"""
<div class="demo-item">
<h3>{demo_key}</h3>
<p><strong>文本:</strong> {demo_info['text']}</p>
<p><strong>状态:</strong> <span class="{status_class}">{status_text}</span></p>
"""
if demo_info["success"]:
html_content += f"""
<audio controls>
<source src="{lang_code}/{demo_info['file']}" type="audio/wav">
您的浏览器不支持音频播放。
</audio>
"""
html_content += "</div>"
html_content += "</div>"
html_content += """
<div style="margin-top: 40px; padding: 20px; background: #e8f5e9; border-radius: 8px;">
<h3>📊 统计信息</h3>
<p>生成时间: <span id="generation-time"></span></p>
<p>总演示数: <span id="total-demos"></span></p>
<p>成功率: <span id="success-rate"></span></p>
</div>
<script>
// 计算统计信息
const demos = document.querySelectorAll('.demo-item');
const total = demos.length;
const success = document.querySelectorAll('.success').length;
const successRate = total > 0 ? Math.round((success / total) * 100) : 0;
document.getElementById('total-demos').textContent = total;
document.getElementById('success-rate').textContent = successRate + '%';
document.getElementById('generation-time').textContent = new Date().toLocaleString();
</script>
</body>
</html>
"""
index_path = output_dir / "index.html"
with open(index_path, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"\n📄 索引文件已生成: {index_path}")
# 使用示例
if __name__ == "__main__":
generator = MultilingualDemoGenerator()
results = generator.generate_all_demos()
# 打印统计
total_demos = sum(len(demos) for demos in results.values())
successful_demos = sum(
sum(1 for demo in demos.values() if demo["success"])
for demos in results.values()
)
print(f"\n📊 多语言演示生成完成:")
print(f" 总语言数: {len(results)}")
print(f" 总演示数: {total_demos}")
print(f" 成功数: {successful_demos}")
print(f" 成功率: {successful_demos/total_demos*100:.1f}%")
print(f" 打开 index.html 查看所有演示")
5.3 案例三：自动化测试流水线
最后，让我们创建一个完整的CI/CD流水线：
# .github/workflows/test.yml
name: Fish Speech CI
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
schedule:
- cron: '0 2 * * *' # 每天凌晨2点运行
jobs:
test:
runs-on: ubuntu-latest
services:
fish-speech:
image: your-registry/fish-speech:latest
ports:
- 7861:7861
options: >-
--gpus all
--shm-size=8g
env:
CUDA_VISIBLE_DEVICES: 0
steps:
- uses: actions/checkout@v3
- name: 设置Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: 安装依赖
run: |
pip install -e .
pip install pytest pytest-cov pytest-html
pip install requests soundfile
- name: 等待服务就绪
run: |
echo "等待Fish Speech服务启动..."
for i in {1..30}; do
if curl -s http://localhost:7861/docs > /dev/null; then
echo "✅ 服务已就绪"
break
fi
echo "等待服务... ($i/30)"
sleep 2
done
- name: 运行单元测试
run: |
pytest tests/test_unit/ -v --junitxml=unit-test-results.xml
- name: 运行集成测试
run: |
pytest tests/test_integration/ -v --junitxml=integration-test-results.xml
- name: 运行回归测试
run: |
pytest tests/test_regression/ -v --junitxml=regression-test-results.xml
- name: 生成测试报告
run: |
pytest --cov=fish_tts --cov-report=xml --cov-report=html
- name: 上传测试报告
uses: actions/upload-artifact@v3
with:
name: test-reports
path: |
*.xml
htmlcov/
- name: 上传覆盖率到Codecov
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
flags: unittests
name: codecov-umbrella
6. 总结
通过本文的实践，我们完成了一个完整的Fish Speech 1.5开发者工具链建设：
6.1 工具链核心价值
- 效率提升：从手动操作到自动化处理，批量生成效率提升10倍以上
- 可靠性保障：通过自动化测试框架，确保每次更新都不会破坏核心功能
- 易用性改善：命令行工具让非开发者也能轻松使用TTS能力
- 集成便利：标准的API接口和客户端库，方便集成到各种系统中
6.2 关键收获
- CLI工具设计：学会了如何设计用户友好的命令行接口，支持多种输入输出方式
- 测试框架构建：掌握了从单元测试到性能测试的完整测试策略
- 实际应用开发：通过有声书生成、多语言演示等案例，展示了工具链的实际价值
- CI/CD集成：了解了如何将测试自动化集成到开发流程中
6.3 下一步建议
-
扩展更多功能：
- 支持更多音频格式（MP3、OGG等）
- 添加语音效果处理（回声、变速、变调）
- 实现实时流式语音生成
-
优化性能：
- 添加缓存机制，避免重复生成相同内容
- 实现并行处理，充分利用多核CPU
- 添加GPU内存优化，支持更大批量处理
-
完善生态系统：
- 开发VS Code插件，在编辑器中直接生成语音
- 创建Webhook服务，支持事件驱动的语音生成
- 构建Docker镜像，一键部署完整工具链
-
社区贡献：
- 将工具开源，吸引更多开发者贡献
- 编写详细文档和教程
- 收集用户反馈，持续改进工具
6.4 开始行动
现在，你已经掌握了构建专业级TTS工具链的所有技能。无论是为自己的项目添加语音功能，还是为企业构建语音合成平台，这套工具链都能为你提供坚实的基础。
记住，好的工具不是一次建成的，而是在实际使用中不断迭代完善的。从今天开始，用这套工具链解决一个实际的语音合成问题，你会发现开发效率和质量都会有质的提升。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
2
0- 0
已为社区贡献110条内容
所有评论(0)