| |
| """Build the public catalog and hydrate artifact sizes and checksums.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as source: |
| for block in iter(lambda: source.read(1024 * 1024), b""): |
| digest.update(block) |
| return digest.hexdigest() |
|
|
|
|
| def hydrate(value: Any) -> Any: |
| if isinstance(value, list): |
| return [hydrate(item) for item in value] |
| if not isinstance(value, dict): |
| return value |
| if "$artifact" in value: |
| if len(value) != 1 or not isinstance(value["$artifact"], dict): |
| raise ValueError("$artifact must be the only key and contain an object") |
| artifact = dict(value["$artifact"]) |
| relative_path = Path(artifact["path"]) |
| if relative_path.is_absolute() or ".." in relative_path.parts: |
| raise ValueError(f"Artifact path must stay inside the repository: {relative_path}") |
| path = ROOT / relative_path |
| if not path.is_file(): |
| raise FileNotFoundError(path) |
| artifact["sizeBytes"] = path.stat().st_size |
| artifact["sha256"] = sha256(path) |
| return artifact |
| return {key: hydrate(item) for key, item in value.items()} |
|
|
|
|
| template = json.loads((ROOT / "catalog.template.json").read_text(encoding="utf-8")) |
|
|
| upstream_inventory = json.loads( |
| (ROOT / "sources/upstream-assets.json").read_text(encoding="utf-8") |
| ) |
| voices_by_language: dict[str, list[dict[str, Any]]] = {} |
| for source_voice in upstream_inventory["voices"]: |
| voices_by_language.setdefault(source_voice["languageId"], []).append( |
| { |
| "id": source_voice["id"], |
| "displayName": source_voice["displayName"], |
| "gender": source_voice["gender"], |
| "modelId": source_voice["modelId"], |
| "artifact": { |
| "$artifact": { |
| "path": source_voice["artifactPath"], |
| "mediaType": "application/octet-stream", |
| } |
| }, |
| } |
| ) |
|
|
| for language in template["languages"]: |
| if language.pop("$includeUpstreamVoices", False): |
| upstream_voices = voices_by_language.pop(language["id"], []) |
| language["voices"].extend(upstream_voices) |
| language["voices"].sort(key=lambda voice: voice["id"]) |
|
|
| if voices_by_language: |
| raise RuntimeError( |
| f"Upstream voices refer to unknown languages: {sorted(voices_by_language)}" |
| ) |
|
|
| catalog = hydrate(template) |
| (ROOT / "catalog.json").write_text( |
| json.dumps(catalog, ensure_ascii=False, indent=2) + "\n", |
| encoding="utf-8", |
| ) |
| voice_count = sum(len(language["voices"]) for language in catalog["languages"]) |
| print(f"{ROOT / 'catalog.json'} ({len(catalog['languages'])} profiles, {voice_count} voices)") |
|
|
