# ╔══════════════════════════════════════════════════════════════════╗
# ║ Pro Voice Studio (XTTS + NLP + Resemble Enhance Integration) ║
# ╚══════════════════════════════════════════════════════════════════╝
import os, sys, time, json, uuid, shutil
import uvicorn
from fastapi import FastAPI, Form, File, UploadFile, HTTPException
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import torch
import torchaudio
# === Model 1 Imports (NLP) ===
import pyarabic.araby as araby
# === Model 2 Imports (TTS) ===
from TTS.api import TTS
# === Model 3 Imports (Enhancement/Super Resolution) ===
from resemble_enhance.enhancer.inference import enhance
# ── Env ──────────────────────────────────────────────────────────────
os.environ["COQUI_TOS_AGREED"] = "1"
os.environ.setdefault("HF_HOME", "/data/huggingface")
VOICE_LIB = "/data/voice_library"
OUTPUT_DIR = "/data/outputs"
HISTORY_FILE = "/data/history.json"
for d in [VOICE_LIB, OUTPUT_DIR]:
os.makedirs(d, exist_ok=True)
# ── Hardware ─────────────────────────────────────────────────────────
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"[*] Pipeline running on {device.upper()} …")
# ── Load Core TTS ────────────────────────────────────────────────────
print("[*] Loading Model 2: XTTS v2 …")
xtts_engine = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device)
print("[✓] XTTS Model ready.")
# ── Text Normalization Wrapper (Model 1) ─────────────────────────────
def normalize_arabic_text(text: str) -> str:
"""Pre-process text to guide the TTS model better."""
text = araby.strip_tashkeel(text) # Optional: Can be modified based on need
text = araby.normalize_ligature(text)
# Add subtle punctuation spacing to give the acoustic model breathing room
text = text.replace("،", "، ").replace(".", ". ")
return " ".join(text.split())
# ── Audio Enhancer Wrapper (Model 3) ─────────────────────────────────
def enhance_audio_quality(input_path: str, output_path: str):
"""Pass XTTS output through Resemble Enhance for ElevenLabs-like quality."""
print("[*] Running Model 3: Resemble Enhance (Super Resolution)...")
dwav, sr = torchaudio.load(input_path)
# Apply inference enhancement (NFE=32 is a good balance of speed/quality)
hwav, new_sr = enhance(dwav=dwav, sr=sr, device=device, nfe=32)
# Save the polished audio
torchaudio.save(output_path, hwav, new_sr)
print("[✓] Enhancement complete.")
# ── History helpers ──────────────────────────────────────────────────
def load_history():
if os.path.exists(HISTORY_FILE):
try: return json.load(open(HISTORY_FILE))
except: pass
return []
def save_history(h):
json.dump(h, open(HISTORY_FILE, "w"), ensure_ascii=False, indent=2)
# ── FastAPI app ──────────────────────────────────────────────────────
app = FastAPI(title="Pro Voice Studio")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
LANGUAGES = {
"ar": "العربية", "en": "English", "es": "Español", "fr": "Français",
"de": "Deutsch", "it": "Italiano", "pt": "Português","ru": "Русский",
"zh-cn": "中文", "ja": "日本語"
}
# ══════════════════════════════════════════════════════════════════════
# HTML Frontend
# ══════════════════════════════════════════════════════════════════════
HTML = r"""
Pro Voice Studio | Multi-Model
"""
# ══════════════════════════════════════════════════════════════════════
# Routes
# ══════════════════════════════════════════════════════════════════════
@app.get("/", response_class=HTMLResponse)
async def ui():
return HTML.replace("DEVICE_PLACEHOLDER", device)
@app.post("/generate")
async def generate(
text: str = Form(...),
language: str = Form("ar"),
temperature: float = Form(0.75),
speed: float = Form(1.0),
enhance_audio: str = Form("false"),
voice_name: str = Form(None),
files: list[UploadFile] = File(default=[]),
):
if not text.strip(): raise HTTPException(400, "النص فارغ.")
# 1. NLP Processing (Model 1)
processed_text = normalize_arabic_text(text) if language == "ar" else text
# Prepare voice references
ref_paths, tmp_files = [], []
for f in files:
path = f"/tmp/ref_{uuid.uuid4().hex}_{f.filename}"
with open(path, "wb") as buf: shutil.copyfileobj(f.file, buf)
ref_paths.append(path)
tmp_files.append(path)
if voice_name:
lib_dir = os.path.join(VOICE_LIB, voice_name)
if os.path.isdir(lib_dir):
ref_paths += [os.path.join(lib_dir, fn) for fn in os.listdir(lib_dir) if fn.endswith((".wav", ".mp3"))]
if not ref_paths: raise HTTPException(400, "يجب تحديد عينة صوتية.")
# 2. TTS Generation (Model 2)
base_out_name = f"gen_base_{uuid.uuid4().hex[:8]}.wav"
base_out_path = os.path.join(OUTPUT_DIR, base_out_name)
final_out_name = base_out_name
try:
xtts_engine.tts_to_file(
text=processed_text, speaker_wav=ref_paths, language=language,
file_path=base_out_path, temperature=float(temperature), speed=float(speed)
)
# 3. Super Resolution & Enhancement (Model 3)
if enhance_audio.lower() == "true":
enhanced_name = f"gen_pro_{uuid.uuid4().hex[:8]}.wav"
enhanced_path = os.path.join(OUTPUT_DIR, enhanced_name)
enhance_audio_quality(base_out_path, enhanced_path)
final_out_name = enhanced_name
finally:
for p in tmp_files:
try: os.remove(p)
except: pass
hist = load_history()
hist.append({"filename": final_out_name, "text": processed_text[:100], "ts": int(time.time())})
save_history(hist)
return {"filename": final_out_name}
@app.get("/audio/{filename}")
def get_audio(filename: str):
path = os.path.join(OUTPUT_DIR, filename)
if not os.path.exists(path): raise HTTPException(404, "Not found")
return FileResponse(path, media_type="audio/wav")
@app.get("/history")
def get_history(): return JSONResponse(load_history())
@app.get("/voices")
def list_voices(): return [d for d in os.listdir(VOICE_LIB) if os.path.isdir(os.path.join(VOICE_LIB, d))]
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")