# ╔══════════════════════════════════════════════════════════════════════╗
# ║ XTTS v2 Voice Studio — Professional Edition v11 ║
# ║ • Faster-Whisper • Soft Clipping • Real Progress • Cancel Jobs ║
# ║ • Audio Post-Editor • Full Mobile Responsive UI ║
# ║ • ZeroGPU (gradio.Server backend, no Gradio UI) • Per-user Login ║
# ║ • Privacy Notice ║
# ╚══════════════════════════════════════════════════════════════════════╝
# ── ZeroGPU: `spaces` MUST be imported before torch / any CUDA-touching
# library. It monkey-patches torch.cuda.* so GPU allocation can be
# deferred and handed out on-demand by Hugging Face's ZeroGPU pool. ──
import spaces
import os, re, time, json, uuid, shutil, asyncio, threading, hashlib, secrets
import numpy as np
import soundfile as sf
# ── gradio.Server: a FastAPI subclass. It gives ZeroGPU eligibility,
# request queuing/concurrency control on Spaces, but ships NO Gradio
# UI components — every route below is a plain FastAPI/Starlette
# route serving our own HTML/JSON, exactly like `FastAPI()` would. ──
from gradio import Server
from fastapi import Form, File, UploadFile, HTTPException, BackgroundTasks, Request
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse, RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
import torch
# ── Environment ───────────────────────────────────────────────────────
os.environ["COQUI_TOS_AGREED"] = "1"
os.environ.setdefault("HF_HOME", "/home/user/.cache/huggingface")
os.environ["OMP_NUM_THREADS"] = "4"
os.environ["MKL_NUM_THREADS"] = "4"
# ── 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()
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ A U T H & U S E R M A N A G E M E N T ║
# ║ Accounts are stored in the persistent /data mount so they survive ║
# ║ restarts. Passwords are never stored in plain text (PBKDF2-SHA256 ║
# ║ with a per-user random salt, 200k iterations). ║
# ╚══════════════════════════════════════════════════════════════════════╝
USERS_FILE = os.path.join(PERSISTENT_DIR, "users.json")
SECRET_FILE = os.path.join(PERSISTENT_DIR, ".session_secret")
if os.path.exists(SECRET_FILE):
SESSION_SECRET = open(SECRET_FILE, "r", encoding="utf-8").read().strip()
else:
SESSION_SECRET = secrets.token_hex(32)
with open(SECRET_FILE, "w", encoding="utf-8") as f:
f.write(SESSION_SECRET)
PASSWORD_RULE = re.compile(r'^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[^A-Za-z0-9]).{8,}$')
def is_strong_password(pw: str) -> bool:
"""8+ chars, upper+lower+digit+symbol. Required for every account."""
return bool(PASSWORD_RULE.match(pw or ""))
def _hash_password(password: str, salt: bytes = None) -> str:
salt = salt or secrets.token_bytes(16)
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 200_000)
return salt.hex() + "$" + dk.hex()
def _verify_password(password: str, stored: str) -> bool:
try:
salt_hex, hash_hex = stored.split("$")
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt_hex), 200_000)
return secrets.compare_digest(dk.hex(), hash_hex)
except Exception:
return False
def load_users() -> dict:
try:
return json.load(open(USERS_FILE, encoding="utf-8")) if os.path.exists(USERS_FILE) else {}
except Exception:
return {}
def save_users(u: dict):
json.dump(u, open(USERS_FILE, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
trigger_cloud_backup()
def safe_username(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_.\-]", "_", (name or "").strip())[:40]
def bootstrap_admin():
"""Create the developer/admin account on first run if none exists."""
users = load_users()
if users:
return
admin_user = safe_username(os.environ.get("ADMIN_USERNAME", "admin")) or "admin"
admin_pass = os.environ.get("ADMIN_PASSWORD", "")
if not is_strong_password(admin_pass):
admin_pass = secrets.token_urlsafe(9) + "Aa1!"
print("=" * 72)
print("[!] لم يتم ضبط كلمة مرور قوية عبر متغيّر البيئة ADMIN_PASSWORD.")
print("[!] تم إنشاء حساب المطوّر تلقائياً بالبيانات التالية (سجّل الدخول وغيّرها فوراً):")
print(f" اسم المستخدم : {admin_user}")
print(f" كلمة المرور : {admin_pass}")
print("[!] يُفضّل ضبط ADMIN_USERNAME / ADMIN_PASSWORD كمتغيّرات بيئة سرّية (Secrets) بدلاً من ذلك.")
print("=" * 72)
users[admin_user] = {
"password": _hash_password(admin_pass), "role": "admin", "created": int(time.time())
}
save_users(users)
bootstrap_admin()
def get_session_user(request: Request):
return request.session.get("user")
# ── Chatterbox Multilingual V3 Model ───────────────────────────────────
# Replaces XTTS v2. MIT-licensed, ~500M params, native Arabic support,
# and — per Resemble AI's own Podonos blind evaluations — preferred over
# ElevenLabs by listeners. Internal variable names (xtts / xtts_ready /
# xtts_loading / xtts_error) are kept as-is on purpose: they're referenced
# throughout /generate, /sts, /dub, /sfx and /model_status, and renaming
# them everywhere would add risk for zero benefit — they now simply hold
# the Chatterbox model instead of XTTS.
from chatterbox.mtl_tts import ChatterboxMultilingualTTS
import inspect as _inspect
device = "cuda" if torch.cuda.is_available() else "cpu"
# ── Model loading ────────────────────────────────────────────────────
# NOTE: on ZeroGPU, `spaces`'s CUDA-emulation shim only intercepts
# `.to("cuda")` (or `device="cuda"` passed into `from_pretrained`) when it
# runs synchronously at module scope in the main thread — this is exactly
# HF's documented pattern (`pipe.to("cuda")` right after `from_pretrained`,
# no threading involved). Loading the model inside a background
# `threading.Thread` (so the server could start immediately) escapes that
# emulation hook, so a *real* low-level CUDA init is reached in a
# stateless-GPU process, and ZeroGPU aborts with:
# "Low-level CUDA init (`torch._C._cuda_init`) reached ..."
# So the model is loaded synchronously here, before the server starts —
# same pattern as before, just a different model class.
xtts = None # holds the ChatterboxMultilingualTTS instance
xtts_ready = False
xtts_loading = True
xtts_error = None
# Cache of the keyword arguments xtts.generate() actually accepts, so we
# never crash on a param name mismatch between Chatterbox releases — we
# just silently drop anything the installed version doesn't support.
_generate_valid_params = set()
try:
print(f"[*] Loading Chatterbox Multilingual V3 on {device.upper()} …")
try:
# "t3_model=v3" selects the newer V3 checkpoint on releases that
# support it. Older/newer chatterbox-tts builds may not accept
# this kwarg at all — fall back to the plain call if so, rather
# than failing the whole model load over one optional argument.
xtts = ChatterboxMultilingualTTS.from_pretrained(device=device, t3_model="v3")
except TypeError:
print("[i] Installed chatterbox-tts build doesn't accept t3_model= — loading default checkpoint.")
xtts = ChatterboxMultilingualTTS.from_pretrained(device=device)
try:
_generate_valid_params = set(_inspect.signature(xtts.generate).parameters.keys())
except (TypeError, ValueError):
_generate_valid_params = {"language_id", "audio_prompt_path", "exaggeration",
"cfg_weight", "temperature"}
xtts_ready = True
print("[✓] Chatterbox ready.")
except Exception as e:
xtts_error = str(e)
print(f"[!] Chatterbox loading failed: {e}")
finally:
xtts_loading = False
# 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")
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, owner="unknown"):
h = load_history()
h.append(dict(filename=filename, text=text[:150], language=language,
mode=mode, char_count=char_count, ts=int(time.time()), owner=owner))
if len(h) > MAX_HISTORY:
h = h[-MAX_HISTORY:]
save_history(h)
_prune_outputs()
def owns_output_file(filename: str, username: str) -> bool:
"""True if `username` generated `filename` (per the history log)."""
for entry in load_history():
if entry.get("filename") == filename:
return entry.get("owner") == username
return False
# ── Parameter Engineering ─────────────────────────────────────────────
# Chatterbox Multilingual exposes three real generation knobs:
# exaggeration : emotional intensity / expressiveness (default 0.5)
# cfg_weight : how tightly generation follows the reference voice —
# lower = looser & more dramatic, higher = literal (default 0.5)
# temperature : sampling randomness (default ~0.8)
# `speed` is NOT a native Chatterbox parameter — we apply it ourselves as
# a post-generation time-stretch (see _apply_speed_factor / edit_audio).
#
# Our four legacy UI sliders (Stability / Clarity / Style / Speed) are kept
# as-is in the UI (people are used to them) and mapped onto the three real
# knobs below:
# Stability (0=expressive .. 1=consistent) → inverse → temperature
# Clarity (0=varied .. 1=matches ref) → direct → cfg_weight
# Style Exaggeration (0=natural .. 1=dramatic) → direct → exaggeration
def stability_to_temperature(s): return round(0.95 - (float(s) * 0.55), 3) # 0.95 .. 0.40
def clarity_to_cfg_weight(c): return round(0.15 + (float(c) * 0.65), 3) # 0.15 .. 0.80
def style_to_exaggeration(st): return round(0.20 + (float(st) * 1.10), 3) # 0.20 .. 1.30
# ── Expression Presets ────────────────────────────────────────────────
# Each preset overrides exaggeration/cfg_weight/temperature to push
# Chatterbox toward a specific emotional character.
EXPRESSION_PRESETS = {
"natural": dict(exaggeration=0.50, cfg_weight=0.50, temperature=0.80, speed_mult=1.00),
"warm": dict(exaggeration=0.55, cfg_weight=0.55, temperature=0.82, speed_mult=0.97),
"excited": dict(exaggeration=0.85, cfg_weight=0.35, temperature=0.90, speed_mult=1.10),
"sad": dict(exaggeration=0.35, cfg_weight=0.60, temperature=0.70, speed_mult=0.88),
"serious": dict(exaggeration=0.30, cfg_weight=0.65, temperature=0.60, speed_mult=0.95),
"story": dict(exaggeration=0.65, cfg_weight=0.45, temperature=0.85, speed_mult=0.96),
"news": dict(exaggeration=0.30, cfg_weight=0.60, temperature=0.62, speed_mult=1.02),
"child": dict(exaggeration=0.80, cfg_weight=0.35, temperature=0.92, speed_mult=1.05),
}
def apply_expression_preset(preset_name: str, speed: float) -> dict:
p = EXPRESSION_PRESETS.get(preset_name, EXPRESSION_PRESETS["natural"])
return {
"exaggeration": p["exaggeration"],
"cfg_weight": p["cfg_weight"],
"temperature": p["temperature"],
"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. ║
# ║ Chatterbox parameters (exaggeration, cfg_weight, speed) shift ║
# ║ per-phrase so the delivery feels alive, not machine-read. ║
# ║ Post-generation volume envelopes reinforce the prosodic contour. ║
# ╚══════════════════════════════════════════════════════════════════════╝
# ── Per-emotion Chatterbox deltas ──────────────────────────────────────
# (applied on top of the base params; scaled by user's variation_strength)
# exaggeration_delta : shifts Chatterbox's own expressiveness knob
# speed_mult : multiplies our post-generation time-stretch
# cfg_weight_delta : shifts how tightly generation follows the reference
# (lower = looser/more dramatic, higher = more literal)
EMOTION_DELTAS = {
# exagg speed_mult cfg_weight silence_ms
"neutral": ( 0.00, 1.000, 0.00, 110 ),
"question": (-0.05, 0.960, +0.15, 140 ), # careful, rising uncertainty
"exclaim": (+0.35, 1.080, -0.20, 75 ), # faster, more energy
"emphasis": (+0.15, 0.900, +0.10, 155 ), # deliberate, weighty
"sadness": (-0.15, 0.870, +0.10, 185 ), # slow, heavy, restrained
"joy": (+0.30, 1.060, -0.15, 85 ), # bright, forward
"anger": (+0.25, 1.040, -0.10, 95 ), # tense, clipped
"soft": (-0.20, 0.920, +0.15, 160 ), # tender, gentle
"urgency": (+0.30, 1.120, -0.15, 65 ), # fast, alert
"story": (+0.15, 0.950, -0.05, 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 Chatterbox 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 applied to
Chatterbox's own expressiveness knobs (exaggeration / cfg_weight),
plus our own post-generation `speed` (time-stretch applied after
generation, since Chatterbox has no native speed parameter).
"""
if variation < 0.02:
return base.copy()
d = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"])
exagg_d, speed_m, cfg_d, _ = d
strength = conf * variation # 0 … ~1.84
p = base.copy()
p["exaggeration"] = float(np.clip(
base.get("exaggeration", 0.5) + exagg_d * strength, 0.05, 1.8))
p["cfg_weight"] = float(np.clip(
base.get("cfg_weight", 0.5) + cfg_d * strength, 0.0, 1.0))
p["speed"] = float(np.clip(
base.get("speed", 1.0) * (1.0 + (speed_m - 1.0) * strength), 0.60, 1.65))
# ── Position-based natural drift ────────────────────────────────
# Opening: slightly more controlled (lower exaggeration, slower)
if pos_ratio < 0.08:
p["exaggeration"] = float(np.clip(p["exaggeration"] - 0.05 * variation, 0.05, 1.8))
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["exaggeration"] = float(np.clip(p["exaggeration"] - 0.04 * variation, 0.05, 1.8))
# ── Micro-random jitter: prevent identical params on consecutive ─
# neutral phrases (which would sound machine-even)
if emotion == "neutral" and variation > 0.3:
h = abs(hash(base.get("_phrase_hint", ""))) % 1000 / 1000.0
p["exaggeration"] += (h - 0.5) * 0.08 * variation
p["speed"] *= 1.0 + (h - 0.5) * 0.04 * variation
p["exaggeration"] = float(np.clip(p["exaggeration"], 0.05, 1.8))
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 generation.
This adds prosodic *weight* that model 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]
# ── Post-generation speed control ──────────────────────────────────────
# Chatterbox has no native "speed" knob (unlike XTTS). We apply it
# ourselves via polyphase resampling — the same technique the audio
# editor's Speed slider already uses in /edit_audio, so the two stay
# consistent. Trade-off: this shifts pitch slightly at extreme values,
# same as before.
def _resample_speed(data: np.ndarray, sr: int, factor: float) -> np.ndarray:
if abs(factor - 1.0) < 0.01 or len(data) == 0:
return data
try:
from scipy.signal import resample_poly
from math import gcd
p = max(1, round(factor * 1000))
q = 1000
g = gcd(p, q)
return resample_poly(data, q // g, p // g).astype(np.float32)
except Exception as e:
print(f"[speed] resample failed: {e} — returning unmodified audio")
return data.astype(np.float32)
# ── Voice prompt preparation ───────────────────────────────────────────
# XTTS accepted a LIST of speaker_wav files. Chatterbox's `generate()`
# takes exactly one `audio_prompt_path`. If more than one reference clip
# was supplied (main sample + "extra sample" + library folder with
# several files), we concatenate them into a single prompt clip, capped
# at 30s (plenty for cloning, keeps prompt-encoding fast).
def prepare_voice_prompt(refs: list) -> str:
if not refs:
raise RuntimeError("لا توجد عيّنة صوتية مرجعية.")
if len(refs) == 1:
return refs[0]
MAX_SEC = 30
arrays, sr_out = [], None
for p in refs:
try:
data, sr = sf.read(p)
if data.ndim > 1:
data = data.mean(axis=1)
if sr_out is None:
sr_out = sr
elif sr != sr_out:
from scipy.signal import resample
data = resample(data, int(len(data) * sr_out / sr))
arrays.append(data.astype(np.float32))
arrays.append(np.zeros(int(sr_out * 0.15), dtype=np.float32)) # 150ms gap
except Exception as e:
print(f"[voice_prompt] skipped {p}: {e}")
if not arrays:
return refs[0]
combined = np.concatenate(arrays)
max_samples = int(MAX_SEC * (sr_out or 24000))
if len(combined) > max_samples:
combined = combined[:max_samples]
out_path = f"/tmp/prompt_{uuid.uuid4().hex}.wav"
sf.write(out_path, combined, sr_out or 24000, subtype="PCM_16")
return out_path
# ── Core TTS call (Chatterbox Multilingual) ────────────────────────────
def run_tts(text, refs, lang, tts_params, split, out_path):
"""
`refs` must already be resolved to a single-element list (callers go
through run_tts_chunked, which prepares the voice prompt ONCE and
reuses it across every phrase/chunk instead of re-concatenating
reference clips on every sentence).
`split` is accepted for interface compatibility with the old XTTS
call signature but has no effect — chunking is handled entirely by
run_tts_chunked() before this function is ever called.
"""
if not xtts_ready or xtts is None:
raise RuntimeError("النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.")
prompt_path = prepare_voice_prompt(refs)
speed = float(tts_params.get("speed", 1.0))
gen_kwargs = {
"language_id": lang,
"audio_prompt_path": prompt_path,
"exaggeration": float(tts_params.get("exaggeration", 0.5)),
"cfg_weight": float(tts_params.get("cfg_weight", 0.5)),
"temperature": float(tts_params.get("temperature", 0.8)),
}
# Defensive: only pass kwargs the installed Chatterbox version accepts.
if _generate_valid_params:
gen_kwargs = {k: v for k, v in gen_kwargs.items() if k in _generate_valid_params}
wav = xtts.generate(text, **gen_kwargs)
data = wav.detach().cpu().numpy() if hasattr(wav, "detach") else np.asarray(wav)
if data.ndim > 1:
data = data.mean(axis=0)
data = data.astype(np.float32)
if abs(speed - 1.0) > 0.01:
data = _resample_speed(data, xtts.sr, speed)
sf.write(out_path, data, xtts.sr, subtype="PCM_16")
@spaces.GPU(duration=180)
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 Chatterbox 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"}
# Build the voice prompt ONCE — reused across every chunk instead of
# re-concatenating reference clips on every sentence.
prompt_path = prepare_voice_prompt(refs)
prompt_refs = [prompt_path]
is_temp_prompt = prompt_path.startswith("/tmp/prompt_")
try:
# ── 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, prompt_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, prompt_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
finally:
if is_temp_prompt:
try: os.remove(prompt_path)
except: pass
def user_voice_dir(username: str) -> str:
d = os.path.join(VOICE_LIB, safe_username(username))
os.makedirs(d, exist_ok=True)
return d
def collect_refs(files, voice_name, username):
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(user_voice_dir(username), 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
# ── App (gradio.Server — FastAPI-compatible, no Gradio UI mounted) ─────
app = Server(title="XTTS Voice Studio CPU Optimized")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
PUBLIC_PATHS = {"/login", "/signup", "/favicon.ico", "/model_status"}
@app.middleware("http")
async def auth_gate(request: Request, call_next):
"""Blocks every route (API + pages) unless the visitor has a valid
session, except the login page itself and the health-check endpoint.
Also fences off /admin/* to the developer/admin role only.
`/gradio_api/*` is gradio.Server's own internal plumbing (startup
check, queue, ZeroGPU broker calls, etc.) and must stay open — it's
not user-facing content, and gradio.Server won't even finish
starting if these routes require a login."""
path = request.url.path
if path in PUBLIC_PATHS or path.startswith("/gradio_api"):
return await call_next(request)
user = request.session.get("user")
if not user:
if request.method == "GET" and "text/html" in request.headers.get("accept", ""):
return RedirectResponse(url="/login")
return JSONResponse({"detail": "يجب تسجيل الدخول أولاً."}, status_code=401)
if path.startswith("/admin") and user.get("role") != "admin":
return JSONResponse({"detail": "هذه الصفحة مخصّصة لحساب المطوّر فقط."}, status_code=403)
return await call_next(request)
# SessionMiddleware is added AFTER auth_gate on purpose. Starlette treats
# whichever middleware was registered LAST as the OUTERMOST layer (it
# wraps every other one and therefore runs FIRST on each request) — so
# adding it here guarantees request.session exists by the time auth_gate
# runs. Adding it earlier (before auth_gate) is what caused
# "SessionMiddleware must be installed to access request.session".
# Hugging Face Spaces are normally viewed inside an iframe on
# huggingface.co, so — from the browser's point of view — the Space's
# own hf.space origin is a *cross-site* (third-party) context relative
# to the page in the address bar, even for requests the iframe makes to
# itself. A `same_site="lax"` cookie is dropped in that situation, so
# the session looks empty on the very next request and auth_gate bounces
# the user straight back to /login — a login loop. `same_site="none"`
# fixes this, but browsers require `Secure` on any SameSite=None cookie,
# hence `https_only=True` (Spaces are always served over HTTPS, so this
# is safe; only disable it if you run this file locally over plain http).
app.add_middleware(
SessionMiddleware, secret_key=SESSION_SECRET, session_cookie="xtts_session",
max_age=60 * 60 * 24 * 7, same_site="none", https_only=True,
)
# ── Supported languages ─────────────────────────────────────────────────
# Matches Chatterbox Multilingual V3's 23 supported languages exactly
# (github.com/resemble-ai/chatterbox — "Supported Languages"). Arabic is
# listed first since it's this app's primary audience. `LANG_MAP` is used
# ONLY for the /dub translation step (deep-translator's Google Translate
# codes differ slightly from Chatterbox's language_id codes in a couple
# of cases) — the TTS call itself always uses the raw key below.
LANGUAGES = {
"ar":"العربية","en":"English","es":"Español","fr":"Français",
"de":"Deutsch","it":"Italiano","pt":"Português","ru":"Русский",
"zh":"中文","ja":"日本語","ko":"한국어","tr":"Türkçe",
"nl":"Nederlands","pl":"Polski","hi":"हिन्दी","he":"עברית",
"da":"Dansk","el":"Ελληνικά","fi":"Suomi","ms":"Bahasa Melayu",
"no":"Norsk","sv":"Svenska","sw":"Kiswahili",
}
LANG_MAP = {"zh":"zh-CN"} # deep-translator target-code override (dub only)
# ══════════════════════════════════════════════════════════════════════
# F R O N T E N D
# ══════════════════════════════════════════════════════════════════════
HTML = r"""
Chatterbox Voice Studio
🔒 يتم حفظ الأصوات المُدخلة (العيّنات المرجعية) والمخرجات الصوتية والنصوص المستخدمة في تخزين
دائم خاص بحساب USER_PH فقط. لا يستطيع أي مستخدم آخر رؤية ملفاتك. كل مقطع صوتي مُولَّد يحمل
بصمة تحقق غير مسموعة (PerTh watermark) لإثبات أنه مولَّد بالذكاء الاصطناعي.
ADMIN_LINK_PH
تسجيل الخروج
"""
# ══════════════════════════════════════════════════════════════════════
# R O U T E S
# ══════════════════════════════════════════════════════════════════════
@app.get("/", response_class=HTMLResponse)
async def ui(request: Request):
user = request.session["user"]
admin_link = (''
'لوحة المطوّر') if user.get("role") == "admin" else ""
return (HTML
.replace("DEVICE_PH", device)
.replace("LANGS_PH", json.dumps(LANGUAGES, ensure_ascii=False))
.replace("USER_PH", user["username"])
.replace("ADMIN_LINK_PH", admin_link))
# ── 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(
request: Request,
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."""
user = request.session["user"]
if user.get("role") != "admin" and not owns_output_file(os.path.basename(filename), user["username"]):
raise HTTPException(403, "لا تملك صلاحية الوصول لهذا الملف.")
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
p = round(speed_factor * 1000)
q = 1000
g = gcd(p, q)
data = resample_poly(data, q // g, p // g).astype(np.float32)
except ImportError:
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
# ── 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
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))
append_history(new_filename, f"[تحرير] {filename}", "-", "edit", 0, owner=user["username"])
return {"filename": new_filename}
# ── Humanize endpoint ─────────────────────────────────────────────────
@app.post("/humanize")
async def humanize_endpoint(
request: Request,
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."""
user = request.session["user"]
if user.get("role") != "admin" and not owns_output_file(os.path.basename(filename), user["username"]):
raise HTTPException(403, "لا تملك صلاحية الوصول لهذا الملف.")
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}")
append_history(os.path.basename(out), f"[معالجة بيئية] {filename}", "-", "humanize", 0, owner=user["username"])
return {"filename": os.path.basename(out)}
@app.get("/humanize/presets")
def humanize_presets():
return list(HUMANIZE_PRESETS.keys())
@app.post("/generate")
async def generate(
request: Request,
background_tasks: BackgroundTasks,
text: str = Form(...),
job_id: str = Form(default=""),
language: str = Form("ar"),
stability: float = Form(0.5),
clarity: float = Form(0.5),
style_exaggeration: float = Form(0.5),
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),
cfg_weight=clarity_to_cfg_weight(clarity),
exaggeration=style_to_exaggeration(style_exaggeration),
speed=speed,
)
username = request.session["user"]["username"]
refs = collect_refs(files, voice_name, username)
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), owner=username)
return {"filename": os.path.basename(out)}
# ── STS ───────────────────────────────────────────────────────────────
@app.post("/sts")
async def sts(
request: Request,
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.5),
style_exaggeration: float = Form(0.5),
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
username = request.session["user"]["username"]
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, username)
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),
cfg_weight=clarity_to_cfg_weight(clarity),
exaggeration=style_to_exaggeration(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), owner=username)
return {"filename": os.path.basename(out), "transcribed": text}
# ── DUB ───────────────────────────────────────────────────────────────
@app.post("/dub")
async def dub(
request: Request,
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.5),
style_exaggeration: float = Form(0.5),
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
username = request.session["user"]["username"]
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, username)
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),
cfg_weight=clarity_to_cfg_weight(clarity),
exaggeration=style_to_exaggeration(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), owner=username)
return {"filename": os.path.basename(out), "original": original, "translated": translated}
# ── SFX ───────────────────────────────────────────────────────────────
@app.post("/sfx")
async def sfx(
request: Request,
background_tasks: BackgroundTasks,
description: str = Form(...),
job_id: str = Form(default=""),
stability: float = Form(0.2),
clarity: float = Form(0.4),
style_exaggeration: float = Form(1.1),
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
username = request.session["user"]["username"]
ref_list = [ref_file] if (ref_file and getattr(ref_file, "filename", None)) else []
refs = collect_refs(ref_list, voice_name, username)
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),
cfg_weight=clarity_to_cfg_weight(clarity),
exaggeration=style_to_exaggeration(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), owner=username)
return {"filename": os.path.basename(out)}
# ── Audio / Voice endpoints (per-user privacy scoped) ─────────────────
@app.get("/audio/{filename}")
def audio(filename: str, request: Request):
safe = os.path.basename(filename)
if safe != filename or ".." in safe:
raise HTTPException(400, "اسم ملف غير صالح.")
user = request.session["user"]
if user.get("role") != "admin" and not owns_output_file(safe, user["username"]):
raise HTTPException(403, "لا تملك صلاحية الوصول لهذا الملف.")
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(request: Request):
user = request.session["user"]
h = load_history()
if user.get("role") != "admin":
h = [e for e in h if e.get("owner") == user["username"]]
return JSONResponse(h)
@app.get("/voices")
def list_voices(request: Request):
lib = user_voice_dir(request.session["user"]["username"])
return [d for d in os.listdir(lib) if os.path.isdir(os.path.join(lib, d))]
@app.post("/voices/save")
async def save_voice(request: Request, background_tasks: BackgroundTasks,
name: str = Form(...),
file: UploadFile = File(...),
file2: UploadFile = File(default=None)):
safe = name.strip().replace("/","_").replace("..","_")[:40]
lib = os.path.join(user_voice_dir(request.session["user"]["username"]), 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(request: Request, background_tasks: BackgroundTasks, name: str):
lib = os.path.join(user_voice_dir(request.session["user"]["username"]), 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(request: Request, name: str):
lib = os.path.join(user_voice_dir(request.session["user"]["username"]), 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, "لا توجد ملفات صوتية.")
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ L O G I N / L O G O U T / A D M I N P A N E L ║
# ╚══════════════════════════════════════════════════════════════════════╝
LOGIN_HTML = r"""
تسجيل الدخول — Chatterbox Voice Studio
🔒 تنويه خصوصية: يقوم هذا التطبيق بحفظ الأصوات المُدخلة (العيّنات المرجعية)
والمخرجات الصوتية الناتجة، وكذلك النصوص المستخدمة، وذلك في تخزين دائم خاص بحسابك
لتتمكن من استرجاعها لاحقاً. لا يمكن لأي مستخدم آخر الوصول إلى ملفاتك أو سجلّك.
يمكن لمالك الخدمة (حساب المطوّر) الوصول لأغراض الصيانة والامتثال فقط.
كل مقطع صوتي مولَّد يحمل بصمة تحقق غير مسموعة (PerTh) تثبت أنه من إنتاج الذكاء الاصطناعي.
برجاء عدم رفع تسجيلات صوتية لأشخاص دون موافقتهم.
"""
SIGNUP_ERRORS = {
"badname": "اسم المستخدم غير صالح (3 أحرف على الأقل).",
"exists": "اسم المستخدم مستخدم بالفعل، جرّب اسماً آخر.",
"weak": "كلمة المرور ضعيفة: يجب أن تحتوي 8 أحرف على الأقل، حرف كبير وحرف صغير ورقم ورمز.",
"mismatch": "كلمة المرور وتأكيدها غير متطابقين.",
}
@app.get("/signup", response_class=HTMLResponse)
async def signup_page(error: str = ""):
msg = f'
{SIGNUP_ERRORS[error]}
' if error in SIGNUP_ERRORS else ""
return SIGNUP_HTML.replace("ERROR_PH", msg)
@app.post("/signup")
async def signup_submit(request: Request, username: str = Form(...),
password: str = Form(...), password2: str = Form(...)):
uname = safe_username(username)
if not uname or len(uname) < 3:
return RedirectResponse(url="/signup?error=badname", status_code=303)
if password != password2:
return RedirectResponse(url="/signup?error=mismatch", status_code=303)
if not is_strong_password(password):
return RedirectResponse(url="/signup?error=weak", status_code=303)
users = load_users()
if uname in users:
return RedirectResponse(url="/signup?error=exists", status_code=303)
users[uname] = {"password": _hash_password(password), "role": "user", "created": int(time.time())}
save_users(users)
request.session["user"] = {"username": uname, "role": "user"}
return RedirectResponse(url="/", status_code=303)
ADMIN_HTML = r"""
لوحة المطوّر — Chatterbox Voice Studio