File size: 5,106 Bytes
f9609df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """
participants.py
Local participant records for InsightUX's multi-participant tracking.
A participant is NOT an InsightUX login/profile (see auth.py) — it's a
tracking SUBJECT that an owner creates to keep another person's sessions
separate from their own, e.g. a UX researcher running sessions on Rahul,
Priya, Aman. Participants never authenticate and always live nested inside
their owner's own folder (users/<owner_id>/participants/<id>/), not as a
sibling top-level directory — so a participant's data cannot structurally
exist outside their owner's folder at all.
One profile.json per participant folder, not a shared index file — a
participant folder is always a complete, self-contained, movable/deletable
unit, the same way users/<owner_id>/ already is relative to DATA_DIR.
No heavy imports (stdlib only) — same reasoning as theme.py/auth.py.
"""
import os
import json
import shutil
import uuid
from datetime import datetime, timezone
class ParticipantError(Exception):
"""User-facing failures (missing name, not found, ...)."""
def participant_dir(owner_dir, participant_id):
return os.path.join(owner_dir, "participants", participant_id)
def _profile_path(owner_dir, participant_id):
return os.path.join(participant_dir(owner_dir, participant_id), "profile.json")
def _save_profile(owner_dir, participant_id, record):
path = _profile_path(owner_dir, participant_id)
os.makedirs(os.path.dirname(path), exist_ok=True)
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(record, f, indent=2)
os.replace(tmp, path) # atomic on POSIX and Windows
def create_participant(owner_dir, name, notes=""):
name = (name or "").strip()
if not name:
raise ParticipantError("Name is required.")
participant_id = uuid.uuid4().hex[:12]
record = {
"id": participant_id,
"name": name,
"notes": (notes or "").strip(),
"created_at": datetime.now(timezone.utc).isoformat(),
"session_count": 0,
"last_session_at": None,
}
_save_profile(owner_dir, participant_id, record)
return record
def get_participant(owner_dir, participant_id):
if not participant_id:
return None
path = _profile_path(owner_dir, participant_id)
if not os.path.exists(path):
return None
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return None
def list_participants(owner_dir):
"""Sorted by name — same predictable-picker convention as
auth.list_profiles()."""
root = os.path.join(owner_dir, "participants")
out = []
if os.path.isdir(root):
for name in os.listdir(root):
if not os.path.isdir(os.path.join(root, name)):
continue
record = get_participant(owner_dir, name)
if record:
out.append(record)
out.sort(key=lambda p: p["name"].lower())
return out
def update_participant(owner_dir, participant_id, name=None, notes=None):
"""In-place edit — name/notes are only touched when explicitly passed
(None means "leave as-is"), so a caller updating just one field can't
accidentally blank out the other. id/created_at/session_count/
last_session_at are never touched here; renaming a participant must
never look like a new one to anything reading session_count."""
record = get_participant(owner_dir, participant_id)
if not record:
raise ParticipantError("That participant no longer exists.")
if name is not None:
name = name.strip()
if not name:
raise ParticipantError("Name is required.")
record["name"] = name
if notes is not None:
record["notes"] = notes.strip()
_save_profile(owner_dir, participant_id, record)
return record
def delete_participant(owner_dir, participant_id):
"""Removes the participant's entire folder — profile.json, sessions/,
calibration.pkl, everything nested under it. Irreversible; the caller
(Api.delete_participant()) is expected to have already confirmed with
the user and to have refused this while that participant is the
active tracking subject mid-session."""
if not participant_id:
raise ParticipantError("No participant specified.")
pdir = participant_dir(owner_dir, participant_id)
if not os.path.isdir(pdir):
raise ParticipantError("That participant no longer exists.")
shutil.rmtree(pdir)
def touch_session_stats(owner_dir, participant_id):
"""Called once a session for this participant actually starts — keeps
session_count/last_session_at current without needing to re-scan every
session folder just to show them in the picker/details view later."""
record = get_participant(owner_dir, participant_id)
if not record:
return
record["session_count"] = record.get("session_count", 0) + 1
record["last_session_at"] = datetime.now(timezone.utc).isoformat()
_save_profile(owner_dir, participant_id, record)
|