feat(xtts): Voice Export/Import auf der Gamebox

Zwei neue RVS-Handler in der F5-TTS-Bridge:

  xtts_export_voice
    Packt <name>.wav + <name>.txt aus VOICES_DIR als tar.gz in-memory,
    sendet base64-codiert als xtts_voice_exported zurueck. Diagnostic
    baut daraus den Browser-Download.

  xtts_import_voice
    Empfaengt base64 tar.gz mit <name>.wav (+ optional <name>.txt),
    legt sie in VOICES_DIR ab (sanitized name, Path-Traversal-Schutz),
    sendet xtts_voice_imported zurueck. Anschliessend handle_list_voices
    damit App/Diagnostic die neue Stimme sofort sehen.

So koennen Stimmen zwischen mehreren Gameboxen wandern, ohne die
WAV-Files manuell rumzukopieren.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-11 22:24:35 +02:00
parent d16dcd34cc
commit bfa06d78a7
+74
View File
@@ -661,6 +661,76 @@ async def handle_delete_voice(ws, payload: dict) -> None:
logger.exception("handle_delete_voice Fehler")
async def handle_export_voice(ws, payload: dict) -> None:
"""Packt eine Stimme (.wav + .txt) als tar.gz und sendet sie base64 zurueck."""
name = (payload.get("name") or "").strip()
if not name:
await _send(ws, "xtts_voice_exported", {"ok": False, "error": "name fehlt"})
return
try:
wav, txt = voice_paths(name)
if not wav.exists():
await _send(ws, "xtts_voice_exported", {"ok": False, "name": name, "error": "Stimme nicht gefunden"})
return
import io, tarfile
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
tar.add(wav, arcname=wav.name)
if txt.exists():
tar.add(txt, arcname=txt.name)
data = base64.b64encode(buf.getvalue()).decode("ascii")
logger.info("Voice exportiert: %s (%d KB tar.gz)", name, len(buf.getvalue()) // 1024)
await _send(ws, "xtts_voice_exported", {"ok": True, "name": name, "data": data})
except Exception as e:
logger.exception("handle_export_voice Fehler")
await _send(ws, "xtts_voice_exported", {"ok": False, "name": name, "error": str(e)[:200]})
async def handle_import_voice(ws, payload: dict) -> None:
"""Empfaengt eine tar.gz mit <name>.wav (+ optional <name>.txt) und legt
sie in VOICES_DIR ab. Ueberschreibt bestehende Stimme gleichen Namens."""
name = (payload.get("name") or "").strip()
data_b64 = payload.get("data") or ""
if not name or not data_b64:
await _send(ws, "xtts_voice_imported", {"ok": False, "error": "name/data fehlt"})
return
try:
import io, tarfile
VOICES_DIR.mkdir(parents=True, exist_ok=True)
safe = sanitize_voice_name(name)
data = base64.b64decode(data_b64)
extracted_wav = False
with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
for member in tar.getmembers():
if not member.isfile():
continue
base = Path(member.name).name # Path-Traversal verhindern
if base.lower().endswith(".wav"):
target = VOICES_DIR / f"{safe}.wav"
f = tar.extractfile(member)
if f is None:
continue
with open(target, "wb") as out:
out.write(f.read())
extracted_wav = True
elif base.lower().endswith(".txt"):
target = VOICES_DIR / f"{safe}.txt"
f = tar.extractfile(member)
if f is None:
continue
with open(target, "wb") as out:
out.write(f.read())
if not extracted_wav:
await _send(ws, "xtts_voice_imported", {"ok": False, "name": name, "error": "Kein .wav im Archiv"})
return
logger.info("Voice importiert: %s", name)
await _send(ws, "xtts_voice_imported", {"ok": True, "name": name})
await handle_list_voices(ws)
except Exception as e:
logger.exception("handle_import_voice Fehler")
await _send(ws, "xtts_voice_imported", {"ok": False, "name": name, "error": str(e)[:200]})
# Letzte diagnostisch-gesetzte Voice (verhindert Endlos-Preload bei jedem config)
_last_diag_voice = ""
@@ -781,6 +851,10 @@ async def run_loop(runner: F5Runner) -> None:
asyncio.create_task(handle_list_voices(ws))
elif mtype == "xtts_delete_voice":
asyncio.create_task(handle_delete_voice(ws, payload))
elif mtype == "xtts_export_voice":
asyncio.create_task(handle_export_voice(ws, payload))
elif mtype == "xtts_import_voice":
asyncio.create_task(handle_import_voice(ws, payload))
elif mtype == "voice_preload":
asyncio.create_task(handle_voice_preload(ws, payload, runner))
elif mtype == "stt_response":