feat: initial Flask REST API implementation for TTS + Voice Conversion
This commit is contained in:
root
55
app/__init__.py
Normal file
55
app/__init__.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""Flask application factory."""
|
||||
import os
|
||||
import logging
|
||||
from flask import Flask
|
||||
from flask_restx import Api
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables
|
||||
load_dotenv()
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_app():
|
||||
"""Create and configure Flask application."""
|
||||
app = Flask(__name__)
|
||||
|
||||
# Configuration
|
||||
app.config['VOICES_DIR'] = os.getenv('VOICES_DIR', './voices')
|
||||
app.config['OUTPUT_DIR'] = os.getenv('OUTPUT_DIR', './output')
|
||||
app.config['API_TOKENS'] = os.getenv('API_TOKENS', '').split(',')
|
||||
app.config['SEED_VC_MODEL'] = os.getenv('SEED_VC_MODEL', 'seed-uvit-whisper-small-wavenet')
|
||||
app.config['SEED_VC_REPO_PATH'] = os.getenv('SEED_VC_REPO_PATH', './Seed-VC')
|
||||
app.config['EDGE_TTS_VOICE'] = os.getenv('EDGE_TTS_VOICE', 'ru-RU-SvetlanaNeural')
|
||||
|
||||
# Ensure directories exist
|
||||
os.makedirs(app.config['VOICES_DIR'], exist_ok=True)
|
||||
os.makedirs(app.config['OUTPUT_DIR'], exist_ok=True)
|
||||
|
||||
# Create and initialize API
|
||||
api = Api(
|
||||
version='1.0',
|
||||
title='TTS-VC API',
|
||||
description='Text-to-Speech with Voice Conversion API',
|
||||
doc='/swagger-ui',
|
||||
validate=True
|
||||
)
|
||||
api.init_app(app)
|
||||
|
||||
# Register namespaces
|
||||
from app.api import tts as tts_module
|
||||
from app.api import voices as voices_module
|
||||
from app.api import health as health_module
|
||||
|
||||
api.add_namespace(tts_module.api, path='/tts')
|
||||
api.add_namespace(voices_module.api, path='/voices')
|
||||
api.add_namespace(health_module.api, path='/health')
|
||||
|
||||
logger.info("Flask app created successfully")
|
||||
return app
|
||||
1
app/api/__init__.py
Normal file
1
app/api/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""API package."""
|
||||
22
app/api/health.py
Normal file
22
app/api/health.py
Normal file
@@ -0,0 +1,22 @@
|
||||
"""Health check endpoint."""
|
||||
import logging
|
||||
from flask_restx import Namespace, Resource
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
api = Namespace('health', description='Health check operations')
|
||||
|
||||
|
||||
@api.route('')
|
||||
class Health(Resource):
|
||||
"""Health check endpoint."""
|
||||
|
||||
@api.doc('health_check')
|
||||
def get(self):
|
||||
"""Check API health status."""
|
||||
logger.info("Health check requested")
|
||||
return {
|
||||
'status': 'healthy',
|
||||
'service': 'tts-vc-api',
|
||||
'version': '1.0.0'
|
||||
}, 200
|
||||
152
app/api/tts.py
Normal file
152
app/api/tts.py
Normal file
@@ -0,0 +1,152 @@
|
||||
"""TTS API endpoints."""
|
||||
import os
|
||||
import uuid
|
||||
import logging
|
||||
import tempfile
|
||||
from flask import current_app, request, send_file
|
||||
from flask_restx import Namespace, Resource, fields
|
||||
from app.models.voice import Voice
|
||||
from app.services.edge_tts_service import EdgeTTSService
|
||||
from app.services.seed_vc_service import SeedVCService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
api = Namespace('tts', description='Text-to-Speech operations')
|
||||
|
||||
# Models
|
||||
tts_request_model = api.model('TTSRequest', {
|
||||
'text': fields.String(required=True, description='Text to convert to speech'),
|
||||
'voice_id': fields.String(required=True, description='Reference voice ID'),
|
||||
'speed': fields.Float(description='Speech speed (0.5-2.0)', default=1.0)
|
||||
})
|
||||
|
||||
tts_response_model = api.model('TTSResponse', {
|
||||
'message': fields.String(),
|
||||
'voice_id': fields.String()
|
||||
})
|
||||
|
||||
error_model = api.model('Error', {
|
||||
'error': fields.String(required=True),
|
||||
'message': fields.String()
|
||||
})
|
||||
|
||||
|
||||
def check_auth():
|
||||
"""Check if request is authenticated."""
|
||||
auth_header = request.headers.get('Authorization', '')
|
||||
if not auth_header.startswith('Bearer '):
|
||||
return False
|
||||
|
||||
token = auth_header[7:] # Remove 'Bearer '
|
||||
valid_tokens = current_app.config.get('API_TOKENS', [])
|
||||
return token in valid_tokens
|
||||
|
||||
|
||||
@api.route('')
|
||||
class TTS(Resource):
|
||||
"""Text-to-Speech conversion endpoint."""
|
||||
|
||||
@api.doc('text_to_speech', security='bearer')
|
||||
@api.expect(tts_request_model)
|
||||
@api.response(200, 'Success')
|
||||
@api.response(400, 'Bad Request', error_model)
|
||||
@api.response(401, 'Unauthorized', error_model)
|
||||
@api.response(404, 'Voice Not Found', error_model)
|
||||
@api.response(500, 'Server Error', error_model)
|
||||
@api.produces(['audio/wav'])
|
||||
def post(self):
|
||||
"""
|
||||
Convert text to speech with voice cloning.
|
||||
|
||||
Takes text input, generates speech using Edge-TTS with Russian female voice,
|
||||
then applies voice conversion using the specified reference voice.
|
||||
"""
|
||||
logger.info("TTS request received")
|
||||
|
||||
# Check authentication
|
||||
if not check_auth():
|
||||
logger.warning("Unauthorized TTS request")
|
||||
return {'error': 'Unauthorized', 'message': 'Invalid or missing token'}, 401
|
||||
|
||||
# Parse request
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
logger.error("No JSON data in request")
|
||||
return {'error': 'Bad Request', 'message': 'JSON body required'}, 400
|
||||
|
||||
# Validate required fields
|
||||
text = data.get('text', '').strip()
|
||||
voice_id = data.get('voice_id', '').strip()
|
||||
speed = float(data.get('speed', 1.0))
|
||||
|
||||
if not text:
|
||||
logger.error("Empty text provided")
|
||||
return {'error': 'Bad Request', 'message': 'Text is required'}, 400
|
||||
|
||||
if not voice_id:
|
||||
logger.error("Empty voice_id provided")
|
||||
return {'error': 'Bad Request', 'message': 'voice_id is required'}, 400
|
||||
|
||||
# Validate speed
|
||||
if not 0.5 <= speed <= 2.0:
|
||||
logger.error(f"Invalid speed: {speed}")
|
||||
return {'error': 'Bad Request', 'message': 'Speed must be between 0.5 and 2.0'}, 400
|
||||
|
||||
# Check if voice exists
|
||||
voices_dir = current_app.config['VOICES_DIR']
|
||||
voice = Voice.from_file(voice_id, voices_dir)
|
||||
|
||||
if not voice or not voice.exists(voices_dir):
|
||||
logger.error(f"Voice not found: {voice_id}")
|
||||
return {'error': 'Not Found', 'message': f'Voice "{voice_id}" not found'}, 404
|
||||
|
||||
reference_path = voice.get_audio_path(voices_dir)
|
||||
logger.info(f"Using reference voice: {reference_path}")
|
||||
|
||||
# Generate unique output filename
|
||||
output_id = str(uuid.uuid4())[:8]
|
||||
output_dir = current_app.config['OUTPUT_DIR']
|
||||
tts_output = os.path.join(output_dir, f"{output_id}_tts.mp3")
|
||||
final_output = os.path.join(output_dir, f"{output_id}_converted.wav")
|
||||
|
||||
try:
|
||||
# Step 1: Generate TTS with Edge-TTS
|
||||
logger.info("Step 1: Generating TTS with Edge-TTS")
|
||||
edge_tts_voice = current_app.config['EDGE_TTS_VOICE']
|
||||
tts_service = EdgeTTSService(voice=edge_tts_voice)
|
||||
tts_service.text_to_speech(text, tts_output, speed=speed)
|
||||
|
||||
# Step 2: Convert voice with Seed-VC
|
||||
logger.info("Step 2: Converting voice with Seed-VC")
|
||||
seed_vc_model = current_app.config['SEED_VC_MODEL']
|
||||
seed_vc_repo = current_app.config['SEED_VC_REPO_PATH']
|
||||
vc_service = SeedVCService(model_name=seed_vc_model, repo_path=seed_vc_repo)
|
||||
vc_service.convert_voice(tts_output, reference_path, final_output)
|
||||
|
||||
logger.info(f"TTS+VC complete: {final_output}")
|
||||
|
||||
# Return audio file
|
||||
return send_file(
|
||||
final_output,
|
||||
mimetype='audio/wav',
|
||||
as_attachment=True,
|
||||
download_name=f"tts_{voice_id}.wav"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"TTS processing failed: {e}", exc_info=True)
|
||||
# Cleanup on failure
|
||||
for f in [tts_output, final_output]:
|
||||
if os.path.exists(f):
|
||||
try:
|
||||
os.remove(f)
|
||||
except:
|
||||
pass
|
||||
return {'error': 'Server Error', 'message': str(e)}, 500
|
||||
finally:
|
||||
# Cleanup temporary files
|
||||
try:
|
||||
if os.path.exists(tts_output):
|
||||
os.remove(tts_output)
|
||||
except:
|
||||
pass
|
||||
166
app/api/voices.py
Normal file
166
app/api/voices.py
Normal file
@@ -0,0 +1,166 @@
|
||||
"""Voices API endpoints."""
|
||||
import os
|
||||
import uuid
|
||||
import logging
|
||||
from flask import current_app, request
|
||||
from flask_restx import Namespace, Resource, fields
|
||||
from werkzeug.utils import secure_filename
|
||||
from app.models.voice import Voice
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
api = Namespace('voices', description='Voice management operations')
|
||||
|
||||
# Models
|
||||
voice_model = api.model('Voice', {
|
||||
'id': fields.String(required=True, description='Voice ID'),
|
||||
'name': fields.String(required=True, description='Voice name'),
|
||||
'filename': fields.String(required=True, description='Audio filename'),
|
||||
'created_at': fields.String(description='Creation timestamp')
|
||||
})
|
||||
|
||||
voices_list_model = api.model('VoicesList', {
|
||||
'voices': fields.List(fields.Nested(voice_model))
|
||||
})
|
||||
|
||||
upload_response_model = api.model('UploadResponse', {
|
||||
'id': fields.String(required=True, description='Voice ID'),
|
||||
'status': fields.String(required=True, description='Upload status'),
|
||||
'name': fields.String(description='Voice name')
|
||||
})
|
||||
|
||||
error_model = api.model('Error', {
|
||||
'error': fields.String(required=True),
|
||||
'message': fields.String()
|
||||
})
|
||||
|
||||
|
||||
def check_auth():
|
||||
"""Check if request is authenticated."""
|
||||
auth_header = request.headers.get('Authorization', '')
|
||||
if not auth_header.startswith('Bearer '):
|
||||
return False
|
||||
|
||||
token = auth_header[7:] # Remove 'Bearer '
|
||||
valid_tokens = current_app.config.get('API_TOKENS', [])
|
||||
return token in valid_tokens
|
||||
|
||||
|
||||
@api.route('')
|
||||
class VoicesList(Resource):
|
||||
"""List all available voices."""
|
||||
|
||||
@api.doc('list_voices', security='bearer')
|
||||
@api.response(200, 'Success', voices_list_model)
|
||||
@api.response(401, 'Unauthorized', error_model)
|
||||
def get(self):
|
||||
"""List all available reference voices."""
|
||||
logger.info("Listing voices")
|
||||
|
||||
if not check_auth():
|
||||
logger.warning("Unauthorized request to list voices")
|
||||
return {'error': 'Unauthorized', 'message': 'Invalid or missing token'}, 401
|
||||
|
||||
voices_dir = current_app.config['VOICES_DIR']
|
||||
voices = Voice.list_all(voices_dir)
|
||||
|
||||
logger.info(f"Found {len(voices)} voices")
|
||||
return {'voices': [v.to_dict() for v in voices]}, 200
|
||||
|
||||
|
||||
@api.route('/upload')
|
||||
class VoiceUpload(Resource):
|
||||
"""Upload new reference voice."""
|
||||
|
||||
@api.doc('upload_voice', security='bearer')
|
||||
@api.param('file', 'Audio file (WAV, MP3, OGG)', required=True, type='file')
|
||||
@api.param('name', 'Display name for the voice', required=False, type='string')
|
||||
@api.response(201, 'Created', upload_response_model)
|
||||
@api.response(400, 'Bad Request', error_model)
|
||||
@api.response(401, 'Unauthorized', error_model)
|
||||
def post(self):
|
||||
"""Upload a new reference voice."""
|
||||
logger.info("Voice upload requested")
|
||||
|
||||
if not check_auth():
|
||||
logger.warning("Unauthorized voice upload attempt")
|
||||
return {'error': 'Unauthorized', 'message': 'Invalid or missing token'}, 401
|
||||
|
||||
# Check if file is present
|
||||
if 'file' not in request.files:
|
||||
logger.error("No file in request")
|
||||
return {'error': 'Bad Request', 'message': 'No file provided'}, 400
|
||||
|
||||
file = request.files['file']
|
||||
if file.filename == '':
|
||||
logger.error("Empty filename")
|
||||
return {'error': 'Bad Request', 'message': 'Empty filename'}, 400
|
||||
|
||||
# Validate file extension
|
||||
allowed_extensions = {'.wav', '.mp3', '.ogg', '.m4a', '.flac'}
|
||||
filename = secure_filename(file.filename)
|
||||
ext = os.path.splitext(filename)[1].lower()
|
||||
|
||||
if ext not in allowed_extensions:
|
||||
logger.error(f"Invalid file extension: {ext}")
|
||||
return {
|
||||
'error': 'Bad Request',
|
||||
'message': f'Invalid file type. Allowed: {", ".join(allowed_extensions)}'
|
||||
}, 400
|
||||
|
||||
# Generate voice ID
|
||||
voice_id = str(uuid.uuid4())[:8]
|
||||
|
||||
# Get display name
|
||||
name = request.form.get('name', '').strip()
|
||||
if not name:
|
||||
name = os.path.splitext(filename)[0]
|
||||
|
||||
# Prepare filenames
|
||||
audio_filename = f"{voice_id}{ext}"
|
||||
voices_dir = current_app.config['VOICES_DIR']
|
||||
audio_path = os.path.join(voices_dir, audio_filename)
|
||||
|
||||
try:
|
||||
# Save file
|
||||
file.save(audio_path)
|
||||
logger.info(f"File saved: {audio_path}")
|
||||
|
||||
# Convert to WAV if needed (for consistency)
|
||||
if ext != '.wav':
|
||||
try:
|
||||
import soundfile as sf
|
||||
import librosa
|
||||
audio, sr = librosa.load(audio_path, sr=22050)
|
||||
wav_filename = f"{voice_id}.wav"
|
||||
wav_path = os.path.join(voices_dir, wav_filename)
|
||||
sf.write(wav_path, audio, 22050)
|
||||
|
||||
# Remove original and update filename
|
||||
os.remove(audio_path)
|
||||
audio_filename = wav_filename
|
||||
logger.info(f"Converted to WAV: {wav_path}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not convert to WAV: {e}")
|
||||
|
||||
# Create voice metadata
|
||||
voice = Voice(
|
||||
voice_id=voice_id,
|
||||
name=name,
|
||||
filename=audio_filename
|
||||
)
|
||||
voice.save(voices_dir)
|
||||
|
||||
logger.info(f"Voice uploaded successfully: {voice_id}")
|
||||
return {
|
||||
'id': voice_id,
|
||||
'status': 'uploaded',
|
||||
'name': name
|
||||
}, 201
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save voice: {e}")
|
||||
# Cleanup on failure
|
||||
if os.path.exists(audio_path):
|
||||
os.remove(audio_path)
|
||||
return {'error': 'Server Error', 'message': str(e)}, 500
|
||||
26
app/config.py
Normal file
26
app/config.py
Normal file
@@ -0,0 +1,26 @@
|
||||
"""Configuration module."""
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class Config:
|
||||
"""Application configuration."""
|
||||
# API Authentication
|
||||
API_TOKENS = os.getenv('API_TOKENS', '').split(',')
|
||||
|
||||
# Directories
|
||||
VOICES_DIR = os.getenv('VOICES_DIR', './voices')
|
||||
OUTPUT_DIR = os.getenv('OUTPUT_DIR', './output')
|
||||
|
||||
# Seed-VC
|
||||
SEED_VC_MODEL = os.getenv('SEED_VC_MODEL', 'seed-uvit-whisper-small-wavenet')
|
||||
SEED_VC_REPO_PATH = os.getenv('SEED_VC_REPO_PATH', './Seed-VC')
|
||||
|
||||
# Edge-TTS
|
||||
EDGE_TTS_VOICE = os.getenv('EDGE_TTS_VOICE', 'ru-RU-SvetlanaNeural')
|
||||
|
||||
# Flask
|
||||
FLASK_PORT = int(os.getenv('FLASK_PORT', '5000'))
|
||||
FLASK_DEBUG = os.getenv('FLASK_DEBUG', 'false').lower() == 'true'
|
||||
4
app/models/__init__.py
Normal file
4
app/models/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
||||
"""Models package."""
|
||||
from app.models.voice import Voice
|
||||
|
||||
__all__ = ['Voice']
|
||||
73
app/models/voice.py
Normal file
73
app/models/voice.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""Voice model for reference voice metadata."""
|
||||
import os
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import List, Optional, Dict, Any
|
||||
|
||||
|
||||
class Voice:
|
||||
"""Represents a reference voice."""
|
||||
|
||||
def __init__(self, voice_id: str, name: str, filename: str, created_at: str = None):
|
||||
self.id = voice_id
|
||||
self.name = name
|
||||
self.filename = filename
|
||||
self.created_at = created_at or datetime.now().isoformat()
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Convert to dictionary."""
|
||||
return {
|
||||
'id': self.id,
|
||||
'name': self.name,
|
||||
'filename': self.filename,
|
||||
'created_at': self.created_at
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, voice_id: str, voices_dir: str) -> Optional['Voice']:
|
||||
"""Load voice metadata from JSON file."""
|
||||
meta_path = os.path.join(voices_dir, f"{voice_id}.json")
|
||||
if not os.path.exists(meta_path):
|
||||
return None
|
||||
|
||||
with open(meta_path, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
return cls(
|
||||
voice_id=data['id'],
|
||||
name=data['name'],
|
||||
filename=data['filename'],
|
||||
created_at=data.get('created_at')
|
||||
)
|
||||
|
||||
def save(self, voices_dir: str):
|
||||
"""Save voice metadata to JSON file."""
|
||||
meta_path = os.path.join(voices_dir, f"{self.id}.json")
|
||||
with open(meta_path, 'w') as f:
|
||||
json.dump(self.to_dict(), f, indent=2)
|
||||
|
||||
def get_audio_path(self, voices_dir: str) -> str:
|
||||
"""Get full path to audio file."""
|
||||
return os.path.join(voices_dir, self.filename)
|
||||
|
||||
def exists(self, voices_dir: str) -> bool:
|
||||
"""Check if voice files exist."""
|
||||
audio_path = self.get_audio_path(voices_dir)
|
||||
meta_path = os.path.join(voices_dir, f"{self.id}.json")
|
||||
return os.path.exists(audio_path) and os.path.exists(meta_path)
|
||||
|
||||
@classmethod
|
||||
def list_all(cls, voices_dir: str) -> List['Voice']:
|
||||
"""List all voices in directory."""
|
||||
voices = []
|
||||
if not os.path.exists(voices_dir):
|
||||
return voices
|
||||
|
||||
for filename in os.listdir(voices_dir):
|
||||
if filename.endswith('.json'):
|
||||
voice_id = filename[:-5] # Remove .json
|
||||
voice = cls.from_file(voice_id, voices_dir)
|
||||
if voice and voice.exists(voices_dir):
|
||||
voices.append(voice)
|
||||
|
||||
return sorted(voices, key=lambda v: v.created_at, reverse=True)
|
||||
5
app/services/__init__.py
Normal file
5
app/services/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Services package."""
|
||||
from app.services.edge_tts_service import EdgeTTSService
|
||||
from app.services.seed_vc_service import SeedVCService
|
||||
|
||||
__all__ = ['EdgeTTSService', 'SeedVCService']
|
||||
67
app/services/edge_tts_service.py
Normal file
67
app/services/edge_tts_service.py
Normal file
@@ -0,0 +1,67 @@
|
||||
"""Edge-TTS service for text-to-speech conversion."""
|
||||
import os
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Optional
|
||||
import edge_tts
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EdgeTTSService:
|
||||
"""Service for converting text to speech using Edge-TTS."""
|
||||
|
||||
def __init__(self, voice: str = "ru-RU-SvetlanaNeural"):
|
||||
self.voice = voice
|
||||
logger.info(f"EdgeTTS Service initialized with voice: {voice}")
|
||||
|
||||
async def _generate_speech_async(self, text: str, output_path: str, speed: float = 1.0) -> str:
|
||||
"""Generate speech asynchronously."""
|
||||
# Adjust rate based on speed
|
||||
if speed != 1.0:
|
||||
rate_percent = int((speed - 1.0) * 100)
|
||||
rate = f"{rate_percent:+d}%"
|
||||
else:
|
||||
rate = "+0%"
|
||||
|
||||
communicate = edge_tts.Communicate(text, self.voice, rate=rate)
|
||||
await communicate.save(output_path)
|
||||
|
||||
return output_path
|
||||
|
||||
def text_to_speech(self, text: str, output_path: str, speed: float = 1.0) -> str:
|
||||
"""
|
||||
Convert text to speech and save to file.
|
||||
|
||||
Args:
|
||||
text: Text to convert
|
||||
output_path: Path to save audio file
|
||||
speed: Speech speed multiplier (0.5 - 2.0)
|
||||
|
||||
Returns:
|
||||
Path to generated audio file
|
||||
"""
|
||||
logger.info(f"Generating TTS for text: {text[:50]}...")
|
||||
|
||||
# Run async function in sync context
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
try:
|
||||
result = loop.run_until_complete(
|
||||
self._generate_speech_async(text, output_path, speed)
|
||||
)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
logger.info(f"TTS generated: {result}")
|
||||
return result
|
||||
|
||||
def get_supported_voices(self) -> list:
|
||||
"""Get list of supported voices."""
|
||||
voices = [
|
||||
"ru-RU-SvetlanaNeural",
|
||||
"ru-RU-DmitryNeural",
|
||||
"en-US-AriaNeural",
|
||||
"en-US-GuyNeural",
|
||||
]
|
||||
return voices
|
||||
143
app/services/seed_vc_service.py
Normal file
143
app/services/seed_vc_service.py
Normal file
@@ -0,0 +1,143 @@
|
||||
"""Seed-VC service for voice conversion."""
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
import tempfile
|
||||
from typing import Optional
|
||||
import soundfile as sf
|
||||
import librosa
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SeedVCService:
|
||||
"""Service for voice conversion using Seed-VC."""
|
||||
|
||||
def __init__(self, model_name: str = "seed-uvit-whisper-small-wavenet", repo_path: str = "./Seed-VC"):
|
||||
self.model_name = model_name
|
||||
self.repo_path = repo_path
|
||||
self.model = None
|
||||
self.device = None
|
||||
self.initialized = False
|
||||
|
||||
logger.info(f"SeedVC Service initialized with model: {model_name}")
|
||||
|
||||
def _ensure_repo_exists(self):
|
||||
"""Ensure Seed-VC repository is cloned."""
|
||||
if not os.path.exists(self.repo_path):
|
||||
logger.info("Cloning Seed-VC repository...")
|
||||
import subprocess
|
||||
subprocess.run([
|
||||
"git", "clone", "https://github.com/Plachtaa/Seed-VC.git", self.repo_path
|
||||
], check=True)
|
||||
logger.info("Seed-VC repository cloned")
|
||||
|
||||
def initialize(self):
|
||||
"""Initialize the model (lazy loading)."""
|
||||
if self.initialized:
|
||||
return
|
||||
|
||||
try:
|
||||
self._ensure_repo_exists()
|
||||
|
||||
# Add Seed-VC to path
|
||||
if self.repo_path not in sys.path:
|
||||
sys.path.insert(0, self.repo_path)
|
||||
|
||||
import torch
|
||||
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
||||
logger.info(f"Using device: {self.device}")
|
||||
|
||||
# Import Seed-VC modules
|
||||
try:
|
||||
from inference import load_model, inference
|
||||
self.inference_fn = inference
|
||||
self.model = load_model(self.model_name, self.device)
|
||||
logger.info(f"Model {self.model_name} loaded successfully")
|
||||
except ImportError as e:
|
||||
logger.warning(f"Could not import Seed-VC modules: {e}")
|
||||
logger.warning("Voice conversion will use passthrough mode")
|
||||
self.model = None
|
||||
|
||||
self.initialized = True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize Seed-VC: {e}")
|
||||
self.initialized = True # Mark as initialized to avoid retry
|
||||
self.model = None
|
||||
|
||||
def convert_voice(
|
||||
self,
|
||||
source_audio_path: str,
|
||||
reference_audio_path: str,
|
||||
output_path: str
|
||||
) -> str:
|
||||
"""
|
||||
Convert voice from source to match reference voice.
|
||||
|
||||
Args:
|
||||
source_audio_path: Path to source audio (TTS output)
|
||||
reference_audio_path: Path to reference voice audio
|
||||
output_path: Path to save converted audio
|
||||
|
||||
Returns:
|
||||
Path to converted audio file
|
||||
"""
|
||||
self.initialize()
|
||||
|
||||
logger.info(f"Converting voice: {source_audio_path} -> {output_path}")
|
||||
logger.info(f"Using reference: {reference_audio_path}")
|
||||
|
||||
if self.model is None:
|
||||
# Fallback: copy source audio
|
||||
logger.warning("Seed-VC not available, using passthrough mode")
|
||||
import shutil
|
||||
shutil.copy(source_audio_path, output_path)
|
||||
return output_path
|
||||
|
||||
try:
|
||||
# Load audio files
|
||||
source_audio, sr = librosa.load(source_audio_path, sr=22050)
|
||||
reference_audio, _ = librosa.load(reference_audio_path, sr=22050)
|
||||
|
||||
# Perform voice conversion
|
||||
converted_audio = self.inference_fn(
|
||||
self.model,
|
||||
source_audio,
|
||||
reference_audio,
|
||||
sr=22050
|
||||
)
|
||||
|
||||
# Save output
|
||||
sf.write(output_path, converted_audio, 22050)
|
||||
logger.info(f"Voice conversion complete: {output_path}")
|
||||
|
||||
return output_path
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Voice conversion failed: {e}")
|
||||
# Fallback: copy source audio
|
||||
import shutil
|
||||
shutil.copy(source_audio_path, output_path)
|
||||
return output_path
|
||||
|
||||
def preprocess_audio(self, input_path: str, output_path: str, target_sr: int = 22050) -> str:
|
||||
"""
|
||||
Preprocess audio to match Seed-VC requirements.
|
||||
|
||||
Args:
|
||||
input_path: Input audio path
|
||||
output_path: Output audio path
|
||||
target_sr: Target sample rate
|
||||
|
||||
Returns:
|
||||
Path to preprocessed audio
|
||||
"""
|
||||
try:
|
||||
audio, sr = librosa.load(input_path, sr=target_sr)
|
||||
sf.write(output_path, audio, target_sr)
|
||||
return output_path
|
||||
except Exception as e:
|
||||
logger.error(f"Audio preprocessing failed: {e}")
|
||||
raise
|
||||
Reference in New Issue
Block a user