"""Persona group store: groups hold a sale kit + personas + report. A group is created by an admin from a setup form + optional files. After analyze, it contains `personas` (15 by default = 5 per tier) and a `report`. Groups are editable/re-analyzeable by admins. Trainees only read revealable views of personas and run one-shot sessions (sessions are stored separately). """ from __future__ import annotations import datetime from pathlib import Path from typing import Any from ..storage.store import JsonStore, new_id from .store import ensure_persona_shape DEFAULT_TIERS = ["A", "B", "C"] PERSONAS_PER_TIER = 5 def _now() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat() class GroupStore: def __init__(self, data_dir: Path) -> None: self.groups = JsonStore(data_dir / "groups") def create(self, *, org_id: str, creator_id: str, title: str) -> dict[str, Any]: gid = new_id("group") group = { "id": gid, "org_id": org_id, "creator_id": creator_id, "title": title or "Untitled group", "status": "draft", # draft -> analyzing -> ready | failed "created_at": _now(), "updated_at": _now(), "input": {}, # form fields "sales_kit": None, "personas": [], # full persona dicts "report": None, "error": None, } return self.groups.create(group, key=gid) def get(self, gid: str) -> dict[str, Any]: return self.groups.get(gid) def get_or_none(self, gid: str) -> dict[str, Any] | None: return self.groups.get_or_none(gid) def update(self, gid: str, **fields: Any) -> dict[str, Any]: fields.setdefault("updated_at", _now()) return self.groups.update(gid, **fields) def list_for_org(self, org_id: str | None = None) -> list[dict[str, Any]]: groups = self.groups.all() if org_id: groups = [g for g in groups if g.get("org_id") == org_id] return groups def list_visible_to( self, *, role: str, org_id: str | None = None ) -> list[dict[str, Any]]: """List groups a given role/user can see. Trainees see only ready groups.""" groups = self.groups.all() if org_id: groups = [g for g in groups if g.get("org_id") == org_id] if role == "user": groups = [g for g in groups if g.get("status") == "ready"] return groups # ── personas ──────────────────────────────────────────────────────── def set_personas(self, gid: str, personas: list[dict[str, Any]]) -> dict[str, Any]: personas = [ensure_persona_shape(p) for p in personas] return self.groups.update(gid, personas=personas) def get_persona(self, gid: str, pid: str) -> dict[str, Any] | None: group = self.get(gid) for p in group.get("personas", []): if p.get("id") == pid: return p return None def update_persona(self, gid: str, pid: str, patch: dict[str, Any]) -> dict[str, Any]: group = self.get(gid) found = False for i, p in enumerate(group.get("personas", [])): if p.get("id") == pid: merged = {**p, **patch, "id": pid} group["personas"][i] = ensure_persona_shape(merged) found = True break if not found: raise ValueError("persona not found") return self.groups.replace(gid, group) def delete(self, gid: str) -> None: """Hard-delete a group (personas/report included).""" self.groups.delete(gid)