diff --git "a/app.py" "b/app.py" new file mode 100644--- /dev/null +++ "b/app.py" @@ -0,0 +1,3442 @@ +# ╔══════════════════════════════════════════════════════════════════════╗ +# ║ 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 + +# ══════════════════════════════════════════════════════════════════════ +# 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.22.0 +transformers>=4.33.0,<4.38.0 +fastapi>=0.111.0 +uvicorn[standard]>=0.29.0 +python-multipart>=0.0.9 +torch>=2.2.0,<2.6.0 +torchaudio>=2.2.0,<2.6.0 +numpy>=1.24.0 +soundfile>=0.12.0 +scipy>=1.10.0 +faster-whisper>=0.10.0 +deep-translator>=1.11.0 +huggingface-hub>=0.20.0 +""" + +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 + +torch.set_num_threads(CPU_THREADS) # use all 16 cores for inference +try: + torch.set_num_interop_threads(2) +except RuntimeError: + pass + +# ── Persistent Storage ──────────────────────────────────────────────── +PERSISTENT_DIR = "/data" if os.path.exists("/data") else os.path.join(os.getcwd(), "app_data") +VOICE_LIB = os.path.join(PERSISTENT_DIR, "voice_library") +OUTPUT_DIR = os.path.join(PERSISTENT_DIR, "outputs") +HISTORY_FILE = os.path.join(PERSISTENT_DIR, "history.json") +MAX_CHARS = 5000 + +for d in [VOICE_LIB, OUTPUT_DIR]: + os.makedirs(d, exist_ok=True) + +HF_TOKEN = os.environ.get("HF_TOKEN") +DATASET_REPO_ID = os.environ.get("DATASET_REPO_ID") + +try: + from huggingface_hub import HfApi, snapshot_download + HUB_AVAILABLE = True + if HF_TOKEN and DATASET_REPO_ID: + print("[*] Restoring data from Hugging Face Hub …") + try: + snapshot_download(repo_id=DATASET_REPO_ID, repo_type="dataset", + local_dir=PERSISTENT_DIR, token=HF_TOKEN) + print("[✓] Data restoration complete.") + except Exception as e: + print(f"[!] Warning: Could not download backup ({e})") +except ImportError: + HUB_AVAILABLE = False + +def trigger_cloud_backup(): + if not (HUB_AVAILABLE and HF_TOKEN and DATASET_REPO_ID): return + def _run(): + try: + HfApi(token=HF_TOKEN).upload_folder( + folder_path=PERSISTENT_DIR, repo_id=DATASET_REPO_ID, + repo_type="dataset", commit_message=f"Auto-backup: {int(time.time())}") + except Exception as e: + print(f"[!] Cloud backup failed: {e}") + threading.Thread(target=_run).start() + +# ── XTTS Model ──────────────────────────────────────────────────────── +from TTS.api import TTS +device = "cpu" # GPU intentionally disabled — CPU only + +# ── Background model loading (server starts immediately) ────────────── +xtts = None +xtts_ready = False +xtts_loading = True +xtts_error = None + +def _load_model(): + global xtts, xtts_ready, xtts_loading, xtts_error + try: + print(f"[*] Loading XTTS v2 on {device.upper()} …") + xtts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device) + xtts_ready = True + print("[✓] XTTS ready.") + except Exception as e: + xtts_error = f"{type(e).__name__}: {e}" + print(f"[!] XTTS loading failed: {xtts_error}", flush=True) + traceback.print_exc() + finally: + xtts_loading = False + +threading.Thread(target=_load_model, daemon=True).start() + +# tts_lock is created lazily inside the event loop to avoid asyncio issues +_tts_lock = None +def get_tts_lock(): + global _tts_lock + if _tts_lock is None: + _tts_lock = asyncio.Lock() + return _tts_lock + +def _startup_cleanup(): + for f in os.listdir("/tmp"): + if any(f.startswith(p) for p in ("ch_","ref_","sts_","dub_")): + try: os.remove(os.path.join("/tmp",f)) + except: pass +_startup_cleanup() + +# ── Job Management: Cancellation & Progress ─────────────────────────── +job_progress: dict = {} # job_id -> {current, total, status, ts} +cancel_events: dict = {} # job_id -> threading.Event +JOB_TTL_SEC = 3600 + +def _gc_jobs(): + cutoff = time.time() - JOB_TTL_SEC + for k in [k for k,v in list(job_progress.items()) + if v.get("status") != "running" and v.get("ts",0) < cutoff]: + job_progress.pop(k, None) + +# ── Faster-Whisper (replaces openai-whisper) ────────────────────────── +whisper_mdl = None +WHISPER_OK = False + +def _load_whisper(): + """Load Faster-Whisper in the background. If the default HF cache + (e.g. the /data volume) gives an I/O error, retry from local disk.""" + global whisper_mdl, WHISPER_OK + try: + from faster_whisper import WhisperModel + except Exception as e: + print(f"[!] Faster-Whisper not installed: {e}", flush=True) + return + for root in (None, "/tmp/whisper_models"): + try: + if root: os.makedirs(root, exist_ok=True) + whisper_mdl = WhisperModel("small", device="cpu", compute_type="int8", + cpu_threads=CPU_THREADS, download_root=root) + WHISPER_OK = True + print(f"[✓] Faster-Whisper ready (small · int8 · CPU) [{root or 'default cache'}].", flush=True) + return + except Exception as e: + print(f"[!] Faster-Whisper load failed ({root or 'default cache'}): {e}", flush=True) + print("[!] Faster-Whisper not available.", flush=True) + +threading.Thread(target=_load_whisper, daemon=True).start() + +def transcribe_audio(path: str) -> str: + """Transcribe using Faster-Whisper; returns plain text.""" + if whisper_mdl is None: + raise RuntimeError("Faster-Whisper not loaded") + segments, _ = whisper_mdl.transcribe(path, beam_size=5) + return " ".join(s.text for s in segments).strip() + +# ── Translator ──────────────────────────────────────────────────────── +try: + from deep_translator import GoogleTranslator + TRANSLATOR_OK = True +except Exception: + TRANSLATOR_OK = False + +# ── History helpers ─────────────────────────────────────────────────── +def load_history(): + try: + return json.load(open(HISTORY_FILE)) if os.path.exists(HISTORY_FILE) else [] + except Exception: + return [] + +def save_history(h): + json.dump(h, open(HISTORY_FILE, "w"), ensure_ascii=False, indent=2) + trigger_cloud_backup() + +MAX_HISTORY = 300 +MAX_OUTPUTS = 200 + +def _prune_outputs(): + try: + wavs = sorted( + [f for f in os.listdir(OUTPUT_DIR) if f.endswith(".wav")], + key=lambda f: os.path.getmtime(os.path.join(OUTPUT_DIR, f)) + ) + for old in wavs[:-MAX_OUTPUTS]: + try: os.remove(os.path.join(OUTPUT_DIR, old)) + except: pass + except Exception: + pass + +def append_history(filename, text, language, mode, char_count=0): + h = load_history() + h.append(dict(filename=filename, text=text[:150], language=language, + mode=mode, char_count=char_count, ts=int(time.time()))) + if len(h) > MAX_HISTORY: + h = h[-MAX_HISTORY:] + save_history(h) + _prune_outputs() + +# ── Parameter Engineering ───────────────────────────────────────────── +def stability_to_temperature(s): return round(0.85 - (s * 0.40), 3) +def clarity_to_top_p(c): return round(0.70 + (c * 0.25), 3) +def style_to_rep_penalty(st): return round(2.00 + (st * 2.00), 2) + +# ── Expression Presets ──────────────────────────────────────────────── +# Each preset overrides stability/temperature/top_p/rep_penalty to push +# XTTS toward a specific emotional character. +EXPRESSION_PRESETS = { + "natural": dict(temperature=0.72, top_p=0.88, rep_penalty=2.0, speed_mult=1.00), + "warm": dict(temperature=0.78, top_p=0.90, rep_penalty=1.80, speed_mult=0.97), + "excited": dict(temperature=0.88, top_p=0.92, rep_penalty=1.65, speed_mult=1.10), + "sad": dict(temperature=0.65, top_p=0.85, rep_penalty=2.20, speed_mult=0.88), + "serious": dict(temperature=0.55, top_p=0.82, rep_penalty=2.50, speed_mult=0.95), + "story": dict(temperature=0.82, top_p=0.93, rep_penalty=1.75, speed_mult=0.96), + "news": dict(temperature=0.60, top_p=0.84, rep_penalty=2.40, speed_mult=1.02), + "child": dict(temperature=0.90, top_p=0.95, rep_penalty=1.60, speed_mult=1.05), +} + +def apply_expression_preset(preset_name: str, speed: float) -> dict: + p = EXPRESSION_PRESETS.get(preset_name, EXPRESSION_PRESETS["natural"]) + return { + "temperature": p["temperature"], + "top_p": p["top_p"], + "repetition_penalty": p["rep_penalty"], + "speed": round(speed * p["speed_mult"], 3), + } + +# ── Text Expressiveness Enhancer ────────────────────────────────────── +# XTTS is very sensitive to punctuation and sentence rhythm. +# Adding deliberate pauses, ellipses, and varied sentence endings +# dramatically improves naturalness and eliminates the "robot reading" feel. +def enhance_text_expression(text: str, lang: str, mode: str) -> str: + """ + Enrich text with punctuation cues that guide XTTS prosody. + Does NOT change words — only adds/adjusts punctuation & whitespace. + """ + if mode == "natural" or not text.strip(): + return text + + t = text.strip() + + # 1. Normalise whitespace + t = re.sub(r'[ \t]+', ' ', t) + + # 2. Ensure sentences end with terminal punctuation + # (XTTS pauses longer on . than on nothing) + t = re.sub(r'([^\.\!\?؟،,\n])\n', r'\1.\n', t) + + # 3. Mode-specific rhythm adjustments + if mode == "excited": + # Shorten pauses between sentences → rapid delivery + t = re.sub(r'\.\s+', '. ', t) + # Add emphasis markers via comma-pauses before key conjunctions + for conj in ['لكن','بل','إلا','however','but','yet','still']: + t = re.sub(rf'\b({conj})\b', r', \1', t, flags=re.IGNORECASE) + + elif mode == "sad": + # Extend pauses → slower, heavier delivery + t = re.sub(r'\.\s+', '… ', t) + t = re.sub(r'،\s+', '، ', t) + + elif mode == "story": + # Add brief breath-pause after scene-setting clauses + t = re.sub(r'(كان|كانت|في يوم|ذات يوم|once upon|there was)', + r'\1،', t, flags=re.IGNORECASE) + + elif mode == "news": + # Crisp, no trailing breath + t = re.sub(r'\s*…+\s*', '. ', t) + + elif mode == "child": + # Short bursts with upward inflection hints + t = re.sub(r'\.\s+([A-Zا-ي])', r'! \1', t) + + # 4. Collapse multiple punctuation + t = re.sub(r'([.!?؟،,]){2,}', r'\1', t) + t = re.sub(r'…{2,}', '…', t) + + return t + +# ╔══════════════════════════════════════════════════════════════════════╗ +# ║ P H R A S E - L E V E L P R O S O D Y V A R I A T I O N ║ +# ║ Each sentence is analysed for emotional content. ║ +# ║ XTTS parameters (temperature, speed, repetition_penalty) shift ║ +# ║ per-phrase so the delivery feels alive, not machine-read. ║ +# ║ Post-generation volume envelopes reinforce the prosodic contour. ║ +# ╚══════════════════════════════════════════════════════════════════════╝ + +# ── Per-emotion XTTS deltas ─────────────────────────────────────────── +# (applied on top of the base params; scaled by user's variation_strength) +EMOTION_DELTAS = { + # temp speed_mult rep_penalty silence_ms + "neutral": ( 0.00, 1.000, 0.00, 110 ), + "question": (+0.07, 0.960, -0.10, 140 ), # slower, more uncertain + "exclaim": (+0.12, 1.080, -0.30, 75 ), # faster, more energy + "emphasis": (-0.06, 0.900, +0.30, 155 ), # deliberate, weighty + "sadness": (-0.10, 0.870, +0.35, 185 ), # slow, heavy + "joy": (+0.10, 1.060, -0.25, 85 ), # bright, forward + "anger": (+0.08, 1.040, -0.20, 95 ), # tense, clipped + "soft": (-0.09, 0.920, +0.20, 160 ), # tender, gentle + "urgency": (+0.11, 1.120, -0.25, 65 ), # fast, alert + "story": (+0.06, 0.950, -0.10, 135 ), # varied, narrative +} + +# ── Keyword patterns (Arabic + English) ────────────────────────────── +EMOTION_PATTERNS = { + "joy": [ + # Arabic + r'سعيد|فرح|رائع|مذهل|عظيم|ممتاز|برافو|مبروك|ياه|يا لروع|' + r'جميل|بديع|أحب|نجح|وفق|الله|يسعد|سرور|ابتهاج|بهجة|نشوة', + # English + r'happy|wonderful|amazing|great|excellent|love|joy|fantastic|' + r'brilliant|perfect|congrats|hooray|yay|delightful|terrific', + ], + "sadness": [ + r'حزين|للأسف|مؤسف|وداع|ألم|فقدنا|صعب|مؤلم|بكى|دموع|' + r'خسر|رحل|غاب|اشتاق|أفتقد|حسرة|مأساة|مصيبة|فراق|وجع', + r'sad|unfortunately|sorry|pain|lost|difficult|hard|tears|' + r'miss|regret|grieve|mourn|tragedy|farewell|ache', + ], + "anger": [ + r'غضب|لا أقبل|مستحيل|يكفي|كفى|ظلم|حرام|لا يجوز|رفض|أبداً|' + r'لن أتحمل|سئمت|ضاق|احتج|رفض|نقمة|سخط|فساد', + r'never|impossible|stop|enough|angry|refuse|unacceptable|' + r'outrage|disgusting|unbelievable|ridiculous|furious', + ], + "emphasis": [ + r'بالتأكيد|دون شك|قطعاً|أكيد|لا شك|يجب|ضرورة|مهم|انتبه|' + r'اسمع|لا بد|حتماً|بلا شك|من المؤكد|بوضوح|صراحةً|جداً', + r'definitely|absolutely|certainly|must|important|listen|' + r'crucial|critical|clearly|obviously|indeed|precisely', + ], + "soft": [ + r'ربما|لعل|عسى|أتمنى|أرجو|بلطف|برفق|هدوء|سلام|' + r'هناء|رويداً|بهدوء|بتأن|رفيق|حنان|لطيف|رحمة|أمل', + r'perhaps|maybe|hopefully|gently|softly|peace|calm|' + r'wish|hope|kindly|tenderly|quietly|sweetly', + ], + "urgency": [ + r'الآن|فوراً|سريعاً|عاجل|انتبه|احذر|خطر|أسرع|بسرعة|' + r'عاجلاً|في الحال|لا تتأخر|إنذار|تحذير|يستعجل|عجلة', + r'now|immediately|quickly|urgent|warning|danger|hurry|' + r'fast|alert|emergency|asap|right away|no time', + ], + "story": [ + r'كان يا مكان|ذات يوم|في قديم|يحكى أن|وفي ليلة|' + r'ثم قال|فأجاب|نظر|قرر|انطلق|توجه|وصل|لقي|رأى', + r'once upon|there was|long ago|he said|she replied|' + r'suddenly|then he|and so|meanwhile|the next day', + ], +} + +# ── Phrase emotion detector ─────────────────────────────────────────── +def analyze_phrase_emotion(phrase: str, lang: str) -> tuple[str, float]: + """ + Detect the dominant emotion in a phrase. + Returns (emotion_name, confidence_0_to_1). + Confidence drives how strongly the deltas are applied. + """ + p = phrase.strip() + + # 1. Punctuation-first rules (fastest, highest confidence) + if re.search(r'[؟?]$', p): + return "question", 0.90 + count_exclaim = len(re.findall(r'[!]', p)) + if count_exclaim >= 2: + return "exclaim", 0.95 + if count_exclaim == 1 and len(p) < 60: + return "exclaim", 0.75 + + # 2. Keyword scan + scores: dict[str, float] = {} + for emotion, patterns in EMOTION_PATTERNS.items(): + for pat in patterns: + matches = re.findall(pat, p, re.IGNORECASE) + if matches: + # Score = matches / words — more keywords → higher confidence + words = max(1, len(p.split())) + scores[emotion] = scores.get(emotion, 0) + len(matches) / words + + if scores: + best = max(scores, key=scores.get) + conf = min(0.92, scores[best] * 3.0) # cap at 0.92 + return best, conf + + # 3. Structural heuristics + if len(p) < 20: + return "emphasis", 0.35 # short phrases tend to be punchy + if p.endswith('…') or p.endswith('...'): + return "soft", 0.50 # trailing ellipsis = trailing/wistful + + return "neutral", 0.0 + + +# ── Per-phrase XTTS param builder ───────────────────────────────────── +def make_phrase_params(base: dict, emotion: str, conf: float, + variation: float, pos_ratio: float) -> dict: + """ + variation : user slider 0-2 (0=none, 1=natural, 2=dramatic) + pos_ratio : 0.0 = first phrase, 1.0 = last phrase + Returns a NEW params dict with emotion-driven adjustments. + """ + if variation < 0.02: + return base.copy() + + d = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"]) + temp_d, speed_m, rep_d, _ = d + strength = conf * variation # 0 … ~1.84 + + p = base.copy() + p["temperature"] = float(np.clip( + base["temperature"] + temp_d * strength, 0.40, 0.96)) + p["speed"] = float(np.clip( + base["speed"] * (1.0 + (speed_m - 1.0) * strength), 0.60, 1.65)) + p["repetition_penalty"] = float(np.clip( + base["repetition_penalty"] + rep_d * strength, 1.30, 3.80)) + + # ── Position-based natural drift ──────────────────────────────── + # Opening: slightly more controlled (lower temp, slower) + if pos_ratio < 0.08: + p["temperature"] = float(np.clip(p["temperature"] - 0.04 * variation, 0.40, 0.96)) + p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.04 * variation), 0.60, 1.65)) + + # Closing: natural deceleration (the "landing" of a speech) + if pos_ratio > 0.88: + p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.06 * variation), 0.60, 1.65)) + p["temperature"] = float(np.clip(p["temperature"] - 0.03 * variation, 0.40, 0.96)) + + # ── Micro-random jitter: prevent identical params on consecutive ─ + # neutral phrases (which would sound machine-even) + if emotion == "neutral" and variation > 0.3: + # deterministic pseudo-random from phrase content hash + h = abs(hash(base.get("_phrase_hint", ""))) % 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): + data, sr = sf.read(path) + peak = np.max(np.abs(data)) + if peak > 0: + data = data / peak * 0.95 + # Soft clipping via hyperbolic tangent — smooth, natural saturation + # Unlike np.clip (hard clipping), tanh curves the peaks gently + data = np.tanh(data * 1.8) * (1.0 / np.tanh(1.8)) + sf.write(path, data, sr, subtype='PCM_16') + + +# ╔══════════════════════════════════════════════════════════════════════╗ +# ║ H U M A N I Z E — Pure-NumPy/SciPy DSP Chain ║ +# ║ Simulates recording in a real acoustic environment. ║ +# ║ No extra libraries. No IR files. All synthesized algorithmically. ║ +# ╚══════════════════════════════════════════════════════════════════════╝ + +# ── Environment presets ─────────────────────────────────────────────── +# Each dict drives the parameters of every DSP stage below. +HUMANIZE_PRESETS = { + "studio": dict( + # Clean, slightly padded room — professional voiceover booth + reverb_room=0.06, reverb_damping=0.85, reverb_wet=0.04, + noise_floor=-68, noise_color=0.5, + mic_hp_hz=90, mic_presence_db=1.5, mic_presence_hz=5500, + mic_air_db=-1.0, mic_air_hz=14000, + saturation=0.04, pitch_drift_cents=0.4, jitter_ms=0.3, + breathing=False, + ), + "bedroom": dict( + # Untreated small room — YouTube / indie podcast feel + reverb_room=0.22, reverb_damping=0.60, reverb_wet=0.09, + noise_floor=-62, noise_color=0.7, + mic_hp_hz=80, mic_presence_db=2.0, mic_presence_hz=4800, + mic_air_db=-0.5, mic_air_hz=13000, + saturation=0.08, pitch_drift_cents=0.8, jitter_ms=0.6, + breathing=False, + ), + "podcast": dict( + # Dynamic-mic, slightly driven — warm, intimate + reverb_room=0.10, reverb_damping=0.78, reverb_wet=0.05, + noise_floor=-65, noise_color=0.6, + mic_hp_hz=100, mic_presence_db=2.5, mic_presence_hz=3800, + mic_air_db=-2.5, mic_air_hz=12000, + saturation=0.14, pitch_drift_cents=0.6, jitter_ms=0.4, + breathing=False, + ), + "radio": dict( + # Classic broadcast: bandpass 200–8 kHz, compressed + reverb_room=0.04, reverb_damping=0.90, reverb_wet=0.02, + noise_floor=-72, noise_color=0.3, + mic_hp_hz=200, mic_presence_db=3.0, mic_presence_hz=3000, + mic_air_db=-8.0, mic_air_hz=8000, + saturation=0.18, pitch_drift_cents=0.3, jitter_ms=0.2, + breathing=False, + ), + "phone": dict( + # GSM / telephone: narrow band 300–3400 Hz, heavy saturation + reverb_room=0.02, reverb_damping=0.95, reverb_wet=0.01, + noise_floor=-55, noise_color=0.9, + mic_hp_hz=300, mic_presence_db=1.0, mic_presence_hz=2200, + mic_air_db=-20.0, mic_air_hz=3500, + saturation=0.28, pitch_drift_cents=1.2, jitter_ms=1.2, + breathing=False, + ), + "cafe": dict( + # Mid-sized reverberant room, background murmur + reverb_room=0.42, reverb_damping=0.45, reverb_wet=0.16, + noise_floor=-54, noise_color=0.8, + mic_hp_hz=70, mic_presence_db=1.0, mic_presence_hz=4000, + mic_air_db=-1.0, mic_air_hz=14000, + saturation=0.06, pitch_drift_cents=1.0, jitter_ms=0.8, + breathing=False, + ), + "outdoor": dict( + # Open space — very short reverb, wind-ish noise + reverb_room=0.08, reverb_damping=0.92, reverb_wet=0.03, + noise_floor=-50, noise_color=0.95, + mic_hp_hz=60, mic_presence_db=0.5, mic_presence_hz=3500, + mic_air_db=0.0, mic_air_hz=16000, + saturation=0.05, pitch_drift_cents=1.5, jitter_ms=1.5, + breathing=False, + ), +} + +# ── DSP helpers ─────────────────────────────────────────────────────── + +def _biquad_hp(data: np.ndarray, sr: int, fc: float) -> np.ndarray: + """2nd-order Butterworth high-pass filter — clamped to safe range.""" + from scipy.signal import butter, sosfilt + fc = float(np.clip(fc, 1.0, sr * 0.49)) # never hit Nyquist + sos = butter(2, fc / (sr / 2), btype='high', output='sos') + out = sosfilt(sos, data.astype(np.float64)) + if not np.all(np.isfinite(out)): + return data.astype(np.float32) + return out.astype(np.float32) + +def _biquad_peak(data: np.ndarray, sr: int, fc: float, + gain_db: float, Q: float = 1.4) -> np.ndarray: + """Parametric peak EQ biquad — clamped and guarded.""" + from scipy.signal import sosfilt + fc = float(np.clip(fc, 20.0, sr * 0.45)) + gain_db = float(np.clip(gain_db, -40.0, 20.0)) + Q = float(np.clip(Q, 0.1, 20.0)) + w0 = 2 * np.pi * fc / sr + A = 10 ** (gain_db / 40.0) + cos = np.cos(w0); sin = np.sin(w0) + alpha = sin / (2 * Q) + b0 = 1 + alpha * A; b1 = -2 * cos; b2 = 1 - alpha * A + a0 = 1 + alpha / A; a1 = -2 * cos; a2 = 1 - alpha / A + if abs(a0) < 1e-10: + return data.astype(np.float32) + sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64) + if not np.all(np.isfinite(sos)): + return data.astype(np.float32) + out = sosfilt(sos, data.astype(np.float64)) + if not np.all(np.isfinite(out)): + return data.astype(np.float32) + return out.astype(np.float32) + +def _biquad_shelf_hi(data: np.ndarray, sr: int, fc: float, + gain_db: float) -> np.ndarray: + """ + High-shelf filter (air band roll-off / boost). + Fixed: clamp fc below Nyquist and guard alpha against overflow/NaN. + """ + from scipy.signal import sosfilt + # Clamp cutoff to a safe range (never exceed 90% of Nyquist) + fc = float(np.clip(fc, 20.0, sr * 0.45)) + gain_db = float(np.clip(gain_db, -40.0, 20.0)) + + w0 = 2 * np.pi * fc / sr + A = 10 ** (gain_db / 40.0) + cos = np.cos(w0) + sin = np.sin(w0) + sqA = float(np.sqrt(max(A, 1e-12))) + S = 1.0 + + # Guard the sqrt argument — clamp to avoid negative/NaN + sqrt_arg = (A + 1.0 / max(A, 1e-12)) * (1.0 / max(S, 1e-9) - 1.0) + 2.0 + sqrt_arg = max(sqrt_arg, 0.0) + alpha = sin / 2.0 * np.sqrt(sqrt_arg) + + b0 = A * ((A + 1) + (A - 1) * cos + 2 * sqA * alpha) + b1 = -2 * A * ((A - 1) + (A + 1) * cos) + b2 = A * ((A + 1) + (A - 1) * cos - 2 * sqA * alpha) + a0 = (A + 1) - (A - 1) * cos + 2 * sqA * alpha + a1 = 2 * ((A - 1) - (A + 1) * cos) + a2 = (A + 1) - (A - 1) * cos - 2 * sqA * alpha + + # Guard against degenerate filter (a0 near zero) + if abs(a0) < 1e-10: + return data.astype(np.float32) + + sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64) + + # Final guard: if any coefficient is NaN/Inf, bypass + if not np.all(np.isfinite(sos)): + return data.astype(np.float32) + + result = sosfilt(sos, data.astype(np.float64)) + + # Guard output (should never happen now, but safety net) + if not np.all(np.isfinite(result)): + return data.astype(np.float32) + + return result.astype(np.float32) + +def _pink_noise(n: int, rng: np.random.Generator) -> np.ndarray: + """FFT-based pink noise — O(N log N), no sample loops.""" + white = rng.standard_normal(n).astype(np.float64) + fft_w = np.fft.rfft(white) + freqs = np.arange(1, len(fft_w) + 1, dtype=np.float64) + fft_w /= np.sqrt(freqs) # 1/sqrt(f) → pink (1/f power) + pink = np.fft.irfft(fft_w, n=n) + peak = np.max(np.abs(pink)) + if peak > 0: pink /= peak + return pink.astype(np.float32) + +def _comb_reverb(data: np.ndarray, sr: int, + room: float, damping: float, wet: float) -> np.ndarray: + """ + Schroeder reverb via scipy.signal.lfilter — fully vectorised, no Python loops. + 4 parallel comb filters + 2 series allpass. + room : 0–1 (room size; affects delay lengths) + damping : 0–1 (high-frequency absorption) + wet : 0–1 (wet/dry mix) + """ + if wet < 0.001: + return data + from scipy.signal import lfilter + + comb_delays_ms = [29.7, 37.1, 41.1, 43.7] + comb_base_gains = [0.805, 0.827, 0.783, 0.764] + allpass_delays_ms = [5.0, 1.7] + allpass_gain = 0.7 + scale = 0.5 + room * 0.5 + damp_factor = 1.0 - damping * 0.35 # approximate HF absorption + + sig64 = data.astype(np.float64) + rev = np.zeros_like(sig64) + + for dm, g_base in zip(comb_delays_ms, comb_base_gains): + N = max(2, int(dm * scale * sr / 1000)) + g = min(g_base * (0.5 + room * 0.5) * damp_factor, 0.97) + b = np.zeros(N + 1); b[0] = 1.0 + a = np.zeros(N + 1); a[0] = 1.0; a[N] = -g + rev += lfilter(b, a, sig64) + + rev *= 0.25 + + for dm in allpass_delays_ms: + N = max(2, int(dm * sr / 1000)) + g = allpass_gain + b = np.zeros(N + 1); b[0] = -g; b[N] = 1.0 + a = np.zeros(N + 1); a[0] = 1.0; a[N] = -g + rev = lfilter(b, a, rev) + + return (data * (1.0 - wet) + rev.astype(np.float32) * wet).astype(np.float32) + +def _micro_pitch_drift(data: np.ndarray, sr: int, + cents: float, rng: np.random.Generator) -> np.ndarray: + """ + Subtle continuous pitch drift ±cents via nearest-neighbour resampling. + Creates the organic wobble of a real human voice. + """ + if cents < 0.05: + return data + # Generate a slow-moving drift signal (0.1–0.5 Hz sinusoid + noise) + t = np.arange(len(data)) / sr + freq_hz = 0.2 + rng.uniform(0, 0.15) + drift = np.sin(2 * np.pi * freq_hz * t) + drift += 0.3 * rng.standard_normal(len(data)) # add gentle randomness + # Smooth with a 200 ms window + k = max(1, int(0.2 * sr)) + drift = np.convolve(drift, np.ones(k)/k, mode='same') + # Scale to ±ratio + ratio = 2 ** (cents / 1200) # cents → semitone ratio + factor = 1.0 + (ratio - 1.0) * drift # ±ratio around 1.0 + factor = np.clip(factor, 0.90, 1.10) + + # Nearest-neighbour reindex + indices = np.cumsum(factor) + indices = indices / indices[-1] * (len(data) - 1) + indices = np.clip(indices.astype(np.int32), 0, len(data) - 1) + return data[indices].astype(np.float32) + +def _timing_jitter(data: np.ndarray, sr: int, jitter_ms: float, + rng: np.random.Generator) -> np.ndarray: + """ + Apply block-level timing jitter: randomly shift 20 ms blocks by ±jitter_ms. + Breaks the perfectly even machine cadence of synthetic speech. + """ + if jitter_ms < 0.1: + return data + block = int(0.020 * sr) # 20 ms blocks + out = data.copy() + shift_max = int(jitter_ms * sr / 1000) + for start in range(0, len(data) - block, block): + shift = rng.integers(-shift_max, shift_max + 1) + src_start = max(0, start + shift) + src_end = min(len(data), start + block + shift) + dst_len = min(block, len(data) - start) + src_len = src_end - src_start + n = min(dst_len, src_len) + if n > 0: + out[start:start + n] = data[src_start:src_start + n] + return out + +def _soft_saturate(data: np.ndarray, amount: float) -> np.ndarray: + """ + Soft harmonic saturation — adds 2nd/3rd harmonics like analog tape/tube. + amount: 0 = bypass, 1 = heavy saturation + """ + if amount < 0.005: + return data + drive = 1.0 + amount * 6.0 # 1x … 7x overdrive + driven = data * drive + # tanh saturation, then scale back down + sat = np.tanh(driven) / np.tanh(drive) + return (data * (1.0 - amount) + sat * amount).astype(np.float32) + +# ── Master humanize function ────────────────────────────────────────── +def humanize_audio(src_path: str, out_path: str, preset_name: str, + intensity: float = 1.0, seed: int = 42): + """ + Apply full humanization DSP chain to a WAV file. + intensity: 0.0 = almost dry, 1.0 = full preset, >1.0 = exaggerated. + Each stage is wrapped in a try/except so a failure degrades gracefully + instead of corrupting the output file. + """ + p = HUMANIZE_PRESETS.get(preset_name, HUMANIZE_PRESETS["bedroom"]) + rng = np.random.default_rng(seed) + + data, sr = sf.read(src_path) + if data.ndim > 1: + data = data.mean(axis=1) + data = data.astype(np.float32) + + if len(data) == 0: + raise ValueError("ملف الصوت فارغ.") + + mix = float(np.clip(intensity, 0.0, 2.0)) + + def _safe(fn, label, *args, **kwargs): + """Run fn(*args, **kwargs), return original data on failure.""" + nonlocal data + try: + result = fn(*args, **kwargs) + # Reject NaN / Inf output — fall back to unchanged + if result is not None and len(result) > 0 and np.all(np.isfinite(result)): + data = result.astype(np.float32) + else: + print(f"[humanize] {label}: non-finite output — stage skipped") + except Exception as e: + print(f"[humanize] {label}: {e} — stage skipped") + + # 1. Mic coloring ───────────────────────────────────────────────── + _safe(_biquad_hp, "HP filter", data, sr, p["mic_hp_hz"]) + if abs(p["mic_presence_db"]) > 0.1: + _safe(_biquad_peak, "Peak EQ", data, sr, + p["mic_presence_hz"], p["mic_presence_db"] * mix, Q=1.2) + if abs(p["mic_air_db"]) > 0.1: + _safe(_biquad_shelf_hi, "Hi-shelf", data, sr, + p["mic_air_hz"], p["mic_air_db"] * mix) + + # 2. Harmonic saturation ────────────────────────────────────────── + _safe(_soft_saturate, "Saturation", data, p["saturation"] * mix) + + # 3. Timing jitter ──────────────────────────────────────────────── + _safe(_timing_jitter, "Jitter", data, sr, p["jitter_ms"] * mix, rng) + + # 4. Micro pitch drift ──────────────────────────────────────────── + _safe(_micro_pitch_drift, "Pitch drift", data, sr, + p["pitch_drift_cents"] * mix, rng) + + # 5. Room reverb ────────────────────────────────────────────────── + _safe(_comb_reverb, "Reverb", data, sr, + p["reverb_room"], p["reverb_damping"], p["reverb_wet"] * mix) + + # 6. Noise floor ────────────────────────────────────────────────── + try: + floor_db = p["noise_floor"] + (1.0 - mix) * 20 + floor_db = float(np.clip(floor_db, -96.0, -30.0)) + noise_amp = 10 ** (floor_db / 20.0) + color = float(np.clip(p["noise_color"], 0.0, 1.0)) + pink = _pink_noise(len(data), rng) + white = rng.standard_normal(len(data)).astype(np.float32) + noise_mix = pink * color + white * (1.0 - color) + nm_peak = np.max(np.abs(noise_mix)) + if nm_peak > 1e-9: + noise_mix = noise_mix / nm_peak * noise_amp + candidate = data + noise_mix.astype(np.float32) + if np.all(np.isfinite(candidate)): + data = candidate + except Exception as e: + print(f"[humanize] Noise floor: {e} — stage skipped") + + # 7. Final normalize + soft ceiling ────────────────────────────── + peak = float(np.max(np.abs(data))) + if peak > 0: + data = data / peak * 0.92 + # tanh soft-limiter (can't produce NaN on finite input) + data = (np.tanh(data * 1.1) * (1.0 / np.tanh(1.1))).astype(np.float32) + + sf.write(out_path, data, sr, subtype='PCM_16') + +# ── Text Splitting for Real Progress ───────────────────────────────── +def split_text_sentences(text: str, max_chunk: int = 280) -> list: + """Split text into sentence-sized chunks with a max character limit.""" + parts = re.split(r'(?<=[.!?؟\n،])\s+', text.strip()) + chunks, cur = [], "" + for p in parts: + if not p.strip(): + continue + if len(cur) + len(p) + 1 <= max_chunk: + cur = (cur + " " + p).strip() if cur else p + else: + if cur: + chunks.append(cur) + if len(p) > max_chunk: + # Sub-split on comma/semicolon + sub_parts = re.split(r'(?<=[،,;])\s+', p) + sc = "" + for s in sub_parts: + if len(sc) + len(s) + 1 <= max_chunk: + sc = (sc + " " + s).strip() if sc else s + else: + if sc: chunks.append(sc) + sc = s + if sc: chunks.append(sc) + cur = "" + else: + cur = p + if cur: + chunks.append(cur) + return chunks if chunks else [text] + +# ── Core TTS call ───────────────────────────────────────────────────── +def run_tts(text, refs, lang, tts_params, split, out_path): + if not xtts_ready or xtts is None: + raise RuntimeError("النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.") + p = {k: v for k, v in tts_params.items() if not k.startswith("_")} + xtts.tts_to_file( + text=text, speaker_wav=refs, language=lang, file_path=out_path, + enable_text_splitting=split, **p, + ) + +def run_tts_chunked(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 + +def collect_refs(files, voice_name): + paths = [] + for f in files: + if not f or not getattr(f, "filename", None): continue + p = f"/tmp/ref_{uuid.uuid4().hex}_{f.filename}" + with open(p, "wb") as buf: shutil.copyfileobj(f.file, buf) + paths.append(p) + if voice_name and voice_name not in ("", "null", "none", "undefined"): + lib_dir = os.path.join(VOICE_LIB, voice_name) + if os.path.isdir(lib_dir): + paths += [os.path.join(lib_dir, fn) for fn in sorted(os.listdir(lib_dir)) + if fn.lower().endswith((".wav", ".mp3", ".flac", ".ogg"))] + return paths + +def cleanup(paths): + for p in paths: + if p and p.startswith("/tmp/"): + try: os.remove(p) + except: pass + +# ── FastAPI ─────────────────────────────────────────────────────────── +app = FastAPI(title="XTTS Voice Studio CPU Optimized") +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) + +LANGUAGES = { + "ar":"العربية","en":"English","es":"Español","fr":"Français", + "de":"Deutsch","it":"Italiano","pt":"Português","ru":"Русский", + "zh-cn":"中文","ja":"日本語","ko":"한국어","tr":"Türkçe", + "nl":"Nederlands","pl":"Polski","cs":"Čeština","hi":"हिन्दी", +} +LANG_MAP = {"zh-cn":"zh-CN"} + +# ══════════════════════════════════════════════════════════════════════ +# F R O N T E N D +# ══════════════════════════════════════════════════════════════════════ +HTML = r""" + + + + +XTTS Voice Studio + + + + + + + + +
+ + +""" + +# ══════════════════════════════════════════════════════════════════════ +# R O U T E S +# ══════════════════════════════════════════════════════════════════════ +@app.get("/", response_class=HTMLResponse) +async def ui(): + return (HTML + .replace("DEVICE_PH", device) + .replace("LANGS_PH", json.dumps(LANGUAGES, ensure_ascii=False))) + + +# ── Model status endpoint ───────────────────────────────────────────── +@app.get("/model_status") +def model_status(): + return JSONResponse({ + "loading": xtts_loading, + "ready": xtts_ready, + "error": xtts_error, + "device": device, + }) + +# ── Job control endpoints ───────────────────────────────────────────── +@app.get("/progress/{job_id}") +def get_progress(job_id: str): + _gc_jobs() + p = job_progress.get(job_id, {"current": 0, "total": 1, "status": "unknown"}) + return JSONResponse(p) + +@app.post("/cancel/{job_id}") +def cancel_job(job_id: str): + ev = cancel_events.get(job_id) + if ev: + ev.set() + return {"cancelled": True} + + +# ── Audio post-edit endpoint ────────────────────────────────────────── +@app.post("/edit_audio") +async def edit_audio( + filename: str = Form(...), + volume_db: float = Form(0.0), + speed_factor: float = Form(1.0), + trim_start: float = Form(0.0), + trim_end: float = Form(0.0), + fade_in: float = Form(0.0), # seconds + fade_out: float = Form(0.0), # seconds + normalize: bool = Form(False), # peak-normalize to 0 dBFS +): + """Edit a generated audio file without re-running TTS. + Always writes a NEW file — the original is never modified.""" + src = os.path.join(OUTPUT_DIR, filename) + if not os.path.exists(src): + raise HTTPException(404, "الملف غير موجود.") + + def _edit(): + data, sr = sf.read(src) + if data.ndim > 1: + data = data.mean(axis=1) + + # ── Trim start / end ────────────────────────────────────────── + if trim_start > 0: + data = data[min(int(trim_start * sr), len(data)):] + if trim_end > 0: + end_sample = max(0, len(data) - int(trim_end * sr)) + data = data[:end_sample] + + if len(data) == 0: + raise ValueError("الملف فارغ بعد القطع — قلّل قيم القطع.") + + # ── Speed via polyphase resampling ──────────────────────────── + if abs(speed_factor - 1.0) > 0.01: + try: + from scipy.signal import resample_poly + from math import gcd + # Represent speed as a rational fraction p/q + p = round(speed_factor * 1000) + q = 1000 + g = gcd(p, q) + # resample_poly(data, up, down): output length = len * up/down + # We want output_len = input_len / speed ⟹ up=q, down=p + data = resample_poly(data, q // g, p // g).astype(np.float32) + except ImportError: + # Fallback: crude integer resampling via scipy.signal.resample + try: + from scipy.signal import resample as sp_resample + new_len = max(1, int(len(data) / speed_factor)) + data = sp_resample(data, new_len).astype(np.float32) + except ImportError: + pass # scipy not available + + # ── Normalize (peak to 0 dBFS) ──────────────────────────────── + if normalize: + peak = np.max(np.abs(data)) + if peak > 0: + data = data / peak * 0.98 + + # ── Volume (dB) ─────────────────────────────────────────────── + if abs(volume_db) > 0.01: + factor = 10 ** (volume_db / 20.0) + data = data * factor + # Soft clip via tanh to prevent digital distortion + data = np.tanh(data * 1.2) * (1.0 / np.tanh(1.2)) + + # ── Fade In (linear ramp) ───────────────────────────────────── + if fade_in > 0: + n = min(int(fade_in * sr), len(data)) + data[:n] *= np.linspace(0.0, 1.0, n, dtype=np.float32) + + # ── Fade Out (linear ramp) ──────────────────────────────────── + if fade_out > 0: + n = min(int(fade_out * sr), len(data)) + data[-n:] *= np.linspace(1.0, 0.0, n, dtype=np.float32) + + out = os.path.join(OUTPUT_DIR, f"edit_{uuid.uuid4().hex[:8]}.wav") + sf.write(out, data, sr, subtype='PCM_16') + return os.path.basename(out) + + try: + new_filename = await asyncio.to_thread(_edit) + except ValueError as e: + raise HTTPException(400, str(e)) + return {"filename": new_filename} + + +# ── Humanize endpoint ───────────────────────────────────────────────── +@app.post("/humanize") +async def humanize_endpoint( + filename: str = Form(...), + preset: str = Form("bedroom"), + intensity: float = Form(1.0), + seed: int = Form(42), +): + """Apply acoustic environment simulation to a generated WAV. + Returns a brand-new file — original untouched.""" + src = os.path.join(OUTPUT_DIR, filename) + if not os.path.exists(src): + raise HTTPException(404, "الملف غير موجود.") + if preset not in HUMANIZE_PRESETS: + raise HTTPException(400, f"بيئة غير معروفة: {preset}") + intensity = float(np.clip(intensity, 0.05, 2.0)) + out = os.path.join(OUTPUT_DIR, f"hum_{uuid.uuid4().hex[:8]}.wav") + try: + await asyncio.to_thread(humanize_audio, src, out, preset, intensity, seed) + except Exception as e: + if os.path.exists(out): + os.remove(out) + raise HTTPException(500, f"خطأ في المعالجة: {e}") + return {"filename": os.path.basename(out)} + + +@app.get("/humanize/presets") +def humanize_presets(): + return list(HUMANIZE_PRESETS.keys()) + +@app.post("/generate") +async def generate( + background_tasks: BackgroundTasks, + text: str = Form(...), + job_id: str = Form(default=""), + language: str = Form("ar"), + stability: float = Form(0.5), + clarity: float = Form(0.75), + style_exaggeration: float = Form(0.0), + speed: float = Form(1.0), + speaker_boost: bool = Form(False), + enable_text_splitting: bool = Form(True), + expression_mode: str = Form("natural"), + variation_strength: float = Form(1.0), + voice_name: str = Form(None), + files: list[UploadFile] = File(default=[]), +): + if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.") + if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}") + if not text.strip(): raise HTTPException(400, "النص فارغ.") + if len(text) > MAX_CHARS: raise HTTPException(400, f"النص يتجاوز {MAX_CHARS} حرف.") + if not job_id: + job_id = uuid.uuid4().hex + + enhanced_text = enhance_text_expression(text, language, expression_mode) + tts_params = apply_expression_preset(expression_mode, speed) + if expression_mode == "natural": + tts_params = dict( + temperature=stability_to_temperature(stability), + top_p=clarity_to_top_p(clarity), + repetition_penalty=style_to_rep_penalty(style_exaggeration), + speed=speed, + ) + + refs = collect_refs(files, voice_name) + if not refs: raise HTTPException(400, "يجب تحديد عينة صوتية مرجعية.") + out = os.path.join(OUTPUT_DIR, f"gen_{uuid.uuid4().hex[:10]}.wav") + cancel_events[job_id] = threading.Event() + silence_ms = {"story":160,"sad":180,"news":80,"excited":70}.get(expression_mode, 110) + variation = float(np.clip(variation_strength, 0.0, 2.0)) + 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 + ) + if not success: + if os.path.exists(out): os.remove(out) + raise HTTPException(499, "تم إلغاء التوليد.") + if speaker_boost: + await asyncio.to_thread(apply_speaker_boost, out) + finally: + cleanup(refs) + cancel_events.pop(job_id, None) + + append_history(os.path.basename(out), text, language, "tts", len(text)) + return {"filename": os.path.basename(out)} + + +# ── STS ─────────────────────────────────────────────────────────────── +@app.post("/sts") +async def sts( + background_tasks: BackgroundTasks, + source_audio: UploadFile = File(...), + job_id: str = Form(default=""), + language: str = Form("ar"), + stability: float = Form(0.5), + clarity: float = Form(0.75), + style_exaggeration: float = Form(0.0), + speed: float = Form(1.0), + speaker_boost: bool = Form(False), + variation_strength_sts: float = Form(1.0), + voice_name: str = Form(None), + ref_files: list[UploadFile] = File(default=[]), +): + if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.") + if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}") + if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.") + if not job_id: job_id = uuid.uuid4().hex + src = f"/tmp/sts_{uuid.uuid4().hex}.wav" + with open(src, "wb") as f: shutil.copyfileobj(source_audio.file, f) + refs = [] + cancel_events[job_id] = threading.Event() + try: + text = await asyncio.to_thread(transcribe_audio, src) + if not text: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.") + refs = collect_refs(ref_files, voice_name) + if not refs: raise HTTPException(400, "حدد صوت الهدف.") + out = os.path.join(OUTPUT_DIR, f"sts_{uuid.uuid4().hex[:10]}.wav") + sts_params = dict( + temperature=stability_to_temperature(stability), + top_p=clarity_to_top_p(clarity), + repetition_penalty=style_to_rep_penalty(style_exaggeration), + speed=speed, + ) + var_sts = float(np.clip(variation_strength_sts, 0.0, 2.0)) + async with get_tts_lock(): + success = await asyncio.to_thread( + run_tts_chunked, text, refs, language, sts_params, out, job_id, 110, var_sts + ) + if not success: + if os.path.exists(out): os.remove(out) + raise HTTPException(499, "تم إلغاء التحويل.") + if speaker_boost: + await asyncio.to_thread(apply_speaker_boost, out) + finally: + cleanup([src]); cleanup(refs) + cancel_events.pop(job_id, None) + + append_history(os.path.basename(out), text, language, "sts", len(text)) + return {"filename": os.path.basename(out), "transcribed": text} + + +# ── DUB ─────────────────────────────────────────────────────────────── +@app.post("/dub") +async def dub( + background_tasks: BackgroundTasks, + source_audio: UploadFile = File(...), + job_id: str = Form(default=""), + target_language: str = Form("en"), + stability: float = Form(0.5), + clarity: float = Form(0.75), + style_exaggeration: float = Form(0.0), + speed: float = Form(1.0), + speaker_boost: bool = Form(False), + variation_strength_dub: float = Form(1.0), + voice_name: str = Form(None), + ref_files: list[UploadFile] = File(default=[]), +): + if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.") + if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}") + if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.") + if not TRANSLATOR_OK: raise HTTPException(503, "deep-translator غير مثبّت.") + if not job_id: job_id = uuid.uuid4().hex + src = f"/tmp/dub_{uuid.uuid4().hex}.wav" + with open(src, "wb") as f: shutil.copyfileobj(source_audio.file, f) + refs = [] + cancel_events[job_id] = threading.Event() + try: + original = await asyncio.to_thread(transcribe_audio, src) + if not original: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.") + tgt_code = LANG_MAP.get(target_language, target_language) + translated = await asyncio.to_thread( + GoogleTranslator(source="auto", target=tgt_code).translate, original) + if not translated: raise HTTPException(422, "فشلت الترجمة.") + refs = collect_refs(ref_files, voice_name) + if not refs: raise HTTPException(400, "حدد صوت الدبلجة.") + out = os.path.join(OUTPUT_DIR, f"dub_{uuid.uuid4().hex[:10]}.wav") + dub_params = dict( + temperature=stability_to_temperature(stability), + top_p=clarity_to_top_p(clarity), + repetition_penalty=style_to_rep_penalty(style_exaggeration), + speed=speed, + ) + var_dub = float(np.clip(variation_strength_dub, 0.0, 2.0)) + async with get_tts_lock(): + success = await asyncio.to_thread( + run_tts_chunked, translated, refs, target_language, dub_params, out, job_id, 110, var_dub + ) + if not success: + if os.path.exists(out): os.remove(out) + raise HTTPException(499, "تم إلغاء الدبلجة.") + if speaker_boost: + await asyncio.to_thread(apply_speaker_boost, out) + finally: + cleanup([src]); cleanup(refs) + cancel_events.pop(job_id, None) + + append_history(os.path.basename(out), translated, target_language, "dub", len(translated)) + return {"filename": os.path.basename(out), "original": original, "translated": translated} + + +# ── SFX ─────────────────────────────────────────────────────────────── +@app.post("/sfx") +async def sfx( + background_tasks: BackgroundTasks, + description: str = Form(...), + job_id: str = Form(default=""), + stability: float = Form(0.1), + clarity: float = Form(0.5), + style_exaggeration: float = Form(0.9), + speed: float = Form(0.85), + speaker_boost: bool = Form(False), + voice_name: str = Form(None), + ref_file: UploadFile = File(default=None), +): + if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.") + if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}") + if not description.strip(): raise HTTPException(400, "الوصف فارغ.") + if not job_id: job_id = uuid.uuid4().hex + ref_list = [ref_file] if (ref_file and getattr(ref_file, "filename", None)) else [] + refs = collect_refs(ref_list, voice_name) + if not refs: raise HTTPException(400, "حدد صوتاً مرجعياً.") + out = os.path.join(OUTPUT_DIR, f"sfx_{uuid.uuid4().hex[:10]}.wav") + cancel_events[job_id] = threading.Event() + sfx_params = dict( + temperature=stability_to_temperature(stability), + top_p=clarity_to_top_p(clarity), + repetition_penalty=style_to_rep_penalty(style_exaggeration), + speed=speed, + ) + try: + async with get_tts_lock(): + success = await asyncio.to_thread( + run_tts_chunked, description, refs, "en", sfx_params, out, job_id + ) + if not success: + if os.path.exists(out): os.remove(out) + raise HTTPException(499, "تم إلغاء التوليد.") + if speaker_boost: + await asyncio.to_thread(apply_speaker_boost, out) + finally: + cleanup(refs) + cancel_events.pop(job_id, None) + + append_history(os.path.basename(out), description, "en", "sfx", len(description)) + return {"filename": os.path.basename(out)} + + +# ── Audio / Voice endpoints ─────────────────────────────────────────── +@app.get("/audio/{filename}") +def audio(filename: str): + safe = os.path.basename(filename) + if safe != filename or ".." in safe: + raise HTTPException(400, "اسم ملف غير صالح.") + p = os.path.join(OUTPUT_DIR, safe) + if not os.path.exists(p): raise HTTPException(404, "ملف غير موجود.") + return FileResponse(p, media_type="audio/wav", + headers={"Accept-Ranges": "bytes", "Cache-Control": "no-cache"}) + +@app.get("/history") +def get_history(): + return JSONResponse(load_history()) + +@app.get("/voices") +def list_voices(): + if not os.path.isdir(VOICE_LIB): return [] + return [d for d in os.listdir(VOICE_LIB) if os.path.isdir(os.path.join(VOICE_LIB, d))] + +@app.post("/voices/save") +async def save_voice(background_tasks: BackgroundTasks, + name: str = Form(...), + file: UploadFile = File(...), + file2: UploadFile = File(default=None)): + safe = name.strip().replace("/","_").replace("..","_")[:40] + lib = os.path.join(VOICE_LIB, safe) + os.makedirs(lib, exist_ok=True) + for f in ([file, file2] if file2 and file2.filename else [file]): + safe_fn = re.sub(r"[^A-Za-z0-9_.\-]", "_", os.path.basename(f.filename))[:80] + with open(os.path.join(lib, safe_fn), "wb") as buf: + shutil.copyfileobj(f.file, buf) + background_tasks.add_task(trigger_cloud_backup) + return {"name": safe} + +@app.delete("/voices/{name}") +def delete_voice(background_tasks: BackgroundTasks, name: str): + lib = os.path.join(VOICE_LIB, name) + if os.path.isdir(lib): shutil.rmtree(lib) + background_tasks.add_task(trigger_cloud_backup) + return {"deleted": name} + +@app.get("/voices/{name}/preview") +def preview_voice(name: str): + lib = os.path.join(VOICE_LIB, name) + if not os.path.isdir(lib): raise HTTPException(404, "صوت غير موجود.") + for fn in os.listdir(lib): + if fn.lower().endswith((".wav",".mp3",".flac",".ogg")): + mt = "audio/wav" if fn.endswith(".wav") else "audio/mpeg" + return FileResponse(os.path.join(lib, fn), media_type=mt) + raise HTTPException(404, "لا توجد ملفات صوتية.") + + +# ══════════════════════════════════════════════════════════════════════ +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info") \ No newline at end of file