86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
from flask import Flask, request, Response
|
|
import subprocess
|
|
import os
|
|
import time
|
|
import random
|
|
import re
|
|
from datetime import datetime
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
|
|
app = Flask(__name__)
|
|
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
|
|
|
|
# Configuration
|
|
MAX_TEXT_LENGTH = 1000
|
|
TEXT_DIR = 'data/texts'
|
|
AUDIO_DIR = 'data/audio'
|
|
|
|
# Ensure directories exist
|
|
os.makedirs(TEXT_DIR, exist_ok=True)
|
|
os.makedirs(AUDIO_DIR, exist_ok=True)
|
|
|
|
def sanitize_text(text):
|
|
"""Remove potentially dangerous characters"""
|
|
return re.sub(r'[;$`|]', '', text)
|
|
|
|
def generate_filename():
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
random_num = random.randint(1000, 9999)
|
|
return f"{timestamp}_{random_num}"
|
|
|
|
@app.route('/tts', methods=['POST'])
|
|
def tts():
|
|
if not request.is_json:
|
|
return {"error": "Request must be JSON"}, 400
|
|
|
|
text = sanitize_text(request.json.get('text', '').strip())
|
|
|
|
if not text:
|
|
return {"error": "No text provided"}, 400
|
|
if len(text) > MAX_TEXT_LENGTH:
|
|
return {"error": f"Text too long (max {MAX_TEXT_LENGTH} characters)"}, 400
|
|
|
|
base_filename = generate_filename()
|
|
text_filename = os.path.join(TEXT_DIR, f"{base_filename}.txt")
|
|
wav_filename = os.path.join(AUDIO_DIR, f"{base_filename}.wav")
|
|
|
|
try:
|
|
# Save input text
|
|
with open(text_filename, 'w', encoding='utf-8') as f:
|
|
f.write(text)
|
|
|
|
# SAFE Piper execution (no shell=True)
|
|
piper_cmd = [
|
|
'./piper/piper',
|
|
'--model', './model/en_US-amy-medium.onnx',
|
|
'--output_file', wav_filename
|
|
]
|
|
|
|
process = subprocess.run(
|
|
piper_cmd,
|
|
input=text.encode('utf-8'),
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE
|
|
)
|
|
|
|
if not os.path.exists(wav_filename):
|
|
raise Exception("Audio file not created")
|
|
|
|
with open(wav_filename, 'rb') as f:
|
|
return Response(
|
|
f.read(),
|
|
mimetype='audio/wav',
|
|
headers={'Content-Disposition': f'attachment; filename={base_filename}.wav'}
|
|
)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
app.logger.error(f"Piper failed: {e.stderr.decode()}")
|
|
return {"error": "TTS generation failed"}, 500
|
|
except Exception as e:
|
|
app.logger.error(f"Unexpected error: {str(e)}")
|
|
return {"error": "Processing failed"}, 500
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=4005) # Remove debug=True for production
|