260 lines
8.1 KiB
Python
260 lines
8.1 KiB
Python
import os
|
|
import subprocess
|
|
import tempfile
|
|
import re
|
|
import logging
|
|
import glob
|
|
from pathlib import Path
|
|
from flask import Flask, request, jsonify
|
|
from werkzeug.utils import secure_filename
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = Flask(__name__)
|
|
|
|
# Configuration from environment variables
|
|
OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "output")
|
|
PORT = int(os.environ.get("PORT", 5012))
|
|
API_KEY = os.environ.get("TTS_API_KEY", "")
|
|
|
|
# Issue #10: Whitelist allowed TTS commands
|
|
ALLOWED_TTS_COMMANDS = {"kokoro-tts", "/usr/local/bin/kokoro-tts", "/usr/bin/kokoro-tts"}
|
|
_tts_command = os.environ.get("TTS_COMMAND", "kokoro-tts")
|
|
if _tts_command not in ALLOWED_TTS_COMMANDS:
|
|
logger.warning(
|
|
f"TTS_COMMAND '{_tts_command}' not in whitelist, using default 'kokoro-tts'"
|
|
)
|
|
TTS_COMMAND = "kokoro-tts"
|
|
else:
|
|
TTS_COMMAND = _tts_command
|
|
|
|
# Issue #12: Maximum number of output files to keep per title
|
|
MAX_OUTPUT_FILES_PER_TITLE = 5
|
|
|
|
# Ensure output directory exists
|
|
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
|
|
|
# Available voices
|
|
VOICES = ["bm_fable", "bm_lewis", "bm_george"]
|
|
|
|
|
|
def parse_speaker_text(text):
|
|
"""Parse text into speaker: complete lines pairs for full conversation"""
|
|
lines = [line.strip() for line in text.strip().split("\n") if line.strip()]
|
|
|
|
speakers = {}
|
|
current_speaker = None
|
|
speaker_content = []
|
|
tracks = []
|
|
|
|
for line in lines:
|
|
match = re.match(r"^(.*?): (.*)$", line)
|
|
if match:
|
|
if current_speaker and speaker_content:
|
|
speakers[current_speaker] = "\n".join(speaker_content)
|
|
|
|
current_speaker = match.group(1).strip()
|
|
speaker_content = [match.group(2).strip()]
|
|
tracks += [(current_speaker, speaker_content[0])]
|
|
else:
|
|
if current_speaker and line.strip():
|
|
speaker_content.append(line.strip())
|
|
|
|
if current_speaker and speaker_content:
|
|
speakers[current_speaker] = "\n".join(speaker_content)
|
|
|
|
return (speakers, tracks)
|
|
|
|
|
|
def assign_voices_to_speakers(speakers):
|
|
"""Assign voices to speakers (same voice for each speaker throughout conversation)"""
|
|
speaker_voice_map = {}
|
|
used_voices = set()
|
|
|
|
for speaker in speakers.keys():
|
|
if not speaker_voice_map.get(speaker):
|
|
available_voices = [voice for voice in VOICES if voice not in used_voices]
|
|
if available_voices:
|
|
selected_voice = available_voices[0]
|
|
else:
|
|
selected_voice = VOICES[0]
|
|
|
|
speaker_voice_map[speaker] = selected_voice
|
|
used_voices.add(selected_voice)
|
|
|
|
return speaker_voice_map
|
|
|
|
|
|
def clean_old_outputs(title):
|
|
"""Issue #12: Remove old output files for a given title, keeping only the newest."""
|
|
pattern = os.path.join(OUTPUT_DIR, f"{title}_*.wav")
|
|
files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True)
|
|
for old_file in files[MAX_OUTPUT_FILES_PER_TITLE:]:
|
|
try:
|
|
os.unlink(old_file)
|
|
logger.info(f"Removed old output file: {old_file}")
|
|
except OSError as e:
|
|
logger.warning(f"Failed to remove old output file {old_file}: {e}")
|
|
|
|
|
|
def generate_wav_files(speakers, speaker_voice_map, title, tracks):
|
|
"""Generate individual WAV files for each speaker"""
|
|
wav_files = []
|
|
turn = 0
|
|
|
|
for speaker, text in tracks:
|
|
safe_speaker = secure_filename(speaker)
|
|
filepath = os.path.join(OUTPUT_DIR, f"{title}_{turn}.wav")
|
|
turn += 1
|
|
|
|
temp_input_path = None
|
|
try:
|
|
# Issue #11: Create temporary input file with proper cleanup
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".txt", delete=False
|
|
) as temp_file:
|
|
temp_file.write(text)
|
|
temp_input_path = temp_file.name
|
|
|
|
cmd = [
|
|
TTS_COMMAND,
|
|
temp_input_path,
|
|
filepath,
|
|
"--voice",
|
|
speaker_voice_map[speaker],
|
|
]
|
|
subprocess.run(cmd, check=True, capture_output=True)
|
|
|
|
wav_files.append(filepath)
|
|
except Exception as e:
|
|
logger.error(f"Error generating audio for {speaker}: {e}")
|
|
return None
|
|
finally:
|
|
# Issue #11: Always clean up temp file, using specific exception type
|
|
if temp_input_path and os.path.exists(temp_input_path):
|
|
try:
|
|
os.unlink(temp_input_path)
|
|
except OSError as cleanup_err:
|
|
logger.warning(
|
|
f"Failed to clean up temp file {temp_input_path}: {cleanup_err}"
|
|
)
|
|
|
|
return wav_files
|
|
|
|
|
|
def merge_wav_files(wav_files, final_output_path):
|
|
"""Merge multiple WAV files into a single file using sox"""
|
|
try:
|
|
if not wav_files:
|
|
return None
|
|
|
|
if len(wav_files) == 1:
|
|
import shutil
|
|
|
|
shutil.copy2(wav_files[0], final_output_path)
|
|
return final_output_path
|
|
|
|
cmd = ["sox"] + wav_files + [final_output_path]
|
|
subprocess.run(cmd, check=True, capture_output=True)
|
|
|
|
# Clean up intermediate files
|
|
for wav_file in wav_files:
|
|
try:
|
|
os.unlink(wav_file)
|
|
except OSError as e:
|
|
logger.warning(f"Failed to remove intermediate file {wav_file}: {e}")
|
|
|
|
return final_output_path
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error merging WAV files: {e}")
|
|
if wav_files and len(wav_files) > 0:
|
|
import shutil
|
|
|
|
try:
|
|
shutil.copy2(wav_files[-1], final_output_path)
|
|
return final_output_path
|
|
except OSError as fallback_err:
|
|
logger.error(
|
|
f"Fallback copy failed for {wav_files[-1]}: {fallback_err}"
|
|
)
|
|
return None
|
|
|
|
|
|
def check_api_key():
|
|
"""Issue #9: Validate API key from Authorization header."""
|
|
if not API_KEY:
|
|
return True # No key configured = skip auth (dev mode)
|
|
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
return False
|
|
provided_key = auth_header[len("Bearer "):]
|
|
return provided_key == API_KEY
|
|
|
|
|
|
@app.route("/tts", methods=["POST"])
|
|
def text_to_speech():
|
|
"""Endpoint to convert full conversation to audio podcast"""
|
|
# Issue #9: API key authentication
|
|
if not check_api_key():
|
|
return jsonify({"error": "Unauthorized. Provide valid API key."}), 401
|
|
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return jsonify({"error": "No JSON data provided"}), 400
|
|
|
|
text = data.get("text")
|
|
title = data.get("title")
|
|
|
|
if not text or not title:
|
|
return jsonify({"error": "Both 'text' and 'title' fields are required"}), 400
|
|
|
|
# Sanitize title to prevent path traversal
|
|
safe_title = secure_filename(title)
|
|
if not safe_title:
|
|
return jsonify({"error": "Invalid title format"}), 400
|
|
|
|
(speakers, tracks) = parse_speaker_text(text)
|
|
|
|
if not speakers:
|
|
return jsonify({"error": "No speaker lines found in text"}), 400
|
|
|
|
speaker_voice_map = assign_voices_to_speakers(speakers)
|
|
|
|
# Issue #12: Clean old outputs before generating new ones
|
|
clean_old_outputs(safe_title)
|
|
|
|
wav_files = generate_wav_files(speakers, speaker_voice_map, safe_title, tracks)
|
|
|
|
if not wav_files:
|
|
return jsonify({"error": "Failed to generate audio files"}), 500
|
|
|
|
final_wav_path = os.path.join(OUTPUT_DIR, f"{safe_title}.wav")
|
|
|
|
merged_file = merge_wav_files(wav_files, final_wav_path)
|
|
if not merged_file:
|
|
return jsonify({"error": "Failed to merge audio files"}), 500
|
|
|
|
return jsonify(
|
|
{
|
|
"file_path": final_wav_path,
|
|
"message": "Full conversation podcast generated successfully",
|
|
"speakers": list(speakers.keys()),
|
|
"total_speakers": len(speakers),
|
|
"total_lines": len(text.split("\n")),
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/health", methods=["GET"])
|
|
def health_check():
|
|
"""Health check endpoint"""
|
|
return jsonify({"status": "healthy"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=PORT, debug=False)
|