Create voices.py

voices.py

@@ -0,0 +1,66 @@
+import os
+import json
+import requests
+from pathlib import Path
+
+HF_VOICES_JSON_URL = os.getenv("HF_VOICES_JSON_URL", "")
+HF_VOICES_BASE_URL = os.getenv("HF_VOICES_BASE_URL", "")
+LOCAL_VOICES_DIR = Path(os.getenv("LOCAL_VOICES_DIR", "/tmp/voices"))
+
+LOCAL_VOICES_DIR.mkdir(parents=True, exist_ok=True)
+
+class VoiceRegistry:
+    def __init__(self):
+        self._voices = None
+
+    def load(self):
+        if self._voices is not None:
+            return self._voices
+        if not HF_VOICES_JSON_URL:
+            raise RuntimeError("HF_VOICES_JSON_URL is not set")
+
+        r = requests.get(HF_VOICES_JSON_URL, timeout=60)
+        r.raise_for_status()
+        self._voices = r.json()
+        return self._voices
+
+    def list_voice_ids(self):
+        v = self.load()
+        return sorted(list(v.keys()))
+
+    def ensure_voice(self, voice_id: str) -> dict:
+        voices = self.load()
+        if voice_id not in voices:
+            raise KeyError(f"Unknown voiceId: {voice_id}")
+
+        meta = voices[voice_id]
+        rel_model = meta["model"]  # e.g. "voices/en_US-amy.onnx"
+        rel_json = meta.get("config")  # optional: "voices/en_US-amy.onnx.json"
+
+        model_path = LOCAL_VOICES_DIR / Path(rel_model).name
+        if not model_path.exists():
+            if not HF_VOICES_BASE_URL:
+                raise RuntimeError("HF_VOICES_BASE_URL is not set")
+            url = f"{HF_VOICES_BASE_URL.rstrip('/')}/{rel_model}"
+            self._download(url, model_path)
+
+        config_path = None
+        if rel_json:
+            config_path = LOCAL_VOICES_DIR / Path(rel_json).name
+            if not config_path.exists():
+                url = f"{HF_VOICES_BASE_URL.rstrip('/')}/{rel_json}"
+                self._download(url, config_path)
+
+        return {
+            "voiceId": voice_id,
+            "modelPath": str(model_path),
+            "configPath": str(config_path) if config_path else None,
+            **meta,
+        }
+
+    def _download(self, url: str, dst: Path):
+        r = requests.get(url, timeout=120)
+        r.raise_for_status()
+        dst.write_bytes(r.content)
+
+registry = VoiceRegistry()