Spaces:
Running on Zero
Running on Zero
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ 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 = "2"
# ── 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 _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 _stretches(mask, keep_gap=90):
"""Group voiced segments separated by pauses ≤ 0.9 s into 'stretches', so natural
phrasing (sentence-to-sentence pauses) stays exactly as recorded."""
out = []
for s, e in _segments(mask):
s, e = int(s), int(e)
if out and s - out[-1][1] <= keep_gap: out[-1][1] = e
else: out.append([s, e])
return out
def _split_stretch(db, s, e, max_frames):
"""Cut an over-long stretch at its deepest-energy frames (= inside pauses).
Nothing is ever joined afterwards, so a cut can never become an audible splice."""
smooth = uniform_filter(db, size=7, mode="nearest")
cuts, pos = [], s
while e - pos > max_frames:
lo, hi = pos + int(0.55 * max_frames), pos + max_frames
c = lo + int(np.argmin(smooth[lo:hi]))
cuts.append((pos, c)); pos = c
cuts.append((pos, e))
return cuts
def _score_window(x, db, w, h, a, b, mask):
seg, v = db[a:b], mask[a:b]
snr = float(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
clip = float(np.mean(np.abs(x[a * h: b * h + w]) > 0.98))
sp = float(v.sum() * 0.01)
return snr - 0.5 * sd - 300.0 * clip + 0.5 * min(sp, 20.0), sp
def _finish_segment(x, sr):
"""Level-match one CONTIGUOUS segment (voiced RMS → −20 dBFS) and fade its ends
inside the surrounding silence."""
z = x.astype(np.float32).copy()
db, w, h = frame_db(z, sr)
m = _voiced_mask(db)
vs = [z[s * h: e * h + w] for s, e in _segments(m)]
lvl = float(np.sqrt(np.mean(np.square(np.concatenate(vs)))) + 1e-9) if vs else float(np.sqrt(np.mean(z ** 2)) + 1e-9)
z = soft_limit(z * (10 ** (-20 / 20) / lvl), 0.95, 0.85)
return _fade(z, sr, 15, 45)
def enhance_reference(path, strength=1.0, denoise=True, max_seg_sec=28.0, min_speech_sec=1.5):
"""Clean one recording and return CONTIGUOUS candidate segments (no cutting-and-
pasting): → (candidates [{wav, speech, score}], report)."""
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))
x = raw / (float(np.max(np.abs(raw))) + 1e-9) * 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
after = _analyze(x, fs)
db, w, h = frame_db(x, REF_SR)
mask = _voiced_mask(db)
speech_sec_all = float(mask.sum() * 0.01)
cands = []
lead, tail = int(0.15 * REF_SR), int(0.30 * REF_SR)
for s, e in _stretches(mask):
for a, b in _split_stretch(db, s, e, int(max_seg_sec * 100)):
sc, sp = _score_window(x, db, w, h, a, b, mask)
if sp < min_speech_sec: continue
i0 = max(0, a * h - (lead if a == s else 0))
i1 = min(len(x), b * h + w + (tail if b == e else 0))
cands.append(dict(wav=_finish_segment(x[i0:i1], REF_SR), speech=sp, score=sc))
if not cands: # no clear phrasing: keep one contiguous voiced span
idx = np.where(mask)[0]
i0, i1 = (max(0, idx[0] * h - lead), min(len(x), idx[-1] * h + w + tail)) if len(idx) else (0, len(x))
i1 = min(i1, i0 + int(max_seg_sec * REF_SR))
cands.append(dict(wav=_finish_segment(x[i0:i1], REF_SR), speech=speech_sec_all, score=0.0))
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 - max(0, 20 - before["snr_db"]) * 1.5 - min(30, clip_pct * 10) \
- max(0, 6 - speech_sec_all) * 5 - (15 if before["bandwidth_hz"] < 6000 else 0)
report = dict(duration_in=round(dur_in, 1), speech_sec=round(speech_sec_all, 1), clipping_pct=round(clip_pct, 2),
denoised=did_denoise, before=before, after=after, score=int(np.clip(score, 0, 100)),
warnings=warns, tips=tips, candidates=len(cands))
return cands, 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_segments=4, denoise=True):
"""Clean the recordings and pick the best CONTIGUOUS segments (≤ max_segments, ≈target_sec
of speech in total). Segments are written as SEPARATE files and never concatenated —
each one's conditioning latents are computed on their own and averaged by the caller.
→ ([segment_wav_paths], [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_segments}|{denoise}|{VQ_VERSION}").encode()).hexdigest()[:20]
meta = os.path.join(cache_dir, f"ref_{key}.json")
if os.path.exists(meta):
d = json.load(open(meta, encoding="utf-8"))
if d["paths"] and all(os.path.exists(p) for p in d["paths"]):
return d["paths"], d["reports"], d["summary"]
pool, reports = [], []
for fi, p in enumerate(paths):
try:
cands, rep = enhance_reference(p, strength, denoise)
rep["file"] = os.path.basename(p)[-40:]
reports.append(rep)
for c in cands: c["fscore"] = rep["score"]; pool.append(c)
except Exception as e:
reports.append(dict(file=os.path.basename(p)[-40:], error=str(e), score=0,
warnings=[f"تعذّرت معالجة الملف: {e}"], tips=[]))
if not pool:
raise ValueError("تعذّرت معالجة كل ملفات العينة: " + "; ".join(r.get("error", "?") for r in reports))
chosen, total = [], 0.0
for c in sorted(pool, key=lambda c: c["score"] + 0.1 * c["fscore"], reverse=True):
if len(chosen) >= max_segments or total >= target_sec: break
chosen.append(c); total += c["speech"]
out_paths = []
for i, c in enumerate(chosen):
op = os.path.join(cache_dir, f"ref_{key}_{i}.wav")
sf.write(op, c["wav"], REF_SR, subtype="PCM_16"); out_paths.append(op)
good = [r for r in reports if "error" not in r]
best = max(good, key=lambda r: r["score"])
summary = dict(score=int(np.mean([r["score"] for r in good])),
duration=round(sum(len(c["wav"]) for c in chosen) / REF_SR, 1),
speech=round(total, 1), segments=len(chosen), 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(paths=out_paths, reports=reports, summary=summary), open(meta, "w", encoding="utf-8"), ensure_ascii=False)
return out_paths, 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()
CPS_RANGE = {"zh": (1.5, 14.0), "ja": (2.0, 16.0), "ko": (3.0, 20.0)} # chars/sec sanity bounds
def take_info(wav, sr, text, asr_fn=None, lang="ar"):
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)
lo, hi = CPS_RANGE.get(lang.split("-")[0], (4.0, 28.0))
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 lo <= cps <= hi
return dict(dur=dur, cps=cps, peak=peak, sim=sim, ok=ok, secs=None,
clip=float(np.mean(np.abs(wav) > 0.99)) if len(wav) else 0.0)
def _take_score(i, med):
s = (i["sim"] * 100.0) if i["sim"] is not None else 60.0
if i.get("secs") is not None:
s += 80.0 * (i["secs"] - 0.6) # speaker-identity consistency
s -= 25.0 * abs(math.log(max(i["dur"], 1e-3) / max(med, 1e-3)))
s -= 300.0 * i["clip"]
s -= 0.0 if i["ok"] else 30.0
return s
def synth_best(gen_fn, text, attempts=1, asr_fn=None, sr=XTTS_SR, seed=1234,
accept_sim=0.90, secs_fn=None, lang="ar", guard_retries=2, min_secs=0.45):
"""gen_fn(seed:int) → float32 wav. Returns (wav, info, n_generated).
attempts==1 : cheap self-healing guard — regenerate (≤ guard_retries) only if the
take is obviously broken (silent / absurd speech-rate / clipped / off-speaker).
attempts>1 : best-of-N ranked by ASR similarity + speaker similarity + duration.
`secs_fn` (speaker-encoder cosine similarity) is a single extra forward pass — cheap
enough to run even when attempts==1, and it is the only check that catches a take
that drifted away from the reference speaker's timbre (silence/clip/rate checks don't)."""
multi = attempts > 1
takes, k = [], 0
while True:
wav = trim_take(gen_fn(seed + 7919 * k), sr)
info = take_info(wav, sr, text, asr_fn if multi else None, lang)
if secs_fn is not None and len(wav) > int(0.3 * sr):
try: info["secs"] = float(secs_fn(wav, sr))
except Exception: info["secs"] = None
takes.append((wav, info)); k += 1
if multi:
if k >= attempts: break
if info["ok"] and info["sim"] is not None and info["sim"] >= accept_sim: break
else:
secs_ok = info.get("secs") is None or info["secs"] >= min_secs
if (info["ok"] and info["clip"] < 0.005 and secs_ok) or k > guard_retries: break
if len(takes) == 1: return takes[0][0], takes[0][1], 1
med = float(np.median([i["dur"] for _, i in takes]))
best = max(takes, key=lambda t: _take_score(t[1], med))
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)
# ═══════════════════════════════════════════════════════════════════════
# 6. DEPTH / SPACE ("مجسّم") — stereo room, body, air, room-tone
# XTTS output is DRY, MONO, 24 kHz and gapped with digital silence: that
# is what reads as "flat". This adds the missing dimensions:
# · body/presence EQ · harmonic "air" above XTTS's 12 kHz limit
# · stereo early-reflections + decorrelated late tail (IR convolution)
# · continuous room-tone bed (no dead digital silence between phrases)
# ═══════════════════════════════════════════════════════════════════════
SPACE_PRESETS = {
# rt60 predelay er damp wet_db body presence air tone_db
"studio": dict(rt60=0.30, pre=8, er=0.55, damp=0.70, wet_db=-17, body=1.5, pres=1.2, air=1.0, tone=-68),
"room": dict(rt60=0.50, pre=14, er=0.60, damp=0.50, wet_db=-14, body=1.5, pres=1.0, air=0.8, tone=-64),
"hall": dict(rt60=0.95, pre=22, er=0.45, damp=0.40, wet_db=-12, body=1.0, pres=0.5, air=0.8, tone=-62),
}
DEPTH_SR = 48000
def _rbj(kind, fs, f0, gain_db, Q=0.9):
A = 10 ** (gain_db / 40.0); w0 = 2 * math.pi * f0 / fs
cw, sw = math.cos(w0), math.sin(w0); alpha = sw / (2 * Q)
if kind == "peak":
b = [1 + alpha * A, -2 * cw, 1 - alpha * A]; a = [1 + alpha / A, -2 * cw, 1 - alpha / A]
else:
sA = 2 * math.sqrt(A) * alpha
if kind == "hishelf":
b = [A * ((A + 1) + (A - 1) * cw + sA), -2 * A * ((A - 1) + (A + 1) * cw), A * ((A + 1) + (A - 1) * cw - sA)]
a = [(A + 1) - (A - 1) * cw + sA, 2 * ((A - 1) - (A + 1) * cw), (A + 1) - (A - 1) * cw - sA]
else: # loshelf
b = [A * ((A + 1) - (A - 1) * cw + sA), 2 * A * ((A - 1) - (A + 1) * cw), A * ((A + 1) - (A - 1) * cw - sA)]
a = [(A + 1) + (A - 1) * cw + sA, -2 * ((A - 1) + (A + 1) * cw), (A + 1) + (A - 1) * cw - sA]
return np.array(b) / a[0], np.array(a) / a[0]
def _eq(x, fs, kind, f0, gain_db, Q=0.9):
if abs(gain_db) < 0.05: return x
b, a = _rbj(kind, fs, f0, gain_db, Q)
return signal.lfilter(b, a, x)
def _band_split_noise(n, fs, rng, rt60, damp):
"""Noise with frequency-dependent decay: lows ring longer, highs die first."""
noise = rng.standard_normal(n)
t = np.arange(n) / fs
out = np.zeros(n)
for lo, hi, k in ((None, 500, 1.15), (500, 3000, 1.0), (3000, None, max(0.22, 1.0 - 0.7 * damp))):
if lo is None: sos = signal.butter(2, hi, "lowpass", fs=fs, output="sos")
elif hi is None: sos = signal.butter(2, lo, "highpass", fs=fs, output="sos")
else: sos = signal.butter(2, [lo, hi], "bandpass", fs=fs, output="sos")
out += signal.sosfilt(sos, noise) * np.exp(-6.9078 * t / (rt60 * k))
return out
def make_room_ir(fs=DEPTH_SR, rt60=0.5, pre=12, er=0.5, damp=0.5, seed=1):
"""Synthetic stereo room impulse response WITHOUT the direct sound.
→ (irL, irR), each unit-energy. Early reflections differ per ear."""
rng = np.random.default_rng(seed)
n = int(fs * min(max(rt60 * 1.2, 0.25), 1.8))
sos_hp = signal.butter(2, 180, "highpass", fs=fs, output="sos") # keep bass centred
er_ms = [6.3, 9.7, 14.1, 19.6, 26.2, 33.5, 41.8]
er_g = [0.9, 0.75, 0.60, 0.50, 0.40, 0.30, 0.22]
irs = []
for ch in range(2):
tail = _band_split_noise(n, fs, rng, rt60, damp)
tail *= np.minimum(1.0, np.arange(n) / (0.025 * fs)) # bloom after the ERs
p = int(fs * pre / 1000)
ir = np.zeros(n); ir[p:] = tail[: n - p]
ir = signal.sosfilt(sos_hp, ir)
e = np.sqrt(np.sum(ir ** 2)) + 1e-12
ir = ir / e # unit-energy tail
sos_lp = signal.butter(2, 7000, "lowpass", fs=fs, output="sos")
er_ir = np.zeros(n)
for tm, g in zip(er_ms, er_g):
j = tm * (1.0 + (0.16 if ch else -0.16)) + rng.uniform(-0.8, 0.8) # L/R differ
idx = int(fs * (pre + j) / 1000)
if idx < n: er_ir[idx] += g * er * (1 if rng.random() > 0.35 else -1)
er_ir = signal.sosfilt(sos_lp, er_ir)
er_ir = signal.sosfilt(sos_hp, er_ir) * 0.12 # ERs relative to tail
irs.append((ir + er_ir).astype(np.float64))
return irs[0], irs[1]
def _rms_db(x):
return 20 * math.log10(float(np.sqrt(np.mean(np.square(x))) + 1e-9))
def _exciter(x, fs, amount):
"""Regenerate 'air' above XTTS's ~12 kHz ceiling from its 5.5 kHz+ band."""
if amount < 0.02: return np.zeros_like(x)
hp = signal.sosfilt(signal.butter(4, 5500, "highpass", fs=fs, output="sos"), x)
y = np.tanh(4.0 * hp)
y = signal.sosfilt(signal.butter(4, 9000, "highpass", fs=fs, output="sos"), y)
ry, rh = np.sqrt(np.mean(y ** 2)) + 1e-9, np.sqrt(np.mean(hp ** 2)) + 1e-9
return y / ry * rh * 0.22 * amount
def _room_tone(n, fs, rng, level_db):
"""Pink-ish, low-passed, decorrelated L/R bed — the 'air of the room'."""
out = []
sos = signal.butter(2, 5500, "lowpass", fs=fs, output="sos")
for _ in range(2):
w = rng.standard_normal(n); F = np.fft.rfft(w); F /= np.sqrt(np.arange(1, len(F) + 1))
p = np.fft.irfft(F, n=n); p = signal.sosfilt(sos, p)
p = p / (np.sqrt(np.mean(p ** 2)) + 1e-9) * 10 ** (level_db / 20)
out.append(p)
fade = int(0.35 * fs)
env = np.ones(n); env[:fade] = np.linspace(0, 1, fade); env[-fade:] = np.linspace(1, 0, fade)
return out[0] * env, out[1] * env
def spatialize(wav, sr, preset="studio", amount=0.5, seed=7):
"""mono float wav @sr → (stereo float32 [n,2], DEPTH_SR). amount 0…1 (0 = bypass)."""
wav = np.asarray(wav, np.float64).reshape(-1)
a = float(np.clip(amount, 0.0, 1.0))
if a < 0.02 or len(wav) == 0:
return np.stack([wav, wav], 1).astype(np.float32), sr
P = SPACE_PRESETS.get(preset, SPACE_PRESETS["studio"])
fs = DEPTH_SR
if sr != fs:
g = math.gcd(int(sr), fs); wav = signal.resample_poly(wav, fs // g, int(sr) // g)
ref_lvl = _rms_db(wav)
rng = np.random.default_rng(seed)
# tone: warmth / de-mud / presence / air
x = _eq(wav, fs, "loshelf", 190, P["body"] * a, 0.8)
x = _eq(x, fs, "peak", 380, -1.0 * a, 0.9)
x = _eq(x, fs, "peak", 3200, P["pres"] * a, 0.8)
x = x + _exciter(x, fs, a * P["air"])
# space: stereo IR convolution, tail rings out after the last word
irL, irR = make_room_ir(fs, P["rt60"], P["pre"], P["er"], P["damp"], seed)
tail = int(fs * min(P["rt60"] * 0.9, 1.2))
xp = np.concatenate([x, np.zeros(tail)])
wetL = signal.fftconvolve(xp, irL)[: len(xp)]
wetR = signal.fftconvolve(xp, irR)[: len(xp)]
wg = 10 ** (P["wet_db"] / 20) * a * 2.2 * (np.sqrt(np.mean(x ** 2)) / (np.sqrt(np.mean(wetL ** 2)) + 1e-9)) * 1.0
L, R = xp + wg * wetL, xp + wg * wetR
# room tone under everything (incl. pauses)
tl, tr = _room_tone(len(xp), fs, rng, P["tone"] + 6 * (1 - a))
L, R = L + tl, R + tr
st = np.stack([L, R], 1)
st *= 10 ** ((ref_lvl - _rms_db(st)) / 20) # keep the original loudness
st[-int(0.15 * fs):] *= np.linspace(1, 0, int(0.15 * fs))[:, None]
st = soft_limit(st, 0.95, 0.80)
return st.astype(np.float32), fs
'''
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
_backup_lock = threading.Lock()
_backup_state = {"running": False, "pending": False, "last": 0.0}
BACKUP_MIN_INTERVAL = 90 # seconds between uploads
def trigger_cloud_backup():
"""Single-flight + coalesced + rate-limited. (Previously every save/delete spawned
its own thread uploading the whole data folder, so uploads could pile up.)"""
if not (HUB_AVAILABLE and HF_TOKEN and DATASET_REPO_ID): return
with _backup_lock:
if _backup_state["running"]:
_backup_state["pending"] = True
return
_backup_state["running"] = True
def _run():
while True:
wait = BACKUP_MIN_INTERVAL - (time.time() - _backup_state["last"])
if wait > 0: time.sleep(wait)
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())}",
ignore_patterns=["*.tmp", "*.corrupt", ".cache/*"])
except Exception as e:
print(f"[!] Cloud backup failed: {e}", flush=True)
with _backup_lock:
_backup_state["last"] = time.time()
if _backup_state["pending"]:
_backup_state["pending"] = False
continue
_backup_state["running"] = False
return
threading.Thread(target=_run, daemon=True, name="cloud-backup").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 ───────────────────────────────────────────────────
_history_lock = threading.RLock()
MAX_HISTORY = 300
MAX_OUTPUTS = 200
def load_history():
with _history_lock:
try:
if not os.path.exists(HISTORY_FILE): return []
with open(HISTORY_FILE, encoding="utf-8") as f:
return json.load(f)
except Exception:
try: shutil.copyfile(HISTORY_FILE, HISTORY_FILE + ".corrupt") # keep it, don't silently overwrite
except Exception: pass
return []
def save_history(h):
with _history_lock: # atomic: write temp → replace
tmp = HISTORY_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(h, f, ensure_ascii=False, indent=2)
os.replace(tmp, HISTORY_FILE)
trigger_cloud_backup()
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 Exception: pass
except Exception:
pass
def append_history(filename, text, language, mode, char_count=0):
with _history_lock: # read-modify-write must be one unit
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:
"""Room reverb via a synthetic impulse response (dense, non-metallic).
Replaces the old 4-comb Schroeder network, which rang at 30-44 ms delays."""
if wet < 0.001:
return data
from scipy.signal import fftconvolve
irL, irR = vq.make_room_ir(sr, rt60=0.12 + room * 0.9, pre=8, er=0.5, damp=damping)
rev = fftconvolve(data.astype(np.float64), (irL + irR) / 2.0)[: len(data)]
rev *= (np.sqrt(np.mean(data ** 2)) + 1e-9) / (np.sqrt(np.mean(rev ** 2)) + 1e-9)
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/4. Timing jitter & micro pitch drift — REMOVED: measured to add
# -11…-2 dB (jitter) / -40…-32 dB (drift) of non-harmonic distortion
# (block copies without crossfade; nearest-neighbour resampling).
# 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 = vq.soft_limit(data, 0.92, 0.75)
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:
paths, _reports, summary = vq.enhance_reference_set(refs, REF_CACHE_DIR)
return list(paths), 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):
"""Conditioning latents for one or several reference files.
With several files each one is encoded ON ITS OWN and the results are averaged
(duration-weighted). XTTS' own multi-file path glues the waveforms together with
torch.cat before encoding — an audible splice that the model then imitates."""
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
kw = dict(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)))
paths = list(ref_paths)
if len(paths) == 1:
lat = model.get_conditioning_latents(audio_path=paths, **kw)
else:
parts, weights = [], []
for p in paths:
parts.append(model.get_conditioning_latents(audio_path=[p], **kw))
try: dur = float(sf.info(p).duration)
except Exception: dur = float(kw["gpt_cond_len"])
weights.append(max(0.5, min(dur, float(kw["gpt_cond_len"]))))
wsum = sum(weights)
lat = (sum(w * g for w, (g, _) in zip(weights, parts)) / wsum,
sum(w * s for w, (_, s) in zip(weights, parts)) / wsum)
_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
DRY_CACHE_DIR = os.path.join(tempfile.gettempdir(), "vs_drycache")
os.makedirs(DRY_CACHE_DIR, exist_ok=True)
os.makedirs(REF_CACHE_DIR, exist_ok=True)
RECENT_JOBS = collections.deque(maxlen=20)
def _prune_dir(path, max_files, max_bytes):
try:
fs = [os.path.join(path, f) for f in os.listdir(path)]
fs = sorted((f for f in fs if os.path.isfile(f)), key=os.path.getmtime)
total = sum(os.path.getsize(f) for f in fs)
while fs and (len(fs) > max_files or total > max_bytes):
f = fs.pop(0); total -= os.path.getsize(f); os.remove(f)
except Exception:
pass
_prune_dir(REF_CACHE_DIR, 40, 300 << 20)
_prune_dir(DRY_CACHE_DIR, 60, 400 << 20)
def _dry_cache_key(txt, lang, ref_paths, base_params, silence_ms, variation, attempts, asr_on):
"""Everything that changes the synthesised (dry) audio. Depth / boost are post-FX
and deliberately NOT part of the key → changing them re-renders instantly."""
payload = dict(v=vq.VQ_VERSION + "/dry2", t=txt, l=lang, r=[vq._file_sha(p) for p in ref_paths],
p={k: round(float(v), 4) for k, v in base_params.items() if isinstance(v, (int, float))},
s=int(silence_ms), var=round(float(variation), 3), a=int(attempts), asr=bool(asr_on))
return hashlib.sha1(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:24]
def _record_job(stats_out, job_id, **kw):
rec = dict(job=str(job_id)[:8], ts=int(time.time()), **kw)
if rec.get("audio_sec") and rec.get("wall") is not None:
rec["rtf"] = round(rec["wall"] / rec["audio_sec"], 2)
for k in ("wall", "audio_sec"):
if k in rec: rec[k] = round(float(rec[k]), 1)
RECENT_JOBS.append(rec)
if stats_out is not None: stats_out.update(rec)
print(f"[job {rec['job']}] " + ", ".join(f"{k}={v}" for k, v in rec.items() if k not in ("job", "ts")), flush=True)
def _secs_fn(model, ref_spk):
"""Cosine similarity between a take and the reference, using XTTS's own speaker encoder."""
ref = ref_spk.detach().flatten().float().cpu()
warned = []
def fn(wav, sr):
try:
with torch.inference_mode():
e = model.get_speaker_embedding(torch.from_numpy(np.ascontiguousarray(wav, dtype=np.float32))[None, :], sr)
return float(torch.nn.functional.cosine_similarity(e.detach().flatten().float().cpu(), ref, dim=0))
except Exception as ex:
if not warned: # log once, never spam
warned.append(1)
print(f"[!] speaker-similarity scoring unavailable ({type(ex).__name__}: {ex}) — ranking without it.", flush=True)
raise
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,
stats_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, stats_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, stats_out=None):
cancel_ev = cancel_events.get(job_id)
job_progress[job_id] = {"current": 0, "total": 1, "status": "running"}
t_start = time.time()
ref_paths, summary = prepare_refs(refs, enhance_ref)
if summary and ref_summary_out is not None:
ref_summary_out.update(summary)
txt = vq.normalize_text(text, lang) if clean_text else text.strip()
attempts = int(np.clip(attempts, 1, 4))
asr = _asr_for_scoring(lang) if attempts > 1 else None
cache_path = os.path.join(DRY_CACHE_DIR, _dry_cache_key(
txt, lang, ref_paths, base_params, silence_ms, variation, attempts, asr is not None) + ".wav")
if os.path.exists(cache_path): # identical request → skip synthesis entirely
shutil.copyfile(cache_path, out_path)
try: os.utime(cache_path, None)
except Exception: pass
job_progress[job_id] = {"current": 1, "total": 1, "status": "done", "ts": time.time()}
_record_job(stats_out, job_id, cached=True, audio_sec=sf.info(out_path).duration,
wall=time.time() - t_start)
return True
lat_gpt, lat_spk = get_latents(model, ref_paths)
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)
try:
# One speaker-encoder forward pass per take — cheap, so run it even at attempts==1;
# this used to be gated behind attempts>1, which meant the default UI setting
# (attempts=1) never actually checked whether a chunk still sounded like the
# reference speaker. That gap is the main source of "sounds like a different
# speaker sometimes" on normal (non best-of-N) generations.
secs_fn = _secs_fn(model, lat_spk)
except Exception as ex: # scoring is a bonus — never block generation
print(f"[!] speaker-similarity scoring disabled: {ex}", flush=True)
secs_fn = None
retries = 0
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,
secs_fn=secs_fn, lang=lang)
retries += _n - 1
wav = _residual_stretch(wav, resid)
# `conf` (how confident analyze_phrase_emotion was) was computed above but never
# reached these two effects — they used to apply full-strength dynamics/pause
# changes even on a low-confidence (often false-positive) keyword match, which is
# what made ordinary neutral sentences suddenly jump in loudness/pacing next to
# their neighbours. Scaling by `conf` makes weak matches barely audible and keeps
# strong matches (question marks, "!!", clear keywords) at full strength.
if use_var and emo != "neutral":
wav = apply_phrase_dynamics(wav, vq.XTTS_SR, emo, variation * 0.5 * conf)
gap = EMOTION_DELTAS.get(emo, EMOTION_DELTAS["neutral"])[3]
emo_factor = 1.0 + (gap / 110.0 - 1.0) * 0.5 * min(variation, 1.0) * conf 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")
try:
shutil.copyfile(out_path, cache_path)
_prune_dir(DRY_CACHE_DIR, 60, 400 << 20)
except Exception:
pass
job_progress[job_id]["status"] = "done"
job_progress[job_id]["ts"] = time.time()
_record_job(stats_out, job_id, cached=False, chunks=total, attempts=attempts, extra_takes=retries,
audio_sec=len(audio) / vq.XTTS_SR, wall=time.time() - t_start)
return True
def apply_depth(path, amount=0.5, space="studio"):
"""Stereo room + body + air + room-tone (see vq.spatialize). Never breaks the file."""
try:
amount = float(np.clip(amount, 0.0, 1.0))
if amount < 0.02:
return
data, sr = sf.read(path, dtype="float32")
if data.ndim > 1: data = data.mean(axis=1)
st, out_sr = vq.spatialize(data, sr, space, amount)
sf.write(path, st, out_sr, subtype="PCM_16")
except Exception:
print("[!] depth processing failed — file left dry:", flush=True)
traceback.print_exc()
MAX_UPLOAD_BYTES = 30 * 1024 * 1024 # per file
MAX_REQUEST_BYTES = 130 * 1024 * 1024 # per request (Content-Length)
MAX_REF_FILES = 4
_AUDIO_EXT = (".wav", ".mp3", ".flac", ".ogg", ".m4a", ".opus", ".webm", ".aac")
def _copy_limited(fobj, dest, limit=MAX_UPLOAD_BYTES):
"""Copy an upload to disk, refusing anything above `limit` bytes."""
size = 0
try:
with open(dest, "wb") as out:
while True:
chunk = fobj.read(1 << 20)
if not chunk: break
size += len(chunk)
if size > limit:
raise HTTPException(413, f"الملف أكبر من {limit >> 20} MB.")
out.write(chunk)
except Exception:
try: os.remove(dest)
except Exception: pass
raise
_VOICE_NAME_RE = re.compile(r"[^\w\- ]", re.UNICODE)
def safe_voice_name(name) -> str:
return _VOICE_NAME_RE.sub("_", (name or "").strip()).strip(" ._")[:40]
def voice_dir(name, must_exist=False) -> str:
"""Resolve a voice-library folder, guaranteed to live directly inside VOICE_LIB
(blocks '..', absolute paths and separators). Names saved by older versions
are still found via an exact-name lookup."""
lib = os.path.realpath(VOICE_LIB)
raw = (name or "").strip()
found = None
for cand in dict.fromkeys([raw, safe_voice_name(raw)]):
if not cand or cand in (".", "..") or any(c in cand for c in "/\\\x00"):
continue
p = os.path.realpath(os.path.join(lib, cand))
if os.path.dirname(p) != lib:
continue
if os.path.isdir(p): return p
found = found or p
if found is None: raise HTTPException(400, "اسم الصوت غير صالح.")
if must_exist: raise HTTPException(404, "صوت غير موجود.")
return found
def _output_path(filename: str) -> str:
safe = os.path.basename(filename or "")
if not safe or safe != filename or ".." in safe:
raise HTTPException(400, "اسم ملف غير صالح.")
return os.path.join(OUTPUT_DIR, safe)
def collect_refs(files, voice_name):
paths = []
try:
for f in files:
if not f or not getattr(f, "filename", None): continue
if len(paths) >= MAX_REF_FILES: break
base = re.sub(r"[^\w.\-]", "_", os.path.basename(f.filename), flags=re.UNICODE)[-60:] or "ref.wav"
p = f"/tmp/ref_{uuid.uuid4().hex}_{base}"
_copy_limited(f.file, p)
paths.append(p)
if voice_name and voice_name not in ("", "null", "none", "undefined"):
lib_dir = voice_dir(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(_AUDIO_EXT)]
except Exception:
cleanup(paths); raise
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=["*"])
@app.middleware("http")
async def _limit_request_size(request, call_next):
cl = request.headers.get("content-length", "")
if request.method in ("POST", "PUT") and cl.isdigit() and int(cl) > MAX_REQUEST_BYTES:
return JSONResponse({"detail": f"الطلب أكبر من {MAX_REQUEST_BYTES >> 20} MB."}, status_code=413)
return await call_next(request)
@app.exception_handler(Exception)
async def _unhandled_error(request, exc):
print(f"[!] Unhandled error on {request.method} {request.url.path}:", flush=True)
traceback.print_exception(type(exc), exc, exc.__traceback__)
return JSONResponse({"detail": f"خطأ داخلي ({type(exc).__name__}). راجع سجلات الخادم."}, status_code=500)
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"""<!DOCTYPE html>
<html lang="ar">
<head>
<meta charset="UTF-8"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>XTTS Voice Studio</title>
<link rel="preconnect" href="https://fonts.googleapis.com"/>
<link href="https://fonts.googleapis.com/css2?family=IBM+Plex+Mono:wght@300;400;500;600&family=Tajawal:wght@300;400;500;700&display=swap" rel="stylesheet"/>
<script src="https://unpkg.com/react@18/umd/react.production.min.js"></script>
<script src="https://unpkg.com/react-dom@18/umd/react-dom.production.min.js"></script>
<script src="https://unpkg.com/@babel/standalone/babel.min.js"></script>
<style>
/* ═══ DESIGN SYSTEM ═══════════════════════════════════════════════ */
:root{
--bg:#090a0d; --bg2:#0f1117; --surface:#141720; --surface2:#1b1f2e;
--border:rgba(255,255,255,.07); --border-hi:rgba(255,255,255,.15);
--amber:#f0a500; --amber2:#ffc94d; --amber-glow:rgba(240,165,0,.15);
--green:#22c55e; --red:#ef4444; --blue:#3b82f6; --purple:#a78bfa;
--text:#e2e8f0; --text2:#94a3b8; --text3:#475569;
--mono:'IBM Plex Mono',monospace; --sans:'Tajawal',sans-serif;
--radius:10px; --radius-sm:6px;
--sidebar:220px; --topbar:56px;
--trans:.18s cubic-bezier(.4,0,.2,1);
}
[data-theme=light]{
--bg:#f1f5f9; --bg2:#e8edf4; --surface:#ffffff; --surface2:#f8fafc;
--border:rgba(0,0,0,.08); --border-hi:rgba(0,0,0,.15);
--text:#1e293b; --text2:#64748b; --text3:#94a3b8;
}
*{box-sizing:border-box;margin:0;padding:0}
html,body,#root{height:100%}
body{font-family:var(--sans);background:var(--bg);color:var(--text);-webkit-font-smoothing:antialiased}
::-webkit-scrollbar{width:4px;height:4px}
::-webkit-scrollbar-track{background:transparent}
::-webkit-scrollbar-thumb{background:var(--border-hi);border-radius:4px}
/* ═══ LAYOUT ══════════════════════════════════════════════════════ */
.app{display:grid;grid-template-columns:var(--sidebar) 1fr;grid-template-rows:var(--topbar) 1fr;height:100%;overflow:hidden}
.sidebar{grid-row:1/-1;background:var(--bg2);border-right:1px solid var(--border);
display:flex;flex-direction:column;padding:16px 0;overflow:hidden;z-index:20;
transition:transform .25s cubic-bezier(.4,0,.2,1)}
.sidebar-logo{padding:0 16px 20px;border-bottom:1px solid var(--border);margin-bottom:8px}
.sidebar-logo-text{font-family:var(--mono);font-size:15px;font-weight:600;letter-spacing:-.02em}
.sidebar-logo-text .hi{color:var(--amber)}
.sidebar-logo-sub{font-family:var(--mono);font-size:9px;color:var(--text3);letter-spacing:.1em;text-transform:uppercase;margin-top:2px}
.sidebar-section{padding:4px 10px;margin-top:4px}
.sidebar-section-label{font-family:var(--mono);font-size:9px;color:var(--text3);letter-spacing:.12em;text-transform:uppercase;padding:0 6px;margin-bottom:4px}
.nav-item{display:flex;align-items:center;gap:10px;padding:9px 10px;border-radius:var(--radius-sm);
cursor:pointer;transition:background var(--trans),color var(--trans);color:var(--text2);font-size:13px;font-weight:500;
border:1px solid transparent;user-select:none}
.nav-item .ni-icon{font-size:15px;width:20px;text-align:center;flex-shrink:0}
.nav-item:hover{background:var(--surface);color:var(--text)}
.nav-item.active{background:var(--amber-glow);color:var(--amber);border-color:rgba(240,165,0,.2)}
.sidebar-footer{margin-top:auto;padding:12px 16px;border-top:1px solid var(--border)}
.status-dot{width:7px;height:7px;border-radius:50%;flex-shrink:0}
.status-dot.green{background:var(--green);box-shadow:0 0 6px var(--green)}
.status-dot.red{background:var(--red)}
.status-row{display:flex;align-items:center;gap:8px;font-family:var(--mono);font-size:10px;color:var(--text3);margin-bottom:4px}
.topbar{grid-column:2;display:flex;align-items:center;justify-content:space-between;
padding:0 24px;border-bottom:1px solid var(--border);background:var(--bg2);z-index:10}
.topbar-title{font-family:var(--mono);font-size:13px;font-weight:500;color:var(--text2);letter-spacing:.05em}
.topbar-title span{color:var(--amber);font-weight:600}
.topbar-right{display:flex;align-items:center;gap:10px}
.hamburger{display:none;flex-direction:column;gap:4px;cursor:pointer;padding:6px;
border:1px solid var(--border);border-radius:var(--radius-sm);background:var(--surface2)}
.hamburger span{display:block;width:18px;height:2px;background:var(--text2);border-radius:1px;
transition:all .2s}
.main{overflow-y:auto;padding:28px;background:var(--bg);grid-column:2}
.main-inner{max-width:900px;margin:0 auto;display:flex;flex-direction:column;gap:20px}
.sidebar-overlay{display:none;position:fixed;inset:0;background:rgba(0,0,0,.55);z-index:19}
/* ═══ PROGRESS REAL ══════════════════════════════════════════════= */
.progress-wrap{margin-bottom:8px}
.progress-label{font-family:var(--mono);font-size:11px;color:var(--text3);margin-bottom:6px;
display:flex;justify-content:space-between;align-items:center}
.progress-track{height:6px;background:var(--surface2);border-radius:3px;overflow:hidden;
border:1px solid var(--border)}
.progress-fill{height:100%;border-radius:3px;background:linear-gradient(90deg,var(--amber),var(--amber2));
transition:width .35s ease;min-width:4px}
.progress-indeterminate{animation:prog-ind 1.4s ease-in-out infinite}
@keyframes prog-ind{0%{width:0%;margin-right:100%}50%{width:50%;margin-right:50%}100%{width:0%;margin-right:0%}}
/* ═══ CARDS ═══════════════════════════════════════════════════════ */
.card{background:var(--surface);border:1px solid var(--border);border-radius:var(--radius);overflow:hidden}
.card-header{display:flex;align-items:center;justify-content:space-between;
padding:14px 18px;border-bottom:1px solid var(--border);gap:12px}
.card-title{font-family:var(--mono);font-size:11px;font-weight:500;color:var(--text2);
letter-spacing:.1em;text-transform:uppercase;display:flex;align-items:center;gap:8px}
.card-body{padding:18px}
/* ═══ FORM ════════════════════════════════════════════════════════ */
.textarea{width:100%;background:transparent;border:none;color:var(--text);
font-family:var(--sans);font-size:15px;line-height:1.75;resize:none;outline:none;
min-height:140px;padding:0}
.textarea::placeholder{color:var(--text3)}
.char-bar{display:flex;justify-content:space-between;align-items:center;
padding:10px 18px;border-top:1px solid var(--border);background:var(--bg2)}
.char-count{font-family:var(--mono);font-size:11px;color:var(--text3)}
.char-count.warn{color:#f59e0b}
.char-count.limit{color:var(--red)}
select{background:var(--surface2);border:1px solid var(--border);color:var(--text);
border-radius:var(--radius-sm);padding:6px 28px 6px 10px;font-family:var(--sans);
font-size:13px;outline:none;cursor:pointer;-webkit-appearance:none;
background-image:url("data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' width='10' height='6'%3E%3Cpath d='M0 0l5 6 5-6z' fill='%2394a3b8'/%3E%3C/svg%3E");
background-repeat:no-repeat;background-position:right 8px center}
select:focus{border-color:var(--amber)}
input[type=text]{background:var(--surface2);border:1px solid var(--border);color:var(--text);
border-radius:var(--radius-sm);padding:8px 12px;font-family:var(--sans);font-size:13px;
outline:none;width:100%;transition:border-color var(--trans)}
input[type=text]:focus{border-color:var(--amber)}
input[type=text]::placeholder{color:var(--text3)}
/* ═══ SLIDERS ════════════════════════════════════════════════════= */
.slider-track{position:relative;height:4px;border-radius:2px;background:var(--surface2);cursor:pointer}
.slider-fill{position:absolute;left:0;top:0;height:100%;border-radius:2px;
background:linear-gradient(90deg,var(--amber),var(--amber2));pointer-events:none;transition:width .05s}
input[type=range]{position:absolute;inset:0;width:100%;opacity:0;height:4px;cursor:pointer;margin:0;-webkit-appearance:none}
.slider-wrap{padding:2px 0}
.slider-meta{display:flex;justify-content:space-between;align-items:center;margin-bottom:8px}
.slider-label{font-family:var(--mono);font-size:11px;color:var(--text2)}
.slider-value{font-family:var(--mono);font-size:11px;color:var(--amber);
background:var(--amber-glow);border:1px solid rgba(240,165,0,.25);
border-radius:4px;padding:1px 7px}
.slider-hints{display:flex;justify-content:space-between;margin-top:6px}
.slider-hint{font-family:var(--mono);font-size:9px;color:var(--text3);letter-spacing:.05em}
/* ═══ TOGGLE ═════════════════════════════════════════════════════= */
.toggle{display:flex;align-items:center;gap:10px;cursor:pointer;user-select:none}
.toggle-track{width:38px;height:21px;border-radius:11px;background:var(--surface2);
border:1px solid var(--border);position:relative;transition:background var(--trans),border-color var(--trans);flex-shrink:0}
.toggle-track.on{background:var(--amber);border-color:var(--amber)}
.toggle-thumb{position:absolute;width:15px;height:15px;border-radius:50%;background:#fff;
top:2px;left:2px;transition:left var(--trans);box-shadow:0 1px 3px rgba(0,0,0,.4)}
.toggle-track.on .toggle-thumb{left:19px}
.toggle-label{font-size:13px;color:var(--text2)}
/* ═══ BUTTONS ════════════════════════════════════════════════════= */
.btn-primary{background:var(--amber);color:#000;font-family:var(--sans);font-weight:700;
font-size:15px;border:none;border-radius:var(--radius);padding:14px 28px;cursor:pointer;
width:100%;display:flex;align-items:center;justify-content:center;gap:10px;
transition:opacity var(--trans),transform var(--trans);letter-spacing:.01em}
.btn-primary:hover:not(:disabled){opacity:.88}
.btn-primary:active:not(:disabled){transform:scale(.98)}
.btn-primary:disabled{opacity:.35;cursor:not-allowed}
.btn-cancel{background:rgba(239,68,68,.12);color:var(--red);font-family:var(--sans);font-weight:600;
font-size:14px;border:1px solid rgba(239,68,68,.35);border-radius:var(--radius);padding:10px 20px;
cursor:pointer;width:100%;display:flex;align-items:center;justify-content:center;gap:8px;
transition:background var(--trans)}
.btn-cancel:hover{background:rgba(239,68,68,.22)}
.btn-secondary{background:var(--surface2);color:var(--text2);font-family:var(--sans);
font-size:13px;border:1px solid var(--border);border-radius:var(--radius-sm);
padding:8px 14px;cursor:pointer;display:flex;align-items:center;gap:6px;
transition:color var(--trans),border-color var(--trans);white-space:nowrap}
.btn-secondary:hover{color:var(--text);border-color:var(--border-hi)}
.btn-sm{padding:5px 12px;font-size:12px}
.btn-icon-round{width:32px;height:32px;border-radius:50%;background:var(--surface2);
border:1px solid var(--border);cursor:pointer;display:flex;align-items:center;
justify-content:center;font-size:14px;transition:border-color var(--trans)}
.btn-icon-round:hover{border-color:var(--amber)}
/* ═══ BADGES ═════════════════════════════════════════════════════= */
.badge{display:inline-flex;align-items:center;gap:4px;font-family:var(--mono);font-size:10px;
border-radius:20px;padding:2px 9px;white-space:nowrap}
.badge-amber{background:var(--amber-glow);border:1px solid rgba(240,165,0,.3);color:var(--amber)}
.badge-green{background:rgba(34,197,94,.1);border:1px solid rgba(34,197,94,.3);color:var(--green)}
.badge-red{background:rgba(239,68,68,.1);border:1px solid rgba(239,68,68,.3);color:var(--red)}
.badge-blue{background:rgba(59,130,246,.1);border:1px solid rgba(59,130,246,.3);color:var(--blue)}
.badge-purple{background:rgba(167,139,250,.1);border:1px solid rgba(167,139,250,.3);color:var(--purple)}
/* ═══ FILE UPLOAD ════════════════════════════════════════════════= */
.dropzone{border:1.5px dashed var(--border);border-radius:var(--radius);
padding:16px;text-align:center;cursor:pointer;position:relative;
transition:border-color var(--trans),background var(--trans)}
.dropzone:hover,.dropzone.drag{border-color:var(--amber);background:var(--amber-glow)}
.dropzone input{position:absolute;inset:0;opacity:0;cursor:pointer;font-size:0}
.dropzone-icon{font-size:22px;margin-bottom:6px}
.dropzone-label{font-size:12px;color:var(--text3)}
.dropzone-hint{font-family:var(--mono);font-size:10px;color:var(--text3);margin-top:3px}
.file-chip{display:flex;align-items:center;gap:8px;padding:8px 12px;
background:var(--surface2);border:1px solid var(--border);border-radius:var(--radius-sm)}
.file-chip-icon{font-size:16px}
.file-chip-name{flex:1;font-size:12px;color:var(--text);overflow:hidden;text-overflow:ellipsis;white-space:nowrap}
.file-chip-size{font-family:var(--mono);font-size:10px;color:var(--text3)}
.file-chip-rm{background:none;border:none;color:var(--text3);cursor:pointer;font-size:14px;
line-height:1;padding:2px;transition:color var(--trans)}
.file-chip-rm:hover{color:var(--red)}
/* ═══ VOICE CARD ═════════════════════════════════════════════════= */
.voice-card{display:flex;align-items:center;gap:12px;padding:12px 14px;
border-radius:var(--radius);border:1px solid var(--border);background:var(--surface2);
cursor:pointer;transition:border-color var(--trans),background var(--trans)}
.voice-card:hover{border-color:var(--border-hi)}
.voice-card.selected{border-color:var(--amber);background:var(--amber-glow)}
.voice-avatar{width:36px;height:36px;border-radius:50%;display:flex;align-items:center;
justify-content:center;font-family:var(--mono);font-size:13px;font-weight:600;
flex-shrink:0;border:1.5px solid var(--border)}
.voice-info{flex:1;min-width:0}
.voice-name{font-size:13px;font-weight:600;color:var(--text);margin-bottom:3px;
overflow:hidden;text-overflow:ellipsis;white-space:nowrap}
.voice-tags{display:flex;gap:4px;flex-wrap:wrap}
.voice-check{font-size:16px;color:var(--amber);flex-shrink:0}
/* ═══ AUDIO PLAYER ═══════════════════════════════════════════════= */
.player{background:var(--surface2);border:1px solid var(--border);border-radius:var(--radius);overflow:hidden}
.player-wave{padding:0 4px;cursor:pointer}
.player-wave canvas{width:100%;height:56px;display:block}
.player-controls{display:flex;align-items:center;gap:12px;padding:12px 16px;border-top:1px solid var(--border)}
.player-play{width:38px;height:38px;border-radius:50%;background:var(--amber);border:none;
cursor:pointer;display:flex;align-items:center;justify-content:center;font-size:14px;flex-shrink:0;
transition:opacity var(--trans);color:#000}
.player-play:hover{opacity:.85}
.player-time{font-family:var(--mono);font-size:11px;color:var(--text2);white-space:nowrap}
.player-seek{flex:1;height:3px;border-radius:2px;background:var(--surface);cursor:pointer;position:relative}
.player-seek-fill{position:absolute;left:0;top:0;height:100%;border-radius:2px;background:var(--amber);transition:width .1s linear}
.player-actions{display:flex;gap:6px}
/* ═══ AUDIO EDITOR ═══════════════════════════════════════════════= */
.editor-card{border:1px solid rgba(240,165,0,.25);background:rgba(240,165,0,.04)}
/* ═══ HISTORY ════════════════════════════════════════════════════= */
.hist-item{display:flex;align-items:center;gap:12px;padding:12px 16px;
border:1px solid var(--border);border-radius:var(--radius);background:var(--surface);
transition:border-color var(--trans)}
.hist-item:hover{border-color:var(--border-hi)}
.hist-text{flex:1;min-width:0}
.hist-preview{font-size:13px;color:var(--text);overflow:hidden;text-overflow:ellipsis;
white-space:nowrap;margin-bottom:5px}
.hist-meta{display:flex;gap:6px;align-items:center;flex-wrap:wrap}
.hist-time{font-family:var(--mono);font-size:10px;color:var(--text3)}
.hist-actions{display:flex;gap:6px;flex-shrink:0}
/* ═══ INFO BOX ═══════════════════════════════════════════════════= */
.info-box{padding:12px 16px;border-radius:var(--radius-sm);background:var(--surface2);
border:1px solid var(--border);font-size:13px;color:var(--text2);line-height:1.7;
display:flex;gap:10px}
.info-box .info-icon{font-size:16px;flex-shrink:0;margin-top:1px}
/* ═══ GRID ═══════════════════════════════════════════════════════= */
.grid-2{display:grid;grid-template-columns:1fr 1fr;gap:14px}
.grid-3{display:grid;grid-template-columns:1fr 1fr 1fr;gap:10px}
/* ═══ TOAST ══════════════════════════════════════════════════════= */
.toast-root{position:fixed;bottom:24px;left:50%;transform:translateX(-50%);
z-index:999;display:flex;flex-direction:column;gap:8px;align-items:center;pointer-events:none}
.toast{display:flex;align-items:center;gap:10px;padding:10px 16px;border-radius:var(--radius);
font-size:13px;font-weight:500;box-shadow:0 8px 32px rgba(0,0,0,.5);
pointer-events:auto;animation:toast-in .25s ease;white-space:nowrap;max-width:90vw}
@keyframes toast-in{from{opacity:0;transform:translateY(10px)}to{opacity:1;transform:translateY(0)}}
.toast.success{background:#166534;color:#dcfce7;border:1px solid #15803d}
.toast.error {background:#7f1d1d;color:#fee2e2;border:1px solid #b91c1c}
.toast.info {background:var(--surface2);color:var(--text);border:1px solid var(--border)}
/* ═══ WAVE ANIMATION ═════════════════════════════════════════════= */
.wave-bars{display:flex;align-items:center;gap:3px;height:18px}
.wave-bars span{width:3px;border-radius:2px;background:currentColor;
animation:wb .9s ease-in-out infinite}
.wave-bars span:nth-child(2){animation-delay:.15s}
.wave-bars span:nth-child(3){animation-delay:.3s}
.wave-bars span:nth-child(4){animation-delay:.15s}
@keyframes wb{0%,100%{height:3px}50%{height:16px}}
/* ═══ MISC ════════════════════════════════════════════════════════ */
.empty{text-align:center;padding:48px 24px;color:var(--text3)}
.empty-icon{font-size:36px;margin-bottom:12px;opacity:.5}
.empty-title{font-size:15px;font-weight:600;color:var(--text2);margin-bottom:6px}
.empty-sub{font-size:13px;line-height:1.6}
.divider{height:1px;background:var(--border);margin:4px 0}
.row{display:flex;align-items:center;gap:10px}
.col{display:flex;flex-direction:column;gap:12px}
.label{font-family:var(--mono);font-size:10px;color:var(--text3);letter-spacing:.1em;text-transform:uppercase;margin-bottom:6px}
.section-gap{display:flex;flex-direction:column;gap:16px}
.text-amber{color:var(--amber)}
.text-muted{color:var(--text2)}
.fz-12{font-size:12px}
.fz-13{font-size:13px}
.mt-4{margin-top:4px}
.mt-8{margin-top:8px}
.mt-12{margin-top:12px}
/* ═══ MOBILE RESPONSIVE ══════════════════════════════════════════= */
@media (max-width:768px){
:root{ --sidebar:240px; --topbar:52px; }
.app{ grid-template-columns:1fr; grid-template-rows:var(--topbar) 1fr; }
.sidebar{
position:fixed; top:0; right:0; height:100%;
transform:translateX(100%); transition:transform .28s cubic-bezier(.4,0,.2,1);
z-index:30; border-right:none; border-left:1px solid var(--border);
width:var(--sidebar);
}
.sidebar.mobile-open{ transform:translateX(0); }
.sidebar-overlay{ display:block; opacity:0; pointer-events:none; transition:opacity .28s; }
.sidebar-overlay.visible{ opacity:1; pointer-events:auto; }
.topbar{ grid-column:1; padding:0 16px; }
.topbar-title{ font-size:11px; }
.hamburger{ display:flex; }
.main{ grid-column:1; padding:16px; }
.main-inner{ max-width:100%; }
.grid-2{ grid-template-columns:1fr; }
.grid-3{ grid-template-columns:1fr 1fr; }
.card-header{ padding:12px 14px; flex-wrap:wrap; gap:8px; }
.card-body{ padding:14px; }
.btn-primary{ font-size:14px; padding:12px 20px; }
.player-controls{ padding:10px 12px; gap:8px; }
.hist-actions{ flex-direction:column; gap:4px; }
.char-bar{ flex-direction:column; gap:8px; align-items:flex-start; }
}
@media (max-width:480px){
.grid-3{ grid-template-columns:1fr; }
.topbar-title .title-long{ display:none; }
}
</style>
</head>
<body>
<div id="root"></div>
<script type="text/babel">
const { useState, useEffect, useRef, useCallback, useMemo } = React;
const DEVICE = "DEVICE_PH";
const LANGS = LANGS_PH;
const MAX_CH = 5000;
const fmt2 = v => parseFloat(v).toFixed(2);
const fmtSize = b => b>1048576 ? (b/1048576).toFixed(1)+"MB" : (b/1024).toFixed(0)+"KB";
const fmtTime = s => `${Math.floor(s/60)}:${String(Math.floor(s%60)).padStart(2,"0")}`;
const genId = () => (typeof crypto!=="undefined" && crypto.randomUUID)
? crypto.randomUUID() : (Date.now().toString(36)+Math.random().toString(36).slice(2));
/* ══ PERSISTENT STATE — survives refresh & tab-switch ══════════════ */
/**
* useLS(key, def) – localStorage-backed useState.
* Values are JSON-serialised. File objects (non-serialisable) are
* intentionally NOT persisted — only primitive / plain-object state.
*/
function useLS(key, def) {
const [val, setVal] = useState(()=>{
try {
const raw = localStorage.getItem(key);
return raw !== null ? JSON.parse(raw) : def;
} catch { return def; }
});
const set = useCallback(v => {
setVal(prev => {
const next = typeof v === "function" ? v(prev) : v;
try { localStorage.setItem(key, JSON.stringify(next)); } catch {}
return next;
});
}, [key]);
return [val, set];
}
/** Persist the last generated result so it reloads after a page refresh. */
function usePersistedResult(storageKey) {
const [result, _setResult] = useLS(storageKey, null);
// Verify the file still exists on the server before restoring
const [verified, setVerified] = useState(false);
useEffect(()=>{
if (!result) { setVerified(true); return; }
fetch(result.url, {method:"HEAD"})
.then(r=>{ if(!r.ok) _setResult(null); setVerified(true); })
.catch(()=>{ _setResult(null); setVerified(true); });
}, []);
const setResult = useCallback(v => { _setResult(v); }, []);
return [verified ? result : null, setResult];
}
/* ══ TOAST ══════════════════════════════════════════════════════════ */
let _setToasts = null;
const toast = (type, msg, ms=3500) => {
const id = Date.now();
_setToasts(t => [...t, {id,type,msg}]);
setTimeout(() => _setToasts(t => t.filter(x=>x.id!==id)), ms);
};
function ToastRoot() {
const [toasts, setToasts] = useState([]);
useEffect(()=>{ _setToasts=setToasts; },[]);
return (
<div className="toast-root">
{toasts.map(t=>(
<div key={t.id} className={`toast ${t.type}`}>
<span>{t.type==="success"?"✓":t.type==="error"?"✕":"ℹ"}</span>
<span>{t.msg}</span>
</div>
))}
</div>
);
}
/* ══ REAL PROGRESS BAR ══════════════════════════════════════════════ */
function ProgressBar({ progress }) {
if (!progress) return null;
const { current, total, status } = progress;
const indeterminate = total <= 1;
const pct = total > 1 ? Math.round((current/total)*100) : 0;
return (
<div className="progress-wrap">
<div className="progress-label">
<span>
{status==="running"
? (total > 1 ? `توليد جملة ${current} من ${total}` : "جاري التوليد…")
: status==="cancelled" ? "تم الإلغاء"
: ""}
</span>
{total > 1 && <span style={{color:"var(--amber)"}}>{pct}%</span>}
</div>
<div className="progress-track">
{indeterminate
? <div className="progress-fill progress-indeterminate" style={{width:"40%"}}/>
: <div className="progress-fill" style={{width:`${pct}%`}}/>
}
</div>
</div>
);
}
/* ══ AUDIO PLAYER ══════════════════════════════════════════════════ */
function AudioPlayer({ url, filename }) {
const canvasRef = useRef(null);
const audioRef = useRef(new Audio());
const actxRef = useRef(null);
const [playing, setPlaying] = useState(false);
const [current, setCurrent] = useState(0);
const [duration, setDuration] = useState(0);
const [drawn, setDrawn] = useState(false);
useEffect(()=>{
const a = audioRef.current;
a.src=url; setPlaying(false); setCurrent(0); setDuration(0); setDrawn(false);
const onMeta = ()=>setDuration(a.duration||0);
const onTime = ()=>setCurrent(a.currentTime);
const onEnd = ()=>{ setPlaying(false); setCurrent(0); a.currentTime=0; };
a.addEventListener("loadedmetadata",onMeta);
a.addEventListener("timeupdate",onTime);
a.addEventListener("ended",onEnd);
drawWaveform(url);
return ()=>{ a.pause();
a.removeEventListener("loadedmetadata",onMeta);
a.removeEventListener("timeupdate",onTime);
a.removeEventListener("ended",onEnd); };
},[url]);
const drawWaveform = async (src)=>{
try{
const res=await fetch(src);
const buf=await res.arrayBuffer();
if(!actxRef.current||actxRef.current.state==="closed")
actxRef.current=new(window.AudioContext||window.webkitAudioContext)();
const actx=actxRef.current;
const decoded=await actx.decodeAudioData(buf.slice(0));
const data=decoded.getChannelData(0);
const canvas=canvasRef.current; if(!canvas) return;
const W=canvas.clientWidth*window.devicePixelRatio||600;
const H=canvas.clientHeight*window.devicePixelRatio||112;
canvas.width=W; canvas.height=H;
const ctx2d=canvas.getContext("2d");
ctx2d.clearRect(0,0,W,H);
const step=Math.ceil(data.length/W), mid=H/2;
ctx2d.strokeStyle="rgba(240,165,0,0.6)"; ctx2d.lineWidth=1;
for(let i=0;i<W;i++){
let min=1,max=-1;
for(let j=0;j<step;j++){const d=data[i*step+j]||0;if(d<min)min=d;if(d>max)max=d;}
ctx2d.beginPath();
ctx2d.moveTo(i,mid+min*mid*0.9);
ctx2d.lineTo(i,mid+max*mid*0.9);
ctx2d.stroke();
}
setDrawn(true);
}catch(e){}
};
const togglePlay=()=>{
const a=audioRef.current;
if(playing){a.pause();setPlaying(false);}
else{a.play().catch(()=>{});setPlaying(true);}
};
const seek=(e)=>{
const rect=e.currentTarget.getBoundingClientRect();
const pct=(e.clientX-rect.left)/rect.width;
audioRef.current.currentTime=pct*(duration||0);
};
const pct=duration>0?(current/duration*100).toFixed(1):0;
return (
<div className="player">
<div className="player-wave" onClick={seek}>
<canvas ref={canvasRef} style={{height:56}}/>
{!drawn&&(
<div style={{display:"flex",alignItems:"center",justifyContent:"center",
height:56,color:"var(--text3)",fontFamily:"var(--mono)",fontSize:11}}>
جاري تحليل الموجة…
</div>
)}
</div>
<div className="player-controls">
<button className="player-play" onClick={togglePlay}>{playing?"⏸":"▶"}</button>
<span className="player-time">{fmtTime(current)}</span>
<div className="player-seek" onClick={seek}>
<div className="player-seek-fill" style={{width:`${pct}%`}}/>
</div>
<span className="player-time">{fmtTime(duration)}</span>
<div className="player-actions">
<a href={url} download={filename} style={{textDecoration:"none"}}>
<button className="btn-secondary btn-sm" title="تحميل">⬇ تحميل</button>
</a>
</div>
</div>
</div>
);
}
/* ══ AUDIO EDITOR ══════════════════════════════════════════════════ */
function AudioEditor({ filename, onEdited }) {
const [volume, setVolume] = useState(0);
const [speed, setSpeed] = useState(1.0);
const [trimStart, setTrimStart] = useState(0);
const [trimEnd, setTrimEnd] = useState(0);
const [fadeIn, setFadeIn] = useState(0);
const [fadeOut, setFadeOut] = useState(0);
const [normalize, setNormalize] = useState(false);
const [applying, setApplying] = useState(false);
const dirty = volume!==0||speed!==1||trimStart!==0||trimEnd!==0
||fadeIn!==0||fadeOut!==0||normalize;
const reset = ()=>{
setVolume(0);setSpeed(1);setTrimStart(0);setTrimEnd(0);
setFadeIn(0);setFadeOut(0);setNormalize(false);
};
const apply = async ()=>{
setApplying(true);
const fd=new FormData();
fd.append("filename",filename); fd.append("volume_db",volume);
fd.append("speed_factor",speed); fd.append("trim_start",trimStart);
fd.append("trim_end",trimEnd); fd.append("fade_in",fadeIn);
fd.append("fade_out",fadeOut); fd.append("normalize",normalize);
try{
const r=await fetch("/edit_audio",{method:"POST",body:fd});
if(!r.ok){const e=await r.json();throw new Error(e.detail||"خطأ");}
const data=await r.json();
onEdited({url:`/audio/${data.filename}`,filename:data.filename});
reset(); toast("success","✅ تم حفظ ملف معدَّل جديد");
}catch(e){toast("error",e.message);}
finally{setApplying(false);}
};
return (
<div style={{paddingTop:16}}>
<div className="row" style={{marginBottom:14,justifyContent:"space-between"}}>
<span style={{fontFamily:"var(--mono)",fontSize:10,color:"var(--text2)",
letterSpacing:".1em",textTransform:"uppercase"}}>✂ تعديل الصوت</span>
{dirty&&<button className="btn-secondary btn-sm"
style={{fontSize:11,color:"var(--text3)"}} onClick={reset}>↺ إعادة ضبط</button>}
</div>
<div className="col" style={{gap:14}}>
<div className="grid-2" style={{gap:12}}>
<Slider label="مستوى الصوت" min={-20} max={20} step={1}
value={volume} onChange={setVolume} left="-20 dB" right="+20 dB"/>
<Slider label="سرعة التشغيل" min={0.5} max={2.0} step={0.05}
value={speed} onChange={setSpeed} left="×0.5" right="×2.0"/>
</div>
<div className="grid-2" style={{gap:12}}>
<Slider label="قطع من البداية (ث)" min={0} max={20} step={0.1}
value={trimStart} onChange={setTrimStart} left="0" right="20 ث"/>
<Slider label="قطع من النهاية (ث)" min={0} max={20} step={0.1}
value={trimEnd} onChange={setTrimEnd} left="0" right="20 ث"/>
</div>
<div className="grid-2" style={{gap:12}}>
<Slider label="Fade In (ث)" min={0} max={5} step={0.1}
value={fadeIn} onChange={setFadeIn} left="0" right="5 ث"/>
<Slider label="Fade Out (ث)" min={0} max={5} step={0.1}
value={fadeOut} onChange={setFadeOut} left="0" right="5 ث"/>
</div>
<div className="row" style={{justifyContent:"space-between",flexWrap:"wrap",gap:10}}>
<Toggle label="Normalize — رفع أعلى ذروة إلى 0 dB"
value={normalize} onChange={setNormalize}/>
<button onClick={apply} disabled={applying} style={{
background:dirty?"var(--amber)":"var(--surface2)",
color:dirty?"#000":"var(--text3)",
border:dirty?"none":"1px solid var(--border)",
fontFamily:"var(--sans)",fontWeight:700,fontSize:13,
borderRadius:"var(--radius-sm)",padding:"9px 20px",
cursor:applying?"not-allowed":"pointer",
display:"flex",alignItems:"center",gap:8,
opacity:applying?.6:1,whiteSpace:"nowrap",flexShrink:0,
}}>
{applying
? <><div className="wave-bars">{[1,2,3].map(i=><span key={i}/>)}</div>جاري التعديل…</>
: "✅ تطبيق وحفظ كملف جديد"}
</button>
</div>
</div>
</div>
);
}
/* ══ HUMANIZE PANEL ═════════════════════════════════════════════════ */
const HUMANIZE_PRESETS_UI = [
{ id:"studio", icon:"🎙", label:"استوديو", desc:"بوث احترافي — نظيف، هادئ، موثوق" },
{ id:"bedroom", icon:"🏠", label:"غرفة", desc:"غرفة صغيرة غير معالَجة — يوتيوب / بودكاست هواة" },
{ id:"podcast", icon:"🎧", label:"بودكاست", desc:"ميكروفون ديناميكي دافئ — قريب وحميمي" },
{ id:"radio", icon:"📻", label:"راديو", desc:"بث كلاسيكي — نطاق ترددي محدود، ضغط مرتفع" },
{ id:"phone", icon:"📱", label:"هاتف", desc:"GSM / تيليفون — نطاق ضيق 300–3400 هرتز" },
{ id:"cafe", icon:"☕", label:"مقهى", desc:"غرفة واسعة مع ضجيج خفيف في الخلفية" },
{ id:"outdoor", icon:"🌿", label:"خارجي", desc:"فضاء مفتوح مع ريح وأجواء طبيعية" },
];
function HumanizePanel({ filename, onHumanized, storageKey }) {
const sk = storageKey || "x";
const [preset, setPreset] = useLS("hum.preset." + sk, "bedroom");
const [intensity, setIntensity] = useLS("hum.intensity." + sk, 0.8);
const [applying, setApplying] = useState(false);
const [lastOut, setLastOut] = useState(null); // last humanized result
const apply = async ()=>{
setApplying(true);
const fd=new FormData();
fd.append("filename",filename);
fd.append("preset",preset);
fd.append("intensity",intensity);
fd.append("seed", Math.floor(Math.random()*9999));
try{
const r=await fetch("/humanize",{method:"POST",body:fd});
if(!r.ok){const e=await r.json();throw new Error(e.detail||"خطأ");}
const data=await r.json();
const res={url:`/audio/${data.filename}`,filename:data.filename};
setLastOut(res);
onHumanized(res);
toast("success","🎙 تمت المحاكاة الصوتية — استمع إلى الفرق");
}catch(e){toast("error",e.message);}
finally{setApplying(false);}
};
return (
<div style={{paddingTop:16}}>
{/* Section header */}
<div className="row" style={{marginBottom:16,gap:10}}>
<span style={{fontFamily:"var(--mono)",fontSize:10,color:"var(--text2)",
letterSpacing:".1em",textTransform:"uppercase"}}>🎙 بيئة التسجيل الصوتية</span>
<span className="badge badge-purple">DSP</span>
</div>
{/* Info box */}
<div style={{
padding:"10px 14px",borderRadius:"var(--radius-sm)",
background:"rgba(167,139,250,.07)",border:"1px solid rgba(167,139,250,.2)",
fontSize:12,color:"var(--text2)",lineHeight:1.7,marginBottom:16,
}}>
تُضيف هذه الأداة طابع التسجيل البشري الحقيقي: ضجيج الغرفة الطبيعي، دفء ميكروفون
حقيقي، انجراف طفيف في النبرة، ارتعاش في التوقيت، تشبّع هارمونيكي (tape/tube)،
وصدى بيئي — كل ذلك بدون إعادة توليد الصوت.
</div>
{/* Preset grid */}
<div className="label" style={{marginBottom:10}}>اختر البيئة الصوتية</div>
<div style={{
display:"grid",
gridTemplateColumns:"repeat(auto-fill,minmax(130px,1fr))",
gap:8, marginBottom:20,
}}>
{HUMANIZE_PRESETS_UI.map(p=>(
<div key={p.id} onClick={()=>setPreset(p.id)} title={p.desc}
style={{
display:"flex",flexDirection:"column",gap:5,
padding:"10px 12px",borderRadius:"var(--radius-sm)",
border: preset===p.id
? "1.5px solid var(--amber)"
: "1px solid var(--border)",
background: preset===p.id ? "var(--amber-glow)" : "var(--surface2)",
cursor:"pointer",
transition:"border-color var(--trans),background var(--trans)",
}}>
<div style={{fontSize:22,lineHeight:1}}>{p.icon}</div>
<div style={{
fontSize:13,fontWeight:700,
color:preset===p.id?"var(--amber)":"var(--text)",
}}>{p.label}</div>
<div style={{fontSize:10,color:"var(--text3)",lineHeight:1.4}}>{p.desc}</div>
</div>
))}
</div>
{/* Intensity slider */}
<Slider
label="شدة التأثير"
min={0.1} max={2.0} step={0.05}
value={intensity} onChange={setIntensity}
left="خفيف جداً" right="مبالَغ فيه"
hint="1.0 = القيمة الموصى بها. أكثر من 1.5 لتأثيرات درامية (هاتف قديم / راديو خمسينات)."
/>
{/* Apply button */}
<button onClick={apply} disabled={applying} style={{
marginTop:18, width:"100%",
background:"linear-gradient(135deg,rgba(167,139,250,.85),rgba(139,92,246,.9))",
color:"#fff", border:"none",
fontFamily:"var(--sans)", fontWeight:700, fontSize:15,
borderRadius:"var(--radius)", padding:"13px 20px",
cursor:applying?"not-allowed":"pointer",
display:"flex", alignItems:"center", justifyContent:"center", gap:10,
opacity:applying?.55:1,
transition:"opacity var(--trans)",
boxShadow:"0 4px 20px rgba(139,92,246,.3)",
}}>
{applying
? <><div className="wave-bars" style={{color:"#fff"}}>
{[1,2,3,4].map(i=><span key={i}/>)}
</div>جاري محاكاة البيئة الصوتية…</>
: <>🎙 تطبيق البيئة الصوتية — ملف جديد</>}
</button>
{/* last result mini-badge */}
{lastOut && (
<div style={{marginTop:10,display:"flex",alignItems:"center",gap:8}}>
<span className="badge badge-purple">✓ آخر نتيجة</span>
<a href={lastOut.url} download={lastOut.filename}
style={{fontSize:12,color:"var(--purple)",textDecoration:"none"}}>
⬇ {lastOut.filename}
</a>
</div>
)}
</div>
);
}
/* ══ RESULT CARD — shared across all tabs ═══════════════════════════ */
/**
* Shows the player + two tabs: "✂ تعديل" and "🎙 بيئة"
* onUpdate(result): called when either editor or humanizer produces a new file.
*/
function ResultCard({ result, onUpdate, badge="✓ تم التوليد", title="🔊 النتيجة", storageKey="gen" }) {
const [panel, setPanel] = useLS("rc.panel." + storageKey, "edit");
const [editorKey, setEditorKey] = useState(0);
// When the file changes (e.g. humanizer produces a new one), reset editor
const handleUpdate = (r)=>{ onUpdate(r); setEditorKey(k=>k+1); };
const TAB_BTN = (id, label) => (
<button onClick={()=>setPanel(id)} style={{
padding:"6px 14px", borderRadius:"var(--radius-sm)",
border: panel===id ? "1.5px solid var(--amber)" : "1px solid var(--border)",
background: panel===id ? "var(--amber-glow)" : "transparent",
color: panel===id ? "var(--amber)" : "var(--text2)",
fontFamily:"var(--sans)", fontSize:12, fontWeight:600,
cursor:"pointer", transition:"all var(--trans)",
}}>{label}</button>
);
return (
<div className="card">
<div className="card-header">
<span className="card-title">{title}</span>
<div className="row" style={{gap:6,flexWrap:"wrap"}}>
<span className="badge badge-green">{badge}</span>
{TAB_BTN("edit","✂ تعديل")}
{TAB_BTN("humanize","🎙 بيئة صوتية")}
</div>
</div>
<div className="card-body">
<AudioPlayer url={result.url} filename={result.filename}/>
<div style={{borderTop:"1px solid var(--border)",marginTop:14}}>
{panel==="edit"
? <AudioEditor key={editorKey} filename={result.filename} onEdited={handleUpdate}/>
: <HumanizePanel filename={result.filename} onHumanized={handleUpdate} storageKey={storageKey}/>}
</div>
</div>
</div>
);
}
/* ══ SLIDER ════════════════════════════════════════════════════════ */
function Slider({ label, min, max, step, value, onChange, left, right, hint }) {
const pct = ((value-min)/(max-min)*100).toFixed(1);
return (
<div>
<div className="slider-meta">
<span className="slider-label">{label}</span>
<span className="slider-value">{fmt2(value)}</span>
</div>
<div className="slider-wrap">
<div className="slider-track">
<div className="slider-fill" style={{width:`${pct}%`}}/>
<input type="range" min={min} max={max} step={step} value={value}
onChange={e=>onChange(parseFloat(e.target.value))}/>
</div>
</div>
{(left||right)&&<div className="slider-hints">
<span className="slider-hint">{left}</span>
<span className="slider-hint">{right}</span>
</div>}
{hint&&<div style={{marginTop:5,fontSize:11,color:"var(--text3)",lineHeight:1.5}}>{hint}</div>}
</div>
);
}
/* ══ TOGGLE ════════════════════════════════════════════════════════ */
function Toggle({ label, value, onChange, hint }) {
return (
<div>
<div className="toggle" onClick={()=>onChange(!value)}>
<div className={`toggle-track ${value?"on":""}`}><div className="toggle-thumb"/></div>
<span className="toggle-label">{label}</span>
</div>
{hint&&<div style={{marginTop:4,fontSize:11,color:"var(--text3)",lineHeight:1.5,marginRight:48}}>{hint}</div>}
</div>
);
}
/* ══ FILE UPLOAD ════════════════════════════════════════════════════ */
function FileUpload({ label, file, onFile, hint }) {
const [drag, setDrag] = useState(false);
return (
<div>
{label&&<div className="label">{label}</div>}
{file ? (
<div className="file-chip">
<span className="file-chip-icon">🎙</span>
<span className="file-chip-name">{file.name}</span>
<span className="file-chip-size">{fmtSize(file.size)}</span>
<button className="file-chip-rm" onClick={()=>onFile(null)}>✕</button>
</div>
) : (
<div className={`dropzone ${drag?"drag":""}`}
onDragOver={e=>{e.preventDefault();setDrag(true)}}
onDragLeave={()=>setDrag(false)}
onDrop={e=>{e.preventDefault();setDrag(false);onFile(e.dataTransfer.files[0]);}}>
<input type="file" accept="audio/*" onChange={e=>onFile(e.target.files[0])}/>
<div className="dropzone-icon">🎵</div>
<div className="dropzone-label">اسحب ملفاً أو انقر للاختيار</div>
{hint&&<div className="dropzone-hint">{hint}</div>}
</div>
)}
</div>
);
}
/* ══ VOICE SELECTOR ════════════════════════════════════════════════ */
const AVATAR_COLORS=[
["#fbbf24","#78350f"],["#34d399","#064e3b"],["#818cf8","#1e1b4b"],
["#f472b6","#500724"],["#38bdf8","#082f49"],["#a78bfa","#2e1065"],
];
function voiceAvatar(name){
const i=name.charCodeAt(0)%AVATAR_COLORS.length;
const [bg,fg]=AVATAR_COLORS[i];
return {bg,fg,initials:name.slice(0,2).toUpperCase()};
}
function VoiceSelector({ voices, selected, onSelect, f1, setF1, f2, setF2,
saveName, setSaveName, onSave, saveStatus }) {
const [open, setOpen] = useState(false);
return (
<div>
<div className="label">الصوت المرجعي</div>
{selected ? (
<div className="voice-card selected" onClick={()=>setOpen(v=>!v)}>
{(()=>{ const av=voiceAvatar(selected); return (
<div className="voice-avatar" style={{background:av.bg,color:av.fg,borderColor:av.bg}}>
{av.initials}
</div>
);})()}
<div className="voice-info">
<div className="voice-name">{selected}</div>
<div className="voice-tags"><span className="badge badge-green">من المكتبة</span></div>
</div>
<span className="voice-check">✓</span>
<button className="btn-secondary btn-sm"
onClick={e=>{e.stopPropagation();onSelect(null);}}>تغيير</button>
</div>
) : (
<FileUpload label={null} file={f1} onFile={setF1}
hint="WAV · MP3 · FLAC · مدة 5 – 30 ثانية مثالية"/>
)}
{f1 && !selected && (
<div className="mt-8">
<FileUpload label="عينة إضافية (اختياري — تحسّن الجودة)" file={f2} onFile={setF2}/>
</div>
)}
{f1 && !selected && (
<div className="mt-8 row">
<input type="text" placeholder="احفظ هذا الصوت في المكتبة…"
value={saveName} onChange={e=>setSaveName(e.target.value)} style={{flex:1}}/>
<button className="btn-secondary" onClick={onSave}>حفظ</button>
{saveStatus&&<span style={{fontSize:12,color:"var(--green)"}}>{saveStatus}</span>}
</div>
)}
{voices.length>0 && !selected && (
<div className="mt-12">
<div className="row" style={{marginBottom:10}}>
<div className="label" style={{margin:0,flex:1}}>أو اختر من مكتبة الأصوات</div>
<button className="btn-secondary btn-sm" onClick={()=>setOpen(v=>!v)}>
{open?"▲ إخفاء":"▼ عرض"} ({voices.length})
</button>
</div>
{open&&(
<div className="col" style={{gap:8}}>
{voices.map(v=>{
const av=voiceAvatar(v);
return (
<div key={v} className="voice-card" onClick={()=>{onSelect(v);setOpen(false);}}>
<div className="voice-avatar" style={{background:av.bg,color:av.fg,borderColor:av.bg}}>
{av.initials}
</div>
<div className="voice-info">
<div className="voice-name">{v}</div>
<div className="voice-tags"><span className="badge badge-amber">مكتبة</span></div>
</div>
<button className="btn-secondary btn-sm"
onClick={e=>{e.stopPropagation();onSelect(v);setOpen(false);}}>اختيار</button>
</div>
);
})}
</div>
)}
</div>
)}
</div>
);
}
/* ══ LIVE EMOTION BADGE ════════════════════════════════════════════ */
const EMOJI_MAP={neutral:"😐",question:"❓",exclaim:"❗",emphasis:"💬",
sadness:"😢",joy:"😊",anger:"😡",soft:"🌸",urgency:"⚡",story:"📖"};
const ECOLOR_MAP={neutral:"var(--text3)",question:"var(--blue)",exclaim:"var(--amber)",
emphasis:"var(--purple)",sadness:"#7dd3fc",joy:"#fde047",anger:"var(--red)",
soft:"#f9a8d4",urgency:"#fb923c",story:"#86efac"};
function detectEmotionClient(phrase) {
const p = phrase.trim();
if (!p) return "neutral";
if (/[؟?]$/.test(p)) return "question";
const excl = (p.match(/!/g)||[]).length;
if (excl >= 2) return "exclaim";
if (excl === 1 && p.length < 60) return "exclaim";
const patterns = {
joy: /سعيد|فرح|رائع|مذهل|ممتاز|مبروك|جميل|أحب|نجح|happy|wonderful|amazing|great|love|joy|fantastic/i,
sadness: /حزين|للأسف|مؤسف|وداع|ألم|فقدنا|دموع|اشتاق|sad|unfortunately|sorry|pain|tears|miss|farewell/i,
anger: /غضب|مستحيل|يكفي|ظلم|سئمت|never|impossible|stop|angry|refuse|furious|outrage/i,
emphasis: /بالتأكيد|قطعاً|أكيد|يجب|مهم|انتبه|definitely|absolutely|must|important|crucial|clearly/i,
soft: /ربما|لعل|أتمنى|أرجو|هدوء|سلام|حنان|perhaps|maybe|hopefully|gently|peace|calm|tenderly/i,
urgency: /الآن|فوراً|عاجل|احذر|خطر|أسرع|now|immediately|quickly|urgent|warning|danger|hurry/i,
story: /كان يا مكان|ذات يوم|يحكى|ثم قال|once upon|there was|long ago|suddenly|meanwhile/i,
};
for (const [em, re] of Object.entries(patterns)) { if (re.test(p)) return em; }
if (/[…]$/.test(p) || p.endsWith("...")) return "soft";
return "neutral";
}
function LiveEmotionBadge({ text, variation }) {
const emotions = useMemo(()=>{
if (!text.trim() || variation < 0.05) return {};
const sentences = text.replace(/\n+/g,' ').split(/[.!?\u061F\u060C]+/).map(s=>s.trim()).filter(Boolean);
const counts = {};
for (const s of sentences) { const em=detectEmotionClient(s); counts[em]=(counts[em]||0)+1; }
if (Object.keys(counts).length > 1) delete counts.neutral;
return counts;
}, [text, variation]);
const entries = Object.entries(emotions).sort((a,b)=>b[1]-a[1]).slice(0,4);
if (!entries.length) return null;
return (
<div style={{display:"flex",alignItems:"center",gap:5,flexWrap:"wrap"}}>
<span style={{fontFamily:"var(--mono)",fontSize:9,color:"var(--text3)",letterSpacing:".08em",textTransform:"uppercase"}}>مكتشَف:</span>
{entries.map(([em,count])=>(
<span key={em} style={{display:"inline-flex",alignItems:"center",gap:3,
padding:"2px 8px",borderRadius:20,background:`${ECOLOR_MAP[em]}22`,
border:`1px solid ${ECOLOR_MAP[em]}55`,color:ECOLOR_MAP[em],
fontFamily:"var(--mono)",fontSize:10}}>
{EMOJI_MAP[em]} {em}{count>1&&<span style={{opacity:.7}}>×{count}</span>}
</span>
))}
</div>
);
}
/* ══ EXPRESSION MODE SELECTOR ══════════════════════════════════════ */
const EXPRESSION_MODES = [
{ id:"natural", icon:"😐", label:"طبيعي", hint:"الإعدادات اليدوية تتحكم بالكامل" },
{ id:"warm", icon:"🤗", label:"دافئ", hint:"لين، ودود، مريح — مثالي للمحتوى التعليمي" },
{ id:"excited", icon:"🎉", label:"متحمس", hint:"إيقاع سريع، طاقة عالية — للإعلانات والترويج" },
{ id:"sad", icon:"😢", label:"حزين", hint:"بطيء، ثقيل، إيقاع هادئ — للقصص العاطفية" },
{ id:"serious", icon:"🎙", label:"رسمي", hint:"واضح، موثوق، محترف — للأخبار والشرح" },
{ id:"story", icon:"📖", label:"قصصي", hint:"تنوع درامي في النبرة — للروايات والحكايات" },
{ id:"news", icon:"📰", label:"إخباري", hint:"حاسم، لا توقفات زائدة — للنشرات الإخبارية" },
{ id:"child", icon:"🧒", label:"أطفال", hint:"خفيف، مفرح، جُمَل قصيرة — للمحتوى التعليمي" },
];
function ExpressionModeSelector({ value, onChange }) {
return (
<div>
<div className="label" style={{marginBottom:10}}>
🎭 طابع التعبير — يؤثر على النبرة والإيقاع والنفَس
</div>
<div style={{display:"grid",gridTemplateColumns:"repeat(auto-fill,minmax(150px,1fr))",gap:8}}>
{EXPRESSION_MODES.map(m=>(
<div key={m.id}
onClick={()=>onChange(m.id)}
title={m.hint}
style={{
display:"flex", alignItems:"center", gap:9,
padding:"9px 12px", borderRadius:"var(--radius-sm)",
border: value===m.id
? "1.5px solid var(--amber)"
: "1px solid var(--border)",
background: value===m.id ? "var(--amber-glow)" : "var(--surface2)",
cursor:"pointer",
transition:"border-color var(--trans),background var(--trans)",
}}>
<span style={{fontSize:18,lineHeight:1,flexShrink:0}}>{m.icon}</span>
<div style={{minWidth:0}}>
<div style={{
fontSize:13, fontWeight:600,
color: value===m.id ? "var(--amber)" : "var(--text)",
whiteSpace:"nowrap",overflow:"hidden",textOverflow:"ellipsis",
}}>{m.label}</div>
<div style={{fontSize:10,color:"var(--text3)",
overflow:"hidden",textOverflow:"ellipsis",whiteSpace:"nowrap",
maxWidth:110}}>{m.hint}</div>
</div>
</div>
))}
</div>
{value !== "natural" && (
<div style={{marginTop:8,fontSize:11,color:"var(--text3)",lineHeight:1.6}}>
⚠ في الأوضاع غير الطبيعية، يتم تجاوز مؤشرات Stability/Clarity/Style
واستخدام إعدادات معدّة مسبقاً للتعبير الإنساني الأمثل.
</div>
)}
</div>
);
}
function VoiceSettings({ stability,setStability,clarity,setClarity,
styleExag,setStyleExag,speed,setSpeed,boost,setBoost,
splitText,setSplitText,variation,setVariation,
refClean,setRefClean,cleanText,setCleanText,attempts,setAttempts,
depth,setDepth,space,setSpace }) {
return (
<div className="col" style={{gap:20}}>
<Slider label="Stability — الثبات" min={0} max={1} step={.05}
value={stability} onChange={setStability} left="تعبيري عاطفي" right="متسق ثابت"
hint="منخفض = ديناميكي مثالي للتمثيل. مرتفع = ثابت للكتب الصوتية."/>
<Slider label="Clarity + Similarity — الوضوح" min={0} max={1} step={.05}
value={clarity} onChange={setClarity} left="متنوع" right="مطابق للمرجع"
hint="مرتفع = التزام أقوى بالصوت المرجعي وإخراج أوضح."/>
<Slider label="Style Exaggeration — الأسلوب" min={0} max={1} step={.05}
value={styleExag} onChange={setStyleExag} left="طبيعي هادئ" right="مسرحي مبالغ"
hint="يضيف دراما وتعبيراً. صفر = أكثر طبيعية."/>
<Slider label="Speed — السرعة" min={.5} max={2} step={.05}
value={speed} onChange={setSpeed} left="بطيء ×0.5" right="سريع ×2.0"/>
<div className="divider"/>
{/* ── Prosody Variation — the key expressiveness knob ── */}
<Slider label="🎭 Prosody Variation — تنوع تلقائي لكل جملة" min={0} max={2} step={.05}
value={variation} onChange={setVariation}
left="مستوٍ (آلي)" right="درامي جداً"
hint="يحلل كل جملة ويغير تلقائياً: سرعة التوليد، حرارة النموذج، عقوبة التكرار، وشكل مستوى الصوت — بناءً على الكلمات المفتاحية والترقيم. 0 = ثابت، 1.0 = طبيعي، 2.0 = مبالَغ."/>
<Toggle label="Speaker Boost — تعزيز الصوت (Soft Clip)" value={boost} onChange={setBoost}
hint="يرفع مستوى الصوت مع تطبيق soft clipping لإزالة التشويه الرقمي."/>
<Toggle label="تقسيم النص التلقائي" value={splitText} onChange={setSplitText}
hint="يقسم النصوص الطويلة إلى جمل ويعرض شريط تقدم حقيقي."/>
<div className="divider"/>
{/* ── Naturalness engine ── */}
<Toggle label="🧼 تنقية العينة الصوتية تلقائياً" value={refClean} onChange={setRefClean}
hint="يزيل الضجيج والصمت الطويل ويختار أنقى 20–25 ثانية من العينة ويوحّد مستواها. أهم عامل في جودة الاستنساخ."/>
<Toggle label="📝 تنظيف النص قبل التوليد" value={cleanText} onChange={setCleanText}
hint="يحذف التشكيل والتطويل، يحوّل الأرقام الهندية، ويحوّل الأقواس والاقتباسات إلى وقفات. عطّله إن كنت تعتمد على التشكيل."/>
<div>
<div className="slider-meta"><span className="slider-label">🎯 اختيار أفضل نسخة (محاولات لكل جملة)</span></div>
<select value={attempts} onChange={e=>setAttempts(parseInt(e.target.value))}>
<option value={1}>1 — سريع</option>
<option value={2}>2 — أدق (وقت ×2)</option>
<option value={3}>3 — أدق جداً (وقت ×3)</option>
</select>
<div style={{marginTop:5,fontSize:11,color:"var(--text3)",lineHeight:1.5}}>يولّد عدة نسخ لكل جملة ويختار الأوضح نطقاً (مقارنة عبر Whisper) — يتجنب الجمل المتلعثمة أو الآلية.</div>
</div>
<div className="divider"/>
{/* ── Depth / space ── */}
<Slider label="🧊 العمق والمجسّم — Depth" min={0} max={1} step={.05}
value={depth} onChange={setDepth} left="جاف مسطّح" right="غرفة ذات أبعاد"
hint="غرفة ستيريو حقيقية + دفء وحضور وهواء + ضجيج غرفة يملأ الفراغات بدل الصمت الرقمي. ابدأ من 0.4–0.6. صفر = الصوت الجاف الأصلي."/>
<div>
<div className="slider-meta"><span className="slider-label">🏠 نوع المساحة</span></div>
<select value={space} onChange={e=>setSpace(e.target.value)}>
<option value="studio">استوديو — قريب ودافئ</option>
<option value="room">غرفة — طبيعي</option>
<option value="hall">قاعة — واسع</option>
</select>
</div>
</div>
);
}
/* ══ HOOK: useJobRunner — manages cancel, progress, beforeunload ════ */
function useJobRunner() {
const abortRef = useRef(null);
const jobIdRef = useRef(null);
const pollRef = useRef(null);
const [running, setRunning] = useState(false);
const [progress, setProgress] = useState(null);
const startPoll = (jid) => {
stopPoll();
pollRef.current = setInterval(async ()=>{
try {
const r = await fetch(`/progress/${jid}`);
if (r.ok) {
const p = await r.json();
setProgress(p);
if (p.status !== "running") stopPoll();
}
} catch {}
}, 600);
};
const stopPoll = () => {
if (pollRef.current) { clearInterval(pollRef.current); pollRef.current=null; }
};
const cancel = useCallback(async () => {
if (abortRef.current) { abortRef.current.abort(); abortRef.current=null; }
if (jobIdRef.current) {
try { await fetch(`/cancel/${jobIdRef.current}`, {method:"POST"}); } catch {}
}
stopPoll();
setRunning(false);
setProgress(null);
toast("info","⛔ تم إلغاء العملية");
}, []);
// beforeunload: cancel silently via sendBeacon
useEffect(()=>{
const handler = ()=>{
if (jobIdRef.current) {
navigator.sendBeacon(`/cancel/${jobIdRef.current}`);
}
};
window.addEventListener("beforeunload", handler);
return ()=>window.removeEventListener("beforeunload", handler);
}, []);
const run = useCallback(async (url, formData) => {
if (abortRef.current) { abortRef.current.abort(); }
const jid = genId();
jobIdRef.current = jid;
formData.set("job_id", jid);
const controller = new AbortController();
abortRef.current = controller;
setRunning(true); setProgress({current:0,total:1,status:"running"});
startPoll(jid);
try {
const r = await fetch(url, {method:"POST", body:formData, signal:controller.signal});
if (!r.ok) { const e=await r.json(); throw new Error(e.detail||"خطأ في الخادم"); }
const data = await r.json();
stopPoll();
setProgress({current:1,total:1,status:"done"});
return data;
} catch(e) {
stopPoll();
setProgress(null);
if (e.name==="AbortError") return null; // user cancelled
throw e;
} finally {
setRunning(false);
abortRef.current = null;
jobIdRef.current = null;
}
}, []);
return { running, progress, run, cancel };
}
/* ══ GENERATE TAB ══════════════════════════════════════════════════ */
function GenerateTab({ voices, onVoicesRefresh, sharedSettings }) {
const { stability,setStability,clarity,setClarity,styleExag,setStyleExag,
speed,setSpeed,boost,setBoost,splitText,setSplitText,
variation,setVariation,refClean,cleanText,attempts,depth,space } = sharedSettings;
const [refCheck, setRefCheck] = useState(null);
const [checking, setChecking] = useState(false);
/* ── persisted across refresh ── */
const [text, setText] = useLS("gen.text", "");
const [lang, setLang] = useLS("gen.lang", "ar");
const [selVoice, setSelVoice] = useLS("gen.voice", null);
const [exprMode, setExprMode] = useLS("gen.expr", "warm");
const [result, setResult] = usePersistedResult("gen.result");
/* ── session-only (files can't be serialised) ── */
const [f1, setF1] = useState(null);
const [f2, setF2] = useState(null);
const [saveName, setSaveName] = useState("");
const [saveStatus,setSaveStatus]= useState("");
const { running, progress, run, cancel } = useJobRunner();
const isRTL = ["ar","fa","he","ur"].includes(lang);
const chars = text.length;
useEffect(()=>{
const handler=e=>{if((e.ctrlKey||e.metaKey)&&e.key==="Enter") generate();};
window.addEventListener("keydown",handler);
return ()=>window.removeEventListener("keydown",handler);
});
const saveVoice = async ()=>{
if (!saveName.trim()||!f1) return;
const fd=new FormData();
fd.append("name",saveName.trim()); fd.append("file",f1);
if(f2) fd.append("file2",f2);
try{
const r=await fetch("/voices/save",{method:"POST",body:fd});
if(!r.ok) throw new Error("فشل الحفظ");
setSaveStatus("✓ محفوظ"); onVoicesRefresh();
setTimeout(()=>setSaveStatus(""),2000);
} catch(e){ toast("error",e.message); }
};
const checkRef = async ()=>{
if (!f1&&!selVoice) return toast("error","حدد صوتاً مرجعياً أولاً.");
const fd=new FormData();
if(f1) fd.append("files",f1);
if(f2) fd.append("files",f2);
if(selVoice) fd.append("voice_name",selVoice);
setChecking(true); setRefCheck(null);
try{
const r=await fetch("/enhance_ref",{method:"POST",body:fd});
const j=await r.json().catch(()=>({}));
if(!r.ok) throw new Error(j.detail||"فشل فحص العينة");
setRefCheck(j);
} catch(e){ toast("error",e.message); }
setChecking(false);
};
const generate = async ()=>{
if (!text.trim()) return toast("error","أدخل النص أولاً.");
if (!f1&&!selVoice) return toast("error","حدد صوتاً مرجعياً.");
if (chars>MAX_CH) return toast("error",`النص يتجاوز ${MAX_CH} حرف.`);
const fd=new FormData();
fd.append("text",text); fd.append("language",lang);
fd.append("stability",stability); fd.append("clarity",clarity);
fd.append("style_exaggeration",styleExag); fd.append("speed",speed);
fd.append("speaker_boost",boost); fd.append("enable_text_splitting",splitText);
fd.append("expression_mode", exprMode);
fd.append("variation_strength", variation);
fd.append("enhance_ref", refClean); fd.append("clean_text", cleanText);
fd.append("attempts", attempts);
fd.append("depth", depth); fd.append("space", space);
if(f1) fd.append("files",f1);
if(f2) fd.append("files",f2);
if(selVoice) fd.append("voice_name",selVoice);
try {
const data = await run("/generate", fd);
if (!data) return; // cancelled
setResult({url:`/audio/${data.filename}`,filename:data.filename});
toast("success","✅ تم توليد الصوت بنجاح");
if(data.cached) toast("info","⚡ أُعيدت المعالجة النهائية فقط من الذاكرة المؤقتة (لم يُعاد توليد الصوت).",5000);
const rr=data.ref_report;
if(rr&&(rr.score<70||(rr.warnings||[]).length))
toast("info",`جودة العينة ${rr.score}/100 — ${(rr.warnings||[])[0]||"جرّب عينة أنقى"}`,8000);
} catch(e) { toast("error",e.message); }
};
return (
<div className="col" style={{gap:16}}>
{running && <ProgressBar progress={progress}/>}
<div className="card">
<div className="card-header">
<span className="card-title">✏ النص المراد تحويله</span>
<div className="row" style={{gap:8}}>
<span className="fz-12 text-muted">اللغة:</span>
<select value={lang} onChange={e=>setLang(e.target.value)}>
{Object.entries(LANGS).map(([k,v])=><option key={k} value={k}>{v}</option>)}
</select>
</div>
</div>
<div className="card-body" style={{padding:0}}>
<textarea className="textarea" dir={isRTL?"rtl":"ltr"} style={{padding:"16px 18px"}}
rows={7} placeholder={isRTL?"اكتب أو الصق النص هنا…":"Type or paste text here…"}
value={text} onChange={e=>setText(e.target.value.slice(0,MAX_CH+50))}/>
<div className="char-bar">
<span className={`char-count ${chars>MAX_CH?"limit":chars>MAX_CH*.8?"warn":""}`}>
{chars.toLocaleString()} / {MAX_CH.toLocaleString()} حرف
</span>
<div className="row" style={{gap:8}}>
{text&&<button className="btn-secondary btn-sm" onClick={()=>setText("")}>مسح</button>}
<span className="fz-12 text-muted" style={{fontFamily:"var(--mono)"}}>Ctrl+Enter للتوليد</span>
</div>
</div>
</div>
</div>
<div className="grid-2">
<div className="card">
<div className="card-header"><span className="card-title">🎙 الصوت المرجعي</span></div>
<div className="card-body">
<VoiceSelector voices={voices} selected={selVoice} onSelect={setSelVoice}
f1={f1} setF1={setF1} f2={f2} setF2={setF2}
saveName={saveName} setSaveName={setSaveName}
onSave={saveVoice} saveStatus={saveStatus}/>
<div className="col" style={{gap:8,marginTop:14}}>
<button className="btn-secondary btn-sm" onClick={checkRef} disabled={checking}>
{checking?"… جارٍ الفحص":"🔍 فحص وتنقية العينة والاستماع إليها"}
</button>
{refCheck&&refCheck.report&&(
<div className="col" style={{gap:6}}>
<div className="fz-12"><b>جودة العينة: {refCheck.report.score}/100</b> · المدة بعد التنقية {refCheck.report.duration}ث{refCheck.segments>1?" ("+refCheck.segments+" مقاطع متجاورة، المعروض أفضلها)":""} · SNR {refCheck.report.snr_before}→{refCheck.report.snr_after} dB{refCheck.report.denoised?" · تم تقليل الضجيج":""}</div>
{(refCheck.report.warnings||[]).map((w,i)=><div key={"w"+i} className="fz-12" style={{color:"#f59e0b"}}>⚠ {w}</div>)}
{(refCheck.report.tips||[]).map((w,i)=><div key={"t"+i} className="fz-12 text-muted">💡 {w}</div>)}
<audio controls src={refCheck.url} style={{width:"100%"}}/>
</div>
)}
</div>
</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">⚙ إعدادات الصوت</span></div>
<div className="card-body"><VoiceSettings {...sharedSettings}/></div>
</div>
</div>
{/* Expression mode — full width card */}
<div className="card">
<div className="card-header"><span className="card-title">🎭 طابع التعبير والإلقاء</span>{text.trim()&&<LiveEmotionBadge text={text} variation={variation}/>}</div>
<div className="card-body">
<ExpressionModeSelector value={exprMode} onChange={setExprMode}/>
</div>
</div>
{!running
? <button className="btn-primary" onClick={generate}>⚡ توليد الصوت</button>
: <button className="btn-cancel" onClick={cancel}>
<div className="wave-bars">{[1,2,3,4].map(i=><span key={i}/>)}</div>
إلغاء التوليد
</button>
}
{result&&(
<ResultCard result={result} onUpdate={setResult}
title="🔊 النتيجة" badge="✓ تم التوليد" storageKey="gen"/>
)}
</div>
);
}
/* ══ STS TAB ═══════════════════════════════════════════════════════ */
function STSTab({ voices, sharedSettings }) {
const [src, setSrc] = useState(null);
const [ref1, setRef1] = useState(null);
const [ref2, setRef2] = useState(null);
const [selVoice, setSelVoice] = useState(null);
const [outLang, setOutLang] = useState("ar");
const [result, setResult] = useState(null);
const [transcribed, setTranscribed] = useState("");
const { stability,clarity,styleExag,speed,boost,depth,space } = sharedSettings;
const { running, progress, run, cancel } = useJobRunner();
const doRun = async ()=>{
if(!src) return toast("error","ارفع الملف الصوتي المصدر.");
if(!ref1&&!selVoice) return toast("error","حدد صوت الهدف.");
const fd=new FormData();
fd.append("source_audio",src); fd.append("language",outLang);
fd.append("stability",stability); fd.append("clarity",clarity);
fd.append("style_exaggeration",styleExag); fd.append("speed",speed);
fd.append("speaker_boost",boost); fd.append("variation_strength_sts",variation); fd.append("depth",depth); fd.append("space",space);
if(ref1) fd.append("ref_files",ref1);
if(ref2) fd.append("ref_files",ref2);
if(selVoice) fd.append("voice_name",selVoice);
try{
const data = await run("/sts", fd);
if(!data) return;
setResult({url:`/audio/${data.filename}`,filename:data.filename});
if(data.transcribed) setTranscribed(data.transcribed);
toast("success","تم تحويل الصوت بنجاح");
} catch(e){ toast("error",e.message); }
};
return (
<div className="col" style={{gap:16}}>
{running&&<ProgressBar progress={progress}/>}
<div className="info-box">
<span className="info-icon">🎭</span>
<div><strong style={{color:"var(--text)"}}>Speech-to-Speech</strong> — يرفع الملف الصوتي،
ينسخه Faster-Whisper تلقائياً، ثم يعيد توليده بصوت الهدف.</div>
</div>
<div className="grid-2">
<div className="card">
<div className="card-header"><span className="card-title">🎙 الصوت المصدر</span></div>
<div className="card-body"><FileUpload file={src} onFile={setSrc} hint="أي ملف صوتي — كلام بأي لغة"/></div>
</div>
<div className="card">
<div className="card-header">
<span className="card-title">🎯 صوت الهدف</span>
<div className="row" style={{gap:6}}>
<span className="fz-12 text-muted">لغة الإخراج:</span>
<select value={outLang} onChange={e=>setOutLang(e.target.value)}>
{Object.entries(LANGS).map(([k,v])=><option key={k} value={k}>{v}</option>)}
</select>
</div>
</div>
<div className="card-body">
<VoiceSelector voices={voices} selected={selVoice} onSelect={setSelVoice}
f1={ref1} setF1={setRef1} f2={ref2} setF2={setRef2}
saveName="" setSaveName={()=>{}} onSave={()=>{}} saveStatus=""/>
</div>
</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">⚙ إعدادات الصوت</span></div>
<div className="card-body"><VoiceSettings {...sharedSettings}/></div>
</div>
{!running
? <button className="btn-primary" onClick={doRun}>🎭 تحويل الصوت</button>
: <button className="btn-cancel" onClick={cancel}>
<div className="wave-bars">{[1,2,3,4].map(i=><span key={i}/>)}</div>
إلغاء التحويل
</button>
}
{transcribed&&(
<div className="info-box">
<span className="info-icon">📝</span>
<div><strong style={{color:"var(--text)"}}>النص المنسوخ: </strong>{transcribed}</div>
</div>
)}
{result&&(
<ResultCard result={result} onUpdate={setResult}
title="🔊 النتيجة" badge="✓ تم" storageKey="sts"/>
)}
</div>
);
}
/* ══ DUB TAB ═══════════════════════════════════════════════════════ */
function DubTab({ voices, sharedSettings }) {
const [src, setSrc] = useState(null);
const [ref1, setRef1] = useState(null);
const [ref2, setRef2] = useState(null);
const [selVoice, setSelVoice] = useState(null);
const [tgtLang, setTgtLang] = useState("en");
const [result, setResult] = useState(null);
const [original, setOriginal] = useState("");
const [translated, setTranslated] = useState("");
const { stability,clarity,styleExag,speed,boost,depth,space } = sharedSettings;
const { running, progress, run, cancel } = useJobRunner();
const doRun = async ()=>{
if(!src) return toast("error","ارفع الملف الصوتي.");
if(!ref1&&!selVoice) return toast("error","حدد صوت الدبلجة.");
const fd=new FormData();
fd.append("source_audio",src); fd.append("target_language",tgtLang);
fd.append("stability",stability); fd.append("clarity",clarity);
fd.append("style_exaggeration",styleExag); fd.append("speed",speed);
fd.append("speaker_boost",boost); fd.append("variation_strength_dub",variation); fd.append("depth",depth); fd.append("space",space);
if(ref1) fd.append("ref_files",ref1);
if(ref2) fd.append("ref_files",ref2);
if(selVoice) fd.append("voice_name",selVoice);
try{
const data = await run("/dub", fd);
if(!data) return;
setResult({url:`/audio/${data.filename}`,filename:data.filename});
if(data.original) setOriginal(data.original);
if(data.translated) setTranslated(data.translated);
toast("success","تمت الدبلجة بنجاح");
} catch(e){ toast("error",e.message); }
};
return (
<div className="col" style={{gap:16}}>
{running&&<ProgressBar progress={progress}/>}
<div className="info-box">
<span className="info-icon">🌐</span>
<div><strong style={{color:"var(--text)"}}>الدبلجة متعددة اللغات</strong> — ينسخ Faster-Whisper
الكلام، يترجمه إلى اللغة المختارة، ثم يولّد الصوت بالصوت المرجعي.</div>
</div>
<div className="grid-2">
<div className="card">
<div className="card-header"><span className="card-title">🎙 الملف المصدر</span></div>
<div className="card-body"><FileUpload file={src} onFile={setSrc} hint="صوت أو مقطع فيديو"/></div>
</div>
<div className="card">
<div className="card-header">
<span className="card-title">🌍 اللغة الهدف</span>
<select value={tgtLang} onChange={e=>setTgtLang(e.target.value)}>
{Object.entries(LANGS).map(([k,v])=><option key={k} value={k}>{v}</option>)}
</select>
</div>
<div className="card-body">
<VoiceSelector voices={voices} selected={selVoice} onSelect={setSelVoice}
f1={ref1} setF1={setRef1} f2={ref2} setF2={setRef2}
saveName="" setSaveName={()=>{}} onSave={()=>{}} saveStatus=""/>
</div>
</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">⚙ إعدادات الصوت</span></div>
<div className="card-body"><VoiceSettings {...sharedSettings}/></div>
</div>
{!running
? <button className="btn-primary" onClick={doRun}>🌐 بدء الدبلجة</button>
: <button className="btn-cancel" onClick={cancel}>
<div className="wave-bars">{[1,2,3,4].map(i=><span key={i}/>)}</div>
إلغاء الدبلجة
</button>
}
{(original||translated)&&(
<div className="grid-2" style={{gap:12}}>
{original&&<div className="info-box"><span className="info-icon">📄</span><div><strong style={{color:"var(--text)"}}>الأصلي:</strong> {original}</div></div>}
{translated&&<div className="info-box"><span className="info-icon">✍</span><div><strong style={{color:"var(--text)"}}>المترجم:</strong> {translated}</div></div>}
</div>
)}
{result&&(
<ResultCard result={result} onUpdate={setResult}
title="🔊 النتيجة المدبلجة" badge="✓ تم" storageKey="dub"/>
)}
</div>
);
}
/* ══ SFX TAB ═══════════════════════════════════════════════════════ */
function SFXTab({ voices, sharedSettings }) {
const [desc, setDesc] = useState("");
const [ref, setRef] = useState(null);
const [selVoice,setSelVoice]= useState(null);
const [result, setResult] = useState(null);
const { stability,clarity,styleExag,speed,boost,depth,space } = sharedSettings;
const { running, progress, run, cancel } = useJobRunner();
const EXAMPLES=["ضربة طبل عميقة وبطيئة","صوت تساقط المطر الغزير","رياح عاتية وعواء ذئاب","موجات بحر تتكسر على الصخور","صوت محرك سيارة يُشعل"];
const doRun = async ()=>{
if(!desc.trim()) return toast("error","اكتب وصف المؤثر الصوتي.");
if(!ref&&!selVoice) return toast("error","حدد صوتاً مرجعياً لضبط الطابع.");
const fd=new FormData();
fd.append("description",desc);
fd.append("stability",stability); fd.append("clarity",clarity);
fd.append("style_exaggeration",styleExag); fd.append("speed",speed);
fd.append("speaker_boost",boost);
if(ref) fd.append("ref_file",ref);
if(selVoice) fd.append("voice_name",selVoice);
try{
const data = await run("/sfx", fd);
if(!data) return;
setResult({url:`/audio/${data.filename}`,filename:data.filename});
toast("success","تم توليد المؤثر الصوتي");
} catch(e){ toast("error",e.message); }
};
return (
<div className="col" style={{gap:16}}>
{running&&<ProgressBar progress={progress}/>}
<div className="info-box">
<span className="info-icon">💥</span>
<div><strong style={{color:"var(--text)"}}>المؤثرات الصوتية</strong> — صف المؤثر الذي
تريده نصياً وسيُولّد النظام صوتاً يعكسه.</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">✏ وصف المؤثر</span></div>
<div className="card-body">
<textarea className="textarea" dir="rtl" rows={4}
placeholder="صف المؤثر الصوتي بتفصيل…"
value={desc} onChange={e=>setDesc(e.target.value)} style={{minHeight:90}}/>
<div style={{marginTop:12}}>
<div className="label">أمثلة سريعة</div>
<div style={{display:"flex",flexWrap:"wrap",gap:6}}>
{EXAMPLES.map(ex=>(
<button key={ex} className="btn-secondary btn-sm" onClick={()=>setDesc(ex)}>{ex}</button>
))}
</div>
</div>
</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">🎙 الصوت المرجعي</span></div>
<div className="card-body">
<VoiceSelector voices={voices} selected={selVoice} onSelect={setSelVoice}
f1={ref} setF1={setRef} f2={null} setF2={()=>{}}
saveName="" setSaveName={()=>{}} onSave={()=>{}} saveStatus=""/>
</div>
</div>
<div className="card">
<div className="card-header"><span className="card-title">⚙ إعدادات</span></div>
<div className="card-body"><VoiceSettings {...sharedSettings}/></div>
</div>
{!running
? <button className="btn-primary" onClick={doRun}>💥 توليد المؤثر الصوتي</button>
: <button className="btn-cancel" onClick={cancel}>
<div className="wave-bars">{[1,2,3,4].map(i=><span key={i}/>)}</div>
إلغاء التوليد
</button>
}
{result&&(
<ResultCard result={result} onUpdate={setResult}
title="🔊 المؤثر الصوتي" badge="✓ تم" storageKey="sfx"/>
)}
</div>
);
}
/* ══ LIBRARY TAB ═══════════════════════════════════════════════════ */
function LibraryTab({ voices, onRefresh }) {
const [f1,f1Set]=useState(null); const [f2,f2Set]=useState(null);
const [name,setName]=useState(""); const [saving,setSaving]=useState(false);
const save=async()=>{
if(!name.trim()) return toast("error","أدخل اسم الصوت.");
if(!f1) return toast("error","ارفع عينة صوتية أولاً.");
setSaving(true);
const fd=new FormData();
fd.append("name",name.trim()); fd.append("file",f1);
if(f2) fd.append("file2",f2);
try{
const r=await fetch("/voices/save",{method:"POST",body:fd});
if(!r.ok) throw new Error("فشل الحفظ");
toast("success",`تم حفظ الصوت "${name.trim()}"`);
f1Set(null); f2Set(null); setName(""); onRefresh();
} catch(e){ toast("error",e.message); }
finally{ setSaving(false); }
};
const del=async(v)=>{
if(!confirm(`حذف الصوت "${v}"؟`)) return;
await fetch(`/voices/${v}`,{method:"DELETE"});
toast("info",`تم حذف "${v}"`); onRefresh();
};
return (
<div className="col" style={{gap:16}}>
<div className="card">
<div className="card-header"><span className="card-title">➕ إضافة صوت جديد</span></div>
<div className="card-body">
<div className="grid-2" style={{marginBottom:14}}>
<FileUpload label="عينة صوتية 1 (مطلوبة)" file={f1} onFile={f1Set} hint="5–30 ثانية كلام واضح"/>
<FileUpload label="عينة صوتية 2 (اختياري)" file={f2} onFile={f2Set} hint="تحسّن دقة الاستنساخ"/>
</div>
<div className="row">
<input type="text" placeholder="اسم الصوت في المكتبة…" value={name}
onChange={e=>setName(e.target.value)} style={{flex:1}}/>
<button className="btn-secondary" onClick={save} disabled={saving}>
{saving?"جاري الحفظ…":"حفظ في المكتبة"}
</button>
</div>
</div>
</div>
{voices.length===0 ? (
<div className="empty">
<div className="empty-icon">🎙</div>
<div className="empty-title">المكتبة فارغة</div>
<div className="empty-sub">أضف عينات صوتية لتظهر هنا.</div>
</div>
) : (
<div className="col" style={{gap:10}}>
<div className="label">الأصوات المحفوظة ({voices.length})</div>
{voices.map(v=>{
const av=voiceAvatar(v);
return (
<div key={v} className="voice-card" style={{cursor:"default"}}>
<div className="voice-avatar" style={{background:av.bg,color:av.fg,borderColor:av.bg}}>{av.initials}</div>
<div className="voice-info">
<div className="voice-name">{v}</div>
<div className="voice-tags"><span className="badge badge-amber">مكتبة</span></div>
</div>
<button className="btn-secondary btn-sm"
style={{color:"var(--red)",borderColor:"rgba(239,68,68,.3)"}}
onClick={()=>del(v)}>حذف</button>
</div>
);
})}
</div>
)}
</div>
);
}
/* ══ HISTORY TAB ═══════════════════════════════════════════════════ */
function HistoryTab() {
const [history, setHistory] = useState([]);
const [search, setSearch] = useState("");
const [activeUrl, setActiveUrl] = useState(null);
const [activeFn, setActiveFn] = useState(null); // filename for editor
const [editKey, setEditKey] = useState(0); // force-remount editor on new file
useEffect(()=>{
fetch("/history").then(r=>r.json()).then(h=>setHistory([...h].reverse())).catch(()=>{});
},[]);
const openPreview = (filename) => {
setActiveUrl(`/audio/${filename}`);
setActiveFn(filename);
setEditKey(k=>k+1); // reset editor state for new file
setTimeout(()=>{
document.getElementById("hist-preview-card")
?.scrollIntoView({behavior:"smooth",block:"start"});
}, 80);
};
const onEdited = ({url, filename}) => {
setActiveUrl(url);
setActiveFn(filename);
setEditKey(k=>k+1);
};
const MODE_INFO={tts:["badge-amber","⚡ TTS"],sts:["badge-purple","🎭 STS"],
dub:["badge-blue","🌐 Dub"],sfx:["badge-green","💥 SFX"]};
const filtered=history.filter(h=>
!search||h.text.includes(search)||h.language.includes(search)||(h.mode||"").includes(search)
);
return (
<div className="col" style={{gap:14}}>
<div className="row">
<input type="text" placeholder="بحث في السجل…" value={search}
onChange={e=>setSearch(e.target.value)} style={{flex:1}}/>
<span className="fz-12 text-muted" style={{fontFamily:"var(--mono)",whiteSpace:"nowrap"}}>
{filtered.length} عملية
</span>
</div>
{/* ── Preview + Editor panel ─────────────────────────────────── */}
{activeUrl&&activeFn&&(
<ResultCard
key={editKey}
result={{url:activeUrl, filename:activeFn}}
onUpdate={onEdited}
title="🔊 المعاينة والتعديل" storageKey="hist"
badge={activeFn}
/>
)}
{filtered.length===0 ? (
<div className="empty">
<div className="empty-icon">🕘</div>
<div className="empty-title">{search?"لا نتائج":"السجل فارغ"}</div>
<div className="empty-sub">{search?"جرب مصطلح بحث آخر":"ستظهر عمليات التوليد هنا تلقائياً."}</div>
</div>
) : filtered.map((h,i)=>{
const [cls,label]=MODE_INFO[h.mode]||["badge-amber","⚡"];
const isActive = activeFn === h.filename;
return (
<div key={i} className="hist-item"
style={isActive?{borderColor:"var(--amber)",background:"var(--amber-glow)"}:{}}>
<div className="hist-text">
<div className="hist-preview" dir="rtl">{h.text}</div>
<div className="hist-meta">
<span className={`badge ${cls}`}>{label}</span>
<span className="badge badge-amber">{LANGS[h.language]||h.language}</span>
{h.char_count>0&&<span className="hist-time">{h.char_count} حرف</span>}
<span className="hist-time">{new Date(h.ts*1000).toLocaleString("ar-EG")}</span>
</div>
</div>
<div className="hist-actions">
<button className={`btn-secondary btn-sm ${isActive?"":"" }`}
style={isActive?{borderColor:"var(--amber)",color:"var(--amber)"}:{}}
onClick={()=> isActive ? setActiveUrl(null)||setActiveFn(null)
: openPreview(h.filename)}>
{isActive ? "▶ نشط" : "▶ تشغيل + تعديل"}
</button>
<a href={`/audio/${h.filename}`} download={h.filename} style={{textDecoration:"none"}}>
<button className="btn-secondary btn-sm">⬇</button>
</a>
</div>
</div>
);
})}
</div>
);
}
/* ══ APP ROOT ══════════════════════════════════════════════════════ */
function App() {
/* ── persisted across refresh ── */
const [tab, setTab] = useLS("app.tab", "generate");
const [theme, setTheme] = useLS("app.theme", "dark");
const [stability, setStability] = useLS("vs.stability", 0.5);
const [clarity, setClarity] = useLS("vs.clarity", 0.75);
const [styleExag, setStyleExag] = useLS("vs.styleExag", 0.0);
const [speed, setSpeed] = useLS("vs.speed", 1.0);
const [boost, setBoost] = useLS("vs.boost", false);
const [splitText, setSplitText] = useLS("vs.splitText", true);
const [variation, setVariation] = useLS("vs.variation", 1.0);
const [refClean, setRefClean] = useLS("vs.refClean", true);
const [cleanText, setCleanText] = useLS("vs.cleanText", true);
const [attempts, setAttempts] = useLS("vs.attempts", 1);
const [depth, setDepth] = useLS("vs.depth", 0.5);
const [space, setSpace] = useLS("vs.space", "studio");
/* ── session-only ── */
const [voices, setVoices] = useState([]);
const [sidebarOpen,setSidebarOpen]= useState(false);
const sharedSettings={stability,setStability,clarity,setClarity,
styleExag,setStyleExag,speed,setSpeed,boost,setBoost,
splitText,setSplitText,variation,setVariation,
refClean,setRefClean,cleanText,setCleanText,attempts,setAttempts,
depth,setDepth,space,setSpace};
useEffect(()=>{ document.documentElement.setAttribute("data-theme",theme); },[theme]);
const refreshVoices=()=>fetch("/voices").then(r=>r.json()).then(setVoices).catch(()=>{});
useEffect(()=>{ refreshVoices(); },[]);
/* ── Model loading status ── */
const [modelStatus, setModelStatus] = useState({loading:true,ready:false,error:null});
useEffect(()=>{
const poll = () => {
fetch("/model_status").then(r=>r.json()).then(s=>{
setModelStatus(s);
if(s.loading) setTimeout(poll, 3000);
}).catch(()=>setTimeout(poll,5000));
};
poll();
},[]);
const TABS=[
{id:"generate",icon:"⚡",label:"توليد صوت"},
{id:"sts", icon:"🎭",label:"STS"},
{id:"dub", icon:"🌐",label:"دبلجة"},
{id:"sfx", icon:"💥",label:"مؤثرات"},
null,
{id:"library", icon:"📚",label:"مكتبة الأصوات"},
{id:"history", icon:"🕘",label:"السجل"},
];
const TAB_TITLES={
generate:"⚡ توليد الصوت",sts:"🎭 Speech-to-Speech",
dub:"🌐 الدبلجة متعددة اللغات",sfx:"💥 المؤثرات الصوتية",
library:"📚 مكتبة الأصوات",history:"🕘 سجل العمليات",
};
const selectTab=(id)=>{ setTab(id); setSidebarOpen(false); };
return (
<>
<ToastRoot/>
{/* Mobile overlay */}
<div className={`sidebar-overlay ${sidebarOpen?"visible":""}`}
onClick={()=>setSidebarOpen(false)}/>
<div className="app">
{/* ── Sidebar ── */}
<aside className={`sidebar ${sidebarOpen?"mobile-open":""}`}>
<div className="sidebar-logo">
<div className="sidebar-logo-text"><span className="hi">XTTS</span> Studio</div>
<div className="sidebar-logo-sub">v2 Multilingual</div>
</div>
<div className="sidebar-section">
<div className="sidebar-section-label">الأدوات</div>
{TABS.map((t,i)=>t===null
? <div key={i} className="divider" style={{margin:"8px 10px"}}/>
: (
<div key={t.id} className={`nav-item ${tab===t.id?"active":""}`}
onClick={()=>selectTab(t.id)}>
<span className="ni-icon">{t.icon}</span>
<span>{t.label}</span>
</div>
)
)}
</div>
<div className="sidebar-footer">
<div className="status-row">
<div className={`status-dot green`}/>
<span>{DEVICE.toUpperCase()}</span>
</div>
<div className="status-row">
<div className="status-dot" style={{
background: modelStatus.error?"var(--red)":modelStatus.ready?"var(--green)":"var(--amber)",
animation: modelStatus.loading?"wb 1s ease-in-out infinite":undefined
}}/>
<span style={{color:modelStatus.error?"var(--red)":modelStatus.ready?"var(--green)":"var(--amber)"}}>
{modelStatus.error?"❌ فشل التحميل":modelStatus.loading?"⏳ جاري تحميل النموذج…":"✓ XTTS v2 جاهز"}
</span>
</div>
{modelStatus.ready&&<div className="status-row">
<div className="status-dot" style={{background:"var(--amber)"}}/>
<span>{voices.length} صوت محفوظ</span>
</div>}
{modelStatus.loading&&<div style={{marginTop:6,fontSize:10,color:"var(--text3)",fontFamily:"var(--mono)"}}>
الواجهة تعمل — الميزات ستُفعَّل بعد اكتمال التحميل
</div>}
</div>
</aside>
{/* ── Topbar ── */}
<header className="topbar">
<div className="row" style={{gap:12}}>
<div className="hamburger" onClick={()=>setSidebarOpen(v=>!v)}>
<span/><span/><span/>
</div>
<span className="topbar-title">
<span>XTTS</span> <span className="title-long">Voice Studio — </span>{TAB_TITLES[tab]}
</span>
</div>
<div className="topbar-right">
<button className="btn-icon-round"
onClick={()=>setTheme(t=>t==="dark"?"light":"dark")}
title="تبديل الثيم">
{theme==="dark"?"☀️":"🌙"}
</button>
</div>
</header>
{/* ── Main ── */}
<main className="main">
<div className="main-inner">
{tab==="generate"&&<GenerateTab voices={voices} onVoicesRefresh={refreshVoices} sharedSettings={sharedSettings}/>}
{tab==="sts" &&<STSTab voices={voices} sharedSettings={sharedSettings}/>}
{tab==="dub" &&<DubTab voices={voices} sharedSettings={sharedSettings}/>}
{tab==="sfx" &&<SFXTab voices={voices} sharedSettings={sharedSettings}/>}
{tab==="library" &&<LibraryTab voices={voices} onRefresh={refreshVoices}/>}
{tab==="history" &&<HistoryTab/>}
</div>
</main>
</div>
</>
);
}
ReactDOM.createRoot(document.getElementById("root")).render(<App/>);
</script>
</body>
</html>"""
# ══════════════════════════════════════════════════════════════════════
# 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,
"whisper": bool(WHISPER_OK),
"zerogpu": bool(ZEROGPU),
})
_STARTED_AT = time.time()
@app.get("/health")
def health():
"""Always HTTP 200 (so health checks never restart a model that is still loading)."""
return {"status": "ok" if xtts_ready else ("loading" if xtts_loading else "error"),
"model_ready": bool(xtts_ready), "whisper_ready": bool(WHISPER_OK),
"busy": get_tts_lock().locked(), "uptime_s": int(time.time() - _STARTED_AT)}
@app.get("/diagnostics")
def diagnostics():
import importlib.metadata as md
def ver(n):
try: return md.version(n)
except Exception: return "n/a"
def folder(d):
try:
fs = [os.path.join(d, f) for f in os.listdir(d)]
return {"files": len(fs), "mb": round(sum(os.path.getsize(f) for f in fs if os.path.isfile(f)) / 1e6, 1)}
except Exception:
return None
return {
"versions": {k: ver(k) for k in ("torch", "coqui-tts", "transformers", "faster-whisper", "numpy", "spaces")},
"device": device, "zerogpu": bool(ZEROGPU), "cpu_threads": CPU_THREADS,
"model_ready": bool(xtts_ready), "model_error": xtts_error, "whisper_ready": bool(WHISPER_OK),
"caches": {"reference": folder(REF_CACHE_DIR), "dry_audio": folder(DRY_CACHE_DIR), "outputs": folder(OUTPUT_DIR)},
"latent_cache_entries": len(_LAT_CACHE),
"engine": "voice-quality v" + vq.VQ_VERSION,
"recent_jobs": list(RECENT_JOBS),
}
# ── 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 = _output_path(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 = _output_path(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),
depth: float = Form(0.5),
space: str = Form("studio"),
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 language not in LANGUAGES: raise HTTPException(400, f"لغة غير مدعومة: {language}")
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, job_stats = {}, {}
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, job_stats
)
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)
await asyncio.to_thread(apply_depth, out, depth, space)
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,
"cached": bool(job_stats.get("cached")), "stats": job_stats 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:
paths, _reports, summary = await asyncio.to_thread(vq.enhance_reference_set, refs, REF_CACHE_DIR)
primary = paths[0] # best contiguous segment (others are used too, never joined)
fn = "refclean_" + os.path.basename(primary)[4:-4] + ".wav"
shutil.copyfile(primary, os.path.join(OUTPUT_DIR, fn))
return {"url": f"/audio/{fn}", "report": summary, "segments": len(paths)}
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),
depth: float = Form(0.5),
space: str = Form("studio"),
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"
_copy_limited(source_audio.file, src)
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)
await asyncio.to_thread(apply_depth, out, depth, space)
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),
depth: float = Form(0.5),
space: str = Form("studio"),
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"
_copy_limited(source_audio.file, src)
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)
await asyncio.to_thread(apply_depth, out, depth, space)
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 = safe_voice_name(name)
if not safe: raise HTTPException(400, "اسم الصوت غير صالح.")
uploads = [f for f in ([file, file2] if file2 and file2.filename else [file]) if f and f.filename]
names = []
for i, f in enumerate(uploads):
fn = re.sub(r"[^\w.\-]", "_", os.path.basename(f.filename), flags=re.UNICODE)[:80] or f"ref{i}.wav"
if not fn.lower().endswith(_AUDIO_EXT):
raise HTTPException(400, "صيغة الملف غير مدعومة (wav / mp3 / flac / ogg / m4a).")
names.append(fn)
if not uploads: raise HTTPException(400, "لا يوجد ملف صوتي.")
lib = voice_dir(safe)
os.makedirs(lib, exist_ok=True)
for f, fn in zip(uploads, names):
_copy_limited(f.file, os.path.join(lib, fn))
background_tasks.add_task(trigger_cloud_backup)
return {"name": os.path.basename(lib)}
@app.delete("/voices/{name}")
def delete_voice(background_tasks: BackgroundTasks, name: str):
lib = voice_dir(name, must_exist=True)
shutil.rmtree(lib)
background_tasks.add_task(trigger_cloud_backup)
return {"deleted": name}
@app.get("/voices/{name}/preview")
def preview_voice(name: str):
lib = voice_dir(name, must_exist=True)
for fn in sorted(os.listdir(lib)):
if fn.lower().endswith(_AUDIO_EXT):
mt = {"wav": "audio/wav", "mp3": "audio/mpeg", "flac": "audio/flac", "ogg": "audio/ogg",
"m4a": "audio/mp4", "opus": "audio/ogg", "webm": "audio/webm", "aac": "audio/aac"}[fn.rsplit(".", 1)[-1].lower()]
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")
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ 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
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"),