diff --git "a/app.py" "b/app.py"
new file mode 100644--- /dev/null
+++ "b/app.py"
@@ -0,0 +1,4132 @@
+# ╔══════════════════════════════════════════════════════════════════════╗
+# ║ XTTS v2 Voice Studio — Professional Edition v11 ║
+# ║ SINGLE-FILE BUILD (app.py + requirements.txt + Dockerfile) ║
+# ║ • CPU-only (16 vCPU / 104 GB RAM) • @GPU defined, never used ║
+# ║ • Faster-Whisper • Soft Clipping • Real Progress • Cancel Jobs ║
+# ║ • Audio Post-Editor • Full Mobile Responsive UI ║
+# ╚══════════════════════════════════════════════════════════════════════╝
+#
+# Run: python app.py
+# This file now contains everything the old Dockerfile / requirements.txt
+# did. On start-up it (1) sets the environment, (2) installs system
+# packages when running as root, (3) installs the pinned Python packages
+# once, (4) creates the runtime folders, then (5) starts the server on
+# port 7860. Set SKIP_BOOTSTRAP=1 to skip steps 2-3.
+#
+# Minimal Dockerfile that is still needed (a script cannot create its own
+# container):
+# FROM python:3.10-slim
+# RUN useradd -m -u 1000 user && mkdir -p /data && chown 1000:1000 /data
+# USER user
+# WORKDIR /home/user/app
+# COPY --chown=user app.py .
+# EXPOSE 7860
+# CMD ["python", "app.py"]
+
+import os, sys, re, time, json, uuid, shutil, asyncio, threading
+import subprocess, hashlib, tempfile, ctypes.util, traceback, zlib, math, collections, types
+
+# ══════════════════════════════════════════════════════════════════════
+# ZeroGPU COMPATIBILITY
+# Hugging Face ZeroGPU Spaces kill the app at start-up with
+# "No @spaces.GPU function detected" unless at least one function is
+# decorated with @spaces.GPU. `import spaces` must happen BEFORE torch.
+# The stub below is never called — the app itself still runs on CPU only.
+# (If you switch the Space hardware to a normal CPU tier this is harmless.)
+# ══════════════════════════════════════════════════════════════════════
+try:
+ import spaces
+ @spaces.GPU
+ def _zerogpu_stub():
+ return True
+ ZEROGPU = True
+except Exception as _e: # not on Spaces / package missing
+ spaces, ZEROGPU = None, False
+ print(f"[i] `spaces` not used: {_e}", flush=True)
+
+# ══════════════════════════════════════════════════════════════════════
+# HARDWARE PROFILE — 16 CPU cores · 104 GB RAM · NO GPU
+# ══════════════════════════════════════════════════════════════════════
+CPU_THREADS = int(os.environ.get("CPU_THREADS", "16"))
+
+# ══════════════════════════════════════════════════════════════════════
+# ENVIRONMENT (from Dockerfile ENV block)
+# ══════════════════════════════════════════════════════════════════════
+os.environ["PATH"] = os.path.expanduser("~/.local/bin") + os.pathsep + os.environ.get("PATH", "")
+os.environ["COQUI_TOS_AGREED"] = "1" # accept Coqui TOS automatically
+os.environ["TOKENIZERS_PARALLELISM"] = "false" # avoid tokenizer warnings
+if not ZEROGPU:
+ os.environ["CUDA_VISIBLE_DEVICES"] = "" # GPU is never used
+for _v in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS",
+ "NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS"):
+ os.environ[_v] = str(CPU_THREADS)
+
+# HF model cache lives on the persistent /data mount when it is writable
+if os.path.isdir("/data") and os.access("/data", os.W_OK):
+ os.environ.setdefault("HF_HOME", "/data/huggingface")
+else:
+ os.environ.setdefault("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
+
+# ══════════════════════════════════════════════════════════════════════
+# @GPU — placeholder decorator (like `spaces.GPU`).
+# It is defined only for compatibility and is INTENTIONALLY NEVER APPLIED
+# anywhere in this file. Everything runs on CPU.
+# ══════════════════════════════════════════════════════════════════════
+def GPU(func=None, **_kwargs):
+ if func is None:
+ return lambda f: f
+ return func
+
+# ══════════════════════════════════════════════════════════════════════
+# REQUIREMENTS (from requirements.txt)
+# ══════════════════════════════════════════════════════════════════════
+REQUIREMENTS = """
+coqui-tts==0.27.5
+transformers==4.57.6
+torch==2.8.0
+torchaudio==2.8.0
+soundfile>=0.12.0
+scipy>=1.10.0
+faster-whisper>=1.0.0
+deep-translator>=1.11.0
+fastapi>=0.111.0
+uvicorn[standard]>=0.29.0
+python-multipart>=0.0.9
+"""
+
+SYSTEM_PACKAGES = { # apt package -> "is it present?" check
+ "build-essential": lambda: shutil.which("gcc"),
+ "libsndfile1": lambda: ctypes.util.find_library("sndfile"),
+ "ffmpeg": lambda: shutil.which("ffmpeg"),
+ "git": lambda: shutil.which("git"),
+ "libgomp1": lambda: ctypes.util.find_library("gomp"),
+}
+
+def _sh(cmd):
+ print("[bootstrap] $", " ".join(cmd), flush=True)
+ return subprocess.run(cmd).returncode
+
+def _bootstrap():
+ """Everything the Dockerfile used to do at build time."""
+ if os.environ.get("SKIP_BOOTSTRAP") == "1":
+ return
+ if os.environ.get("SPACE_ID") and os.environ.get("FORCE_BOOTSTRAP") != "1":
+ print("[bootstrap] Running on Hugging Face Spaces — dependencies come "
+ "from requirements.txt, skipping runtime pip.", flush=True)
+ return
+
+ # ── System deps (only possible as root with apt) ──────────────────
+ missing = [p for p, ok in SYSTEM_PACKAGES.items() if not ok()]
+ if missing:
+ if hasattr(os, "geteuid") and os.geteuid() == 0 and shutil.which("apt-get"):
+ _sh(["apt-get", "update"])
+ _sh(["apt-get", "install", "-y", "--no-install-recommends"] + missing)
+ else:
+ print(f"[bootstrap] Not root — cannot apt-install: {', '.join(missing)} "
+ "(the pip wheels bundle what the app needs).", flush=True)
+
+ # ── Python deps (once per requirements version) ───────────────────
+ tag = hashlib.md5(REQUIREMENTS.encode()).hexdigest()[:10]
+ marker = os.path.join(os.path.dirname(os.path.abspath(__file__)), f".deps_ok_{tag}")
+ if os.path.exists(marker):
+ return
+ with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f:
+ f.write(REQUIREMENTS)
+ req_path = f.name
+ try:
+ _sh([sys.executable, "-m", "pip", "install", "--no-cache-dir", "--upgrade", "pip"])
+ rc = _sh([sys.executable, "-m", "pip", "install", "--no-cache-dir", "-r", req_path])
+ if rc == 0:
+ open(marker, "w").close()
+ else:
+ print("[bootstrap] pip install failed — continuing anyway.", flush=True)
+ finally:
+ try: os.remove(req_path)
+ except OSError: pass
+
+def _make_dirs():
+ """Runtime folders (from Dockerfile mkdir lines) — best effort."""
+ here = os.path.dirname(os.path.abspath(__file__))
+ for d in ("/data/huggingface", "/data/voice_library", "/data/outputs",
+ os.path.join(here, "voice_library"), os.path.join(here, "outputs")):
+ try: os.makedirs(d, exist_ok=True)
+ except OSError: pass
+
+_bootstrap()
+_make_dirs()
+
+# ══════════════════════════════════════════════════════════════════════
+# APPLICATION
+# ══════════════════════════════════════════════════════════════════════
+import numpy as np
+import soundfile as sf
+import uvicorn
+from fastapi import FastAPI, Form, File, UploadFile, HTTPException, BackgroundTasks
+from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+import torch
+
+# ══ VOICE QUALITY ENGINE (embedded — keeps the app a single file) ══
+VQ_SOURCE = r'''# ═══════════════════════════════════════════════════════════════════════
+# VOICE QUALITY ENGINE
+# 1. enhance_reference_set : clean / trim / select the best part of the sample
+# 2. normalize_text : make text digestible for XTTS (esp. Arabic)
+# 3. split_for_xtts : split by the model's REAL per-language char limit
+# 4. synth_best : N candidate takes → pick the best one (ASR check)
+# 5. assemble / finalize : natural pauses, no tanh distortion
+# Pure numpy/scipy/soundfile — model & ASR are injected, so it is testable.
+# ═══════════════════════════════════════════════════════════════════════
+import os, re, io, json, math, hashlib, subprocess, tempfile, difflib, unicodedata
+import numpy as np
+import soundfile as sf
+from scipy import signal
+from scipy.ndimage import uniform_filter, binary_closing, binary_opening
+
+REF_SR = 22050 # XTTS conditioning sample-rate
+XTTS_SR = 24000 # XTTS output sample-rate
+VQ_VERSION = "1"
+
+# ── generic helpers ───────────────────────────────────────────────────
+def _db(x): return 20.0 * np.log10(np.maximum(x, 1e-9))
+
+def load_mono(path, sr=REF_SR, max_sec=180):
+ """→ (float32 mono @sr, original_sample_rate)."""
+ try:
+ data, fs = sf.read(path, dtype="float32", always_2d=True)
+ except Exception:
+ tmp = tempfile.mktemp(suffix=".wav")
+ try:
+ subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", path,
+ "-ac", "1", tmp], check=True, timeout=120)
+ data, fs = sf.read(tmp, dtype="float32", always_2d=True)
+ finally:
+ if os.path.exists(tmp): os.remove(tmp)
+ x = data.mean(axis=1).astype(np.float32)
+ x = x[: int(fs * max_sec)]
+ if fs != sr:
+ g = math.gcd(int(fs), int(sr))
+ x = signal.resample_poly(x, int(sr) // g, int(fs) // g).astype(np.float32)
+ return x, int(fs)
+
+def frame_db(x, sr, win=0.025, hop=0.010):
+ w, h = int(sr * win), int(sr * hop)
+ if len(x) < w:
+ x = np.pad(x, (0, w - len(x)))
+ n = 1 + (len(x) - w) // h
+ idx = np.arange(w)[None, :] + h * np.arange(n)[:, None]
+ rms = np.sqrt(np.mean(x[idx] ** 2, axis=1) + 1e-12)
+ return _db(rms), w, h
+
+def _highpass(x, sr, fc=70.0, order=4):
+ sos = signal.butter(order, fc, btype="highpass", fs=sr, output="sos")
+ return signal.sosfiltfilt(sos, x).astype(np.float32)
+
+def _fade(x, sr, fin_ms=8, fout_ms=25):
+ x = x.copy()
+ a, b = int(sr * fin_ms / 1000), int(sr * fout_ms / 1000)
+ if a and len(x) > a: x[:a] *= np.linspace(0, 1, a, dtype=np.float32)
+ if b and len(x) > b: x[-b:] *= np.linspace(1, 0, b, dtype=np.float32)
+ return x
+
+def soft_limit(x, ceiling=0.89, knee=0.70):
+ """Transparent below `knee`, smooth tanh-shaped compression above it
+ (no colouring of normal signal — unlike a global tanh(x*1.8))."""
+ y = x.copy()
+ a = np.abs(y)
+ m = a > knee
+ if m.any():
+ over = (a[m] - knee) / max(ceiling - knee, 1e-6)
+ y[m] = np.sign(y[m]) * (knee + (ceiling - knee) * np.tanh(over))
+ return y.astype(np.float32)
+
+# ═══════════════════════════════════════════════════════════════════════
+# 1. REFERENCE ENHANCEMENT
+# ═══════════════════════════════════════════════════════════════════════
+def spectral_denoise(x, sr, strength=1.0):
+ """Spectral-subtraction gate; noise profile = quietest 15 % of frames.
+ Smoothed in time+frequency to avoid 'musical noise'."""
+ nper, nov = 1024, 768
+ _, _, X = signal.stft(x, sr, nperseg=nper, noverlap=nov)
+ mag = np.abs(X)
+ fr = _db(mag.mean(axis=0))
+ quiet = fr <= np.percentile(fr, 15)
+ if quiet.sum() < 3:
+ return x
+ mu = mag[:, quiet].mean(axis=1, keepdims=True)
+ alpha = 1.0 + 1.2 * strength
+ gain = np.clip((mag - alpha * mu) / (mag + 1e-9), 0.0, 1.0)
+ floor = 10 ** (-(10 + 10 * strength) / 20) # −20 … −30 dB
+ gain = np.maximum(gain, floor)
+ gain = uniform_filter(gain, size=(3, 5), mode="nearest")
+ _, y = signal.istft(X * gain, sr, nperseg=nper, noverlap=nov)
+ y = y[: len(x)]
+ if len(y) < len(x): y = np.pad(y, (0, len(x) - len(y)))
+ return y.astype(np.float32)
+
+def _voiced_mask(db, min_gap=15, min_seg=30):
+ noise, peak = np.percentile(db, 10), np.percentile(db, 95)
+ thr = max(noise + 10.0, peak - 28.0)
+ v = db > thr
+ v = binary_closing(v, structure=np.ones(min_gap, bool)) # bridge ≤150 ms
+ v = binary_opening(v, structure=np.ones(min_seg, bool)) # drop < 300 ms
+ return v
+
+def _segments(mask):
+ d = np.diff(np.concatenate([[0], mask.astype(np.int8), [0]]))
+ return list(zip(np.where(d == 1)[0], np.where(d == -1)[0]))
+
+def _snap_low(db, i, span=50):
+ lo, hi = max(0, i - span), min(len(db), i + span + 1)
+ return lo + int(np.argmin(db[lo:hi]))
+
+def _compact_speech(x, sr, db, w, h, mask):
+ """Keep speech + short natural pauses; squeeze long dead air."""
+ segs = _segments(mask)
+ if not segs: return x
+ pad = int(0.06 * sr)
+ parts, prev_end = [], None
+ for s, e in segs:
+ a = max(0, s * h - pad); b = min(len(x), e * h + w + pad)
+ if prev_end is not None and a > prev_end:
+ gap = x[prev_end:a]
+ keep = int(0.30 * sr)
+ gap = gap if len(gap) <= keep else np.concatenate([gap[: keep // 2], gap[-keep // 2:]])
+ parts.append(_fade(gap, sr, 10, 10))
+ parts.append(x[max(a, prev_end or 0):b])
+ prev_end = b
+ return np.concatenate(parts).astype(np.float32)
+
+def _best_blocks(z, sr, target_sec, block_sec=6.0):
+ """Pick the cleanest ~target_sec of audio (blocks cut on low-energy frames)."""
+ if len(z) <= target_sec * sr: return z
+ db, w, h = frame_db(z, sr)
+ nfr = len(db)
+ per = int(block_sec * 100) # frames per block (10 ms hop)
+ cuts = sorted({0, nfr, *[_snap_low(db, i) for i in range(per, nfr - per // 2, per)]})
+ blocks = [(a, b) for a, b in zip(cuts[:-1], cuts[1:]) if b - a > 150]
+ if not blocks: return z[: int(target_sec * sr)]
+ scores = []
+ for a, b in blocks:
+ seg = db[a:b]; v = seg > max(np.percentile(seg, 10) + 10, np.percentile(seg, 95) - 28)
+ clip = float(np.mean(np.abs(z[a * h:b * h + w]) > 0.98))
+ snr = (seg[v].mean() - np.percentile(seg, 10)) if v.any() else 0.0
+ sd = float(seg[v].std()) if v.sum() > 3 else 12.0
+ scores.append(snr - 0.5 * sd - 300 * clip)
+ order = np.argsort(scores)[::-1]
+ chosen, tot = [], 0.0
+ for i in order:
+ a, b = blocks[i]; d = (b - a) / 100.0
+ if tot >= target_sec: break
+ chosen.append(i); tot += d
+ chosen.sort()
+ return np.concatenate([z[blocks[i][0] * h: blocks[i][1] * h + w] for i in chosen]).astype(np.float32)
+
+def _analyze(x_native_norm, fs_native, sr=REF_SR):
+ """Diagnostics on a (peak-normalised) signal."""
+ db, w, h = frame_db(x_native_norm, sr)
+ noise, peak = float(np.percentile(db, 10)), float(np.percentile(db, 95))
+ f, p = signal.welch(x_native_norm, sr, nperseg=2048)
+ cum = np.cumsum(p) / max(p.sum(), 1e-12)
+ rolloff = float(f[np.searchsorted(cum, 0.995)])
+ return dict(noise_floor_db=round(noise, 1), snr_db=round(peak - noise, 1),
+ bandwidth_hz=int(min(rolloff, fs_native / 2)), native_sr=int(fs_native))
+
+def enhance_reference(path, strength=1.0, target_sec=24.0, denoise=True):
+ """→ (clean float32 @22050, report dict)."""
+ raw, fs = load_mono(path)
+ if len(raw) < 0.5 * REF_SR:
+ raise ValueError("العينة قصيرة جداً (أقل من نصف ثانية).")
+ dur_in = len(raw) / REF_SR
+ clip_pct = float(np.mean(np.abs(raw) >= 0.995) * 100)
+ raw = raw - float(np.mean(raw))
+ pk = float(np.max(np.abs(raw)) + 1e-9)
+ x = raw / pk * 0.7
+ before = _analyze(x, fs)
+
+ x = _highpass(x, REF_SR, 70)
+ did_denoise = False
+ if denoise and before["noise_floor_db"] > -62:
+ s = float(np.clip((before["noise_floor_db"] + 62) / 20.0, 0.35, 1.0)) * strength
+ x = spectral_denoise(x, REF_SR, s); did_denoise = True
+
+ db, w, h = frame_db(x, REF_SR)
+ mask = _voiced_mask(db)
+ speech_sec_all = float(mask.sum() * 0.01)
+ z = _compact_speech(x, REF_SR, db, w, h, mask) if mask.any() else x
+ z = _best_blocks(z, REF_SR, target_sec)
+
+ dbz, wz, hz = frame_db(z, REF_SR)
+ mz = _voiced_mask(dbz)
+ lvl = float(np.sqrt(np.mean(np.square(np.concatenate(
+ [z[s * hz:e * hz + wz] for s, e in _segments(mz)]) if mz.any() else z))) + 1e-9)
+ z = z * (10 ** (-20 / 20) / lvl) # voiced RMS → −20 dBFS
+ z = soft_limit(z, 0.95, 0.85)
+ z = _fade(z, REF_SR, 10, 30)
+ after = _analyze(z, fs)
+
+ warns, tips = [], []
+ if speech_sec_all < 6:
+ warns.append(f"الكلام الفعلي في العينة {speech_sec_all:.1f} ث فقط."); tips.append("استخدم عينة من 10–25 ثانية كلام متصل.")
+ if clip_pct > 0.5:
+ warns.append(f"تشويه قص (clipping) {clip_pct:.1f}٪ — لا يمكن إصلاحه."); tips.append("أعد التسجيل بمستوى أقل.")
+ if before["snr_db"] < 18:
+ warns.append(f"ضجيج عالٍ (SNR≈{before['snr_db']:.0f} dB)."); tips.append("سجّل في غرفة هادئة قريباً من الميكروفون.")
+ if before["bandwidth_hz"] < 6000:
+ warns.append(f"جودة ترددية منخفضة (حتى {before['bandwidth_hz']} Hz) — غالباً هاتف/ضغط شديد."); tips.append("استخدم WAV/FLAC أو تسجيلاً بجودة أعلى.")
+ if before["native_sr"] < 16000:
+ warns.append(f"معدل العينة الأصلي {before['native_sr']} Hz منخفض.")
+ score = 100.0
+ score -= max(0, 20 - before["snr_db"]) * 1.5
+ score -= min(30, clip_pct * 10)
+ score -= max(0, 6 - speech_sec_all) * 5
+ score -= 15 if before["bandwidth_hz"] < 6000 else 0
+ score = int(np.clip(score, 0, 100))
+
+ report = dict(duration_in=round(dur_in, 1), duration_out=round(len(z) / REF_SR, 1),
+ speech_sec=round(speech_sec_all, 1), clipping_pct=round(clip_pct, 2),
+ denoised=did_denoise, before=before, after=after,
+ score=score, warnings=warns, tips=tips)
+ return z.astype(np.float32), report
+
+def _file_sha(path):
+ h = hashlib.sha1()
+ with open(path, "rb") as f:
+ for b in iter(lambda: f.read(1 << 20), b""): h.update(b)
+ return h.hexdigest()
+
+def enhance_reference_set(paths, cache_dir, strength=1.0, target_sec=24.0,
+ max_total=30.0, denoise=True):
+ """Clean every file, merge best-first into ONE reference (XTTS only reads
+ the first `gpt_cond_len` s of the concatenation, so ordering matters).
+ → (merged_wav_path, [per-file reports], summary)."""
+ os.makedirs(cache_dir, exist_ok=True)
+ key = hashlib.sha1(("|".join(_file_sha(p) for p in paths) +
+ f"|{strength}|{target_sec}|{max_total}|{denoise}|{VQ_VERSION}").encode()).hexdigest()[:20]
+ out, meta = os.path.join(cache_dir, f"ref_{key}.wav"), os.path.join(cache_dir, f"ref_{key}.json")
+ if os.path.exists(out) and os.path.exists(meta):
+ d = json.load(open(meta, encoding="utf-8"))
+ return out, d["reports"], d["summary"]
+
+ per = target_sec if len(paths) == 1 else max(8.0, target_sec / len(paths) * 1.25)
+ items = []
+ for p in paths:
+ try:
+ z, rep = enhance_reference(p, strength, per, denoise)
+ rep["file"] = os.path.basename(p)[-40:]
+ items.append((z, rep))
+ except Exception as e:
+ items.append((None, dict(file=os.path.basename(p)[-40:], error=str(e), score=0,
+ warnings=[f"تعذّرت معالجة الملف: {e}"], tips=[])))
+ good = [(z, r) for z, r in items if z is not None]
+ if not good:
+ raise ValueError("تعذّرت معالجة كل ملفات العينة: " + "; ".join(r.get("error", "?") for _, r in items))
+ good.sort(key=lambda t: t[1]["score"], reverse=True)
+ gap = np.zeros(int(0.25 * REF_SR), np.float32)
+ merged = np.concatenate([np.concatenate([z, gap]) for z, _ in good])[: int(max_total * REF_SR)]
+ sf.write(out, merged, REF_SR, subtype="PCM_16")
+ reports = [r for _, r in items]
+ best = good[0][1]
+ summary = dict(score=int(np.mean([r["score"] for _, r in good])),
+ duration=round(len(merged) / REF_SR, 1),
+ files=len(paths), used=len(good),
+ warnings=[w for _, r in good for w in r["warnings"]][:6],
+ tips=list(dict.fromkeys(t for _, r in good for t in r["tips"]))[:4],
+ snr_before=best["before"]["snr_db"], snr_after=best["after"]["snr_db"],
+ denoised=any(r["denoised"] for _, r in good))
+ json.dump(dict(reports=reports, summary=summary), open(meta, "w", encoding="utf-8"), ensure_ascii=False)
+ return out, reports, summary
+
+# ═══════════════════════════════════════════════════════════════════════
+# 2. TEXT NORMALISATION
+# ═══════════════════════════════════════════════════════════════════════
+_AR_DIAC = re.compile("[\u0610-\u061A\u064B-\u065F\u0670\u06D6-\u06ED]")
+_INVIS = re.compile("[\u200b\u200c\u200d\u200e\u200f\u202a-\u202e\u2066-\u2069\ufeff]")
+_EASTERN = str.maketrans("٠١٢٣٤٥٦٧٨٩۰۱۲۳۴۵۶۷۸۹", "01234567890123456789")
+
+def normalize_text(text, lang="ar", strip_diacritics=True):
+ t = unicodedata.normalize("NFKC" if lang.startswith("ar") else "NFC", text)
+ t = _INVIS.sub("", t).translate(_EASTERN)
+ if lang.startswith("ar"):
+ t = t.replace("\u0640", "")
+ if strip_diacritics: t = _AR_DIAC.sub("", t)
+ t = re.sub(r"[“”„«»\"‹›]", "", t)
+ t = t.replace("’", "'").replace("‘", "'")
+ t = re.sub(r"[*_#`~|^<>\[\]{}]", " ", t)
+ comma = "، " if lang.startswith("ar") else ", "
+ t = re.sub(r"\s*[—–]\s*|\s+-\s+", comma, t)
+ t = re.sub(r"\s*[()]\s*", comma, t)
+ t = re.sub(r"\.{4,}", "...", t)
+ t = re.sub(r"([!?؟])\1+", r"\1", t)
+ t = re.sub(r"[ \t]+", " ", t)
+ t = re.sub(r"\s*([,،;؛:])\s*(?=[,،;؛:.!?؟])", r"\1", t) # ", ," / ", ."
+ t = re.sub(r"[,،]\s*([.!?؟])", r"\1", t)
+ t = re.sub(r"\n{3,}", "\n\n", t).strip()
+ if t and not re.search(r"[.!?؟…。！？]$", t):
+ t += "。" if lang in ("zh", "zh-cn", "ja") else "."
+ return t
+
+# ════════════════════════════════════════════════════════════════════��══
+# 3. SPLITTING (real per-language limits)
+# ═══════════════════════════════════════════════════════════════════════
+DEFAULT_CHAR_LIMITS = {"en": 250, "de": 253, "fr": 273, "es": 239, "it": 213, "pt": 203,
+ "pl": 224, "zh": 82, "ar": 166, "cs": 186, "ru": 182, "nl": 251,
+ "tr": 226, "ja": 71, "hu": 224, "ko": 95, "hi": 150}
+PAUSE_MS = {".": 400, "!": 380, "?": 420, "؟": 420, "…": 520, "。": 400, "！": 380, "？": 420,
+ "none": 300, "clause": 170, "para": 760}
+_CLAUSE = re.compile(r"(?<=[,،;؛:])\s+")
+
+def _split_long(s, limit):
+ """Split an over-long sentence on clause marks, else on spaces."""
+ if len(s) <= limit: return [s]
+ mid = len(s) // 2
+ cands = [m.end() for m in _CLAUSE.finditer(s) if 15 <= m.end() <= len(s) - 15]
+ if not cands:
+ cands = [m.end() for m in re.finditer(r"\s+", s) if 15 <= m.end() <= len(s) - 15]
+ if not cands:
+ return [s[:limit]] + _split_long(s[limit:], limit)
+ cut = min(cands, key=lambda c: abs(c - mid))
+ return _split_long(s[:cut].strip(), limit) + _split_long(s[cut:].strip(), limit)
+
+def split_for_xtts(text, lang="ar", char_limits=None, safety=0.85, min_chunk=28):
+ """→ [(chunk_text, pause_after_ms)] (pause 0 for the very last chunk)."""
+ lim = dict(DEFAULT_CHAR_LIMITS); lim.update(char_limits or {})
+ base = "zh" if lang.startswith("zh") else lang.split("-")[0]
+ limit = max(30, int(lim.get(base, 250) * safety))
+ out = []
+ paras = [p for p in re.split(r"\n\s*\n", text.strip()) if p.strip()]
+ for pi, para in enumerate(paras):
+ sents = [s.strip() for s in re.split(r"(?<=[.!?؟…])\s+|(?<=[。！？])|\n", para) if s and s.strip()]
+ items = []
+ for s in sents:
+ end = s[-1] if s[-1] in PAUSE_MS else "none"
+ pieces = _split_long(s, limit)
+ for j, pc in enumerate(pieces):
+ items.append((pc, PAUSE_MS[end] if j == len(pieces) - 1 else PAUSE_MS["clause"]))
+ # merge tiny fragments into a neighbour (very short chunks sound odd)
+ merged = []
+ for txt, pz in items:
+ if merged and len(merged[-1][0]) < min_chunk and len(merged[-1][0]) + 1 + len(txt) <= limit:
+ merged[-1] = (merged[-1][0] + " " + txt, pz)
+ else:
+ merged.append((txt, pz))
+ if len(merged) > 1 and len(merged[-1][0]) < min_chunk and len(merged[-2][0]) + 1 + len(merged[-1][0]) <= limit:
+ merged[-2] = (merged[-2][0] + " " + merged[-1][0], merged[-1][1]); merged.pop()
+ if merged and pi < len(paras) - 1:
+ merged[-1] = (merged[-1][0], PAUSE_MS["para"])
+ out.extend(merged)
+ if out: out[-1] = (out[-1][0], 0)
+ return out or [(text.strip(), 0)]
+
+# ═══════════════════════════════════════════════════════════════════════
+# 4. TAKE SELECTION
+# ═══════════════════════════════════════════════════════════════════════
+def trim_take(wav, sr=XTTS_SR, rel_db=-42.0, lead_ms=40, tail_ms=70):
+ wav = np.asarray(wav, np.float32).reshape(-1)
+ if len(wav) < int(0.1 * sr): return wav
+ db, w, h = frame_db(wav, sr)
+ idx = np.where(db > db.max() + rel_db)[0]
+ if len(idx) == 0: return wav
+ s = max(0, idx[0] * h - int(lead_ms * sr / 1000))
+ e = min(len(wav), idx[-1] * h + w + int(tail_ms * sr / 1000))
+ return _fade(wav[s:e], sr, 8, 25)
+
+_NORM_RE = re.compile(r"[^\w]+", re.UNICODE)
+def norm_for_compare(s):
+ s = _AR_DIAC.sub("", unicodedata.normalize("NFKC", s)).replace("\u0640", "")
+ s = re.sub("[إأآ]", "ا", s).replace("ى", "ي").replace("ة", "ه")
+ return _NORM_RE.sub("", s.lower().translate(_EASTERN))
+
+def text_similarity(a, b):
+ a, b = norm_for_compare(a), norm_for_compare(b)
+ if not a or not b: return 0.0
+ return difflib.SequenceMatcher(None, a, b, autojunk=False).ratio()
+
+def take_info(wav, sr, text, asr_fn=None):
+ dur = len(wav) / sr
+ peak = float(np.max(np.abs(wav))) if len(wav) else 0.0
+ cps = len(norm_for_compare(text)) / max(dur, 1e-3)
+ sim = None
+ if asr_fn is not None and dur > 0.2:
+ try: sim = text_similarity(text, asr_fn(wav, sr))
+ except Exception: sim = None
+ ok = peak > 0.02 and 4.0 <= cps <= 28.0
+ return dict(dur=dur, cps=cps, peak=peak, sim=sim, ok=ok,
+ clip=float(np.mean(np.abs(wav) > 0.99)) if len(wav) else 0.0)
+
+def synth_best(gen_fn, text, attempts=1, asr_fn=None, sr=XTTS_SR, seed=1234, accept_sim=0.90):
+ """gen_fn(seed:int) → float32 wav. Returns (wav, info, n_generated)."""
+ takes = []
+ for k in range(max(1, attempts)):
+ wav = trim_take(gen_fn(seed + 7919 * k), sr)
+ info = take_info(wav, sr, text, asr_fn)
+ takes.append((wav, info))
+ if attempts > 1 and info["ok"] and info["sim"] is not None and info["sim"] >= accept_sim:
+ break
+ if len(takes) == 1: return takes[0][0], takes[0][1], 1
+ med = float(np.median([i["dur"] for _, i in takes]))
+ def score(i):
+ s = (i["sim"] * 100.0) if i["sim"] is not None else 60.0
+ s -= 25.0 * abs(math.log(max(i["dur"], 1e-3) / max(med, 1e-3)))
+ s -= 300.0 * i["clip"]
+ s -= 0 if i["ok"] else 30.0
+ return s
+ best = max(takes, key=lambda t: score(t[1]))
+ return best[0], best[1], len(takes)
+
+# ═══════════════════════════════════════════════════════════════════════
+# 5. ASSEMBLY / FINALISATION
+# ═══════════════════════════════════════════════════════════════════════
+def assemble(takes, pauses_ms, sr=XTTS_SR, pause_scale=1.0, seed=7, jitter=0.12):
+ """Join takes with punctuation-aware pauses (seeded jitter, ±12 %)."""
+ rng = np.random.default_rng(seed)
+ parts = []
+ for wav, pz in zip(takes, pauses_ms):
+ parts.append(np.asarray(wav, np.float32))
+ if pz > 0:
+ ms = pz * pause_scale * (1.0 + rng.uniform(-jitter, jitter))
+ parts.append(np.zeros(int(sr * ms / 1000), np.float32))
+ return np.concatenate(parts) if parts else np.zeros(sr, np.float32)
+
+def finalize_audio(wav, sr=XTTS_SR, target_rms_db=-19.0):
+ wav = _highpass(np.asarray(wav, np.float32), sr, 60, 2)
+ db, w, h = frame_db(wav, sr)
+ v = db > max(np.percentile(db, 10) + 10, np.percentile(db, 95) - 30)
+ if v.any():
+ lvl = 10 ** (np.percentile(db[v], 50) / 20)
+ gain = np.clip(10 ** (target_rms_db / 20) / max(lvl, 1e-6), 0.25, 6.0)
+ wav = wav * gain
+ return soft_limit(wav, 0.95, 0.80)
+'''
+vq = types.ModuleType("voice_quality")
+exec(compile(VQ_SOURCE, "voice_quality", "exec"), vq.__dict__)
+
+torch.set_num_threads(CPU_THREADS) # use all 16 cores for inference
+try:
+ torch.set_num_interop_threads(2)
+except RuntimeError:
+ pass
+
+# ── Persistent Storage ────────────────────────────────────────────────
+PERSISTENT_DIR = "/data" if os.path.exists("/data") else os.path.join(os.getcwd(), "app_data")
+VOICE_LIB = os.path.join(PERSISTENT_DIR, "voice_library")
+OUTPUT_DIR = os.path.join(PERSISTENT_DIR, "outputs")
+HISTORY_FILE = os.path.join(PERSISTENT_DIR, "history.json")
+MAX_CHARS = 5000
+
+for d in [VOICE_LIB, OUTPUT_DIR]:
+ os.makedirs(d, exist_ok=True)
+
+HF_TOKEN = os.environ.get("HF_TOKEN")
+DATASET_REPO_ID = os.environ.get("DATASET_REPO_ID")
+
+try:
+ from huggingface_hub import HfApi, snapshot_download
+ HUB_AVAILABLE = True
+ if HF_TOKEN and DATASET_REPO_ID:
+ print("[*] Restoring data from Hugging Face Hub …")
+ try:
+ snapshot_download(repo_id=DATASET_REPO_ID, repo_type="dataset",
+ local_dir=PERSISTENT_DIR, token=HF_TOKEN)
+ print("[✓] Data restoration complete.")
+ except Exception as e:
+ print(f"[!] Warning: Could not download backup ({e})")
+except ImportError:
+ HUB_AVAILABLE = False
+
+def trigger_cloud_backup():
+ if not (HUB_AVAILABLE and HF_TOKEN and DATASET_REPO_ID): return
+ def _run():
+ try:
+ HfApi(token=HF_TOKEN).upload_folder(
+ folder_path=PERSISTENT_DIR, repo_id=DATASET_REPO_ID,
+ repo_type="dataset", commit_message=f"Auto-backup: {int(time.time())}")
+ except Exception as e:
+ print(f"[!] Cloud backup failed: {e}")
+ threading.Thread(target=_run).start()
+
+# ── XTTS Model ────────────────────────────────────────────────────────
+from TTS.api import TTS
+device = "cpu" # GPU intentionally disabled — CPU only
+
+# ── Background model loading (server starts immediately) ──────────────
+xtts = None
+xtts_ready = False
+xtts_loading = True
+xtts_error = None
+
+def _load_model():
+ global xtts, xtts_ready, xtts_loading, xtts_error
+ try:
+ print(f"[*] Loading XTTS v2 on {device.upper()} …")
+ xtts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device)
+ xtts_ready = True
+ print("[✓] XTTS ready.")
+ except Exception as e:
+ xtts_error = f"{type(e).__name__}: {e}"
+ print(f"[!] XTTS loading failed: {xtts_error}", flush=True)
+ traceback.print_exc()
+ finally:
+ xtts_loading = False
+
+threading.Thread(target=_load_model, daemon=True).start()
+
+# tts_lock is created lazily inside the event loop to avoid asyncio issues
+_tts_lock = None
+def get_tts_lock():
+ global _tts_lock
+ if _tts_lock is None:
+ _tts_lock = asyncio.Lock()
+ return _tts_lock
+
+def _startup_cleanup():
+ for f in os.listdir("/tmp"):
+ if any(f.startswith(p) for p in ("ch_","ref_","sts_","dub_")):
+ try: os.remove(os.path.join("/tmp",f))
+ except: pass
+_startup_cleanup()
+
+# ── Job Management: Cancellation & Progress ───────────────────────────
+job_progress: dict = {} # job_id -> {current, total, status, ts}
+cancel_events: dict = {} # job_id -> threading.Event
+JOB_TTL_SEC = 3600
+
+def _gc_jobs():
+ cutoff = time.time() - JOB_TTL_SEC
+ for k in [k for k,v in list(job_progress.items())
+ if v.get("status") != "running" and v.get("ts",0) < cutoff]:
+ job_progress.pop(k, None)
+
+# ── Faster-Whisper (replaces openai-whisper) ──────────────────────────
+whisper_mdl = None
+WHISPER_OK = False
+
+def _load_whisper():
+ """Load Faster-Whisper in the background. If the default HF cache
+ (e.g. the /data volume) gives an I/O error, retry from local disk."""
+ global whisper_mdl, WHISPER_OK
+ try:
+ from faster_whisper import WhisperModel
+ except Exception as e:
+ print(f"[!] Faster-Whisper not installed: {e}", flush=True)
+ return
+ for root in (None, "/tmp/whisper_models"):
+ try:
+ if root: os.makedirs(root, exist_ok=True)
+ whisper_mdl = WhisperModel("small", device="cpu", compute_type="int8",
+ cpu_threads=CPU_THREADS, download_root=root)
+ WHISPER_OK = True
+ print(f"[✓] Faster-Whisper ready (small · int8 · CPU) [{root or 'default cache'}].", flush=True)
+ return
+ except Exception as e:
+ print(f"[!] Faster-Whisper load failed ({root or 'default cache'}): {e}", flush=True)
+ print("[!] Faster-Whisper not available.", flush=True)
+
+threading.Thread(target=_load_whisper, daemon=True).start()
+
+def transcribe_audio(path: str) -> str:
+ """Transcribe using Faster-Whisper; returns plain text."""
+ if whisper_mdl is None:
+ raise RuntimeError("Faster-Whisper not loaded")
+ segments, _ = whisper_mdl.transcribe(path, beam_size=5)
+ return " ".join(s.text for s in segments).strip()
+
+# ── Translator ────────────────────────────────────────────────────────
+try:
+ from deep_translator import GoogleTranslator
+ TRANSLATOR_OK = True
+except Exception:
+ TRANSLATOR_OK = False
+
+# ── History helpers ───────────────────────────────────────────────────
+def load_history():
+ try:
+ return json.load(open(HISTORY_FILE)) if os.path.exists(HISTORY_FILE) else []
+ except Exception:
+ return []
+
+def save_history(h):
+ json.dump(h, open(HISTORY_FILE, "w"), ensure_ascii=False, indent=2)
+ trigger_cloud_backup()
+
+MAX_HISTORY = 300
+MAX_OUTPUTS = 200
+
+def _prune_outputs():
+ try:
+ wavs = sorted(
+ [f for f in os.listdir(OUTPUT_DIR) if f.endswith(".wav")],
+ key=lambda f: os.path.getmtime(os.path.join(OUTPUT_DIR, f))
+ )
+ for old in wavs[:-MAX_OUTPUTS]:
+ try: os.remove(os.path.join(OUTPUT_DIR, old))
+ except: pass
+ except Exception:
+ pass
+
+def append_history(filename, text, language, mode, char_count=0):
+ h = load_history()
+ h.append(dict(filename=filename, text=text[:150], language=language,
+ mode=mode, char_count=char_count, ts=int(time.time())))
+ if len(h) > MAX_HISTORY:
+ h = h[-MAX_HISTORY:]
+ save_history(h)
+ _prune_outputs()
+
+# ── Parameter Engineering ─────────────────────────────────────────────
+def stability_to_temperature(s): return round(0.85 - (s * 0.40), 3)
+def clarity_to_top_p(c): return round(0.70 + (c * 0.25), 3)
+def style_to_rep_penalty(st): return round(2.00 + (st * 2.00), 2)
+
+# ── Expression Presets ────────────────────────────────────────────────
+# Each preset overrides stability/temperature/top_p/rep_penalty to push
+# XTTS toward a specific emotional character.
+EXPRESSION_PRESETS = {
+ "natural": dict(temperature=0.72, top_p=0.88, rep_penalty=2.0, speed_mult=1.00),
+ "warm": dict(temperature=0.78, top_p=0.90, rep_penalty=1.80, speed_mult=0.97),
+ "excited": dict(temperature=0.88, top_p=0.92, rep_penalty=1.65, speed_mult=1.10),
+ "sad": dict(temperature=0.65, top_p=0.85, rep_penalty=2.20, speed_mult=0.88),
+ "serious": dict(temperature=0.55, top_p=0.82, rep_penalty=2.50, speed_mult=0.95),
+ "story": dict(temperature=0.82, top_p=0.93, rep_penalty=1.75, speed_mult=0.96),
+ "news": dict(temperature=0.60, top_p=0.84, rep_penalty=2.40, speed_mult=1.02),
+ "child": dict(temperature=0.90, top_p=0.95, rep_penalty=1.60, speed_mult=1.05),
+}
+
+def apply_expression_preset(preset_name: str, speed: float) -> dict:
+ p = EXPRESSION_PRESETS.get(preset_name, EXPRESSION_PRESETS["natural"])
+ return {
+ "temperature": p["temperature"],
+ "top_p": p["top_p"],
+ "repetition_penalty": p["rep_penalty"],
+ "speed": round(speed * p["speed_mult"], 3),
+ }
+
+# ── Text Expressiveness Enhancer ──────────────────────────────────────
+# XTTS is very sensitive to punctuation and sentence rhythm.
+# Adding deliberate pauses, ellipses, and varied sentence endings
+# dramatically improves naturalness and eliminates the "robot reading" feel.
+def enhance_text_expression(text: str, lang: str, mode: str) -> str:
+ """
+ Enrich text with punctuation cues that guide XTTS prosody.
+ Does NOT change words — only adds/adjusts punctuation & whitespace.
+ """
+ if mode == "natural" or not text.strip():
+ return text
+
+ t = text.strip()
+
+ # 1. Normalise whitespace
+ t = re.sub(r'[ \t]+', ' ', t)
+
+ # 2. Ensure sentences end with terminal punctuation
+ # (XTTS pauses longer on . than on nothing)
+ t = re.sub(r'([^\.\!\?؟،,\n])\n', r'\1.\n', t)
+
+ # 3. Mode-specific rhythm adjustments
+ if mode == "excited":
+ # Shorten pauses between sentences → rapid delivery
+ t = re.sub(r'\.\s+', '. ', t)
+ # Add emphasis markers via comma-pauses before key conjunctions
+ for conj in ['لكن','بل','إلا','however','but','yet','still']:
+ t = re.sub(rf'\b({conj})\b', r', \1', t, flags=re.IGNORECASE)
+
+ elif mode == "sad":
+ # Extend pauses → slower, heavier delivery
+ t = re.sub(r'\.\s+', '… ', t)
+ t = re.sub(r'،\s+', '، ', t)
+
+ elif mode == "story":
+ # Add brief breath-pause after scene-setting clauses
+ t = re.sub(r'(كان|كانت|في يوم|ذات يوم|once upon|there was)',
+ r'\1،', t, flags=re.IGNORECASE)
+
+ elif mode == "news":
+ # Crisp, no trailing breath
+ t = re.sub(r'\s*…+\s*', '. ', t)
+
+ elif mode == "child":
+ # Short bursts with upward inflection hints
+ t = re.sub(r'\.\s+([A-Zا-ي])', r'! \1', t)
+
+ # 4. Collapse multiple punctuation
+ t = re.sub(r'([.!?؟،,]){2,}', r'\1', t)
+ t = re.sub(r'…{2,}', '…', t)
+
+ return t
+
+# ╔══════════════════════════════════════════════════════════════════════╗
+# ║ P H R A S E - L E V E L P R O S O D Y V A R I A T I O N ║
+# ║ Each sentence is analysed for emotional content. ║
+# ║ XTTS parameters (temperature, speed, repetition_penalty) shift ║
+# ║ per-phrase so the delivery feels alive, not machine-read. ║
+# ║ Post-generation volume envelopes reinforce the prosodic contour. ║
+# ╚══════════════════════════════════════════════════════════════════════╝
+
+# ── Per-emotion XTTS deltas ───────────────────────────────────────────
+# (applied on top of the base params; scaled by user's variation_strength)
+EMOTION_DELTAS = {
+ # temp speed_mult rep_penalty silence_ms
+ "neutral": ( 0.00, 1.000, 0.00, 110 ),
+ "question": (+0.07, 0.960, -0.10, 140 ), # slower, more uncertain
+ "exclaim": (+0.12, 1.080, -0.30, 75 ), # faster, more energy
+ "emphasis": (-0.06, 0.900, +0.30, 155 ), # deliberate, weighty
+ "sadness": (-0.10, 0.870, +0.35, 185 ), # slow, heavy
+ "joy": (+0.10, 1.060, -0.25, 85 ), # bright, forward
+ "anger": (+0.08, 1.040, -0.20, 95 ), # tense, clipped
+ "soft": (-0.09, 0.920, +0.20, 160 ), # tender, gentle
+ "urgency": (+0.11, 1.120, -0.25, 65 ), # fast, alert
+ "story": (+0.06, 0.950, -0.10, 135 ), # varied, narrative
+}
+
+# ── Keyword patterns (Arabic + English) ──────────────────────────────
+EMOTION_PATTERNS = {
+ "joy": [
+ # Arabic
+ r'سعيد|فرح|رائع|مذهل|عظيم|ممتاز|برافو|مبروك|ياه|يا لروع|'
+ r'جميل|بديع|أحب|نجح|وفق|الله|يسعد|سرور|ابتهاج|بهجة|نشوة',
+ # English
+ r'happy|wonderful|amazing|great|excellent|love|joy|fantastic|'
+ r'brilliant|perfect|congrats|hooray|yay|delightful|terrific',
+ ],
+ "sadness": [
+ r'حزين|للأسف|مؤسف|وداع|ألم|فقدنا|صعب|مؤلم|بكى|دموع|'
+ r'خسر|رحل|غاب|اشتاق|أفتقد|حسرة|مأساة|مصيبة|فراق|وجع',
+ r'sad|unfortunately|sorry|pain|lost|difficult|hard|tears|'
+ r'miss|regret|grieve|mourn|tragedy|farewell|ache',
+ ],
+ "anger": [
+ r'غضب|لا أقبل|مستحيل|يكفي|كفى|ظلم|حرام|لا يجوز|رفض|أبداً|'
+ r'لن أتحمل|سئمت|ضاق|احتج|رفض|نقمة|سخط|فساد',
+ r'never|impossible|stop|enough|angry|refuse|unacceptable|'
+ r'outrage|disgusting|unbelievable|ridiculous|furious',
+ ],
+ "emphasis": [
+ r'بالتأكيد|دون شك|قطعاً|أكيد|لا شك|يجب|ضرورة|مهم|انتبه|'
+ r'اسمع|لا بد|حتماً|بلا شك|من المؤكد|بوضوح|صراحةً|جداً',
+ r'definitely|absolutely|certainly|must|important|listen|'
+ r'crucial|critical|clearly|obviously|indeed|precisely',
+ ],
+ "soft": [
+ r'ربما|لعل|عسى|أتمنى|أرجو|بلطف|برفق|هدوء|سلام|'
+ r'هناء|رويداً|بهدوء|بتأن|رفيق|حنان|لطيف|رحمة|أمل',
+ r'perhaps|maybe|hopefully|gently|softly|peace|calm|'
+ r'wish|hope|kindly|tenderly|quietly|sweetly',
+ ],
+ "urgency": [
+ r'الآن|فوراً|سريعاً|عاجل|انتبه|احذر|خطر|أسرع|بسرعة|'
+ r'عاجلاً|في الحال|لا تتأخر|إنذار|تحذير|يستعجل|عجلة',
+ r'now|immediately|quickly|urgent|warning|danger|hurry|'
+ r'fast|alert|emergency|asap|right away|no time',
+ ],
+ "story": [
+ r'كان يا مكان|ذات يوم|في قديم|يحكى أن|وفي ليلة|'
+ r'ثم قال|فأجاب|نظر|قرر|انطلق|توجه|وصل|لقي|رأى',
+ r'once upon|there was|long ago|he said|she replied|'
+ r'suddenly|then he|and so|meanwhile|the next day',
+ ],
+}
+
+# ── Phrase emotion detector ───────────────────────────────────────────
+def analyze_phrase_emotion(phrase: str, lang: str) -> tuple[str, float]:
+ """
+ Detect the dominant emotion in a phrase.
+ Returns (emotion_name, confidence_0_to_1).
+ Confidence drives how strongly the deltas are applied.
+ """
+ p = phrase.strip()
+
+ # 1. Punctuation-first rules (fastest, highest confidence)
+ if re.search(r'[؟?]$', p):
+ return "question", 0.90
+ count_exclaim = len(re.findall(r'[!]', p))
+ if count_exclaim >= 2:
+ return "exclaim", 0.95
+ if count_exclaim == 1 and len(p) < 60:
+ return "exclaim", 0.75
+
+ # 2. Keyword scan
+ scores: dict[str, float] = {}
+ for emotion, patterns in EMOTION_PATTERNS.items():
+ for pat in patterns:
+ matches = re.findall(pat, p, re.IGNORECASE)
+ if matches:
+ # Score = matches / words — more keywords → higher confidence
+ words = max(1, len(p.split()))
+ scores[emotion] = scores.get(emotion, 0) + len(matches) / words
+
+ if scores:
+ best = max(scores, key=scores.get)
+ conf = min(0.92, scores[best] * 3.0) # cap at 0.92
+ return best, conf
+
+ # 3. Structural heuristics
+ if len(p) < 20:
+ return "emphasis", 0.35 # short phrases tend to be punchy
+ if p.endswith('…') or p.endswith('...'):
+ return "soft", 0.50 # trailing ellipsis = trailing/wistful
+
+ return "neutral", 0.0
+
+
+# ── Per-phrase XTTS param builder ─────────────────────────────────────
+def make_phrase_params(base: dict, emotion: str, conf: float,
+ variation: float, pos_ratio: float) -> dict:
+ """
+ variation : user slider 0-2 (0=none, 1=natural, 2=dramatic)
+ pos_ratio : 0.0 = first phrase, 1.0 = last phrase
+ Returns a NEW params dict with emotion-driven adjustments.
+ """
+ if variation < 0.02:
+ return base.copy()
+
+ d = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"])
+ temp_d, speed_m, rep_d, _ = d
+ strength = conf * variation # 0 … ~1.84
+
+ p = base.copy()
+ p["temperature"] = float(np.clip(
+ base["temperature"] + temp_d * strength, 0.40, 0.96))
+ p["speed"] = float(np.clip(
+ base["speed"] * (1.0 + (speed_m - 1.0) * strength), 0.60, 1.65))
+ p["repetition_penalty"] = float(np.clip(
+ base["repetition_penalty"] + rep_d * strength, 1.30, 3.80))
+
+ # ── Position-based natural drift ─────────���──────────────────────
+ # Opening: slightly more controlled (lower temp, slower)
+ if pos_ratio < 0.08:
+ p["temperature"] = float(np.clip(p["temperature"] - 0.04 * variation, 0.40, 0.96))
+ p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.04 * variation), 0.60, 1.65))
+
+ # Closing: natural deceleration (the "landing" of a speech)
+ if pos_ratio > 0.88:
+ p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.06 * variation), 0.60, 1.65))
+ p["temperature"] = float(np.clip(p["temperature"] - 0.03 * variation, 0.40, 0.96))
+
+ # ── Micro-random jitter: prevent identical params on consecutive ─
+ # neutral phrases (which would sound machine-even)
+ if emotion == "neutral" and variation > 0.3:
+ # deterministic pseudo-random from phrase content hash
+ h = (zlib.crc32(base.get("_phrase_hint", "").encode("utf-8")) % 1000) / 1000.0
+ p["temperature"] += (h - 0.5) * 0.06 * variation
+ p["speed"] *= 1.0 + (h - 0.5) * 0.04 * variation
+ p["temperature"] = float(np.clip(p["temperature"], 0.40, 0.96))
+ p["speed"] = float(np.clip(p["speed"], 0.60, 1.65))
+
+ return p
+
+
+# ── Post-generation phrase volume envelope ────────────────────────────
+def apply_phrase_dynamics(data: np.ndarray, sr: int,
+ emotion: str, variation: float) -> np.ndarray:
+ """
+ Shape amplitude envelope after XTTS generation.
+ This adds prosodic *weight* that XTTS parameter variation alone can't provide.
+ variation : 0-2
+ """
+ if variation < 0.02 or len(data) == 0:
+ return data
+
+ n = len(data)
+ t = np.linspace(0.0, 1.0, n, dtype=np.float32)
+ env = np.ones(n, dtype=np.float32)
+
+ v = min(variation, 2.0)
+
+ if emotion == "question":
+ # Rising volume in last 25% — mimics upward intonation contour
+ rise_start = 0.75
+ env += np.where(t > rise_start,
+ (t - rise_start) / (1.0 - rise_start) * 0.07 * v, 0)
+
+ elif emotion == "exclaim":
+ # Forward energy burst — louder in first half, taper
+ env += (1.0 - t) * 0.05 * v
+ env += np.sin(t * np.pi * 2) * 0.03 * v # subtle tremolo feel
+
+ elif emotion == "emphasis":
+ # Bell curve — maximum weight in the centre
+ env += np.sin(t * np.pi) * 0.10 * v
+
+ elif emotion == "sadness":
+ # Gradual fade throughout — speech drains away
+ env -= t * 0.08 * v
+ env = np.clip(env, 0.70, 1.0)
+
+ elif emotion == "joy":
+ # Bright forward motion
+ env += t * 0.04 * v + 0.02 * v
+
+ elif emotion == "anger":
+ # Sharp, compressed — no gentle tails
+ attack = np.where(t < 0.05, t / 0.05, 1.0)
+ env *= attack
+ env += 0.04 * v
+
+ elif emotion == "soft":
+ # Gentle, backed-off
+ env -= 0.06 * v
+
+ elif emotion == "urgency":
+ # Flat high energy — no decay
+ env += 0.06 * v
+
+ elif emotion == "story":
+ # Slow arc: quiet start, rise, settle
+ env += np.sin(t * np.pi * 0.7) * 0.05 * v
+
+ # Smooth envelope (25 ms window to avoid clicks)
+ k = max(1, int(0.025 * sr))
+ env = np.convolve(env, np.ones(k) / k, mode='same').astype(np.float32)
+ env = np.clip(env, 0.50, 1.30)
+
+ result = data * env
+ # Soft-limit peaks
+ peak = np.max(np.abs(result))
+ if peak > 0.95:
+ result = result / peak * 0.93
+ return result.astype(np.float32)
+
+
+def silence_samples(sr: int, ms: int = 120) -> np.ndarray:
+ """Short silence gap inserted between sentence chunks."""
+ return np.zeros(int(sr * ms / 1000), dtype=np.float32)
+
+# ── Soft Clipping (tanh) — prevents digital distortion ────────────────
+def apply_speaker_boost(path):
+ """Loudness lift + transparent soft-knee limiter.
+ (The old tanh(x*1.8) saturated the whole signal → harsh, 'digital' timbre.)"""
+ data, sr = sf.read(path)
+ if data.ndim > 1: data = data.mean(axis=1)
+ data = vq.finalize_audio(data.astype(np.float32), sr, target_rms_db=-17.0)
+ sf.write(path, data, sr, subtype='PCM_16')
+
+
+# ╔══════════════════════════════════════════════════════════════════════╗
+# ║ H U M A N I Z E — Pure-NumPy/SciPy DSP Chain ║
+# ║ Simulates recording in a real acoustic environment. ║
+# ║ No extra libraries. No IR files. All synthesized algorithmically. ║
+# ╚══════════════════════════════════════════════════════════════════════╝
+
+# ── Environment presets ───────────────────────────────────────────────
+# Each dict drives the parameters of every DSP stage below.
+HUMANIZE_PRESETS = {
+ "studio": dict(
+ # Clean, slightly padded room — professional voiceover booth
+ reverb_room=0.06, reverb_damping=0.85, reverb_wet=0.04,
+ noise_floor=-68, noise_color=0.5,
+ mic_hp_hz=90, mic_presence_db=1.5, mic_presence_hz=5500,
+ mic_air_db=-1.0, mic_air_hz=14000,
+ saturation=0.04, pitch_drift_cents=0.4, jitter_ms=0.3,
+ breathing=False,
+ ),
+ "bedroom": dict(
+ # Untreated small room — YouTube / indie podcast feel
+ reverb_room=0.22, reverb_damping=0.60, reverb_wet=0.09,
+ noise_floor=-62, noise_color=0.7,
+ mic_hp_hz=80, mic_presence_db=2.0, mic_presence_hz=4800,
+ mic_air_db=-0.5, mic_air_hz=13000,
+ saturation=0.08, pitch_drift_cents=0.8, jitter_ms=0.6,
+ breathing=False,
+ ),
+ "podcast": dict(
+ # Dynamic-mic, slightly driven — warm, intimate
+ reverb_room=0.10, reverb_damping=0.78, reverb_wet=0.05,
+ noise_floor=-65, noise_color=0.6,
+ mic_hp_hz=100, mic_presence_db=2.5, mic_presence_hz=3800,
+ mic_air_db=-2.5, mic_air_hz=12000,
+ saturation=0.14, pitch_drift_cents=0.6, jitter_ms=0.4,
+ breathing=False,
+ ),
+ "radio": dict(
+ # Classic broadcast: bandpass 200–8 kHz, compressed
+ reverb_room=0.04, reverb_damping=0.90, reverb_wet=0.02,
+ noise_floor=-72, noise_color=0.3,
+ mic_hp_hz=200, mic_presence_db=3.0, mic_presence_hz=3000,
+ mic_air_db=-8.0, mic_air_hz=8000,
+ saturation=0.18, pitch_drift_cents=0.3, jitter_ms=0.2,
+ breathing=False,
+ ),
+ "phone": dict(
+ # GSM / telephone: narrow band 300–3400 Hz, heavy saturation
+ reverb_room=0.02, reverb_damping=0.95, reverb_wet=0.01,
+ noise_floor=-55, noise_color=0.9,
+ mic_hp_hz=300, mic_presence_db=1.0, mic_presence_hz=2200,
+ mic_air_db=-20.0, mic_air_hz=3500,
+ saturation=0.28, pitch_drift_cents=1.2, jitter_ms=1.2,
+ breathing=False,
+ ),
+ "cafe": dict(
+ # Mid-sized reverberant room, background murmur
+ reverb_room=0.42, reverb_damping=0.45, reverb_wet=0.16,
+ noise_floor=-54, noise_color=0.8,
+ mic_hp_hz=70, mic_presence_db=1.0, mic_presence_hz=4000,
+ mic_air_db=-1.0, mic_air_hz=14000,
+ saturation=0.06, pitch_drift_cents=1.0, jitter_ms=0.8,
+ breathing=False,
+ ),
+ "outdoor": dict(
+ # Open space — very short reverb, wind-ish noise
+ reverb_room=0.08, reverb_damping=0.92, reverb_wet=0.03,
+ noise_floor=-50, noise_color=0.95,
+ mic_hp_hz=60, mic_presence_db=0.5, mic_presence_hz=3500,
+ mic_air_db=0.0, mic_air_hz=16000,
+ saturation=0.05, pitch_drift_cents=1.5, jitter_ms=1.5,
+ breathing=False,
+ ),
+}
+
+# ── DSP helpers ───────────────────────────────────────────────────────
+
+def _biquad_hp(data: np.ndarray, sr: int, fc: float) -> np.ndarray:
+ """2nd-order Butterworth high-pass filter — clamped to safe range."""
+ from scipy.signal import butter, sosfilt
+ fc = float(np.clip(fc, 1.0, sr * 0.49)) # never hit Nyquist
+ sos = butter(2, fc / (sr / 2), btype='high', output='sos')
+ out = sosfilt(sos, data.astype(np.float64))
+ if not np.all(np.isfinite(out)):
+ return data.astype(np.float32)
+ return out.astype(np.float32)
+
+def _biquad_peak(data: np.ndarray, sr: int, fc: float,
+ gain_db: float, Q: float = 1.4) -> np.ndarray:
+ """Parametric peak EQ biquad — clamped and guarded."""
+ from scipy.signal import sosfilt
+ fc = float(np.clip(fc, 20.0, sr * 0.45))
+ gain_db = float(np.clip(gain_db, -40.0, 20.0))
+ Q = float(np.clip(Q, 0.1, 20.0))
+ w0 = 2 * np.pi * fc / sr
+ A = 10 ** (gain_db / 40.0)
+ cos = np.cos(w0); sin = np.sin(w0)
+ alpha = sin / (2 * Q)
+ b0 = 1 + alpha * A; b1 = -2 * cos; b2 = 1 - alpha * A
+ a0 = 1 + alpha / A; a1 = -2 * cos; a2 = 1 - alpha / A
+ if abs(a0) < 1e-10:
+ return data.astype(np.float32)
+ sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64)
+ if not np.all(np.isfinite(sos)):
+ return data.astype(np.float32)
+ out = sosfilt(sos, data.astype(np.float64))
+ if not np.all(np.isfinite(out)):
+ return data.astype(np.float32)
+ return out.astype(np.float32)
+
+def _biquad_shelf_hi(data: np.ndarray, sr: int, fc: float,
+ gain_db: float) -> np.ndarray:
+ """
+ High-shelf filter (air band roll-off / boost).
+ Fixed: clamp fc below Nyquist and guard alpha against overflow/NaN.
+ """
+ from scipy.signal import sosfilt
+ # Clamp cutoff to a safe range (never exceed 90% of Nyquist)
+ fc = float(np.clip(fc, 20.0, sr * 0.45))
+ gain_db = float(np.clip(gain_db, -40.0, 20.0))
+
+ w0 = 2 * np.pi * fc / sr
+ A = 10 ** (gain_db / 40.0)
+ cos = np.cos(w0)
+ sin = np.sin(w0)
+ sqA = float(np.sqrt(max(A, 1e-12)))
+ S = 1.0
+
+ # Guard the sqrt argument — clamp to avoid negative/NaN
+ sqrt_arg = (A + 1.0 / max(A, 1e-12)) * (1.0 / max(S, 1e-9) - 1.0) + 2.0
+ sqrt_arg = max(sqrt_arg, 0.0)
+ alpha = sin / 2.0 * np.sqrt(sqrt_arg)
+
+ b0 = A * ((A + 1) + (A - 1) * cos + 2 * sqA * alpha)
+ b1 = -2 * A * ((A - 1) + (A + 1) * cos)
+ b2 = A * ((A + 1) + (A - 1) * cos - 2 * sqA * alpha)
+ a0 = (A + 1) - (A - 1) * cos + 2 * sqA * alpha
+ a1 = 2 * ((A - 1) - (A + 1) * cos)
+ a2 = (A + 1) - (A - 1) * cos - 2 * sqA * alpha
+
+ # Guard against degenerate filter (a0 near zero)
+ if abs(a0) < 1e-10:
+ return data.astype(np.float32)
+
+ sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64)
+
+ # Final guard: if any coefficient is NaN/Inf, bypass
+ if not np.all(np.isfinite(sos)):
+ return data.astype(np.float32)
+
+ result = sosfilt(sos, data.astype(np.float64))
+
+ # Guard output (should never happen now, but safety net)
+ if not np.all(np.isfinite(result)):
+ return data.astype(np.float32)
+
+ return result.astype(np.float32)
+
+def _pink_noise(n: int, rng: np.random.Generator) -> np.ndarray:
+ """FFT-based pink noise — O(N log N), no sample loops."""
+ white = rng.standard_normal(n).astype(np.float64)
+ fft_w = np.fft.rfft(white)
+ freqs = np.arange(1, len(fft_w) + 1, dtype=np.float64)
+ fft_w /= np.sqrt(freqs) # 1/sqrt(f) → pink (1/f power)
+ pink = np.fft.irfft(fft_w, n=n)
+ peak = np.max(np.abs(pink))
+ if peak > 0: pink /= peak
+ return pink.astype(np.float32)
+
+def _comb_reverb(data: np.ndarray, sr: int,
+ room: float, damping: float, wet: float) -> np.ndarray:
+ """
+ Schroeder reverb via scipy.signal.lfilter — fully vectorised, no Python loops.
+ 4 parallel comb filters + 2 series allpass.
+ room : 0–1 (room size; affects delay lengths)
+ damping : 0–1 (high-frequency absorption)
+ wet : 0–1 (wet/dry mix)
+ """
+ if wet < 0.001:
+ return data
+ from scipy.signal import lfilter
+
+ comb_delays_ms = [29.7, 37.1, 41.1, 43.7]
+ comb_base_gains = [0.805, 0.827, 0.783, 0.764]
+ allpass_delays_ms = [5.0, 1.7]
+ allpass_gain = 0.7
+ scale = 0.5 + room * 0.5
+ damp_factor = 1.0 - damping * 0.35 # approximate HF absorption
+
+ sig64 = data.astype(np.float64)
+ rev = np.zeros_like(sig64)
+
+ for dm, g_base in zip(comb_delays_ms, comb_base_gains):
+ N = max(2, int(dm * scale * sr / 1000))
+ g = min(g_base * (0.5 + room * 0.5) * damp_factor, 0.97)
+ b = np.zeros(N + 1); b[0] = 1.0
+ a = np.zeros(N + 1); a[0] = 1.0; a[N] = -g
+ rev += lfilter(b, a, sig64)
+
+ rev *= 0.25
+
+ for dm in allpass_delays_ms:
+ N = max(2, int(dm * sr / 1000))
+ g = allpass_gain
+ b = np.zeros(N + 1); b[0] = -g; b[N] = 1.0
+ a = np.zeros(N + 1); a[0] = 1.0; a[N] = -g
+ rev = lfilter(b, a, rev)
+
+ return (data * (1.0 - wet) + rev.astype(np.float32) * wet).astype(np.float32)
+
+def _micro_pitch_drift(data: np.ndarray, sr: int,
+ cents: float, rng: np.random.Generator) -> np.ndarray:
+ """
+ Subtle continuous pitch drift ±cents via nearest-neighbour resampling.
+ Creates the organic wobble of a real human voice.
+ """
+ if cents < 0.05:
+ return data
+ # Generate a slow-moving drift signal (0.1–0.5 Hz sinusoid + noise)
+ t = np.arange(len(data)) / sr
+ freq_hz = 0.2 + rng.uniform(0, 0.15)
+ drift = np.sin(2 * np.pi * freq_hz * t)
+ drift += 0.3 * rng.standard_normal(len(data)) # add gentle randomness
+ # Smooth with a 200 ms window
+ k = max(1, int(0.2 * sr))
+ drift = np.convolve(drift, np.ones(k)/k, mode='same')
+ # Scale to ±ratio
+ ratio = 2 ** (cents / 1200) # cents → semitone ratio
+ factor = 1.0 + (ratio - 1.0) * drift # ±ratio around 1.0
+ factor = np.clip(factor, 0.90, 1.10)
+
+ # Nearest-neighbour reindex
+ indices = np.cumsum(factor)
+ indices = indices / indices[-1] * (len(data) - 1)
+ indices = np.clip(indices.astype(np.int32), 0, len(data) - 1)
+ return data[indices].astype(np.float32)
+
+def _timing_jitter(data: np.ndarray, sr: int, jitter_ms: float,
+ rng: np.random.Generator) -> np.ndarray:
+ """
+ Apply block-level timing jitter: randomly shift 20 ms blocks by ±jitter_ms.
+ Breaks the perfectly even machine cadence of synthetic speech.
+ """
+ if jitter_ms < 0.1:
+ return data
+ block = int(0.020 * sr) # 20 ms blocks
+ out = data.copy()
+ shift_max = int(jitter_ms * sr / 1000)
+ for start in range(0, len(data) - block, block):
+ shift = rng.integers(-shift_max, shift_max + 1)
+ src_start = max(0, start + shift)
+ src_end = min(len(data), start + block + shift)
+ dst_len = min(block, len(data) - start)
+ src_len = src_end - src_start
+ n = min(dst_len, src_len)
+ if n > 0:
+ out[start:start + n] = data[src_start:src_start + n]
+ return out
+
+def _soft_saturate(data: np.ndarray, amount: float) -> np.ndarray:
+ """
+ Soft harmonic saturation — adds 2nd/3rd harmonics like analog tape/tube.
+ amount: 0 = bypass, 1 = heavy saturation
+ """
+ if amount < 0.005:
+ return data
+ drive = 1.0 + amount * 6.0 # 1x … 7x overdrive
+ driven = data * drive
+ # tanh saturation, then scale back down
+ sat = np.tanh(driven) / np.tanh(drive)
+ return (data * (1.0 - amount) + sat * amount).astype(np.float32)
+
+# ── Master humanize function ──────────────────────────────────────────
+def humanize_audio(src_path: str, out_path: str, preset_name: str,
+ intensity: float = 1.0, seed: int = 42):
+ """
+ Apply full humanization DSP chain to a WAV file.
+ intensity: 0.0 = almost dry, 1.0 = full preset, >1.0 = exaggerated.
+ Each stage is wrapped in a try/except so a failure degrades gracefully
+ instead of corrupting the output file.
+ """
+ p = HUMANIZE_PRESETS.get(preset_name, HUMANIZE_PRESETS["bedroom"])
+ rng = np.random.default_rng(seed)
+
+ data, sr = sf.read(src_path)
+ if data.ndim > 1:
+ data = data.mean(axis=1)
+ data = data.astype(np.float32)
+
+ if len(data) == 0:
+ raise ValueError("ملف الصوت فارغ.")
+
+ mix = float(np.clip(intensity, 0.0, 2.0))
+
+ def _safe(fn, label, *args, **kwargs):
+ """Run fn(*args, **kwargs), return original data on failure."""
+ nonlocal data
+ try:
+ result = fn(*args, **kwargs)
+ # Reject NaN / Inf output — fall back to unchanged
+ if result is not None and len(result) > 0 and np.all(np.isfinite(result)):
+ data = result.astype(np.float32)
+ else:
+ print(f"[humanize] {label}: non-finite output — stage skipped")
+ except Exception as e:
+ print(f"[humanize] {label}: {e} — stage skipped")
+
+ # 1. Mic coloring ─────────────────────────────────────────────────
+ _safe(_biquad_hp, "HP filter", data, sr, p["mic_hp_hz"])
+ if abs(p["mic_presence_db"]) > 0.1:
+ _safe(_biquad_peak, "Peak EQ", data, sr,
+ p["mic_presence_hz"], p["mic_presence_db"] * mix, Q=1.2)
+ if abs(p["mic_air_db"]) > 0.1:
+ _safe(_biquad_shelf_hi, "Hi-shelf", data, sr,
+ p["mic_air_hz"], p["mic_air_db"] * mix)
+
+ # 2. Harmonic saturation ──────────────────────────────────────────
+ _safe(_soft_saturate, "Saturation", data, p["saturation"] * mix)
+
+ # 3. Timing jitter ────────────────────────────────────────────────
+ _safe(_timing_jitter, "Jitter", data, sr, p["jitter_ms"] * mix, rng)
+
+ # 4. Micro pitch drift ────────────────────────────────────────────
+ _safe(_micro_pitch_drift, "Pitch drift", data, sr,
+ p["pitch_drift_cents"] * mix, rng)
+
+ # 5. Room reverb ──────────────────────────────────────────────────
+ _safe(_comb_reverb, "Reverb", data, sr,
+ p["reverb_room"], p["reverb_damping"], p["reverb_wet"] * mix)
+
+ # 6. Noise floor ──────────────────────────────────────────────────
+ try:
+ floor_db = p["noise_floor"] + (1.0 - mix) * 20
+ floor_db = float(np.clip(floor_db, -96.0, -30.0))
+ noise_amp = 10 ** (floor_db / 20.0)
+ color = float(np.clip(p["noise_color"], 0.0, 1.0))
+ pink = _pink_noise(len(data), rng)
+ white = rng.standard_normal(len(data)).astype(np.float32)
+ noise_mix = pink * color + white * (1.0 - color)
+ nm_peak = np.max(np.abs(noise_mix))
+ if nm_peak > 1e-9:
+ noise_mix = noise_mix / nm_peak * noise_amp
+ candidate = data + noise_mix.astype(np.float32)
+ if np.all(np.isfinite(candidate)):
+ data = candidate
+ except Exception as e:
+ print(f"[humanize] Noise floor: {e} — stage skipped")
+
+ # 7. Final normalize + soft ceiling ──────────────────────────────
+ peak = float(np.max(np.abs(data)))
+ if peak > 0:
+ data = data / peak * 0.92
+ # tanh soft-limiter (can't produce NaN on finite input)
+ data = (np.tanh(data * 1.1) * (1.0 / np.tanh(1.1))).astype(np.float32)
+
+ sf.write(out_path, data, sr, subtype='PCM_16')
+
+# ── Text Splitting for Real Progress ─────────────────────────────────
+def split_text_sentences(text: str, max_chunk: int = 280) -> list:
+ """Split text into sentence-sized chunks with a max character limit."""
+ parts = re.split(r'(?<=[.!?؟\n،])\s+', text.strip())
+ chunks, cur = [], ""
+ for p in parts:
+ if not p.strip():
+ continue
+ if len(cur) + len(p) + 1 <= max_chunk:
+ cur = (cur + " " + p).strip() if cur else p
+ else:
+ if cur:
+ chunks.append(cur)
+ if len(p) > max_chunk:
+ # Sub-split on comma/semicolon
+ sub_parts = re.split(r'(?<=[،,;])\s+', p)
+ sc = ""
+ for s in sub_parts:
+ if len(sc) + len(s) + 1 <= max_chunk:
+ sc = (sc + " " + s).strip() if sc else s
+ else:
+ if sc: chunks.append(sc)
+ sc = s
+ if sc: chunks.append(sc)
+ cur = ""
+ else:
+ cur = p
+ if cur:
+ chunks.append(cur)
+ return chunks if chunks else [text]
+
+# ── Core TTS call ─────────────────────────────────────────────────────
+def run_tts(text, refs, lang, tts_params, split, out_path):
+ if not xtts_ready or xtts is None:
+ raise RuntimeError("النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.")
+ p = {k: v for k, v in tts_params.items() if not k.startswith("_")}
+ xtts.tts_to_file(
+ text=text, speaker_wav=refs, language=lang, file_path=out_path,
+ enable_text_splitting=split, **p,
+ )
+
+def _run_tts_chunked_legacy(text, refs, lang, base_params,
+ out_path, job_id,
+ silence_ms: int = 110,
+ variation: float = 0.0,
+ ) -> bool:
+ """
+ Split text → per-phrase emotion analysis → per-phrase XTTS params
+ → per-phrase dynamics shaping → silence gaps → merge.
+ variation=0 → all chunks use base_params (original behaviour).
+ Returns True on success, False if cancelled.
+ """
+ cancel_ev = cancel_events.get(job_id)
+ chunks = split_text_sentences(text) if len(text) > 280 else [text]
+ total = len(chunks)
+ job_progress[job_id] = {"current": 0, "total": total, "status": "running"}
+
+ # ── Single-chunk fast path ────────────────────────────────────────
+ if total == 1:
+ if cancel_ev and cancel_ev.is_set():
+ job_progress[job_id]["status"] = "cancelled"; return False
+ phrase_p = base_params
+ sc_emotion = "neutral"
+ if variation > 0.02:
+ sc_emotion, sc_conf = analyze_phrase_emotion(text, lang) # ONE call
+ phrase_p = make_phrase_params(dict(base_params, _phrase_hint=text[:30]),
+ sc_emotion, sc_conf, variation, 0.5)
+ run_tts(text, refs, lang, phrase_p, True, out_path)
+ if variation > 0.02 and sc_emotion != "neutral":
+ data, sr = sf.read(out_path)
+ if data.ndim > 1: data = data.mean(axis=1)
+ data = apply_phrase_dynamics(data.astype(np.float32), sr, sc_emotion, variation)
+ sf.write(out_path, data, sr, subtype="PCM_16")
+ job_progress[job_id] = {"current": 1, "total": 1, "status": "done", "ts": time.time()}
+ return True
+
+ # ── Multi-chunk expressive path ───────────────────────────────────
+ emotions = [analyze_phrase_emotion(c, lang) for c in chunks]
+
+ tmp_files, arrays, sr_out = [], [], None
+ try:
+ for i, (chunk, (emotion, conf)) in enumerate(zip(chunks, emotions)):
+ if cancel_ev and cancel_ev.is_set():
+ job_progress[job_id]["status"] = "cancelled"; return False
+
+ pos = i / max(total - 1, 1)
+ phrase_p = make_phrase_params(
+ dict(base_params, _phrase_hint=chunk[:30]),
+ emotion, conf, variation, pos)
+
+ gap_ms = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"])[3]
+ actual_gap = int(silence_ms * 0.4 + gap_ms * 0.6) if variation > 0.1 else silence_ms
+
+ tmp = f"/tmp/ch_{job_id}_{i}.wav"
+ run_tts(chunk, refs, lang, phrase_p, False, tmp)
+ tmp_files.append(tmp)
+
+ data, sr = sf.read(tmp)
+ if sr_out is None: sr_out = sr
+ if data.ndim > 1: data = data.mean(axis=1)
+ data = data.astype(np.float32)
+ if variation > 0.02:
+ data = apply_phrase_dynamics(data, sr, emotion, variation)
+ arrays.append(data)
+ arrays.append(silence_samples(sr, actual_gap))
+ job_progress[job_id]["current"] = i + 1
+
+ combined = np.concatenate(arrays) if arrays else np.zeros(sr_out or 22050, np.float32)
+ sf.write(out_path, combined, sr_out or 22050, subtype="PCM_16")
+ job_progress[job_id]["status"] = "done"
+ job_progress[job_id]["ts"] = time.time()
+ return True
+ finally:
+ for tf in tmp_files:
+ try: os.remove(tf)
+ except: pass
+
+
+# ╔══════════════════════════════════════════════════════════════════════╗
+# ║ QUALITY PIPELINE (reference cleaning · real char limits · best-of-N) ║
+# ╚══════════════════════════════════════════════════════════════════════╝
+REF_CACHE_DIR = os.path.join(tempfile.gettempdir(), "vs_refcache")
+_LAT_CACHE = collections.OrderedDict()
+
+def _tts_model():
+ return getattr(getattr(xtts, "synthesizer", None), "tts_model", None)
+
+def _model_defaults(model):
+ c = getattr(model, "config", None)
+ g = lambda k, d: type(d)(getattr(c, k, d)) if c is not None else d
+ return dict(temperature=g("temperature", 0.75), length_penalty=g("length_penalty", 1.0),
+ repetition_penalty=g("repetition_penalty", 5.0), top_k=g("top_k", 50),
+ top_p=g("top_p", 0.85))
+
+def prepare_refs(refs, enhance=True):
+ """Clean + merge the reference samples → ([path], summary|None)."""
+ if not enhance:
+ return list(refs), None
+ try:
+ path, _reports, summary = vq.enhance_reference_set(refs, REF_CACHE_DIR)
+ return [path], summary
+ except Exception as e:
+ print(f"[!] Reference enhancement failed — using the raw sample: {e}", flush=True)
+ return list(refs), None
+
+def get_latents(model, ref_paths):
+ key = tuple((p, os.path.getsize(p), int(os.path.getmtime(p))) for p in ref_paths)
+ if key in _LAT_CACHE:
+ _LAT_CACHE.move_to_end(key)
+ return _LAT_CACHE[key]
+ c = model.config
+ lat = model.get_conditioning_latents(
+ audio_path=list(ref_paths),
+ gpt_cond_len=int(getattr(c, "gpt_cond_len", 30)),
+ gpt_cond_chunk_len=int(getattr(c, "gpt_cond_chunk_len", 4)),
+ max_ref_length=int(getattr(c, "max_ref_len", 30)),
+ sound_norm_refs=bool(getattr(c, "sound_norm_refs", False)))
+ _LAT_CACHE[key] = lat
+ while len(_LAT_CACHE) > 6:
+ _LAT_CACHE.popitem(last=False)
+ return lat
+
+def _gen_kwargs(p, d):
+ """UI/preset params → safe XTTS sampling args, centred on the model's own defaults.
+ XTTS 'speed' stretches latents: beyond ±15 % it degrades, so the excess
+ is applied afterwards as a time-stretch."""
+ speed = float(np.clip(p.get("speed", 1.0), 0.5, 2.0))
+ native = float(np.clip(speed, 0.85, 1.20))
+ rep0 = d["repetition_penalty"]
+ rep = float(np.clip(rep0 * float(p.get("repetition_penalty", 2.0)) / 2.0, rep0 * 0.7, 12.0))
+ kw = dict(temperature=float(np.clip(p.get("temperature", d["temperature"]), 0.60, 0.90)),
+ top_p=float(np.clip(p.get("top_p", d["top_p"]), 0.80, 0.95)),
+ top_k=d["top_k"], length_penalty=d["length_penalty"],
+ repetition_penalty=rep, speed=native)
+ return kw, speed / native
+
+def _tame_phrase_params(base, p):
+ """Per-sentence emotion tweaks must stay small — big jumps in temperature/speed
+ between neighbouring sentences make the voice sound like different takes."""
+ q = dict(p)
+ q["temperature"] = float(np.clip(p["temperature"], base["temperature"] - 0.06, base["temperature"] + 0.06))
+ q["speed"] = float(np.clip(p["speed"], base["speed"] * 0.94, base["speed"] * 1.06))
+ q["repetition_penalty"] = float(np.clip(p["repetition_penalty"],
+ base["repetition_penalty"] - 0.3, base["repetition_penalty"] + 0.3))
+ return q
+
+def _residual_stretch(wav, rate):
+ if abs(rate - 1.0) < 0.03:
+ return wav
+ try:
+ import librosa
+ return librosa.effects.time_stretch(np.asarray(wav, np.float32), rate=rate)
+ except Exception:
+ return wav
+
+def _asr_for_scoring(lang):
+ """Whisper round-trip used only to compare candidate takes."""
+ if not (WHISPER_OK and whisper_mdl is not None):
+ return None
+ code = "zh" if lang.startswith("zh") else lang.split("-")[0]
+ from scipy.signal import resample_poly
+ def fn(wav, sr):
+ a = np.asarray(wav, np.float32)
+ if sr != 16000:
+ g = math.gcd(int(sr), 16000)
+ a = resample_poly(a, 16000 // g, int(sr) // g).astype(np.float32)
+ segs, _ = whisper_mdl.transcribe(a, language=code, beam_size=1, vad_filter=False,
+ condition_on_previous_text=False)
+ return " ".join(s.text for s in segs)
+ return fn
+
+def run_tts_chunked(text, refs, lang, base_params, out_path, job_id,
+ silence_ms: int = 110, variation: float = 0.0,
+ attempts: int = 1, enhance_ref: bool = True,
+ clean_text: bool = True, ref_summary_out: dict = None) -> bool:
+ """Quality pipeline. Falls back to the legacy pipeline if anything unexpected
+ goes wrong, so a bug here can never block generation."""
+ model = _tts_model()
+ if model is None or not hasattr(model, "get_conditioning_latents"):
+ return _run_tts_chunked_legacy(text, refs, lang, base_params, out_path, job_id, silence_ms, variation)
+ try:
+ return _run_tts_quality(model, text, refs, lang, base_params, out_path, job_id,
+ silence_ms, variation, attempts, enhance_ref, clean_text, ref_summary_out)
+ except Exception:
+ print("[!] Quality pipeline failed — falling back to legacy pipeline:", flush=True)
+ traceback.print_exc()
+ return _run_tts_chunked_legacy(text, refs, lang, base_params, out_path, job_id, silence_ms, variation)
+
+def _run_tts_quality(model, text, refs, lang, base_params, out_path, job_id,
+ silence_ms, variation, attempts, enhance_ref, clean_text, ref_summary_out):
+ cancel_ev = cancel_events.get(job_id)
+ job_progress[job_id] = {"current": 0, "total": 1, "status": "running"}
+ ref_paths, summary = prepare_refs(refs, enhance_ref)
+ if summary and ref_summary_out is not None:
+ ref_summary_out.update(summary)
+ lat_gpt, lat_spk = get_latents(model, ref_paths)
+
+ txt = vq.normalize_text(text, lang) if clean_text else text.strip()
+ limits = getattr(getattr(model, "tokenizer", None), "char_limits", None)
+ chunks = vq.split_for_xtts(txt, lang, limits)
+ total = len(chunks)
+ job_progress[job_id] = {"current": 0, "total": total, "status": "running"}
+
+ d = _model_defaults(model)
+ attempts = int(np.clip(attempts, 1, 4))
+ asr = _asr_for_scoring(lang) if attempts > 1 else None
+ use_var = variation > 0.02
+ emotions = [analyze_phrase_emotion(c, lang) for c, _ in chunks] if use_var else [("neutral", 0.0)] * total
+ seed0 = zlib.crc32(txt.encode("utf-8")) & 0x7FFFFFFF
+ scale = silence_ms / 110.0
+
+ takes, pauses = [], []
+ for i, ((chunk, pz), (emo, conf)) in enumerate(zip(chunks, emotions)):
+ if cancel_ev and cancel_ev.is_set():
+ job_progress[job_id]["status"] = "cancelled"; return False
+ p = dict(base_params)
+ if use_var:
+ p = make_phrase_params(dict(base_params, _phrase_hint=chunk[:30]),
+ emo, conf, variation, i / max(total - 1, 1))
+ p = _tame_phrase_params(base_params, p)
+ kw, resid = _gen_kwargs(p, d)
+
+ def gen(seed, chunk=chunk, kw=kw):
+ torch.manual_seed(seed)
+ out = model.inference(chunk, lang, lat_gpt, lat_spk, enable_text_splitting=False, **kw)
+ return np.asarray(out["wav"], np.float32).reshape(-1)
+
+ wav, _info, _n = vq.synth_best(gen, chunk, attempts, asr, vq.XTTS_SR, seed0 + 101 * i)
+ wav = _residual_stretch(wav, resid)
+ if use_var and emo != "neutral":
+ wav = apply_phrase_dynamics(wav, vq.XTTS_SR, emo, variation * 0.5)
+ gap = EMOTION_DELTAS.get(emo, EMOTION_DELTAS["neutral"])[3]
+ emo_factor = 1.0 + (gap / 110.0 - 1.0) * 0.5 * min(variation, 1.0) if use_var else 1.0
+ takes.append(wav); pauses.append(pz * scale * emo_factor)
+ job_progress[job_id]["current"] = i + 1
+
+ audio = vq.finalize_audio(vq.assemble(takes, pauses, vq.XTTS_SR, 1.0, seed0), vq.XTTS_SR)
+ sf.write(out_path, audio, vq.XTTS_SR, subtype="PCM_16")
+ job_progress[job_id]["status"] = "done"
+ job_progress[job_id]["ts"] = time.time()
+ return True
+
+def collect_refs(files, voice_name):
+ paths = []
+ for f in files:
+ if not f or not getattr(f, "filename", None): continue
+ p = f"/tmp/ref_{uuid.uuid4().hex}_{f.filename}"
+ with open(p, "wb") as buf: shutil.copyfileobj(f.file, buf)
+ paths.append(p)
+ if voice_name and voice_name not in ("", "null", "none", "undefined"):
+ lib_dir = os.path.join(VOICE_LIB, voice_name)
+ if os.path.isdir(lib_dir):
+ paths += [os.path.join(lib_dir, fn) for fn in sorted(os.listdir(lib_dir))
+ if fn.lower().endswith((".wav", ".mp3", ".flac", ".ogg"))]
+ return paths
+
+def cleanup(paths):
+ for p in paths:
+ if p and p.startswith("/tmp/"):
+ try: os.remove(p)
+ except: pass
+
+# ── FastAPI ───────────────────────────────────────────────────────────
+app = FastAPI(title="XTTS Voice Studio CPU Optimized")
+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":"日本語","ko":"한국어","tr":"Türkçe",
+ "nl":"Nederlands","pl":"Polski","cs":"Čeština","hi":"हिन्दी",
+}
+LANG_MAP = {"zh-cn":"zh-CN"}
+
+# ══════════════════════════════════════════════════════════════════════
+# F R O N T E N D
+# ══════════════════════════════════════════════════════════════════════
+HTML = r"""
+
+
+
+
+XTTS Voice Studio
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+# ══════════════════════════════════════════════════════════════════════
+# R O U T E S
+# ══════════════════════════════════════════════════════════════════════
+@app.get("/", response_class=HTMLResponse)
+async def ui():
+ return (HTML
+ .replace("DEVICE_PH", device)
+ .replace("LANGS_PH", json.dumps(LANGUAGES, ensure_ascii=False)))
+
+
+# ── Model status endpoint ─────────────────────────────────────────────
+@app.get("/model_status")
+def model_status():
+ return JSONResponse({
+ "loading": xtts_loading,
+ "ready": xtts_ready,
+ "error": xtts_error,
+ "device": device,
+ })
+
+# ── Job control endpoints ─────────────────────────────────────────────
+@app.get("/progress/{job_id}")
+def get_progress(job_id: str):
+ _gc_jobs()
+ p = job_progress.get(job_id, {"current": 0, "total": 1, "status": "unknown"})
+ return JSONResponse(p)
+
+@app.post("/cancel/{job_id}")
+def cancel_job(job_id: str):
+ ev = cancel_events.get(job_id)
+ if ev:
+ ev.set()
+ return {"cancelled": True}
+
+
+# ── Audio post-edit endpoint ──────────────────────────────────────────
+@app.post("/edit_audio")
+async def edit_audio(
+ filename: str = Form(...),
+ volume_db: float = Form(0.0),
+ speed_factor: float = Form(1.0),
+ trim_start: float = Form(0.0),
+ trim_end: float = Form(0.0),
+ fade_in: float = Form(0.0), # seconds
+ fade_out: float = Form(0.0), # seconds
+ normalize: bool = Form(False), # peak-normalize to 0 dBFS
+):
+ """Edit a generated audio file without re-running TTS.
+ Always writes a NEW file — the original is never modified."""
+ src = os.path.join(OUTPUT_DIR, filename)
+ if not os.path.exists(src):
+ raise HTTPException(404, "الملف غير موجود.")
+
+ def _edit():
+ data, sr = sf.read(src)
+ if data.ndim > 1:
+ data = data.mean(axis=1)
+
+ # ── Trim start / end ──────────────────────────────────���───────
+ if trim_start > 0:
+ data = data[min(int(trim_start * sr), len(data)):]
+ if trim_end > 0:
+ end_sample = max(0, len(data) - int(trim_end * sr))
+ data = data[:end_sample]
+
+ if len(data) == 0:
+ raise ValueError("الملف فارغ بعد القطع — قلّل قيم القطع.")
+
+ # ── Speed via polyphase resampling ────────────────────────────
+ if abs(speed_factor - 1.0) > 0.01:
+ try:
+ from scipy.signal import resample_poly
+ from math import gcd
+ # Represent speed as a rational fraction p/q
+ p = round(speed_factor * 1000)
+ q = 1000
+ g = gcd(p, q)
+ # resample_poly(data, up, down): output length = len * up/down
+ # We want output_len = input_len / speed ⟹ up=q, down=p
+ data = resample_poly(data, q // g, p // g).astype(np.float32)
+ except ImportError:
+ # Fallback: crude integer resampling via scipy.signal.resample
+ try:
+ from scipy.signal import resample as sp_resample
+ new_len = max(1, int(len(data) / speed_factor))
+ data = sp_resample(data, new_len).astype(np.float32)
+ except ImportError:
+ pass # scipy not available
+
+ # ── Normalize (peak to 0 dBFS) ────────────────────────────────
+ if normalize:
+ peak = np.max(np.abs(data))
+ if peak > 0:
+ data = data / peak * 0.98
+
+ # ── Volume (dB) ───────────────────────────────────────────────
+ if abs(volume_db) > 0.01:
+ factor = 10 ** (volume_db / 20.0)
+ data = data * factor
+ # Soft clip via tanh to prevent digital distortion
+ data = np.tanh(data * 1.2) * (1.0 / np.tanh(1.2))
+
+ # ── Fade In (linear ramp) ─────────────────────────────────────
+ if fade_in > 0:
+ n = min(int(fade_in * sr), len(data))
+ data[:n] *= np.linspace(0.0, 1.0, n, dtype=np.float32)
+
+ # ── Fade Out (linear ramp) ────────────────────────────────────
+ if fade_out > 0:
+ n = min(int(fade_out * sr), len(data))
+ data[-n:] *= np.linspace(1.0, 0.0, n, dtype=np.float32)
+
+ out = os.path.join(OUTPUT_DIR, f"edit_{uuid.uuid4().hex[:8]}.wav")
+ sf.write(out, data, sr, subtype='PCM_16')
+ return os.path.basename(out)
+
+ try:
+ new_filename = await asyncio.to_thread(_edit)
+ except ValueError as e:
+ raise HTTPException(400, str(e))
+ return {"filename": new_filename}
+
+
+# ── Humanize endpoint ─────────────────────────────────────────────────
+@app.post("/humanize")
+async def humanize_endpoint(
+ filename: str = Form(...),
+ preset: str = Form("bedroom"),
+ intensity: float = Form(1.0),
+ seed: int = Form(42),
+):
+ """Apply acoustic environment simulation to a generated WAV.
+ Returns a brand-new file — original untouched."""
+ src = os.path.join(OUTPUT_DIR, filename)
+ if not os.path.exists(src):
+ raise HTTPException(404, "الملف غير موجود.")
+ if preset not in HUMANIZE_PRESETS:
+ raise HTTPException(400, f"بيئة غير معروفة: {preset}")
+ intensity = float(np.clip(intensity, 0.05, 2.0))
+ out = os.path.join(OUTPUT_DIR, f"hum_{uuid.uuid4().hex[:8]}.wav")
+ try:
+ await asyncio.to_thread(humanize_audio, src, out, preset, intensity, seed)
+ except Exception as e:
+ if os.path.exists(out):
+ os.remove(out)
+ raise HTTPException(500, f"خطأ في المعالجة: {e}")
+ return {"filename": os.path.basename(out)}
+
+
+@app.get("/humanize/presets")
+def humanize_presets():
+ return list(HUMANIZE_PRESETS.keys())
+
+@app.post("/generate")
+async def generate(
+ background_tasks: BackgroundTasks,
+ text: str = Form(...),
+ job_id: str = Form(default=""),
+ language: str = Form("ar"),
+ stability: float = Form(0.5),
+ clarity: float = Form(0.75),
+ style_exaggeration: float = Form(0.0),
+ speed: float = Form(1.0),
+ speaker_boost: bool = Form(False),
+ enable_text_splitting: bool = Form(True),
+ expression_mode: str = Form("natural"),
+ variation_strength: float = Form(1.0),
+ enhance_ref: bool = Form(True),
+ clean_text: bool = Form(True),
+ attempts: int = Form(1),
+ voice_name: str = Form(None),
+ files: list[UploadFile] = File(default=[]),
+):
+ if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.")
+ if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
+ if not text.strip(): raise HTTPException(400, "النص فارغ.")
+ if len(text) > MAX_CHARS: raise HTTPException(400, f"النص يتجاوز {MAX_CHARS} حرف.")
+ if not job_id:
+ job_id = uuid.uuid4().hex
+
+ enhanced_text = enhance_text_expression(text, language, expression_mode)
+ tts_params = apply_expression_preset(expression_mode, speed)
+ if expression_mode == "natural":
+ tts_params = dict(
+ temperature=stability_to_temperature(stability),
+ top_p=clarity_to_top_p(clarity),
+ repetition_penalty=style_to_rep_penalty(style_exaggeration),
+ speed=speed,
+ )
+
+ refs = collect_refs(files, voice_name)
+ if not refs: raise HTTPException(400, "يجب تحديد عينة صوتية مرجعية.")
+ out = os.path.join(OUTPUT_DIR, f"gen_{uuid.uuid4().hex[:10]}.wav")
+ cancel_events[job_id] = threading.Event()
+ silence_ms = {"story":160,"sad":180,"news":80,"excited":70}.get(expression_mode, 110)
+ variation = float(np.clip(variation_strength, 0.0, 2.0))
+ ref_summary = {}
+ try:
+ async with get_tts_lock():
+ success = await asyncio.to_thread(
+ run_tts_chunked, enhanced_text, refs, language,
+ tts_params, out, job_id, silence_ms, variation,
+ attempts, enhance_ref, clean_text, ref_summary
+ )
+ if not success:
+ if os.path.exists(out): os.remove(out)
+ raise HTTPException(499, "تم إلغاء التوليد.")
+ if speaker_boost:
+ await asyncio.to_thread(apply_speaker_boost, out)
+ finally:
+ cleanup(refs)
+ cancel_events.pop(job_id, None)
+
+ append_history(os.path.basename(out), text, language, "tts", len(text))
+ return {"filename": os.path.basename(out), "ref_report": ref_summary or None}
+
+
+# ── Reference check: clean the sample and let the user LISTEN to it ───
+@app.post("/enhance_ref")
+async def enhance_ref_endpoint(
+ voice_name: str = Form(None),
+ files: list[UploadFile] = File(default=[]),
+):
+ refs = collect_refs(files, voice_name)
+ if not refs: raise HTTPException(400, "حدد عينة صوتية أولاً.")
+ try:
+ path, _reports, summary = await asyncio.to_thread(vq.enhance_reference_set, refs, REF_CACHE_DIR)
+ fn = "refclean_" + os.path.basename(path)[4:-4] + ".wav"
+ shutil.copyfile(path, os.path.join(OUTPUT_DIR, fn))
+ return {"url": f"/audio/{fn}", "report": summary}
+ except ValueError as e:
+ raise HTTPException(422, str(e))
+ finally:
+ cleanup(refs)
+
+
+# ── STS ───────────────────────────────────────────────────────────────
+@app.post("/sts")
+async def sts(
+ background_tasks: BackgroundTasks,
+ source_audio: UploadFile = File(...),
+ job_id: str = Form(default=""),
+ language: str = Form("ar"),
+ stability: float = Form(0.5),
+ clarity: float = Form(0.75),
+ style_exaggeration: float = Form(0.0),
+ speed: float = Form(1.0),
+ speaker_boost: bool = Form(False),
+ variation_strength_sts: float = Form(1.0),
+ voice_name: str = Form(None),
+ ref_files: list[UploadFile] = File(default=[]),
+):
+ if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
+ if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
+ if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.")
+ if not job_id: job_id = uuid.uuid4().hex
+ src = f"/tmp/sts_{uuid.uuid4().hex}.wav"
+ with open(src, "wb") as f: shutil.copyfileobj(source_audio.file, f)
+ refs = []
+ cancel_events[job_id] = threading.Event()
+ try:
+ text = await asyncio.to_thread(transcribe_audio, src)
+ if not text: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.")
+ refs = collect_refs(ref_files, voice_name)
+ if not refs: raise HTTPException(400, "حدد صوت الهدف.")
+ out = os.path.join(OUTPUT_DIR, f"sts_{uuid.uuid4().hex[:10]}.wav")
+ sts_params = dict(
+ temperature=stability_to_temperature(stability),
+ top_p=clarity_to_top_p(clarity),
+ repetition_penalty=style_to_rep_penalty(style_exaggeration),
+ speed=speed,
+ )
+ var_sts = float(np.clip(variation_strength_sts, 0.0, 2.0))
+ async with get_tts_lock():
+ success = await asyncio.to_thread(
+ run_tts_chunked, text, refs, language, sts_params, out, job_id, 110, var_sts
+ )
+ if not success:
+ if os.path.exists(out): os.remove(out)
+ raise HTTPException(499, "تم إلغاء التحويل.")
+ if speaker_boost:
+ await asyncio.to_thread(apply_speaker_boost, out)
+ finally:
+ cleanup([src]); cleanup(refs)
+ cancel_events.pop(job_id, None)
+
+ append_history(os.path.basename(out), text, language, "sts", len(text))
+ return {"filename": os.path.basename(out), "transcribed": text}
+
+
+# ── DUB ───────────────────────────────────────────────────────────────
+@app.post("/dub")
+async def dub(
+ background_tasks: BackgroundTasks,
+ source_audio: UploadFile = File(...),
+ job_id: str = Form(default=""),
+ target_language: str = Form("en"),
+ stability: float = Form(0.5),
+ clarity: float = Form(0.75),
+ style_exaggeration: float = Form(0.0),
+ speed: float = Form(1.0),
+ speaker_boost: bool = Form(False),
+ variation_strength_dub: float = Form(1.0),
+ voice_name: str = Form(None),
+ ref_files: list[UploadFile] = File(default=[]),
+):
+ if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
+ if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
+ if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.")
+ if not TRANSLATOR_OK: raise HTTPException(503, "deep-translator غير مثبّت.")
+ if not job_id: job_id = uuid.uuid4().hex
+ src = f"/tmp/dub_{uuid.uuid4().hex}.wav"
+ with open(src, "wb") as f: shutil.copyfileobj(source_audio.file, f)
+ refs = []
+ cancel_events[job_id] = threading.Event()
+ try:
+ original = await asyncio.to_thread(transcribe_audio, src)
+ if not original: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.")
+ tgt_code = LANG_MAP.get(target_language, target_language)
+ translated = await asyncio.to_thread(
+ GoogleTranslator(source="auto", target=tgt_code).translate, original)
+ if not translated: raise HTTPException(422, "فشلت الترجمة.")
+ refs = collect_refs(ref_files, voice_name)
+ if not refs: raise HTTPException(400, "حدد صوت الدبلجة.")
+ out = os.path.join(OUTPUT_DIR, f"dub_{uuid.uuid4().hex[:10]}.wav")
+ dub_params = dict(
+ temperature=stability_to_temperature(stability),
+ top_p=clarity_to_top_p(clarity),
+ repetition_penalty=style_to_rep_penalty(style_exaggeration),
+ speed=speed,
+ )
+ var_dub = float(np.clip(variation_strength_dub, 0.0, 2.0))
+ async with get_tts_lock():
+ success = await asyncio.to_thread(
+ run_tts_chunked, translated, refs, target_language, dub_params, out, job_id, 110, var_dub
+ )
+ if not success:
+ if os.path.exists(out): os.remove(out)
+ raise HTTPException(499, "تم إلغاء الدبلجة.")
+ if speaker_boost:
+ await asyncio.to_thread(apply_speaker_boost, out)
+ finally:
+ cleanup([src]); cleanup(refs)
+ cancel_events.pop(job_id, None)
+
+ append_history(os.path.basename(out), translated, target_language, "dub", len(translated))
+ return {"filename": os.path.basename(out), "original": original, "translated": translated}
+
+
+# ── SFX ───────────────────────────────────────────────────────────────
+@app.post("/sfx")
+async def sfx(
+ background_tasks: BackgroundTasks,
+ description: str = Form(...),
+ job_id: str = Form(default=""),
+ stability: float = Form(0.1),
+ clarity: float = Form(0.5),
+ style_exaggeration: float = Form(0.9),
+ speed: float = Form(0.85),
+ speaker_boost: bool = Form(False),
+ voice_name: str = Form(None),
+ ref_file: UploadFile = File(default=None),
+):
+ if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
+ if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
+ if not description.strip(): raise HTTPException(400, "الوصف فارغ.")
+ if not job_id: job_id = uuid.uuid4().hex
+ ref_list = [ref_file] if (ref_file and getattr(ref_file, "filename", None)) else []
+ refs = collect_refs(ref_list, voice_name)
+ if not refs: raise HTTPException(400, "حدد صوتاً مرجعياً.")
+ out = os.path.join(OUTPUT_DIR, f"sfx_{uuid.uuid4().hex[:10]}.wav")
+ cancel_events[job_id] = threading.Event()
+ sfx_params = dict(
+ temperature=stability_to_temperature(stability),
+ top_p=clarity_to_top_p(clarity),
+ repetition_penalty=style_to_rep_penalty(style_exaggeration),
+ speed=speed,
+ )
+ try:
+ async with get_tts_lock():
+ success = await asyncio.to_thread(
+ run_tts_chunked, description, refs, "en", sfx_params, out, job_id
+ )
+ if not success:
+ if os.path.exists(out): os.remove(out)
+ raise HTTPException(499, "تم إلغاء التوليد.")
+ if speaker_boost:
+ await asyncio.to_thread(apply_speaker_boost, out)
+ finally:
+ cleanup(refs)
+ cancel_events.pop(job_id, None)
+
+ append_history(os.path.basename(out), description, "en", "sfx", len(description))
+ return {"filename": os.path.basename(out)}
+
+
+# ── Audio / Voice endpoints ───────────────────────────────────────────
+@app.get("/audio/{filename}")
+def audio(filename: str):
+ safe = os.path.basename(filename)
+ if safe != filename or ".." in safe:
+ raise HTTPException(400, "اسم ملف غير صالح.")
+ p = os.path.join(OUTPUT_DIR, safe)
+ if not os.path.exists(p): raise HTTPException(404, "ملف غير موجود.")
+ return FileResponse(p, media_type="audio/wav",
+ headers={"Accept-Ranges": "bytes", "Cache-Control": "no-cache"})
+
+@app.get("/history")
+def get_history():
+ return JSONResponse(load_history())
+
+@app.get("/voices")
+def list_voices():
+ if not os.path.isdir(VOICE_LIB): return []
+ return [d for d in os.listdir(VOICE_LIB) if os.path.isdir(os.path.join(VOICE_LIB, d))]
+
+@app.post("/voices/save")
+async def save_voice(background_tasks: BackgroundTasks,
+ name: str = Form(...),
+ file: UploadFile = File(...),
+ file2: UploadFile = File(default=None)):
+ safe = name.strip().replace("/","_").replace("..","_")[:40]
+ lib = os.path.join(VOICE_LIB, safe)
+ os.makedirs(lib, exist_ok=True)
+ for f in ([file, file2] if file2 and file2.filename else [file]):
+ safe_fn = re.sub(r"[^A-Za-z0-9_.\-]", "_", os.path.basename(f.filename))[:80]
+ with open(os.path.join(lib, safe_fn), "wb") as buf:
+ shutil.copyfileobj(f.file, buf)
+ background_tasks.add_task(trigger_cloud_backup)
+ return {"name": safe}
+
+@app.delete("/voices/{name}")
+def delete_voice(background_tasks: BackgroundTasks, name: str):
+ lib = os.path.join(VOICE_LIB, name)
+ if os.path.isdir(lib): shutil.rmtree(lib)
+ background_tasks.add_task(trigger_cloud_backup)
+ return {"deleted": name}
+
+@app.get("/voices/{name}/preview")
+def preview_voice(name: str):
+ lib = os.path.join(VOICE_LIB, name)
+ if not os.path.isdir(lib): raise HTTPException(404, "صوت غير موجود.")
+ for fn in os.listdir(lib):
+ if fn.lower().endswith((".wav",".mp3",".flac",".ogg")):
+ mt = "audio/wav" if fn.endswith(".wav") else "audio/mpeg"
+ return FileResponse(os.path.join(lib, fn), media_type=mt)
+ raise HTTPException(404, "لا توجد ملفات صوتية.")
+
+
+# ══════════════════════════════════════════════════════════════════════
+if __name__ == "__main__":
+ # ZeroGPU: the `spaces` package only reports its @spaces.GPU functions to
+ # the platform from inside gradio's Blocks.launch(). This app serves with
+ # uvicorn directly, so we send that start-up report ourselves; otherwise
+ # the Space is killed with "No @spaces.GPU function detected during startup".
+ if ZEROGPU:
+ try:
+ _startup = getattr(spaces.zero, "startup", None) # only exists on ZeroGPU
+ if callable(_startup):
+ _startup()
+ print("[✓] ZeroGPU start-up report sent.", flush=True)
+ else:
+ print("[i] ZeroGPU not active on this hardware — no start-up report needed.", flush=True)
+ except Exception:
+ print("[!] ZeroGPU start-up report failed:", flush=True)
+ traceback.print_exc()
+ uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")
\ No newline at end of file