diff --git "a/app.py" "b/app.py"
deleted file mode 100644--- "a/app.py"
+++ /dev/null
@@ -1,3399 +0,0 @@
-# ╔══════════════════════════════════════════════════════════════════════╗
-# ║ XTTS v2 Voice Studio — Professional Edition v11 ║
-# ║ SINGLE-FILE BUILD (app.py + requirements.txt + Dockerfile) ║
-# ║ • CPU-only (16 vCPU / 104 GB RAM) • @GPU defined, never used ║
-# ║ • Faster-Whisper • Soft Clipping • Real Progress • Cancel Jobs ║
-# ║ • Audio Post-Editor • Full Mobile Responsive UI ║
-# ╚══════════════════════════════════════════════════════════════════════╝
-#
-# Run: python app.py
-# This file now contains everything the old Dockerfile / requirements.txt
-# did. On start-up it (1) sets the environment, (2) installs system
-# packages when running as root, (3) installs the pinned Python packages
-# once, (4) creates the runtime folders, then (5) starts the server on
-# port 7860. Set SKIP_BOOTSTRAP=1 to skip steps 2-3.
-#
-# Minimal Dockerfile that is still needed (a script cannot create its own
-# container):
-# FROM python:3.10-slim
-# RUN useradd -m -u 1000 user && mkdir -p /data && chown 1000:1000 /data
-# USER user
-# WORKDIR /home/user/app
-# COPY --chown=user app.py .
-# EXPOSE 7860
-# CMD ["python", "app.py"]
-
-import os, sys, re, time, json, uuid, shutil, asyncio, threading
-import subprocess, hashlib, tempfile, ctypes.util
-
-# ══════════════════════════════════════════════════════════════════════
-# 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
-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
-
- # ── 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 = str(e)
- print(f"[!] XTTS loading failed: {e}")
- 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) ──────────────────────────
-try:
- from faster_whisper import WhisperModel
- whisper_mdl = WhisperModel("small", device="cpu", compute_type="int8", cpu_threads=CPU_THREADS)
- WHISPER_OK = True
- print("[✓] Faster-Whisper ready (small · int8 · CPU).")
-except Exception as e:
- WHISPER_OK, whisper_mdl = False, None
- print(f"[!] Faster-Whisper not available: {e}")
-
-def transcribe_audio(path: str) -> str:
- """Transcribe using Faster-Whisper; returns plain text."""
- 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