import os
import re
import time
import sys
import logging
from pathlib import Path
# --- CONFIGURACIÓN DE PERSISTENCIA Y CACHÉ (CRÍTICO PARA SPACES) ---
# Estas líneas permiten que las librerías escriban en el disco persistente /data
os.environ["NUMBA_CACHE_DIR"] = "/data/numba_cache"
os.environ["MPLCONFIGDIR"] = "/data/matplotlib_config"
os.environ["PIP_CACHE_DIR"] = "/data/pip_cache"
os.environ["HF_HOME"] = "/data/huggingface"
os.environ["XDG_DATA_HOME"] = "/data/xdg_data"
try:
import rvc_python
print("rvc-python ya está instalado.")
except ImportError:
print("rvc-python no encontrado. Instalando en runtime...")
try:
from install_rvc import install_rvc
install_rvc()
import rvc_python # re-importar después de instalación
print("rvc-python instalado exitosamente.")
except Exception as e:
print(f"ERROR instalando rvc-python: {e}")
import threading
import subprocess
import glob
# Configuración inicial ANTES de importaciones pesadas
os.environ["COQUI_TOS_AGREED"] = "1"
os.environ["OMP_NUM_THREADS"] = "1"
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["GRADIO_SSR_MODE"] = "false" # FIX: Desactivar SSR para evitar errores con Node
# Configurar logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%H:%M:%S'
)
logger = logging.getLogger(__name__)
logger.info("=" * 60)
logger.info("🎙️ PEDRO LABATTAGLIA TTS - INICIANDO (MODO PERSISTENTE)")
logger.info("=" * 60)
# Importaciones
try:
logger.info("📦 Importando dependencias...")
import torch
import gradio as gr
from TTS.tts.configs.xtts_config import XttsConfig
from TTS.tts.models.xtts import Xtts
from huggingface_hub import hf_hub_download, snapshot_download
import scipy.io.wavfile as wavfile
import warnings
import shutil
import pydantic
logger.info(f"Gradio: {gr.__version__}, Pydantic: {pydantic.__version__}")
logger.info("✅ Todas las dependencias importadas correctamente")
except ImportError as e:
logger.error(f"❌ Error importando dependencias: {e}")
sys.exit(1)
warnings.filterwarnings("ignore")
# --- CONFIGURACIÓN RVC PERSISTENTE ---
# Ruta cambiada a /data para persistencia
RVC_MODELS_DIR = Path("/data/models/rvc_models")
def download_rvc_models_from_hf():
"""Descarga los modelos RVC desde el dataset privado al disco persistente."""
try:
repo_id = "Blakus/Pedro_Lab_RVC_Models"
token = os.environ.get("HF_TOKEN")
if not token:
logger.warning("⚠️ No se encontró HF_TOKEN. No se podrán descargar los modelos RVC privados.")
return
# Verificar si el modelo principal ya existe en /data
if (RVC_MODELS_DIR / "Fresco" / "Fresco.pth").exists():
logger.info("✅ Modelos RVC ya existen en almacenamiento persistente (/data). No se descargarán.")
return
logger.info(f"📥 [STARTUP] Iniciando descarga de modelos RVC a {RVC_MODELS_DIR}...")
# Asegurar que el directorio existe
RVC_MODELS_DIR.mkdir(parents=True, exist_ok=True)
snapshot_download(
repo_id=repo_id,
repo_type="model",
local_dir=RVC_MODELS_DIR,
token=token,
ignore_patterns=["*.git*", "README.md", ".gitattributes"]
)
logger.info("✅ [STARTUP] Modelos RVC descargados y listos en /data.")
except Exception as e:
logger.error(f"❌ Error descargando modelos RVC: {e}")
def get_rvc_models_list():
"""Escanea la carpeta de modelos RVC y devuelve una lista de nombres."""
models = []
if not RVC_MODELS_DIR.exists():
return models
for model_dir in RVC_MODELS_DIR.iterdir():
if model_dir.is_dir():
pth_files = list(model_dir.glob('*.pth'))
if pth_files:
models.append(model_dir.name)
return models
def run_rvc_pipeline(input_path, model_name):
"""Ejecuta el pipeline de RVC (Compatible con Linux/Spaces)"""
try:
model_dir = RVC_MODELS_DIR / model_name
pth_files = list(model_dir.glob('*.pth'))
index_files = list(model_dir.glob('*.index'))
if not pth_files:
logger.error(f"❌ No se encontró archivo .pth para el modelo {model_name}")
return input_path
model_path = str(pth_files[0].absolute())
index_path = str(index_files[0].absolute()) if index_files else ""
# Generar nombre de salida
output_rvc = input_path.replace(".wav", f"_rvc_{model_name}.wav")
# Parámetros RVC
pitch = 0
method = "rmvpe"
index_rate = 0
filter_radius = 3
resample_sr = 0
rms_mix_rate = 0.25
protect = 0.50
device = "cuda:0" if torch.cuda.is_available() else "cpu"
# Comando CLI para RVC en Linux (usando sys.executable)
cmd = [
sys.executable, '-m', 'rvc_python',
'cli',
'--input', str(input_path),
'--model', model_path,
'--pitch', str(pitch),
'--method', method,
'--output', str(output_rvc),
'--index_rate', str(index_rate),
'--device', device,
'--protect', str(protect),
'--filter_radius', str(filter_radius),
'--resample_sr', str(resample_sr),
'--rms_mix_rate', str(rms_mix_rate)
]
if index_path:
cmd.extend(['--index', index_path])
logger.info(f"🔄 Ejecutando RVC con modelo: {model_name}...")
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
if os.path.exists(output_rvc):
logger.info("✅ RVC aplicado exitosamente")
return output_rvc
else:
logger.error("❌ RVC finalizó pero no generó archivo de salida")
if result.stdout: logger.info(f"RVC STDOUT: {result.stdout}")
if result.stderr: logger.error(f"RVC STDERR: {result.stderr}")
return input_path
except subprocess.CalledProcessError as e:
logger.error(f"❌ Error al ejecutar comando RVC. Código: {e.returncode}")
if e.stderr: logger.error(f"RVC STDERR: {e.stderr}")
return input_path
except Exception as e:
logger.error(f"❌ Error general en pipeline RVC: {e}")
return input_path
def get_user_data_dir(app_name="tts"):
"""Directorio para datos de usuario redirigido a /data"""
# Usamos /data/xdg_data para que persista
data_dir = Path("/data/xdg_data") / app_name
data_dir.mkdir(parents=True, exist_ok=True)
return str(data_dir)
def authenticate_user(username, password):
"""Autenticación contra variables de entorno"""
valid_username = os.environ.get("AUTH_USERNAME")
valid_password = os.environ.get("AUTH_PASSWORD")
if not valid_username or not valid_password:
logger.error("❌ AUTH_USERNAME o AUTH_PASSWORD no configurados")
return False
if username == valid_username and password == valid_password:
logger.info(f"✅ Usuario autenticado: {username}")
return True
else:
logger.warning(f"❌ Autenticación fallida para: {username}")
return False
class PedroTTSApp:
def __init__(self):
self.model = None
self.config = None
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self.model_loaded = False
self.loading_lock = threading.Lock()
logger.info(f"🖥️ Dispositivo: {self.device}")
if self.device == "cuda":
logger.info(f"🚀 GPU: {torch.cuda.get_device_name(0)}")
def setup_resources(self):
"""Descarga RVC y configura el modelo XTTS en persistencia"""
with self.loading_lock:
if self.model_loaded:
return
# 1. Descargar RVC (Prioridad)
download_rvc_models_from_hf()
# 2. Configurar XTTS
try:
logger.info("📦 Iniciando configuración del modelo XTTS...")
repo_id = "Blakus/Pedro_Lab_XTTS"
# MODIFICACIÓN: Ruta persistente directa para XTTS
local_dir = Path("/data/models/xtts_v2")
local_dir.mkdir(parents=True, exist_ok=True)
files_to_download = ["config.json", "model.pth", "vocab.json"]
for file_name in files_to_download:
file_path = local_dir / file_name
if not file_path.exists():
logger.info(f"📥 Descargando {file_name} a /data...")
try:
hf_hub_download(
repo_id=repo_id,
filename=file_name,
local_dir=str(local_dir),
local_dir_use_symlinks=False
)
logger.info(f"✅ {file_name} descargado")
except Exception as e:
logger.warning(f"⚠️ Error en descarga directa de {file_name}: {e}")
downloaded_file = hf_hub_download(
repo_id=repo_id,
filename=file_name
)
shutil.copy2(downloaded_file, file_path)
logger.info(f"✅ {file_name} copiado")
else:
logger.info(f"✅ {file_name} ya existe en persistencia")
config_path = str(local_dir / "config.json")
checkpoint_path = str(local_dir / "model.pth")
vocab_path = str(local_dir / "vocab.json")
for path, name in [(config_path, "config"), (checkpoint_path, "model"), (vocab_path, "vocab")]:
if not os.path.exists(path):
raise FileNotFoundError(f"Archivo no encontrado: {name} en {path}")
logger.info("⚙️ Cargando configuración...")
self.config = XttsConfig()
self.config.load_json(config_path)
logger.info("🔧 Inicializando modelo...")
self.model = Xtts.init_from_config(self.config)
logger.info("📂 Cargando checkpoint (esto puede tomar unos minutos)...")
self.model.load_checkpoint(
self.config,
checkpoint_path=checkpoint_path,
vocab_path=vocab_path,
eval=True,
use_deepspeed=False
)
if self.device == "cuda":
self.model.cuda()
logger.info("🚀 Modelo cargado en GPU")
else:
self.model.cpu()
logger.info("🐌 Modelo cargado en CPU")
self.model_loaded = True
logger.info("✅ Modelo XTTS configurado y listo")
except Exception as e:
logger.error(f"❌ Error configurando modelo: {e}")
self.model_loaded = False
import traceback
logger.error(traceback.format_exc())
def load_reference_audios(self):
"""Cargar audios de referencia desde dataset privado a persistencia"""
try:
logger.info("🎵 Cargando audios de referencia...")
dataset_id = os.environ.get("PRIVATE_AUDIO_DATASET", "Blakus/Pedro_Lab_XTTS_Reference_Audios")
hf_token = os.environ.get("HF_TOKEN")
# MODIFICACIÓN: Ruta persistente para audios
local_audio_dir = Path("/data/reference_audios")
local_audio_dir.mkdir(parents=True, exist_ok=True)
audio_files = ["neutral.wav", "serio.wav", "alegre.wav", "neutral_ingles.wav"]
downloaded_audios = []
for audio_file in audio_files:
local_path = local_audio_dir / audio_file
if not local_path.exists():
logger.info(f"📥 Descargando {audio_file} a /data...")
try:
downloaded_file = hf_hub_download(
repo_id=dataset_id,
filename=audio_file,
repo_type="dataset",
token=hf_token,
local_dir_use_symlinks=False
)
shutil.copy2(downloaded_file, local_path)
logger.info(f"✅ {audio_file} descargado")
downloaded_audios.append(str(local_path))
except Exception as e:
logger.error(f"❌ Error con {audio_file}: {e}")
else:
logger.info(f"✅ {audio_file} encontrado en persistencia")
downloaded_audios.append(str(local_path))
logger.info(f"🎵 Total audios disponibles: {len(downloaded_audios)}")
return downloaded_audios
except Exception as e:
logger.error(f"❌ Error cargando audios: {e}")
return []
def generate_speech(self, text, language, reference_audio, speed, temperature, enable_text_splitting, use_rvc, rvc_model):
"""Genera audio de voz + RVC"""
try:
if not self.model_loaded or not self.model:
return None, "⏳ Modelo cargando... Intente en unos minutos o contacte al administrador."
if not text or len(text.strip()) < 2:
return None, "❌ El texto debe tener al menos 2 caracteres"
if not reference_audio or reference_audio == "demo":
return None, "❌ Seleccione un estilo de voz válido"
if not os.path.exists(reference_audio):
return None, "❌ Audio de referencia no encontrado"
text = text.strip()
logger.info(f"🎙️ Generando: '{text[:50]}{'...' if len(text) > 50 else ''}' (RVC: {use_rvc})")
try:
gpt_cond_latent, speaker_embedding = self.model.get_conditioning_latents(
audio_path=reference_audio
)
except Exception as e:
logger.error(f"Error en conditioning latents: {e}")
return None, f"❌ Error procesando audio de referencia: {str(e)}"
start_time = time.time()
out = self.model.inference(
text,
language,
gpt_cond_latent,
speaker_embedding,
temperature=float(temperature),
length_penalty=1.0,
repetition_penalty=5.0,
top_k=50,
top_p=0.85,
speed=float(speed),
enable_text_splitting=enable_text_splitting,
do_sample=True
)
inference_time = time.time() - start_time
if "wav" not in out or out["wav"] is None:
return None, "❌ No se generó audio"
timestamp = int(time.time())
# En Spaces usamos ruta relativa o absoluta segura.
output_path = f"output_{timestamp}.wav"
sample_rate = self.config.audio.get("output_sample_rate", 22050)
wavfile.write(output_path, sample_rate, out["wav"])
audio_length = len(out["wav"]) / sample_rate
metrics = f"""✅ Audio generado exitosamente
🎵 Duración: {audio_length:.1f}s
⏱️ Tiempo: {inference_time:.1f}s
⚡ Velocidad: {speed}x
🎨 Creatividad: {temperature}
📖 Segmentación: {'Sí' if enable_text_splitting else 'No'}
🌐 Idioma: {language.upper()}"""
# --- RVC PIPELINE ---
if use_rvc and rvc_model:
logger.info(f"✨ Iniciando mejora de timbre RVC: {rvc_model}")
rvc_start_time = time.time()
rvc_output_path = run_rvc_pipeline(output_path, rvc_model)
if rvc_output_path != output_path:
output_path = rvc_output_path
rvc_time = time.time() - rvc_start_time
metrics += f"\n\n✨ Mejora Aplicada: {rvc_model}\n⏱️ Tiempo RVC: {rvc_time:.1f}s"
else:
metrics += "\n\n⚠️ Fallo en la mejora del timbre, se retorna audio original."
# --------------------
logger.info(f"✅ Proceso completo.")
return output_path, metrics
except Exception as e:
error_msg = f"❌ Error: {str(e)}"
logger.error(error_msg)
import traceback
logger.error(traceback.format_exc())
return None, error_msg
# Instancia global
app = PedroTTSApp()
def create_interface():
"""Crear interfaz Gradio"""
try:
logger.info("🎨 Creando interfaz...")
available_audios = app.load_reference_audios()
languages = [("Español", "es"), ("English", "en")]
ref_mapping = {
"neutral.wav": "🎭 Neutral",
"serio.wav": "😐 Serio",
"alegre.wav": "😊 Alegre",
"neutral_ingles.wav": "🎭 Neutral (English)"
}
if not available_audios:
logger.warning("⚠️ No hay audios de referencia")
audio_refs = [("❌ No disponible", "demo")]
else:
audio_refs = []
for audio_file in available_audios:
filename = Path(audio_file).name
label = ref_mapping.get(filename, filename)
audio_refs.append((label, audio_file))
# Modelos RVC (intento inicial)
initial_rvc_models = get_rvc_models_list()
custom_css = """
.auth-box {
max-width: 450px;
margin: 40px auto;
padding: 40px;
border-radius: 20px;
background: linear-gradient(145deg, #2d2d2d, #1a1a1a);
box-shadow: 0 8px 32px rgba(0,0,0,0.4);
}
.speaker-info {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px;
border-radius: 15px;
margin-bottom: 20px;
text-align: center;
}
.speaker-image {
width: 180px;
height: 180px;
border-radius: 50%;
margin: 0 auto 15px;
border: 4px solid rgba(255,255,255,0.3);
object-fit: cover;
}
.social-links {
display: flex;
justify-content: center;
gap: 15px;
margin-top: 15px;
}
.social-link {
color: white;
text-decoration: none;
font-size: 16px;
padding: 8px 12px;
border-radius: 20px;
background: rgba(255,255,255,0.2);
transition: all 0.3s ease;
}
.social-link:hover {
background: rgba(255,255,255,0.3);
transform: translateY(-2px);
}
.credits-section {
margin-top: 15px;
text-align: center;
}
.credits-text {
color: #6c757d;
font-size: 12px;
margin: 5px 0;
}
.credits-link {
color: #007bff;
text-decoration: none;
font-size: 11px;
transition: color 0.3s ease;
}
.credits-link:hover {
color: #0056b3;
}
/* CSS para el Info Box en Dark Mode */
.info-box {
font-size: 0.9em;
color: #e0e0e0;
padding: 12px;
background-color: #252525;
border-left: 5px solid #9b59b6;
border-radius: 5px;
margin-top: 8px;
box-shadow: 0 2px 5px rgba(0,0,0,0.2);
}
"""
with gr.Blocks(title="Pedro Labattaglia TTS", theme=gr.themes.Soft(), css=custom_css) as demo:
auth_state = gr.State(False)
# Pantalla de login
with gr.Column(elem_classes="auth-box") as auth_column:
gr.Markdown("""
# 🔐 Acceso Restringido
## Pedro Labattaglia - TTS
Ingrese sus credenciales para acceder al sistema.
""")
username_input = gr.Textbox(
label="👤 Usuario",
placeholder="Usuario"
)
password_input = gr.Textbox(
label="🔑 Contraseña",
placeholder="Contraseña",
type="password"
)
login_btn = gr.Button("🚪 Iniciar Sesión", variant="primary", size="lg")
auth_message = gr.Textbox(label="Estado", value="⏳ Esperando credenciales...", interactive=False)
# Interfaz principal
with gr.Column(visible=False) as main_column:
# Header con info del locutor
with gr.Column(elem_classes="speaker-info"):
gr.HTML("""
Pedro Labattaglia
🎙️ Locutor profesional | +20 años dando voz a marcas líderes en Argentina, LATAM y EE.UU. |
Español rioplatense / neutro | Voice Over | Source Connect: pedrovotalent |
✉️ pedrolabattaglia@gmail.com
""")
with gr.Row():
gr.Markdown("### ✅ Sesión activa")
logout_btn = gr.Button("🚪 Salir", variant="secondary", size="sm")
# Controles de generación
with gr.Row():
with gr.Column(scale=2):
language = gr.Dropdown(
choices=languages,
value="es",
label="🌐 Idioma"
)
reference = gr.Dropdown(
choices=audio_refs,
value=audio_refs[0][1] if audio_refs else "",
label="🎭 Estilo de voz"
)
# --- CONTROLES RVC ---
rvc_model_dd = gr.Dropdown(
choices=initial_rvc_models,
label="🎭✨ Estilo de Timbre (Mejora)",
value=initial_rvc_models[0] if initial_rvc_models else None,
interactive=True,
visible=False
)
use_rvc_cb = gr.Checkbox(label="✨ Activar Mejora de Timbre", value=False)
info_box = gr.HTML("""
ℹ️ Info:
Al activarlo tendrás una mejora en la calidad de la voz a costa de, quizás, un mínimo cambio en la pronunciación y/o tono metálico en las S. Puedes combinar como quieras el estilo de voz y el estilo de timbre. eres libre de experimentar.
""", visible=False)
def toggle_rvc_ui(checked):
current_models = get_rvc_models_list()
new_val = current_models[0] if current_models else None
return (
gr.update(visible=checked, choices=current_models, value=new_val),
gr.update(visible=checked)
)
use_rvc_cb.change(toggle_rvc_ui, inputs=use_rvc_cb, outputs=[rvc_model_dd, info_box])
# ---------------------
gr.Markdown("**Velocidad de reproducción del audio**")
speed = gr.Slider(
0.5, 2.0, 1.0, 0.1,
label="⚡ Velocidad"
)
gr.Markdown("**🛡️ Más estable pero menos creativo/expresivo ← → 🎭 Menos estable pero más creativo/expresivo**")
temperature = gr.Slider(
0.01, 1, 0.75, 0.05,
label="🎨 Creatividad"
)
gr.Markdown("**✅ Desactivarlo puede generar mas naturalidad con textos cortos.**")
enable_text_splitting = gr.Checkbox(
value=True,
label="📖 Segmentación inteligente"
)
text_input = gr.Textbox(
label="📝 Texto a sintetizar",
placeholder="Escriba aquí el texto que desea convertir a voz...",
lines=5
)
generate_btn = gr.Button("🎵 Generar Audio", variant="primary", size="lg")
with gr.Column(scale=1):
audio_output = gr.Audio(
label="🔊 Audio Generado",
show_download_button=True
)
metrics_output = gr.Textbox(
label="📊 Información",
value="✨ Listo para generar audio...",
lines=10
)
# Créditos
with gr.Column(elem_classes="credits-section"):
gr.HTML("""
Desarrollado por Ezequiel Casas
LinkedIn
""")
# Event handler para generación
generate_btn.click(
fn=app.generate_speech,
inputs=[
text_input, language, reference, speed, temperature, enable_text_splitting,
use_rvc_cb, # Boolean RVC
rvc_model_dd # Dropdown RVC
],
outputs=[audio_output, metrics_output]
)
# Funciones de autenticación
def handle_login(username, password):
if authenticate_user(username, password):
return (
True,
gr.update(visible=False),
gr.update(visible=True),
"✅ Acceso concedido",
"", ""
)
return (
False,
gr.update(visible=True),
gr.update(visible=False),
"❌ Credenciales incorrectas",
"", ""
)
def handle_logout():
return (
False,
gr.update(visible=True),
gr.update(visible=False),
"⏳ Sesión cerrada",
"", "", None, "✨ Listo para generar audio...", ""
)
# Eventos
login_btn.click(
fn=handle_login,
inputs=[username_input, password_input],
outputs=[auth_state, auth_column, main_column, auth_message, username_input, password_input]
)
logout_btn.click(
fn=handle_logout,
outputs=[auth_state, auth_column, main_column, auth_message,
username_input, password_input, audio_output, metrics_output, text_input]
)
password_input.submit(
fn=handle_login,
inputs=[username_input, password_input],
outputs=[auth_state, auth_column, main_column, auth_message, username_input, password_input]
)
logger.info("✅ Interfaz creada")
return demo
except Exception as e:
logger.error(f"❌ Error creando interfaz: {e}")
with gr.Blocks() as demo:
gr.Markdown(f"# ❌ Error\n\n{str(e)}")
return demo
def main():
try:
logger.info("🚀 Iniciando aplicación...")
# Detectar entorno
is_spaces = os.environ.get("SPACE_ID") is not None
logger.info(f"🌍 Entorno: {'HuggingFace Spaces' if is_spaces else 'Local'}")
# Verificar credenciales
has_auth = os.environ.get("AUTH_USERNAME") and os.environ.get("AUTH_PASSWORD")
if not has_auth:
logger.warning("⚠️ Credenciales no configuradas en secrets")
else:
logger.info("✅ Credenciales configuradas")
logger.info("🎨 Creando interfaz Gradio...")
demo = create_interface()
logger.info("✅ Interfaz creada")
# Cargar modelos en segundo plano para que la UI abra rápido
logger.info("📦 Cargando recursos (RVC + XTTS) en hilo de fondo...")
model_thread = threading.Thread(target=app.setup_resources, daemon=True)
model_thread.start()
# Configuración de puerto
port = int(os.environ.get("PORT", 7860))
logger.info(f"🌐 Preparando lanzamiento en puerto {port}")
# --- CORRECCIÓN AQUÍ ---
# Siempre usamos 0.0.0.0 para que sea visible desde fuera del contenedor
logger.info("🚀 Lanzando servidor Gradio...")
demo.launch(
server_name="0.0.0.0", # IMPORTANTE: Esto permite que HF vea la app
server_port=port,
share=False, # En Spaces no necesitamos share
ssr_mode=False, # Desactivado para evitar errores de Node
show_error=True
)
except Exception as e:
logger.error(f"💥 Error crítico: {e}")
import traceback
logger.error(traceback.format_exc())
sys.exit(1)
if __name__ == "__main__":
main()