feat: 集成Piper自建TTS并增加朗读回退
This commit is contained in:
@@ -11,6 +11,7 @@ from server.auth import admin_required, authenticate, create_user, login_require
|
||||
from server.db import BASE_DIR, get_db, init_db
|
||||
from server.entries_api import entries_bp
|
||||
from server.settings_api import settings_bp
|
||||
from server.tts_api import tts_bp
|
||||
from server.vocab_api import vocab_bp
|
||||
|
||||
|
||||
@@ -28,6 +29,7 @@ def create_app():
|
||||
app.register_blueprint(entries_bp)
|
||||
app.register_blueprint(vocab_bp)
|
||||
app.register_blueprint(settings_bp)
|
||||
app.register_blueprint(tts_bp)
|
||||
|
||||
@app.post("/api/register")
|
||||
def register():
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
from flask import Blueprint, jsonify, request, send_file
|
||||
|
||||
from server.auth import login_required
|
||||
from server.tts_service import generate_speech, piper_available
|
||||
|
||||
tts_bp = Blueprint("tts", __name__)
|
||||
|
||||
|
||||
@tts_bp.route("/api/tts", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def tts_audio(user):
|
||||
if request.method == "POST":
|
||||
data = request.get_json(silent=True) or {}
|
||||
text = (data.get("text") or "").strip()
|
||||
lang = (data.get("lang") or "es").strip().lower()
|
||||
else:
|
||||
text = (request.args.get("text") or "").strip()
|
||||
lang = (request.args.get("lang") or "es").strip().lower()
|
||||
|
||||
if not text:
|
||||
return jsonify({"error": "朗读内容不能为空"}), 400
|
||||
if len(text) > 1000:
|
||||
return jsonify({"error": "朗读内容不能超过 1000 个字符"}), 400
|
||||
if lang not in ("es", "en", "zh"):
|
||||
return jsonify({"error": "不支持的朗读语言"}), 400
|
||||
if not piper_available():
|
||||
return jsonify({"error": "Piper 未配置或未安装"}), 503
|
||||
|
||||
try:
|
||||
audio_path = generate_speech(text, lang)
|
||||
except Exception as exc:
|
||||
return jsonify({"error": "Piper 生成失败: " + str(exc)}), 500
|
||||
|
||||
return send_file(
|
||||
audio_path,
|
||||
mimetype="audio/wav",
|
||||
as_attachment=False,
|
||||
download_name="piper.wav",
|
||||
conditional=True,
|
||||
)
|
||||
@@ -0,0 +1,59 @@
|
||||
import hashlib
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
SERVER_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
DEFAULT_CACHE_DIR = os.path.join(SERVER_DIR, "data", "tts_cache")
|
||||
|
||||
PIPER_BIN = os.environ.get("PIPER_BIN", "/opt/piper/venv/bin/piper")
|
||||
PIPER_MODELS_DIR = os.environ.get("PIPER_MODELS_DIR", "/opt/piper/voices")
|
||||
TTS_CACHE_DIR = os.environ.get("TTS_CACHE_DIR", DEFAULT_CACHE_DIR)
|
||||
|
||||
VOICES = {
|
||||
"es": "es_ES-davefx-medium.onnx",
|
||||
"en": "en_US-lessac-medium.onnx",
|
||||
"zh": "zh_CN-huayan-medium.onnx",
|
||||
}
|
||||
|
||||
|
||||
def piper_available():
|
||||
return bool(PIPER_BIN) and os.path.isfile(PIPER_BIN)
|
||||
|
||||
|
||||
def generate_speech(text, lang="es"):
|
||||
voice = VOICES.get(lang, VOICES["en"])
|
||||
digest = hashlib.sha256((lang + "\0" + text).encode("utf-8")).hexdigest()[:24]
|
||||
cache_path = os.path.join(TTS_CACHE_DIR, f"{lang}_{digest}.wav")
|
||||
if os.path.isfile(cache_path):
|
||||
return cache_path
|
||||
|
||||
model_path = os.path.join(PIPER_MODELS_DIR, voice)
|
||||
if not os.path.isfile(model_path):
|
||||
raise RuntimeError(
|
||||
f"Piper 模型不存在: {model_path},请先下载语音模型并配置 PIPER_MODELS_DIR"
|
||||
)
|
||||
|
||||
os.makedirs(TTS_CACHE_DIR, exist_ok=True)
|
||||
tmp_path = cache_path + ".tmp"
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
[
|
||||
PIPER_BIN,
|
||||
"--model",
|
||||
model_path,
|
||||
"--output_file",
|
||||
tmp_path,
|
||||
],
|
||||
input=text.encode("utf-8"),
|
||||
capture_output=True,
|
||||
timeout=90,
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise RuntimeError("Piper 生成超时") from exc
|
||||
|
||||
if proc.returncode != 0 or not os.path.isfile(tmp_path):
|
||||
stderr = proc.stderr.decode("utf-8", errors="replace")
|
||||
raise RuntimeError(stderr or "Piper 生成失败")
|
||||
|
||||
os.replace(tmp_path, cache_path)
|
||||
return cache_path
|
||||
Reference in New Issue
Block a user