From 23033384c79b8b1e181336210fbfc429296685b8 Mon Sep 17 00:00:00 2001 From: alf Date: Fri, 19 Jun 2026 16:20:55 +0200 Subject: [PATCH] =?UTF-8?q?feat(voice):=20verrou=20voix=E2=86=94profil=20+?= =?UTF-8?q?=20sync=20verrou-conscient=20+=20d=C3=A9clencheur=20auto?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - store : colonne locked_profile_id (+ migration ALTER idempotente), lock_voice/ unlock_voice. Verrou = assignation exclusive à un profil ; NULL = voix généraliste. - orchestrator : list_voices expose locked_profile_id + used_by_profiles (croise profile.voice_id) ; sync_device enrôle+déploie les voix en attente, REFUSE de déployer une voix verrouillée sur une tablette sans le profil propriétaire. - autosync : VoiceAutoSync — watcher de connexions armable depuis l'UI (désarmé par défaut), statut par tablette, ne fait rien si store verrouillé (PII), delete opt-in. - API : lock/unlock, POST /{serial}/sync, GET/POST /voices/autosync (déclarées AVANT /{serial} pour éviter la capture "autosync" comme serial). - tests : +8 (verrou, sync verrou-conscient, autosync). 39/39. Co-Authored-By: Claude Opus 4.8 (1M context) --- kazeia_central/api/app.py | 52 ++++++++++++++- kazeia_central/store/db.py | 26 ++++++++ kazeia_central/voice/autosync.py | 97 ++++++++++++++++++++++++++++ kazeia_central/voice/orchestrator.py | 50 +++++++++++++- tests/test_api_store.py | 2 +- tests/test_fleet.py | 2 +- tests/test_voice.py | 64 +++++++++++++++++- 7 files changed, 288 insertions(+), 5 deletions(-) create mode 100644 kazeia_central/voice/autosync.py diff --git a/kazeia_central/api/app.py b/kazeia_central/api/app.py index 90bd993..c4406ef 100644 --- a/kazeia_central/api/app.py +++ b/kazeia_central/api/app.py @@ -26,6 +26,7 @@ from ..fleet import device_overview, fan_out from ..provider import ProviderClient from ..store import BadPassword, Store, StoreLocked from ..voice import orchestrator as vo +from ..voice.autosync import VoiceAutoSync from ..voice.bridge import VoiceBridge, VoiceWorkerError _WEB_DIR = Path(__file__).resolve().parent.parent / "web" @@ -52,6 +53,19 @@ class _Enroll(BaseModel): delete_source: bool = False # suppression du WAV device = opt-in explicite +class _Sync(BaseModel): + delete_source: bool = False + + +class _AutoSync(BaseModel): + enabled: bool + delete_source: bool | None = None + + +class _Lock(BaseModel): + profile_id: str + + def _install_basic_auth(app: FastAPI) -> None: """Auth HTTP Basic optionnelle, active SEULEMENT si KAZEIA_USER+KAZEIA_PASS sont définis. Couvre tout (UI statique incluse) via middleware. Pensée pour @@ -78,12 +92,15 @@ def _install_basic_auth(app: FastAPI) -> None: def create_app(adb: Adb | None = None, store: Store | None = None, - bridge: VoiceBridge | None = None) -> FastAPI: + bridge: VoiceBridge | None = None, *, start_autosync: bool = True) -> FastAPI: adb = adb or Adb() if store is None: _STORE_PATH.parent.mkdir(parents=True, exist_ok=True) store = Store(_STORE_PATH) bridge = bridge or VoiceBridge() + autosync = VoiceAutoSync(adb, store, bridge) + if start_autosync: + autosync.start() # désarmé par défaut : ne fait que suivre les connexions app = FastAPI(title="Kazeia-central", version="0.0.1") _install_basic_auth(app) @@ -214,6 +231,20 @@ def create_app(adb: Adb | None = None, store: Store | None = None, def voices_health(): return {"encoder_available": bridge.available(), "cv_python": bridge.cv_python} + # ⚠️ Routes littérales sous /api/voices/ AVANT /{serial} sinon "autosync" est + # capté comme un serial (FastAPI matche dans l'ordre de déclaration). + @app.get("/api/voices/autosync") + def autosync_status(): + return autosync.status() + + @app.post("/api/voices/autosync") + def autosync_set(body: _AutoSync): + autosync.set_enabled(body.enabled, delete_source=body.delete_source) + store.audit("autosync_set", actor=_OPERATOR, + detail={"enabled": body.enabled, "delete_source": autosync.delete_source}, + now=_now_ms()) + return autosync.status() + @app.get("/api/voices/{serial}") def voices_list(serial: str): return _guard(lambda: vo.list_voices(adb, store, serial)) @@ -230,6 +261,25 @@ def create_app(adb: Adb | None = None, store: Store | None = None, adb, store, bridge, serial, voice_id, body.transcription, delete_source=body.delete_source, now=_now_ms(), actor=_OPERATOR)) + @app.post("/api/voices/{serial}/{voice_id}/lock") + def voices_lock(serial: str, voice_id: str, body: _Lock): + _guard(lambda: store.lock_voice(serial, voice_id, body.profile_id, now=_now_ms())) + store.audit("voice_lock", actor=_OPERATOR, target=f"{serial}/{voice_id}", + detail={"profile_id": body.profile_id}, now=_now_ms()) + return {"voice_id": voice_id, "locked_profile_id": body.profile_id} + + @app.post("/api/voices/{serial}/{voice_id}/unlock") + def voices_unlock(serial: str, voice_id: str): + _guard(lambda: store.unlock_voice(serial, voice_id, now=_now_ms())) + store.audit("voice_unlock", actor=_OPERATOR, target=f"{serial}/{voice_id}", now=_now_ms()) + return {"voice_id": voice_id, "locked_profile_id": None} + + @app.post("/api/voices/{serial}/sync") + def voices_sync(serial: str, body: _Sync): + return _guard(lambda: vo.sync_device( + adb, store, bridge, serial, delete_source=body.delete_source, + now=_now_ms(), actor=_OPERATOR)) + # UI web locale servie à la racine — APRÈS les routes /api pour ne pas les # masquer. `html=True` → sert index.html sur "/". if _WEB_DIR.is_dir(): diff --git a/kazeia_central/store/db.py b/kazeia_central/store/db.py index a8c0330..47bea8a 100644 --- a/kazeia_central/store/db.py +++ b/kazeia_central/store/db.py @@ -78,6 +78,7 @@ CREATE TABLE IF NOT EXISTS voices ( enrolled_at INTEGER, -- .cvps produit deployed_at INTEGER, -- .cvps poussé sur la tablette wav_deleted_at INTEGER, -- WAV supprimé de la tablette + locked_profile_id TEXT, -- assignation exclusive à un profil patient (NULL = généraliste) PRIMARY KEY (serial, voice_id) ); """ @@ -102,9 +103,17 @@ class Store: self._db.execute("PRAGMA journal_mode=WAL") self._db.execute("PRAGMA foreign_keys=ON") self._db.executescript(_SCHEMA) + self._migrate() self._db.commit() self._vault: Vault | None = None + def _migrate(self) -> None: + """Migrations légères idempotentes (CREATE TABLE IF NOT EXISTS ne fait pas + évoluer une table déjà créée).""" + cols = {r["name"] for r in self._db.execute("PRAGMA table_info(voices)")} + if "locked_profile_id" not in cols: + self._db.execute("ALTER TABLE voices ADD COLUMN locked_profile_id TEXT") + def close(self) -> None: with self._lock: self._db.close() @@ -330,6 +339,23 @@ class Store: with self._lock: self._voice_upsert(serial, voice_id, wav_deleted_at=now) + def lock_voice(self, serial: str, voice_id: str, profile_id: str, *, now: int) -> None: + """Assigne une voix exclusivement à un profil patient (verrou).""" + with self._lock: + self._voice_upsert(serial, voice_id, locked_profile_id=profile_id) + + def unlock_voice(self, serial: str, voice_id: str, *, now: int) -> None: + """Lève le verrou → voix redevient généraliste.""" + with self._lock: + self._voice_upsert(serial, voice_id, locked_profile_id=None) + + def locked_profile(self, serial: str, voice_id: str) -> str | None: + with self._lock: + row = self._db.execute( + "SELECT locked_profile_id FROM voices WHERE serial=? AND voice_id=?", + (serial, voice_id)).fetchone() + return row["locked_profile_id"] if row else None + def voice_record(self, serial: str, voice_id: str) -> dict[str, Any] | None: v = self._require_vault() with self._lock: diff --git a/kazeia_central/voice/autosync.py b/kazeia_central/voice/autosync.py new file mode 100644 index 0000000..3e11c37 --- /dev/null +++ b/kazeia_central/voice/autosync.py @@ -0,0 +1,97 @@ +"""Déclencheur automatique d'enrôlement à la connexion d'une tablette. + +Piloté depuis l'interface : démarre **désarmé** (`enabled=False`). Quand l'opérateur +l'arme, toute tablette qui se **connecte** déclenche un `sync_device` (enrôle + +déploie les voix en attente, verrou-conscient). Statut par tablette consultable. + +Sécurité : ne fait rien si le store est verrouillé (les PII restent protégées) ; la +suppression du WAV device est opt-in (`delete_source`, défaut False). Détecte les +**nouvelles** connexions ; pour traiter une tablette déjà branchée, l'UI utilise le +bouton « synchroniser » (endpoint sync manuel). +""" + +from __future__ import annotations + +import threading +import time +from typing import Any + +from .orchestrator import sync_device + + +class VoiceAutoSync: + def __init__(self, adb, store, bridge, *, interval: float = 3.0) -> None: + self.adb = adb + self.store = store + self.bridge = bridge + self.interval = interval + self.enabled = False + self.delete_source = False + self._known: set[str] = set() + self._status: dict[str, dict] = {} + self._lock = threading.Lock() + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + @staticmethod + def _now() -> int: + return int(time.time() * 1000) + + # ---- cycle de vie ----------------------------------------------------- + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._loop, name="voice-autosync", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + + def set_enabled(self, enabled: bool, *, delete_source: bool | None = None) -> None: + self.enabled = bool(enabled) + if delete_source is not None: + self.delete_source = bool(delete_source) + + def status(self) -> dict[str, Any]: + with self._lock: + return {"enabled": self.enabled, "delete_source": self.delete_source, + "running": bool(self._thread and self._thread.is_alive()), + "devices": dict(self._status)} + + # ---- boucle ----------------------------------------------------------- + def _loop(self) -> None: + while not self._stop.wait(self.interval): + try: + ready = {d.serial for d in self.adb.ready_devices()} + except Exception: + continue + new = ready - self._known + self._known = ready + if not self.enabled: + continue + for serial in sorted(new): + self.sync_now(serial) + + def sync_now(self, serial: str) -> dict: + """Lance une sync pour une tablette (utilisé par le watcher ET par le bouton + manuel). Ne lève pas : renvoie/enregistre un statut.""" + if not self.store.is_unlocked: + st = {"state": "store_verrouille", "at": self._now()} + self._set(serial, st) + return st + self._set(serial, {"state": "sync_en_cours", "at": self._now()}) + try: + rep = sync_device(self.adb, self.store, self.bridge, serial, + delete_source=self.delete_source, now=self._now(), actor="auto") + ok = sum(1 for r in rep if r.get("ok")) + st = {"state": "termine", "at": self._now(), "enrolled": ok, + "total": len(rep), "report": rep} + except Exception as e: + st = {"state": "erreur", "at": self._now(), "error": str(e)} + self._set(serial, st) + return st + + def _set(self, serial: str, st: dict) -> None: + with self._lock: + self._status[serial] = st diff --git a/kazeia_central/voice/orchestrator.py b/kazeia_central/voice/orchestrator.py index bb01e77..ef8862a 100644 --- a/kazeia_central/voice/orchestrator.py +++ b/kazeia_central/voice/orchestrator.py @@ -47,16 +47,33 @@ def _status(deployed: bool, rec: dict) -> str: return "pending" +def _profiles_using(adb: Adb, serial: str) -> tuple[dict[str, list[str]], set[str]]: + """Renvoie ({voice_id: [profil_id,...]}, {profils présents}). Croise profile.voice_id.""" + used: dict[str, list[str]] = {} + present: set[str] = set() + try: + for p in ProviderClient(adb, serial).profiles(): + present.add(p.id) + if p.voice_id: + used.setdefault(p.voice_id, []).append(p.id) + except Exception: + pass + return used, present + + def list_voices(adb: Adb, store: Store, serial: str) -> list[dict[str, Any]]: - """Inventaire voix de la tablette croisé avec l'archive store (statut par voix).""" + """Inventaire voix de la tablette croisé avec l'archive store (statut, verrou, + profils utilisateurs) par voix.""" rows = _provider_voices(adb, serial) cosy = transfer.cosyvoice_dir_for(rows[0].wav_path) if rows else None deployed = transfer.deployed_cvps(adb, serial, cosy) if cosy else set() recs = {r["voice_id"]: r for r in store.voices(serial)} if store.is_unlocked else {} + used_by, _ = _profiles_using(adb, serial) out = [] for v in rows: rec = recs.get(v.id, {}) is_dep = v.id in deployed + locked = rec.get("locked_profile_id") out.append({ "voice_id": v.id, "wav_path": v.wav_path, @@ -67,11 +84,42 @@ def list_voices(adb: Adb, store: Store, serial: str) -> list[dict[str, Any]]: "wav_deleted": bool(rec.get("wav_deleted_at")), "transcription": rec.get("transcription"), "language": rec.get("language"), + "locked_profile_id": locked, # exclusif à ce profil, ou None = généraliste + "used_by_profiles": used_by.get(v.id, []), # profils qui sélectionnent cette voix "status": _status(is_dep, rec), }) return out +def sync_device(adb: Adb, store: Store, bridge: VoiceBridge, serial: str, *, + delete_source: bool, now: int, actor: str | None = None) -> list[dict[str, Any]]: + """Enrôle + déploie les voix en attente d'une tablette (rapport par voix). + Verrou-conscient : ne déploie pas une voix exclusive sur une tablette dépourvue + du profil propriétaire. Transcrit automatiquement si pas encore relu (la relecture + fine reste possible voix par voix via prepare/enroll).""" + _, present = _profiles_using(adb, serial) + report: list[dict[str, Any]] = [] + for v in list_voices(adb, store, serial): + vid = v["voice_id"] + if v["deployed_on_device"]: + continue + lock = v["locked_profile_id"] + if lock and lock not in present: + report.append({"voice_id": vid, "ok": False, + "skipped": "verrou_profil_absent", "profile": lock}) + continue + try: + text = v["transcription"] + if not text: + text = prepare(adb, store, bridge, serial, vid, now=now, actor=actor)["transcription"] + res = enroll_deploy(adb, store, bridge, serial, vid, text, + delete_source=delete_source, now=now, actor=actor) + report.append({"voice_id": vid, "ok": True, **res}) + except Exception as e: + report.append({"voice_id": vid, "ok": False, "error": str(e)}) + return report + + def _ensure_wav_archived(adb: Adb, store: Store, serial: str, voice_id: str, *, now: int) -> bytes: """Garantit que le WAV est archivé (chiffré) côté PC ; le renvoie déchiffré.""" diff --git a/tests/test_api_store.py b/tests/test_api_store.py index 62faa4a..abfb73c 100644 --- a/tests/test_api_store.py +++ b/tests/test_api_store.py @@ -22,7 +22,7 @@ class _FakeAdb(Adb): def _client(tmp_path): store = Store(tmp_path / "central.db", kdf_ops=_OPS, kdf_mem=_MEM) - return TestClient(create_app(adb=_FakeAdb(), store=store)) + return TestClient(create_app(adb=_FakeAdb(), store=store, start_autosync=False)) def test_lock_unlock_flow(tmp_path): diff --git a/tests/test_fleet.py b/tests/test_fleet.py index 418a4f9..71fa4d1 100644 --- a/tests/test_fleet.py +++ b/tests/test_fleet.py @@ -80,7 +80,7 @@ def test_fleet_overview_endpoint_merges_labels(tmp_path): store = Store(tmp_path / "c.db", kdf_ops=nacl.pwhash.argon2id.OPSLIMIT_MIN, kdf_mem=nacl.pwhash.argon2id.MEMLIMIT_MIN) - c = TestClient(create_app(adb=_ReadyAdb(["D1"], _CANNED), store=store)) + c = TestClient(create_app(adb=_ReadyAdb(["D1"], _CANNED), store=store, start_autosync=False)) c.post("/api/unlock", json={"password": "pw"}) c.put("/api/devices/D1/label", json={"label": "Mme D."}) rep = c.get("/api/fleet/overview").json() diff --git a/tests/test_voice.py b/tests/test_voice.py index 0d07926..bd0e695 100644 --- a/tests/test_voice.py +++ b/tests/test_voice.py @@ -67,10 +67,11 @@ def test_voice_status_marks(tmp_path): # ---- orchestration (faux adb + faux bridge) ------------------------------ class _VoiceAdb(Adb): - def __init__(self, voices, deployed=()): + def __init__(self, voices, deployed=(), profiles=()): super().__init__() self._voices = voices # [(id, wav_path, wav_exists)] self._deployed = list(deployed) + self._profiles = list(profiles) # [(profile_id, voice_id)] self.pushed = [] self.deleted = [] @@ -80,6 +81,8 @@ class _VoiceAdb(Adb): def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True): if path == "voices": return [{"id": i, "wav_path": w, "wav_exists": 1} for i, w, _ in self._voices] + if path == "profiles": + return [{"id": pid, "voice_id": vid} for pid, vid in self._profiles] return [] def shell(self, command, *, serial=None, timeout=None): @@ -145,3 +148,62 @@ def test_enroll_deploy_pushes_and_optional_delete(tmp_path): delete_source=True, now=NOW) assert r2["wav_deleted"] is True and adb.deleted == [_WP] assert s.voice_record("D1", "amir")["wav_deleted_at"] == NOW + + +# ---- verrou voix ↔ profil (assignation exclusive + généralistes) --------- +def test_lock_unlock_voice(tmp_path): + s = _store(tmp_path) + s.archive_voice_wav("D1", "amir", b"x", source_wav_path=None, now=NOW) + s.lock_voice("D1", "amir", "patient_07", now=NOW) + assert s.voice_record("D1", "amir")["locked_profile_id"] == "patient_07" + s.unlock_voice("D1", "amir", now=NOW) + assert s.voice_record("D1", "amir")["locked_profile_id"] is None + + +def test_list_voices_shows_lock_and_users(tmp_path): + s = _store(tmp_path) + adb = _VoiceAdb([("amir", _WP, 1)], profiles=[("p1", "amir"), ("p2", "amir")]) + s.archive_voice_wav("D1", "amir", b"x", source_wav_path=_WP, now=NOW) + s.lock_voice("D1", "amir", "p1", now=NOW) + v = vo.list_voices(adb, s, "D1")[0] + assert v["locked_profile_id"] == "p1" + assert set(v["used_by_profiles"]) == {"p1", "p2"} + + +def test_sync_skips_voice_locked_to_absent_profile(tmp_path): + s = _store(tmp_path) + # voix verrouillée sur un profil ABSENT de la tablette → non déployée + adb = _VoiceAdb([("amir", _WP, 1)], profiles=[("p_present", "x")]) + s.archive_voice_wav("D1", "amir", b"WAV", source_wav_path=_WP, now=NOW) + s.lock_voice("D1", "amir", "p_absent", now=NOW) + rep = vo.sync_device(adb, s, _FakeBridge(), "D1", delete_source=False, now=NOW) + assert rep[0]["ok"] is False and rep[0]["skipped"] == "verrou_profil_absent" + assert adb.pushed == [] # rien poussé + + +def test_sync_deploys_general_voice(tmp_path): + s = _store(tmp_path) + adb = _VoiceAdb([("amir", _WP, 1)]) # généraliste (non verrouillée) + rep = vo.sync_device(adb, s, _FakeBridge(), "D1", delete_source=False, now=NOW) + assert rep[0]["ok"] is True and adb.pushed + assert s.voice_record("D1", "amir")["deployed_at"] == NOW + + +# ---- déclencheur auto ----------------------------------------------------- +def test_autosync_blocked_when_store_locked(tmp_path): + from kazeia_central.store import Store + from kazeia_central.voice.autosync import VoiceAutoSync + locked_store = Store(tmp_path / "locked.db", kdf_ops=_OPS, kdf_mem=_MEM) # PAS unlock + a = VoiceAutoSync(_VoiceAdb([("amir", _WP, 1)]), locked_store, _FakeBridge()) + st = a.sync_now("D1") + assert st["state"] == "store_verrouille" + + +def test_autosync_enabled_runs_sync(tmp_path): + from kazeia_central.voice.autosync import VoiceAutoSync + s = _store(tmp_path) + a = VoiceAutoSync(_VoiceAdb([("amir", _WP, 1)]), s, _FakeBridge()) + a.set_enabled(True, delete_source=False) + assert a.status()["enabled"] is True + st = a.sync_now("D1") + assert st["state"] == "termine" and st["enrolled"] == 1