79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
"""Ingestion des voix enregistrées via l'app admin (manifestes record-time).
|
|
|
|
VOICE_DEPLOYMENT_SPEC §4.2 : une voix fraîchement enregistrée + son manifeste vivent
|
|
dans le **stockage ADMIN** `Android/data/com.kazeia.admin/files/voix/`
|
|
(`VoiceStorage.kt`), PAS dans le provider patient `/voices`. C'est là que `scope`
|
|
(exclusive|global|pending), `owner_profile_id` et `reference_text` (phrase de
|
|
consentement) vivent ensemble. Kazeia-central doit ingérer DEPUIS cette source.
|
|
|
|
Py3.14-safe (adb seulement). Layout (source : `VoiceStorage.save()`) :
|
|
<ADMIN_VOIX_DIR>/<id>.wav + <ADMIN_VOIX_DIR>/<id>.json
|
|
manifeste : {id, name, state, wav, wav_size_bytes, duration_seconds, created_at,
|
|
reference_text, source_device, scope, owner_profile_id?, owner_name?}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import posixpath
|
|
import tempfile
|
|
from typing import Any
|
|
|
|
from kazeia_central.core.adb import Adb
|
|
|
|
# getExternalFilesDir("voix") de com.kazeia.admin. /sdcard = /storage/emulated/0.
|
|
ADMIN_VOIX_DIR = os.environ.get(
|
|
"KAZEIA_ADMIN_VOIX_DIR", "/sdcard/Android/data/com.kazeia.admin/files/voix")
|
|
|
|
VALID_SCOPES = ("exclusive", "global", "pending")
|
|
|
|
|
|
def _normalize(m: dict, *, dir_path: str) -> dict[str, Any]:
|
|
vid = m.get("id")
|
|
scope = m.get("scope") or "pending"
|
|
if scope not in VALID_SCOPES:
|
|
scope = "pending"
|
|
wav = m.get("wav") or (f"{vid}.wav" if vid else None)
|
|
return {
|
|
"id": vid,
|
|
"name": m.get("name") or vid,
|
|
"scope": scope,
|
|
"owner_profile_id": m.get("owner_profile_id"),
|
|
"owner_name": m.get("owner_name"),
|
|
"reference_text": m.get("reference_text") or "",
|
|
"duration_seconds": m.get("duration_seconds"),
|
|
"created_at": m.get("created_at"),
|
|
"source_device": m.get("source_device"),
|
|
"wav_device_path": posixpath.join(dir_path, wav) if wav else None,
|
|
"manifest_device_path": posixpath.join(dir_path, f"{vid}.json") if vid else None,
|
|
}
|
|
|
|
|
|
def list_manifests(adb: Adb, serial: str, *, dir_path: str = ADMIN_VOIX_DIR) -> list[dict[str, Any]]:
|
|
"""Liste + parse les manifestes voix du stockage admin (rien de lourd : on ne
|
|
pull que les `.json`, pas les WAV). Voix sans manifeste valide = ignorée."""
|
|
out = adb.shell(f"ls {dir_path} 2>/dev/null", serial=serial)
|
|
names = [ln.strip() for ln in out.splitlines() if ln.strip().endswith(".json")]
|
|
voices: list[dict[str, Any]] = []
|
|
with tempfile.TemporaryDirectory() as td:
|
|
for n in names:
|
|
local = os.path.join(td, n)
|
|
try:
|
|
adb.pull(posixpath.join(dir_path, n), local, serial=serial)
|
|
with open(local, encoding="utf-8") as f:
|
|
m = json.load(f)
|
|
except Exception:
|
|
continue # manifeste illisible/corrompu → on saute, pas de crash global
|
|
if m.get("id"):
|
|
voices.append(_normalize(m, dir_path=dir_path))
|
|
return voices
|
|
|
|
|
|
def delete_admin_voice(adb: Adb, serial: str, voice: dict, *, dir_path: str = ADMIN_VOIX_DIR) -> None:
|
|
"""Supprime WAV + manifeste du stockage admin (après archivage côté PC)."""
|
|
for key in ("wav_device_path", "manifest_device_path"):
|
|
p = voice.get(key)
|
|
if p:
|
|
adb.shell(f"rm -f {p}", serial=serial)
|