805 lines
32 KiB
Python
805 lines
32 KiB
Python
"""Voice library CRUD, upload, audio serving, trim/process, route-sounds."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import FileResponse, Response
|
|
|
|
from core.config import _load_settings
|
|
from core.constants import _VOICES_DIR_DEFAULT, _OUTPUT_DIR_DEFAULT, _VOICE_TARGET_DBFS, _VOICE_PEAK_DBFS, _BENCHMARK_SENTENCES, _MAX_TTS_OUTPUT_SECONDS, _MAX_UPLOAD_BYTES, _MAX_PICTURE_BYTES, _MAX_SOUND_BYTES
|
|
from core.registry import _registry_put, _registry_get, TEMP_DIR
|
|
from core.validation import _copy_limited, _safe_child_path, _validate_http_url
|
|
from core.audio import (
|
|
_to_wav_24k, _to_wav_16k, _trim, _duration, _export_normalized_wav,
|
|
_loudness_info, _auto_trim_bounds, _normalize_segment,
|
|
)
|
|
from core.voice import (
|
|
_AUDIO_EXTS, _AUDIO_MIME, _UPLOAD_EXTS, _PICTURE_EXTS, _SOUND_ASSET_DIRS,
|
|
_find_voice_audio, _load_meta, _save_meta, _picture_path, _picture_mime,
|
|
_active_voices_dir, _hidden_voices_dir,
|
|
_voice_audio_files, _voice_audio_from_request, _voice_entry,
|
|
_backup_original_voice, _backup_candidates, _backup_audio_suffix,
|
|
_remove_audio_variants, _remove_voice_package, _voice_package_paths,
|
|
_move_voice_package, _read_reference_text, _voice_health,
|
|
_is_internal_voice_file, _benchmark_voice,
|
|
)
|
|
from core.voice_index import (
|
|
indexed_voices,
|
|
refresh_voice_index_background,
|
|
remove_voice_from_index,
|
|
rebuild_voice_index,
|
|
upsert_voice_in_index,
|
|
voice_index_info,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ── Upload ────────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/upload")
|
|
async def upload(file: UploadFile = File(...)):
|
|
suffix = Path(file.filename or "audio").suffix.lower() or ".bin"
|
|
if suffix not in _UPLOAD_EXTS:
|
|
raise HTTPException(400, "Unsupported audio/video file type")
|
|
dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}"
|
|
with dest.open("wb") as f:
|
|
_copy_limited(file.file, f, _MAX_UPLOAD_BYTES)
|
|
try:
|
|
wav = _to_wav_24k(dest)
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Audio conversion failed: {e}")
|
|
fid = uuid.uuid4().hex
|
|
_registry_put(fid, wav)
|
|
return {"id": fid, "duration": _duration(wav), "filename": file.filename}
|
|
|
|
|
|
# ── Route sounds ──────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/route-sounds")
|
|
async def list_route_sounds():
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
sounds = []
|
|
seen = set()
|
|
for dirname in sorted(_SOUND_ASSET_DIRS):
|
|
root = scan_dir / dirname
|
|
if not root.exists() or not root.is_dir():
|
|
continue
|
|
for path in sorted(root.rglob("*"), key=lambda p: str(p).lower()):
|
|
if not path.is_file() or path.suffix.lower() not in _AUDIO_EXTS:
|
|
continue
|
|
try:
|
|
safe = _safe_child_path(scan_dir, path)
|
|
rel = str(safe.relative_to(scan_dir))
|
|
except Exception:
|
|
continue
|
|
if rel in seen:
|
|
continue
|
|
seen.add(rel)
|
|
try:
|
|
size = path.stat().st_size
|
|
except Exception:
|
|
size = None
|
|
sounds.append({
|
|
"path": rel,
|
|
"name": path.name,
|
|
"folder": str(path.parent.relative_to(scan_dir)),
|
|
"duration": None,
|
|
"size": size,
|
|
"type": path.suffix.lower().lstrip("."),
|
|
})
|
|
return {"sounds": sounds}
|
|
|
|
|
|
@router.get("/api/route-sounds/file/{sound_path:path}")
|
|
async def route_sound_file(sound_path: str):
|
|
from core.tts_helpers import _route_sound_path
|
|
try:
|
|
path = _route_sound_path(_load_settings(), sound_path)
|
|
except Exception:
|
|
raise HTTPException(404, "Sound not found")
|
|
if path is None:
|
|
raise HTTPException(404, "Sound not found")
|
|
mime = _AUDIO_MIME.get(path.suffix.lower(), "audio/wav")
|
|
return FileResponse(str(path), media_type=mime, filename=path.name)
|
|
|
|
|
|
@router.post("/api/route-sounds/upload")
|
|
async def upload_route_sound(file: UploadFile = File(...), name: str = Form("")):
|
|
from pydub import AudioSegment
|
|
|
|
suffix = Path(file.filename or "sound").suffix.lower() or ".bin"
|
|
if suffix not in _AUDIO_EXTS:
|
|
raise HTTPException(400, "Unsupported sound file type")
|
|
raw_name = Path(name or file.filename or "sound").stem.strip()
|
|
safe_name = re.sub(r"[^A-Za-z0-9_\-.]+", "_", raw_name).strip("._-")[:80] or f"sound_{uuid.uuid4().hex[:8]}"
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
sounds_dir = scan_dir / "sounds"
|
|
sounds_dir.mkdir(parents=True, exist_ok=True)
|
|
target = sounds_dir / f"{safe_name}.wav"
|
|
tmp = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}"
|
|
with tmp.open("wb") as f:
|
|
_copy_limited(file.file, f, _MAX_SOUND_BYTES)
|
|
try:
|
|
segment = AudioSegment.from_file(str(tmp)).set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
segment.export(str(target), format="wav")
|
|
duration = len(segment) / 1000.0
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Sound conversion failed: {e}")
|
|
finally:
|
|
try:
|
|
tmp.unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
return {"ok": True, "path": str(target.relative_to(scan_dir)), "filename": target.name, "duration": round(duration, 2)}
|
|
|
|
|
|
# ── Audio serving ─────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/audio/{fid}")
|
|
async def serve_audio(fid: str):
|
|
path = _registry_get(fid)
|
|
if path is None or not path.exists():
|
|
raise HTTPException(404, "File not found")
|
|
return FileResponse(str(path), media_type="audio/wav")
|
|
|
|
|
|
# ── Auto-trim / Process ───────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/auto-trim")
|
|
async def auto_trim(request: Request):
|
|
data = await request.json()
|
|
fid: str = data["id"]
|
|
src = _registry_get(fid)
|
|
if src is None or not src.exists():
|
|
raise HTTPException(404, "Source file not found")
|
|
try:
|
|
return _auto_trim_bounds(src)
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Auto trim failed: {e}")
|
|
|
|
|
|
@router.post("/api/process")
|
|
async def process(request: Request):
|
|
data = await request.json()
|
|
fid: str = data["id"]
|
|
start = float(data.get("start", 0))
|
|
end: float | None = data.get("end")
|
|
src = _registry_get(fid)
|
|
if src is None or not src.exists():
|
|
raise HTTPException(404, "Source file not found")
|
|
dur = _duration(src)
|
|
if end is None or end <= start:
|
|
end = dur
|
|
trimmed = _trim(src, start, end)
|
|
nid = uuid.uuid4().hex
|
|
_registry_put(nid, trimmed)
|
|
return {"id": nid, "duration": _duration(trimmed)}
|
|
|
|
|
|
# ── Save voice ────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/save")
|
|
async def save_voice(request: Request):
|
|
data = await request.json()
|
|
fid: str = data["id"]
|
|
voice_id: str = data["voice_id"].strip()
|
|
transcript: str = data.get("transcript", "").strip()
|
|
|
|
if not voice_id:
|
|
raise HTTPException(400, "Voice ID is required")
|
|
if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id):
|
|
raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .")
|
|
|
|
src = _registry_get(fid)
|
|
if src is None or not src.exists():
|
|
raise HTTPException(404, "Processed audio not found")
|
|
|
|
settings = _load_settings()
|
|
out_dir = _active_voices_dir(settings)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
wav_dest = out_dir / f"{voice_id}.wav"
|
|
txt_dest = out_dir / f"{voice_id}.reference.txt"
|
|
_remove_audio_variants(out_dir, voice_id)
|
|
loudness = _export_normalized_wav(src, wav_dest)
|
|
txt_dest.write_text(transcript, encoding="utf-8")
|
|
meta = _load_meta(wav_dest)
|
|
meta["enabled"] = True
|
|
meta["loudness"] = loudness
|
|
_save_meta(wav_dest, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav_dest)
|
|
return {"voice_id": voice_id, "wav": str(wav_dest), "txt": str(txt_dest), "loudness": loudness}
|
|
|
|
|
|
# ── Voice library CRUD ────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/voice-load")
|
|
@router.post("/api/voice/load")
|
|
async def load_voice_for_edit(request: Request):
|
|
data = await request.json()
|
|
voice_id: str = data.get("voice_id", "")
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
audio = _voice_audio_from_request(data, scan_dir)
|
|
if audio is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
|
|
try:
|
|
wav = _to_wav_24k(audio)
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Audio conversion failed: {e}")
|
|
|
|
fid = uuid.uuid4().hex
|
|
_registry_put(fid, wav)
|
|
ref = audio.with_suffix(".reference.txt")
|
|
return {
|
|
"id": fid,
|
|
"voice_id": voice_id or audio.stem,
|
|
"path": str(audio),
|
|
"duration": _duration(wav),
|
|
"transcript": ref.read_text(encoding="utf-8").strip() if ref.exists() else "",
|
|
"file_type": audio.suffix.lower().lstrip("."),
|
|
}
|
|
|
|
|
|
@router.post("/api/voice-replace")
|
|
@router.post("/api/voice/replace")
|
|
async def replace_voice_audio(request: Request):
|
|
data = await request.json()
|
|
fid: str = data["id"]
|
|
voice_id: str = data["voice_id"].strip()
|
|
transcript: str = data.get("transcript", "").strip()
|
|
|
|
if not voice_id:
|
|
raise HTTPException(400, "Voice ID is required")
|
|
if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id):
|
|
raise HTTPException(400, "Voice ID may only contain A-Z, 0-9, _, -, .")
|
|
|
|
src = _registry_get(fid)
|
|
if src is None or not src.exists():
|
|
raise HTTPException(404, "Processed audio not found")
|
|
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
current = _voice_audio_from_request(data, scan_dir)
|
|
if current is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
|
|
parent = current.parent
|
|
old_id = current.stem
|
|
meta = _load_meta(current)
|
|
wav_dest = parent / f"{voice_id}.wav"
|
|
txt_dest = parent / f"{voice_id}.reference.txt"
|
|
if voice_id != old_id:
|
|
old_files = {p.resolve() for p in _voice_package_paths(current)}
|
|
suffixes = _AUDIO_EXTS + [".reference.txt", ".meta.json"] + _PICTURE_EXTS
|
|
for sfx in suffixes:
|
|
target = parent / f"{voice_id}{sfx}"
|
|
if target.exists() and target.resolve() not in old_files:
|
|
raise HTTPException(409, f"Voice '{voice_id}' already exists")
|
|
|
|
backup = _backup_original_voice(current) if voice_id == old_id else None
|
|
_remove_audio_variants(parent, voice_id, keep=wav_dest)
|
|
loudness = _export_normalized_wav(src, wav_dest)
|
|
txt_dest.write_text(transcript, encoding="utf-8")
|
|
meta["loudness"] = loudness
|
|
if backup:
|
|
meta["original_backup"] = str(backup)
|
|
meta["needs_tts_restart"] = True
|
|
_save_meta(wav_dest, meta)
|
|
if voice_id != old_id:
|
|
keep = {wav_dest, txt_dest, wav_dest.with_suffix(".meta.json")}
|
|
for ext in _PICTURE_EXTS:
|
|
old_pic = parent / f"{old_id}{ext}"
|
|
if old_pic.exists():
|
|
new_pic = parent / f"{voice_id}{ext}"
|
|
shutil.copy2(str(old_pic), str(new_pic))
|
|
keep.add(new_pic)
|
|
_remove_voice_package(current, keep)
|
|
remove_voice_from_index(old_id)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav_dest)
|
|
return {"voice_id": voice_id, "wav": str(wav_dest), "txt": str(txt_dest),
|
|
"duration": _duration(wav_dest), "file_type": "wav", "loudness": loudness,
|
|
"backup": str(backup) if backup else None}
|
|
|
|
|
|
@router.get("/api/voices")
|
|
async def list_voices(refresh: bool = False):
|
|
settings = _load_settings()
|
|
return await asyncio.to_thread(indexed_voices, settings, refresh=refresh)
|
|
|
|
|
|
@router.get("/api/voices/index")
|
|
async def get_voice_index_info():
|
|
return voice_index_info()
|
|
|
|
|
|
@router.post("/api/voice/meta")
|
|
async def update_voice_meta(request: Request):
|
|
data = await request.json()
|
|
voice_id: str = data["voice_id"]
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
|
|
wav = _find_voice_audio(voice_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
|
|
meta = _load_meta(wav)
|
|
if "enabled" in data:
|
|
enabled = bool(data["enabled"])
|
|
wav = _move_voice_package(wav, _active_voices_dir(settings) if enabled else _hidden_voices_dir(settings))
|
|
meta = _load_meta(wav)
|
|
meta["enabled"] = enabled
|
|
|
|
for field in ("note", "rating", "flag", "gender", "loudness", "persona", "origin", "group",
|
|
"name", "tag", "avatar"):
|
|
if field in data:
|
|
meta[field] = data[field]
|
|
if "transcript" in data:
|
|
meta["needs_tts_restart"] = True
|
|
_save_meta(wav, meta)
|
|
if "transcript" in data:
|
|
wav.with_suffix(".reference.txt").write_text(data.get("transcript", "").strip(), encoding="utf-8")
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav)
|
|
return {"ok": True, "path": str(wav), "enabled": meta.get("enabled", True),
|
|
"transcript": wav.with_suffix(".reference.txt").read_text(encoding="utf-8").strip()
|
|
if wav.with_suffix(".reference.txt").exists() else ""}
|
|
|
|
|
|
@router.post("/api/voices/sync-folders")
|
|
async def sync_voice_folders():
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
active_dir = _active_voices_dir(settings)
|
|
hidden_dir = _hidden_voices_dir(settings)
|
|
moved = {"active": 0, "hidden": 0}
|
|
conflicts = []
|
|
|
|
if not scan_dir.exists():
|
|
return {"moved": moved, "conflicts": conflicts}
|
|
|
|
seen: set[str] = set()
|
|
all_audio = list(_voice_audio_files(scan_dir))
|
|
|
|
for p in sorted(all_audio, key=lambda x: x.stem.lower()):
|
|
if p.stem in seen or not p.exists():
|
|
continue
|
|
seen.add(p.stem)
|
|
meta = _load_meta(p)
|
|
enabled = meta.get("enabled") is not False
|
|
target_dir = active_dir if enabled else hidden_dir
|
|
try:
|
|
new_audio = _move_voice_package(p, target_dir)
|
|
meta = _load_meta(new_audio)
|
|
meta["enabled"] = enabled
|
|
_save_meta(new_audio, meta)
|
|
moved["active" if enabled else "hidden"] += int(new_audio != p)
|
|
except HTTPException as e:
|
|
conflicts.append({"voice_id": p.stem, "detail": e.detail})
|
|
|
|
await asyncio.to_thread(rebuild_voice_index, settings)
|
|
return {"moved": moved, "conflicts": conflicts,
|
|
"active_dir": str(active_dir), "hidden_dir": str(hidden_dir)}
|
|
|
|
|
|
@router.post("/api/voices/calculate-db")
|
|
async def calculate_voice_db():
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
results = []
|
|
errors = []
|
|
if not scan_dir.exists():
|
|
return {"calculated": 0, "errors": errors, "voices": results}
|
|
|
|
seen: set[str] = set()
|
|
all_audio = list(_voice_audio_files(scan_dir))
|
|
|
|
for audio in sorted(all_audio, key=lambda p: p.stem.lower()):
|
|
if audio.stem in seen:
|
|
continue
|
|
seen.add(audio.stem)
|
|
try:
|
|
loudness = _loudness_info(audio)
|
|
meta = _load_meta(audio)
|
|
meta["loudness"] = loudness
|
|
_save_meta(audio, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, audio)
|
|
results.append({"voice_id": audio.stem, "path": str(audio), "loudness": loudness})
|
|
except Exception as e:
|
|
errors.append({"voice_id": audio.stem, "detail": str(e)})
|
|
|
|
return {"calculated": len(results), "errors": errors, "voices": results,
|
|
"target_dbfs": _VOICE_TARGET_DBFS, "peak_dbfs": _VOICE_PEAK_DBFS}
|
|
|
|
|
|
@router.post("/api/voices/benchmark")
|
|
async def benchmark_voices(request: Request):
|
|
data = await request.json()
|
|
active_only = bool(data.get("active_only", True))
|
|
limit = int(data.get("limit") or 0)
|
|
sample_text = str(data.get("text") or "").strip()
|
|
sentences = [("sample", sample_text)] if sample_text else list(_BENCHMARK_SENTENCES)
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
active_dir = _active_voices_dir(settings)
|
|
results = []
|
|
errors = []
|
|
if not scan_dir.exists():
|
|
return {
|
|
"benchmarked": 0,
|
|
"errors": errors,
|
|
"voices": results,
|
|
"text": sample_text or "short / medium / long",
|
|
"sentences": [{"label": label, "text": sent_text} for label, sent_text in sentences],
|
|
"active_only": active_only,
|
|
"max_tts_output_seconds": _MAX_TTS_OUTPUT_SECONDS,
|
|
}
|
|
|
|
target_voice = str(data.get("voice_id") or "").strip()
|
|
seen: set[str] = set()
|
|
if target_voice:
|
|
audio = _find_voice_audio(target_voice, scan_dir)
|
|
if audio is None:
|
|
raise HTTPException(404, f"Voice '{target_voice}' not found")
|
|
all_audio = [audio]
|
|
else:
|
|
all_audio = list(_voice_audio_files(active_dir if active_only and active_dir.exists() else scan_dir))
|
|
|
|
for audio in sorted(all_audio, key=lambda p: p.stem.lower()):
|
|
if audio.stem in seen:
|
|
continue
|
|
seen.add(audio.stem)
|
|
if limit and len(results) >= limit:
|
|
break
|
|
benchmark = await asyncio.to_thread(_benchmark_voice, audio, settings, sentences)
|
|
try:
|
|
meta = _load_meta(audio)
|
|
meta["benchmark"] = benchmark
|
|
_save_meta(audio, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, audio)
|
|
except Exception as e:
|
|
errors.append({"voice_id": audio.stem, "detail": f"Could not save benchmark: {e}"})
|
|
item = {"voice_id": audio.stem, "path": str(audio), "benchmark": benchmark}
|
|
results.append(item)
|
|
if not benchmark.get("ok", False):
|
|
errors.append({"voice_id": audio.stem, "detail": benchmark.get("error", "Benchmark failed")})
|
|
|
|
return {
|
|
"benchmarked": len(results),
|
|
"errors": errors,
|
|
"voices": results,
|
|
"text": sample_text or "short / medium / long",
|
|
"sentences": [{"label": label, "text": sent_text} for label, sent_text in sentences],
|
|
"active_only": active_only,
|
|
"max_tts_output_seconds": _MAX_TTS_OUTPUT_SECONDS,
|
|
}
|
|
|
|
|
|
@router.post("/api/voices/normalize-active")
|
|
async def normalize_active_voices():
|
|
settings = _load_settings()
|
|
active_dir = _active_voices_dir(settings)
|
|
if not active_dir.exists():
|
|
return {"normalized": 0, "skipped": 0, "errors": [], "target_dbfs": _VOICE_TARGET_DBFS}
|
|
|
|
normalized = []
|
|
skipped = []
|
|
errors = []
|
|
for wav in sorted((p for p in active_dir.rglob("*.wav") if not _is_internal_voice_file(p)), key=lambda p: p.stem.lower()):
|
|
try:
|
|
tmp = wav.with_suffix(".normalized.tmp.wav")
|
|
loudness = _export_normalized_wav(wav, tmp)
|
|
shutil.move(str(tmp), str(wav))
|
|
meta = _load_meta(wav)
|
|
meta["enabled"] = True
|
|
meta["loudness"] = loudness
|
|
_save_meta(wav, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav)
|
|
normalized.append({"voice_id": wav.stem, **loudness})
|
|
except Exception as e:
|
|
errors.append({"voice_id": wav.stem, "detail": str(e)})
|
|
try:
|
|
tmp = wav.with_suffix(".normalized.tmp.wav")
|
|
if tmp.exists():
|
|
tmp.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
for ext in [e for e in _AUDIO_EXTS if e != ".wav"]:
|
|
skipped.extend(str(p) for p in active_dir.rglob(f"*{ext}") if not _is_internal_voice_file(p))
|
|
|
|
return {
|
|
"normalized": len(normalized),
|
|
"skipped": len(skipped),
|
|
"errors": errors,
|
|
"target_dbfs": _VOICE_TARGET_DBFS,
|
|
"peak_dbfs": _VOICE_PEAK_DBFS,
|
|
"voices": normalized,
|
|
}
|
|
|
|
|
|
@router.post("/api/voice/normalize")
|
|
async def normalize_voice(request: Request):
|
|
data = await request.json()
|
|
target_dbfs = float(data.get("target_dbfs", _VOICE_TARGET_DBFS))
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
audio = _voice_audio_from_request(data, scan_dir)
|
|
if audio is None:
|
|
raise HTTPException(404, "Voice not found")
|
|
if audio.suffix.lower() != ".wav":
|
|
raise HTTPException(400, "Only WAV voices can be normalized in place")
|
|
|
|
tmp = audio.with_suffix(".normalized.tmp.wav")
|
|
try:
|
|
loudness = _export_normalized_wav(audio, tmp, target_dbfs=target_dbfs)
|
|
shutil.move(str(tmp), str(audio))
|
|
meta = _load_meta(audio)
|
|
meta["loudness"] = loudness
|
|
meta["needs_tts_restart"] = True
|
|
_save_meta(audio, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, audio)
|
|
return {"ok": True, "voice_id": audio.stem, "path": str(audio),
|
|
"duration": _duration(audio), "file_type": "wav", "loudness": loudness}
|
|
except Exception as e:
|
|
if tmp.exists():
|
|
tmp.unlink()
|
|
raise HTTPException(400, f"Normalize failed: {e}")
|
|
|
|
|
|
@router.post("/api/voice/undo")
|
|
async def undo_voice_edit(request: Request):
|
|
data = await request.json()
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
current = _voice_audio_from_request(data, scan_dir)
|
|
if current is None:
|
|
raise HTTPException(404, "Voice not found")
|
|
meta = _load_meta(current)
|
|
backup = next((p for p in _backup_candidates(current, meta) if p.exists()), None)
|
|
if backup is None:
|
|
raise HTTPException(404, "No original backup found")
|
|
|
|
original_suffix = _backup_audio_suffix(backup, current.stem) or current.suffix.lower()
|
|
restored = current.with_suffix(original_suffix)
|
|
shutil.copy2(str(backup), str(restored))
|
|
if restored.resolve() != current.resolve():
|
|
_remove_audio_variants(current.parent, current.stem, keep=restored)
|
|
try:
|
|
meta["loudness"] = _loudness_info(restored)
|
|
except Exception:
|
|
meta.pop("loudness", None)
|
|
meta["needs_tts_restart"] = True
|
|
_save_meta(restored, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, restored)
|
|
return {
|
|
"ok": True,
|
|
"voice_id": restored.stem,
|
|
"path": str(restored),
|
|
"duration": _duration(restored),
|
|
"file_type": restored.suffix.lower().lstrip("."),
|
|
"loudness": meta.get("loudness", {}),
|
|
}
|
|
|
|
|
|
# ── Voice rename ──────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/voice/rename")
|
|
async def rename_voice(request: Request):
|
|
data = await request.json()
|
|
old_id: str = data["old_id"]
|
|
new_id: str = data["new_id"].strip()
|
|
|
|
if not re.match(r"^[A-Za-z0-9_\-\.]+$", new_id):
|
|
raise HTTPException(400, "Invalid voice ID characters")
|
|
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(old_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, f"Voice '{old_id}' not found")
|
|
|
|
parent = wav.parent
|
|
if any((parent / f"{new_id}{ext}").exists() for ext in _AUDIO_EXTS):
|
|
raise HTTPException(409, f"Voice '{new_id}' already exists")
|
|
|
|
from core.voice import _backup_path, _legacy_backup_path
|
|
meta = _load_meta(wav)
|
|
backup = next((p for p in _backup_candidates(wav, meta) if p.exists()), None)
|
|
new_audio = parent / f"{new_id}{wav.suffix.lower()}"
|
|
new_backup = _backup_path(new_audio)
|
|
if backup and backup.exists() and backup.resolve() != new_backup.resolve() and new_backup.exists():
|
|
raise HTTPException(409, f"Backup already exists for '{new_id}'")
|
|
|
|
suffixes = _AUDIO_EXTS + [".reference.txt", ".meta.json"] + _PICTURE_EXTS
|
|
for sfx in suffixes:
|
|
src = parent / f"{old_id}{sfx}"
|
|
if src.exists():
|
|
src.rename(parent / f"{new_id}{sfx}")
|
|
|
|
if backup and backup.exists():
|
|
if backup.resolve() != new_backup.resolve():
|
|
backup.rename(new_backup)
|
|
|
|
# Also rename TTS config assets if mounted
|
|
tts_config = Path("/tts-config")
|
|
if tts_config.exists():
|
|
old_pt = tts_config / "speakers" / f"{old_id}.pt"
|
|
if old_pt.exists():
|
|
old_pt.rename(tts_config / "speakers" / f"{new_id}.pt")
|
|
|
|
old_seeds = tts_config / "seed_samples" / old_id
|
|
if old_seeds.exists() and old_seeds.is_dir():
|
|
old_seeds.rename(tts_config / "seed_samples" / new_id)
|
|
|
|
new_meta = _load_meta(new_audio)
|
|
if backup and backup.exists():
|
|
new_meta["original_backup"] = str(new_backup)
|
|
new_meta["needs_tts_restart"] = True
|
|
_save_meta(new_audio, new_meta)
|
|
remove_voice_from_index(old_id)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, new_audio)
|
|
return {"new_id": new_id, "path": str(new_audio), "file_type": new_audio.suffix.lower().lstrip(".")}
|
|
|
|
|
|
# ── Voice delete ──────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/voices/delete-group")
|
|
async def delete_voice_group(request: Request):
|
|
"""Delete every voice whose meta `group` matches the given name. Used to bulk-remove
|
|
a rehearser play's auto-designed voices."""
|
|
data = await request.json()
|
|
group = str(data.get("group") or "").strip()
|
|
if not group:
|
|
raise HTTPException(400, "No group specified")
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
deleted_ids = []
|
|
for p in _voice_audio_files(scan_dir):
|
|
meta = _load_meta(p)
|
|
if str(meta.get("group") or "").strip() == group:
|
|
for f in _voice_package_paths(p):
|
|
if f.exists():
|
|
f.unlink()
|
|
deleted_ids.append(p.stem)
|
|
remove_voice_from_index(p.stem)
|
|
return {"deleted": deleted_ids, "count": len(deleted_ids)}
|
|
|
|
|
|
@router.delete("/api/voice/{voice_id}")
|
|
async def delete_voice(voice_id: str):
|
|
if not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id):
|
|
raise HTTPException(400, "Invalid voice ID")
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(voice_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
deleted = []
|
|
for f in _voice_package_paths(wav):
|
|
if f.exists():
|
|
f.unlink()
|
|
deleted.append(f.name)
|
|
remove_voice_from_index(voice_id)
|
|
return {"deleted": deleted}
|
|
|
|
|
|
# ── Voice picture upload / serve ──────────────────────────────────────────────
|
|
|
|
@router.post("/api/voice/picture")
|
|
async def upload_picture(voice_id: str = Form(...), file: UploadFile = File(...)):
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(voice_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
|
|
orig_suffix = Path(file.filename or "photo.jpg").suffix.lower()
|
|
if orig_suffix not in _PICTURE_EXTS:
|
|
orig_suffix = ".jpg"
|
|
|
|
for ext in _PICTURE_EXTS:
|
|
old = wav.with_suffix(ext)
|
|
if old.exists():
|
|
old.unlink()
|
|
|
|
dest = wav.with_suffix(orig_suffix)
|
|
with dest.open("wb") as f:
|
|
_copy_limited(file.file, f, _MAX_PICTURE_BYTES)
|
|
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav)
|
|
return {"ok": True, "path": str(dest)}
|
|
|
|
|
|
@router.post("/api/voice/picture-url")
|
|
async def upload_picture_url(request: Request):
|
|
import requests as req_lib
|
|
from urllib.parse import urlparse
|
|
|
|
data = await request.json()
|
|
voice_id = str(data.get("voice_id") or "").strip()
|
|
image_url = str(data.get("image_url") or "").strip()
|
|
if not image_url or not re.match(r"^https?://", image_url, re.I):
|
|
raise HTTPException(400, "Expected an http(s) URL")
|
|
if not voice_id or not re.match(r"^[A-Za-z0-9_\-\.]+$", voice_id):
|
|
raise HTTPException(400, "Invalid voice ID")
|
|
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(voice_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, f"Voice '{voice_id}' not found")
|
|
|
|
suffix = Path(urlparse(image_url).path).suffix.lower()
|
|
if suffix not in _PICTURE_EXTS:
|
|
suffix = ".jpg"
|
|
dest = wav.with_suffix(suffix)
|
|
_VOICE_SOURCE_HEADERS = {
|
|
"User-Agent": "TTS Voice Creator voice-source-scraper/1.0 (+local)",
|
|
"Accept": "text/html,application/json,text/plain;q=0.9,*/*;q=0.8",
|
|
}
|
|
try:
|
|
with req_lib.get(image_url, headers=_VOICE_SOURCE_HEADERS, timeout=20, stream=True) as r:
|
|
r.raise_for_status()
|
|
content_type = (r.headers.get("content-type") or "").split(";", 1)[0].lower()
|
|
if content_type and not content_type.startswith("image/"):
|
|
raise HTTPException(400, "Image URL did not return an image")
|
|
for ext in _PICTURE_EXTS:
|
|
old = wav.with_suffix(ext)
|
|
if old.exists():
|
|
old.unlink()
|
|
total = 0
|
|
with dest.open("wb") as f:
|
|
for chunk in r.iter_content(256 * 1024):
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > _MAX_PICTURE_BYTES:
|
|
raise HTTPException(413, "Image file is too large")
|
|
f.write(chunk)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Image import failed: {e}")
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav)
|
|
return {"ok": True, "voice_id": voice_id, "path": str(dest)}
|
|
|
|
|
|
@router.get("/api/voice/picture/{voice_id}")
|
|
async def serve_picture(voice_id: str):
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(voice_id, scan_dir)
|
|
if wav is None:
|
|
raise HTTPException(404, "Voice not found")
|
|
pic = _picture_path(wav)
|
|
if pic is None:
|
|
raise HTTPException(404, "No picture")
|
|
return FileResponse(str(pic), media_type=_picture_mime(pic))
|
|
|
|
|
|
@router.get("/api/voice-file")
|
|
async def voice_file(path: str):
|
|
settings = _load_settings()
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
p = _safe_child_path(scan_dir, Path(path))
|
|
if not p.exists() or not p.is_file():
|
|
raise HTTPException(404, "File not found")
|
|
if p.suffix.lower() not in _AUDIO_EXTS:
|
|
raise HTTPException(404, "File not found")
|
|
mime = _AUDIO_MIME.get(p.suffix.lower(), "audio/wav")
|
|
return FileResponse(
|
|
str(p),
|
|
media_type=mime,
|
|
headers={"Cache-Control": "no-store, max-age=0"},
|
|
)
|