Christian Krakau-Louis
304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""
|
|
General utility functions used throughout the application.
|
|
"""
|
|
import secrets
|
|
import string
|
|
import os
|
|
import shutil
|
|
from flask import current_app
|
|
from werkzeug.utils import secure_filename
|
|
import uuid
|
|
import requests
|
|
import json
|
|
|
|
def generate_token(length=32):
|
|
"""
|
|
Generate a secure random token for authentication or validation purposes.
|
|
|
|
Args:
|
|
length (int): The length of the token to generate (default: 32)
|
|
|
|
Returns:
|
|
str: A secure random token string
|
|
"""
|
|
# Use secrets module for cryptographically strong random numbers
|
|
alphabet = string.ascii_letters + string.digits
|
|
token = ''.join(secrets.choice(alphabet) for _ in range(length))
|
|
return token
|
|
|
|
def get_user_mp3_directory(username):
|
|
"""
|
|
Get the directory path for a user's custom MP3 files
|
|
Creates the directory if it doesn't exist
|
|
|
|
Args:
|
|
username (str): Username
|
|
|
|
Returns:
|
|
str: Path to the user's MP3 directory
|
|
"""
|
|
base_dir = os.path.join('/data', 'custommp3')
|
|
user_dir = os.path.join(base_dir, secure_filename(username))
|
|
|
|
# Create directories if they don't exist
|
|
if not os.path.exists(base_dir):
|
|
os.makedirs(base_dir)
|
|
if not os.path.exists(user_dir):
|
|
os.makedirs(user_dir)
|
|
|
|
return user_dir
|
|
|
|
def save_user_mp3(file, username, mp3_type):
|
|
"""
|
|
Save a user's uploaded MP3 file
|
|
|
|
Args:
|
|
file: FileStorage object from Flask request.files
|
|
username (str): Username
|
|
mp3_type (str): Type of MP3 (intro, outro, or replay)
|
|
|
|
Returns:
|
|
str: Path to the saved file relative to data directory
|
|
"""
|
|
if not file or not allowed_file(file.filename):
|
|
return None
|
|
|
|
# Get user directory and create if needed
|
|
user_dir = get_user_mp3_directory(username)
|
|
|
|
# Generate a unique filename to avoid overwriting
|
|
original_filename = secure_filename(file.filename)
|
|
file_extension = os.path.splitext(original_filename)[1] # Should be .mp3
|
|
unique_filename = f"{mp3_type}{file_extension}"
|
|
|
|
# Full path to save the file
|
|
filepath = os.path.join(user_dir, unique_filename)
|
|
|
|
# Save the file
|
|
file.save(filepath)
|
|
|
|
# Return the relative path for database storage
|
|
return os.path.join('custommp3', secure_filename(username), unique_filename)
|
|
|
|
def allowed_file(filename):
|
|
"""
|
|
Check if the file has an allowed extension
|
|
|
|
Args:
|
|
filename (str): Name of the file
|
|
|
|
Returns:
|
|
bool: True if extension is allowed, False otherwise
|
|
"""
|
|
allowed_extensions = {'mp3'}
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions
|
|
|
|
def get_mp3_path(user, mp3_type):
|
|
"""
|
|
Get the path to a user's custom MP3 file or the default if not set
|
|
|
|
Args:
|
|
user: User object from database
|
|
mp3_type (str): Type of MP3 (intro, outro, or replay)
|
|
|
|
Returns:
|
|
str: Path to the MP3 file
|
|
"""
|
|
user_setting_attr = f"{mp3_type}_mp3"
|
|
user_mp3_path = getattr(user, user_setting_attr)
|
|
|
|
if user_mp3_path and os.path.exists(os.path.join('/data', user_mp3_path)):
|
|
return os.path.join('/data', user_mp3_path)
|
|
|
|
# Fall back to the default MP3
|
|
return os.path.join(current_app.root_path, 'static', 'audio', f'{mp3_type}.mp3')
|
|
|
|
def get_available_voices(service='polly'):
|
|
"""
|
|
Get a list of available voices for the specified TTS service
|
|
|
|
Args:
|
|
service (str): TTS service ('polly', 'openai', or 'elevenlabs')
|
|
|
|
Returns:
|
|
list: List of available voice options
|
|
"""
|
|
if service == 'polly':
|
|
# Standard AWS Polly voices - subset of most natural ones
|
|
return [
|
|
{'id': 'Joanna', 'name': 'Joanna (Female, US)', 'gender': 'Female', 'language': 'en-US'},
|
|
{'id': 'Matthew', 'name': 'Matthew (Male, US)', 'gender': 'Male', 'language': 'en-US'},
|
|
{'id': 'Amy', 'name': 'Amy (Female, UK)', 'gender': 'Female', 'language': 'en-GB'},
|
|
{'id': 'Brian', 'name': 'Brian (Male, UK)', 'gender': 'Male', 'language': 'en-GB'},
|
|
{'id': 'Kendra', 'name': 'Kendra (Female, US)', 'gender': 'Female', 'language': 'en-US'},
|
|
{'id': 'Kimberly', 'name': 'Kimberly (Female, US)', 'gender': 'Female', 'language': 'en-US'},
|
|
{'id': 'Salli', 'name': 'Salli (Female, US)', 'gender': 'Female', 'language': 'en-US'},
|
|
{'id': 'Joey', 'name': 'Joey (Male, US)', 'gender': 'Male', 'language': 'en-US'},
|
|
]
|
|
elif service == 'openai':
|
|
return [
|
|
{'id': 'alloy', 'name': 'Alloy (Neutral)', 'gender': 'Neutral', 'language': 'en'},
|
|
{'id': 'echo', 'name': 'Echo (Male)', 'gender': 'Male', 'language': 'en'},
|
|
{'id': 'fable', 'name': 'Fable (Male)', 'gender': 'Male', 'language': 'en'},
|
|
{'id': 'onyx', 'name': 'Onyx (Male)', 'gender': 'Male', 'language': 'en'},
|
|
{'id': 'nova', 'name': 'Nova (Female)', 'gender': 'Female', 'language': 'en'},
|
|
{'id': 'shimmer', 'name': 'Shimmer (Female)', 'gender': 'Female', 'language': 'en'},
|
|
]
|
|
elif service == 'elevenlabs':
|
|
# Check if we have a valid API key
|
|
api_key = current_app.config.get('ELEVENLABS_API_KEY')
|
|
if not api_key:
|
|
return []
|
|
|
|
try:
|
|
# Call the ElevenLabs API to get available voices
|
|
headers = {
|
|
"xi-api-key": api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
voices_data = response.json()
|
|
voices = []
|
|
for voice in voices_data.get('voices', []):
|
|
voice_info = {
|
|
'id': voice.get('voice_id'),
|
|
'name': voice.get('name', 'Unknown'),
|
|
'gender': voice.get('labels', {}).get('gender', 'Unknown'),
|
|
'language': 'en' # Default to English
|
|
}
|
|
voices.append(voice_info)
|
|
return voices
|
|
else:
|
|
current_app.logger.error(f"Error fetching ElevenLabs voices: {response.status_code} - {response.text}")
|
|
return []
|
|
except Exception as e:
|
|
current_app.logger.error(f"Exception fetching ElevenLabs voices: {str(e)}")
|
|
return []
|
|
|
|
# Default fallback
|
|
return []
|
|
|
|
def generate_tts_mp3(text, username, mp3_type, service='polly', voice=None, model=None, stability=None, similarity=None):
|
|
"""
|
|
Generate a text-to-speech MP3 file
|
|
|
|
Args:
|
|
text (str): The text to convert to speech
|
|
username (str): Username
|
|
mp3_type (str): Type of MP3 (intro, outro, or replay)
|
|
service (str): TTS service to use ('polly', 'openai', or 'elevenlabs')
|
|
voice (str): Voice ID to use (defaults to service-specific default if None)
|
|
model (str): Model to use for OpenAI or ElevenLabs (defaults to service-specific default if None)
|
|
stability (float): Voice stability parameter for ElevenLabs (0.0-1.0)
|
|
similarity (float): Voice similarity parameter for ElevenLabs (0.0-1.0)
|
|
|
|
Returns:
|
|
str: Path to the generated file relative to data directory
|
|
"""
|
|
try:
|
|
user_dir = get_user_mp3_directory(username)
|
|
output_filename = f"{mp3_type}.mp3"
|
|
output_path = os.path.join(user_dir, output_filename)
|
|
|
|
if service == 'polly':
|
|
# AWS Polly implementation
|
|
import boto3
|
|
|
|
polly_client = boto3.client('polly',
|
|
region_name=current_app.config.get('AWS_REGION', 'us-east-1'),
|
|
aws_access_key_id=current_app.config.get('AWS_ACCESS_KEY_ID'),
|
|
aws_secret_access_key=current_app.config.get('AWS_SECRET_ACCESS_KEY')
|
|
)
|
|
|
|
# Use provided voice or default
|
|
voice_id = voice or current_app.config.get('AWS_POLLY_VOICE', 'Joanna')
|
|
|
|
# Use neural engine if available
|
|
engine = 'neural' if voice_id in ['Joanna', 'Matthew', 'Amy', 'Emma', 'Brian', 'Kendra'] else 'standard'
|
|
|
|
response = polly_client.synthesize_speech(
|
|
Text=text,
|
|
OutputFormat='mp3',
|
|
VoiceId=voice_id,
|
|
Engine=engine
|
|
)
|
|
|
|
# Write the audio stream to a file
|
|
if "AudioStream" in response:
|
|
with open(output_path, 'wb') as file:
|
|
file.write(response["AudioStream"].read())
|
|
|
|
current_app.logger.info(f"Generated AWS Polly TTS with voice {voice_id} for {username}/{mp3_type}")
|
|
|
|
elif service == 'openai':
|
|
# OpenAI TTS implementation
|
|
import openai
|
|
|
|
openai.api_key = current_app.config.get('OPENAI_API_KEY')
|
|
openai.base_url = current_app.config.get('OPENAI_URL', 'https://api.openai.com/v1')
|
|
|
|
# Use provided voice or default
|
|
voice_id = voice or 'alloy'
|
|
# Use provided model or default
|
|
tts_model = model or 'tts-1' # Options: tts-1, tts-1-hd
|
|
|
|
response = openai.audio.speech.create(
|
|
model=tts_model,
|
|
voice=voice_id,
|
|
input=text
|
|
)
|
|
|
|
response.stream_to_file(output_path)
|
|
current_app.logger.info(f"Generated OpenAI TTS with voice {voice_id} for {username}/{mp3_type}")
|
|
|
|
elif service == 'elevenlabs':
|
|
# ElevenLabs implementation
|
|
api_key = current_app.config.get('ELEVENLABS_API_KEY')
|
|
if not api_key:
|
|
current_app.logger.error("ElevenLabs API key not configured")
|
|
return None
|
|
|
|
# Use provided voice or default
|
|
voice_id = voice or "21m00Tcm4TlvDq8ikWAM" # Default to "Rachel" voice
|
|
|
|
# Set default values for stability and similarity if not provided
|
|
stability_value = stability if stability is not None else 0.5
|
|
similarity_value = similarity if similarity is not None else 0.75
|
|
|
|
headers = {
|
|
"xi-api-key": api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
payload = {
|
|
"text": text,
|
|
"model_id": model or "eleven_monolingual_v1",
|
|
"voice_settings": {
|
|
"stability": stability_value,
|
|
"similarity_boost": similarity_value
|
|
}
|
|
}
|
|
|
|
response = requests.post(
|
|
f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}",
|
|
headers=headers,
|
|
data=json.dumps(payload)
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
with open(output_path, 'wb') as file:
|
|
file.write(response.content)
|
|
current_app.logger.info(f"Generated ElevenLabs TTS with voice {voice_id} for {username}/{mp3_type}")
|
|
else:
|
|
current_app.logger.error(f"ElevenLabs API error: {response.status_code} - {response.text}")
|
|
return None
|
|
|
|
# Return the relative path for database storage
|
|
return os.path.join('custommp3', secure_filename(username), output_filename)
|
|
|
|
except Exception as e:
|
|
current_app.logger.error(f"Error generating TTS MP3: {str(e)}")
|
|
return None
|