diff --git "a/app.py" "b/app.py"
deleted file mode 100644--- "a/app.py"
+++ /dev/null
@@ -1,4132 +0,0 @@
-# ╔══════════════════════════════════════════════════════════════════════╗
-# ║ 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