my_voice_assistant_v3_jamulix/scripts/download_voices.py

#!/usr/bin/env python3
"""Lädt die für das jamulix-Deployment benötigten Piper-Stimmen herunter.

Prüft welche Dateien bereits vorhanden sind und lädt nur fehlende nach.

Aufruf:
    python scripts/download_voices.py
    python scripts/download_voices.py --voices-dir /pfad/zu/voices
    python scripts/download_voices.py --force   # vorhandene überschreiben
"""
import argparse
import sys
import urllib.request
from pathlib import Path

BASE_URL = "https://huggingface.co/rhasspy/piper-voices/resolve/v1.0.0"

# Alle für dieses Deployment benötigten Piper-Stimmen.
# Abgeleitet aus PIPER_VOICE_GENDERED + LANG_TO_PIPER_VOICE in app/dependencies.py.
# Bei Änderungen dort diese Liste synchron halten.
# Stimmen mit '#speaker_id'-Suffix (ES sharvard) sind Multi-Speaker-Modelle —
# nur eine Datei für beide Geschlechter, kein separater Download nötig.
REQUIRED_VOICES = [
    "de_DE-thorsten-high",    # DE ♂
    "de_DE-kerstin-low",      # DE ♀
    "en_US-lessac-high",      # EN ♂
    "en_US-amy-medium",       # EN ♀
    "fr_FR-siwis-medium",     # FR ♀
    "fr_FR-tom-medium",       # FR ♂
    "es_ES-sharvard-medium",  # ES ♂+♀  (Multi-Speaker: voice#0=♂, voice#1=♀)
    "it_IT-paola-medium",     # IT ♀
    "it_IT-riccardo-x_low",   # IT ♂
    "pt_BR-faber-medium",     # PT ♂    (kein ♀ in piper-voices verfügbar)
    "pl_PL-gosia-medium",     # PL ♀
    "pl_PL-darkman-medium",   # PL ♂
    "ar_JO-kareem-medium",    # AR ♂    (kein ♀ in piper-voices verfügbar)
    "ru_RU-irina-medium",     # RU ♀
    "ru_RU-ruslan-medium",    # RU ♂
    "zh_CN-huayan-medium",    # ZH ♀    (kein ♂ in piper-voices verfügbar)
]

_QUALITY_SUFFIXES = {"high", "medium", "low", "x_low"}


def _parse_voice_name(name: str) -> tuple[str, str, str, str]:
    """Zerlegt z. B. 'de_DE-kerstin-low' → (lang_family, lang_code, speaker, quality)."""
    name = name.split("#")[0]
    parts = name.split("-")
    lang_code = parts[0]                          # z. B. "de_DE"
    lang_family = lang_code.split("_")[0].lower() # z. B. "de"
    # Qualität kann aus zwei Teilen bestehen (x_low)
    if len(parts) >= 3 and f"{parts[-2]}_{parts[-1]}" in _QUALITY_SUFFIXES:
        quality = f"{parts[-2]}_{parts[-1]}"      # z. B. "x_low"
        speaker = "-".join(parts[1:-2])
    else:
        quality = parts[-1]                       # z. B. "high"
        speaker = "-".join(parts[1:-1])
    return lang_family, lang_code, speaker, quality


def _hf_url(voice_name: str, ext: str) -> str:
    lang_family, lang_code, speaker, quality = _parse_voice_name(voice_name)
    path = f"{lang_family}/{lang_code}/{speaker}/{quality}/{voice_name}{ext}"
    return f"{BASE_URL}/{path}"


def _download(url: str, dest: Path) -> None:
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".tmp")

    def _progress(block_num, block_size, total_size):
        if total_size > 0:
            pct = min(100, block_num * block_size * 100 // total_size)
            bar = "█" * (pct // 5) + "░" * (20 - pct // 5)
            mb = total_size / 1_048_576
            print(f"\r    [{bar}] {pct:3d}%  ({mb:.0f} MB)", end="", flush=True)

    try:
        urllib.request.urlretrieve(url, tmp, _progress)
        tmp.rename(dest)
        print(f"\r    ✓  {dest.name}            ")
    except Exception:
        tmp.unlink(missing_ok=True)
        raise


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--voices-dir", default="config/voices",
                        help="Zielverzeichnis (Standard: config/voices)")
    parser.add_argument("--force", action="store_true",
                        help="Vorhandene Dateien überschreiben")
    args = parser.parse_args()

    voices_dir = Path(args.voices_dir)

    to_download = []
    already_present = []
    for voice in REQUIRED_VOICES:
        onnx  = voices_dir / f"{voice}.onnx"
        json_ = voices_dir / f"{voice}.onnx.json"
        if args.force or not onnx.exists() or not json_.exists():
            to_download.append(voice)
        else:
            already_present.append(voice)

    print(f"Stimmen-Verzeichnis : {voices_dir.resolve()}")
    print(f"Bereits vorhanden   : {len(already_present)}/{len(REQUIRED_VOICES)}")
    print(f"Herunterzuladen     : {len(to_download)}")

    if not to_download:
        print("\nAlle Stimmen sind bereits vorhanden.")
        return

    print()
    errors: list[tuple[str, str, str]] = []
    for i, voice in enumerate(to_download, 1):
        print(f"[{i}/{len(to_download)}] {voice}")
        for ext in (".onnx", ".onnx.json"):
            url  = _hf_url(voice, ext)
            dest = voices_dir / f"{voice}{ext}"
            if dest.exists() and not args.force:
                print(f"    ✓  {dest.name} (vorhanden)")
                continue
            try:
                _download(url, dest)
            except Exception as e:
                print(f"\n    ✗  FEHLER: {e}")
                errors.append((voice, ext, str(e)))

    print()
    if errors:
        print(f"{len(errors)} Fehler beim Download:")
        for voice, ext, msg in errors:
            print(f"  {voice}{ext}: {msg}")
        sys.exit(1)
    else:
        print(f"Fertig — {len(to_download)} Stimme(n) heruntergeladen.")


if __name__ == "__main__":
    main()