6a2e6a326f
Records written before the visibility field existed carry none; the new fail-closed authorization treated missing visibility as invalid, making every legacy group unlistable and unreadable. Add resolved_visibility(group) that derives effective visibility for legacy records only (owner present => private, absent => public), leaves explicit-malformed visibility fail-closed (None), and never derives demo/hidden. Apply it at every list, authorization, chat, and analytics boundary while keeping demo and hidden-preview paths raw and owner_user_id-based private isolation intact. No persisted data is rewritten. Backend full suite passes 517; frontend 26/26; production build passes.
925 lines
33 KiB
Python
925 lines
33 KiB
Python
"""Admin analytics: aggregate trainee results (supports date-range filter)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import io
|
|
import math
|
|
from typing import Callable, Iterator, cast
|
|
from flask import Blueprint, g, jsonify, request
|
|
|
|
from ..auth.users import AuthError, normalize_identifier
|
|
from ..config import Config
|
|
from ..services.groups import is_ready_group, is_valid_owner_visibility, resolved_visibility, validated_personas
|
|
from ..storage.store import StoreError
|
|
from .helpers import ApiError, current_user, is_valid_tenant_id, require_auth, require_roles
|
|
|
|
analytics_bp = Blueprint("analytics", __name__)
|
|
|
|
|
|
def _stores():
|
|
from flask import current_app
|
|
|
|
return {
|
|
"sessions": current_app.extensions["session_store"],
|
|
"groups": current_app.extensions["group_store"],
|
|
"users": current_app.extensions["user_store"],
|
|
}
|
|
|
|
|
|
def _is_shared_group(group: object) -> bool:
|
|
"""Accept only structurally valid tenant-shared group records."""
|
|
return (
|
|
isinstance(group, dict)
|
|
and is_valid_tenant_id(group.get("id"))
|
|
and is_valid_tenant_id(group.get("org_id"))
|
|
and "owner_user_id" not in group
|
|
and is_valid_owner_visibility(group)
|
|
and resolved_visibility(group) in {"public", "hidden"}
|
|
and is_ready_group(group)
|
|
)
|
|
|
|
|
|
def _shared_ready_contexts(groups: object) -> dict[str, tuple[str, set[str]]]:
|
|
"""Map ready shared groups to their tenant and current persona IDs."""
|
|
contexts: dict[str, tuple[str, set[str]]] = {}
|
|
if not isinstance(groups, list):
|
|
return contexts
|
|
for group in groups:
|
|
if not _is_shared_group(group):
|
|
continue
|
|
gid = group.get("id")
|
|
org_id = group.get("org_id")
|
|
if not isinstance(gid, str) or not isinstance(org_id, str):
|
|
continue
|
|
persona_ids = {
|
|
persona["id"]
|
|
for persona in validated_personas(group.get("personas"))
|
|
if isinstance(persona.get("id"), str)
|
|
}
|
|
if persona_ids:
|
|
contexts[gid] = (org_id, persona_ids)
|
|
return contexts
|
|
|
|
|
|
def _canonical_persona_ids(value: object) -> set[str]:
|
|
ids: set[str] = set()
|
|
if not isinstance(value, list):
|
|
return ids
|
|
for persona in value:
|
|
if not isinstance(persona, dict):
|
|
continue
|
|
pid = persona.get("id")
|
|
if isinstance(pid, str) and pid.strip():
|
|
ids.add(pid.strip())
|
|
return ids
|
|
|
|
|
|
def _is_legacy_trainee_mode(value: object) -> bool:
|
|
"""Accept missing mode for legacy rows, but reject malformed falsey values."""
|
|
return value is None or value == "trainee"
|
|
|
|
|
|
def _session_matches_context(
|
|
session: object,
|
|
contexts: dict[str, tuple[str, set[str]]],
|
|
*,
|
|
org_id: str | None = None,
|
|
) -> bool:
|
|
if not isinstance(session, dict):
|
|
return False
|
|
gid = session.get("group_id")
|
|
pid = session.get("persona_id")
|
|
session_org = session.get("org_id")
|
|
if not (
|
|
is_valid_tenant_id(gid)
|
|
and is_valid_tenant_id(pid)
|
|
and is_valid_tenant_id(session_org)
|
|
):
|
|
return False
|
|
context = contexts.get(gid)
|
|
if context is None:
|
|
return False
|
|
group_org, persona_ids = context
|
|
return (
|
|
is_valid_tenant_id(group_org)
|
|
and session_org == group_org
|
|
and (org_id is None or group_org == org_id)
|
|
and pid in persona_ids
|
|
)
|
|
|
|
|
|
def _log_audit(action: str, subject: str, *, detail: dict | None = None) -> None:
|
|
import json
|
|
import time
|
|
|
|
from ..config import Config
|
|
from .helpers import current_user as _cu
|
|
|
|
actor = _cu()
|
|
entry = {
|
|
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"actor": (actor or {}).get("username") or (actor or {}).get("id"),
|
|
"action": action,
|
|
"subject": subject,
|
|
"detail": detail or {},
|
|
}
|
|
try:
|
|
log_dir = Config.DATA_DIR / "audit"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
with open(log_dir / "audit.jsonl", "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _csv_cell(value: object, *, max_chars: int | None = None) -> str:
|
|
"""Export bounded scalar values without serializing persisted structures."""
|
|
if value is None:
|
|
text = ""
|
|
elif isinstance(value, str):
|
|
text = value
|
|
elif isinstance(value, bool):
|
|
text = ""
|
|
elif isinstance(value, int):
|
|
text = str(value)
|
|
elif isinstance(value, float) and math.isfinite(value):
|
|
text = str(value)
|
|
else:
|
|
text = ""
|
|
if max_chars is not None:
|
|
text = text[:max_chars]
|
|
probe = text
|
|
while probe and (probe[0].isspace() or ord(probe[0]) < 32 or probe[0] == "\ufeff"):
|
|
probe = probe[1:]
|
|
if probe.startswith(("=", "+", "-", "@")):
|
|
return "'" + text
|
|
return text
|
|
|
|
|
|
def _debrief_score(session: dict, default: int | float = 0) -> int | float:
|
|
raw_debrief = session.get("debrief")
|
|
if not isinstance(raw_debrief, dict):
|
|
return default
|
|
score = raw_debrief.get("score", default)
|
|
if isinstance(score, bool) or not isinstance(score, (int, float)):
|
|
return default
|
|
try:
|
|
if not math.isfinite(score):
|
|
return default
|
|
return max(0, min(100, score))
|
|
except (OverflowError, TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _is_finished_outcome(session: dict) -> bool:
|
|
return (
|
|
isinstance(session, dict)
|
|
and session.get("status") == "finished"
|
|
and session.get("outcome") in {"won", "lost"}
|
|
)
|
|
|
|
|
|
def _top_weak_area_label(sessions: list[dict]) -> str | None:
|
|
"""Return the dominant weak-area dimension label for a trainee's sessions.
|
|
|
|
Runs the per-user weak-area analysis over an already user-scoped, finished
|
|
trainee session list and returns the highest-evidence dimension label.
|
|
"""
|
|
from ..services.trainee import analyze_weak_areas
|
|
|
|
try:
|
|
user_ids = {s.get("user_id") for s in sessions if isinstance(s, dict) and isinstance(s.get("user_id"), str)}
|
|
if not user_ids:
|
|
return None
|
|
uid = next(iter(user_ids))
|
|
if not isinstance(uid, str):
|
|
return None
|
|
org_ids = {s.get("org_id") for s in sessions if isinstance(s, dict) and isinstance(s.get("org_id"), str)}
|
|
org_id = next(iter(org_ids)) if org_ids else ""
|
|
analysis = analyze_weak_areas(sessions, user_id=uid, org_id=org_id or "org-default")
|
|
except Exception:
|
|
return None
|
|
dimensions = analysis.get("dimensions") or {}
|
|
best_key = None
|
|
best_count = 0
|
|
for key, dim in dimensions.items():
|
|
if not isinstance(dim, dict):
|
|
continue
|
|
try:
|
|
count = len(dim.get("session_ids") or [])
|
|
except TypeError:
|
|
count = 0
|
|
if count > best_count:
|
|
best_count = count
|
|
best_key = key
|
|
if best_key is None:
|
|
return None
|
|
dim = dimensions.get(best_key) or {}
|
|
label = dim.get("label") if isinstance(dim, dict) else None
|
|
return label or best_key
|
|
|
|
|
|
def _is_safe_export_identity(value: object) -> bool:
|
|
if not isinstance(value, str):
|
|
return False
|
|
try:
|
|
return normalize_identifier(value) == value
|
|
except AuthError:
|
|
return False
|
|
|
|
|
|
def _is_safe_export_token_id(value: object) -> bool:
|
|
if not isinstance(value, str):
|
|
return False
|
|
try:
|
|
normalize_identifier(value)
|
|
return True
|
|
except AuthError:
|
|
return False
|
|
|
|
|
|
# Short-lived signed link for CSV export (HMAC-SHA256, expiring). Not the long-lived JWT.
|
|
import hmac
|
|
import hashlib
|
|
import secrets
|
|
import time as _t
|
|
|
|
_EXPORT_TTL = 300 # 5 minutes
|
|
|
|
|
|
def _sign_export_token(actor: dict) -> str:
|
|
from ..config import Config
|
|
|
|
user_id = actor.get("id") or actor.get("username") or ""
|
|
org_id = actor.get("org_id") or ""
|
|
role = actor.get("role") or ""
|
|
if (
|
|
not _is_safe_export_identity(user_id)
|
|
or role not in Config.ROLES
|
|
or (role != "super_admin" and not is_valid_tenant_id(org_id))
|
|
or (role == "super_admin" and org_id and not is_valid_tenant_id(org_id))
|
|
):
|
|
raise ApiError("invalid export authorization", 403)
|
|
expires_at = int(_t.time()) + _EXPORT_TTL
|
|
jti = secrets.token_urlsafe(18)
|
|
payload = f"{user_id}:{org_id}:{role}:{expires_at}:{jti}"
|
|
_export_token_store().create(
|
|
{
|
|
"id": jti,
|
|
"user_id": user_id,
|
|
"org_id": org_id,
|
|
"role": role,
|
|
"expires_at": expires_at,
|
|
"used": False,
|
|
},
|
|
key=jti,
|
|
)
|
|
sig = hmac.new(Config.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
|
return "{}.{}".format(sig, payload)
|
|
|
|
|
|
def _export_token_store():
|
|
from ..config import Config
|
|
from ..storage.store import JsonStore
|
|
|
|
return JsonStore(Config.DATA_DIR / "export_tokens")
|
|
|
|
|
|
def _verify_export_token(token: str) -> dict | None:
|
|
from ..config import Config
|
|
|
|
try:
|
|
sig, payload = token.split(".", 1)
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
expected = hmac.new(Config.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(sig, expected):
|
|
return None
|
|
parts = payload.split(":")
|
|
if len(parts) != 5:
|
|
return None
|
|
user_id, org_id, role, exp_s, jti = parts
|
|
if (
|
|
not _is_safe_export_identity(user_id)
|
|
or not _is_safe_export_token_id(jti)
|
|
or role not in Config.ROLES
|
|
or (role != "super_admin" and not is_valid_tenant_id(org_id))
|
|
or (role == "super_admin" and org_id and not is_valid_tenant_id(org_id))
|
|
):
|
|
return None
|
|
try:
|
|
expires_at = int(exp_s)
|
|
if expires_at < _t.time():
|
|
return None # expired
|
|
except ValueError:
|
|
return None
|
|
from ..storage.store import StoreError
|
|
|
|
try:
|
|
record = _export_token_store().get_or_none(jti)
|
|
except (OSError, StoreError, TypeError, UnicodeError, ValueError):
|
|
return None
|
|
if (
|
|
record is None
|
|
or not isinstance(record, dict)
|
|
or record.get("used") is not False
|
|
or record.get("user_id") != user_id
|
|
or record.get("org_id") != org_id
|
|
or record.get("role") != role
|
|
or record.get("expires_at") != expires_at
|
|
):
|
|
return None
|
|
return {
|
|
"org_id": org_id or None,
|
|
"user_id": user_id or None,
|
|
"role": role,
|
|
"active": True,
|
|
"jti": jti,
|
|
"expires_at": expires_at,
|
|
}
|
|
|
|
|
|
def _consume_export_token(token_actor: dict) -> bool:
|
|
"""Atomically consume a previously authorized signed export token."""
|
|
if not isinstance(token_actor, dict):
|
|
return False
|
|
jti = token_actor.get("jti")
|
|
expires_at = token_actor.get("expires_at")
|
|
if not isinstance(jti, str) or not isinstance(expires_at, int):
|
|
return False
|
|
try:
|
|
consumed = _export_token_store().update_if(
|
|
jti,
|
|
lambda row: (
|
|
row.get("used") is False
|
|
and row.get("user_id") == token_actor.get("user_id")
|
|
and row.get("org_id") == (token_actor.get("org_id") or "")
|
|
and row.get("role") == token_actor.get("role")
|
|
and row.get("expires_at") == expires_at
|
|
and expires_at >= int(_t.time())
|
|
),
|
|
used=True,
|
|
used_at=int(_t.time()),
|
|
)
|
|
except (OSError, StoreError, TypeError, UnicodeError, ValueError, AttributeError):
|
|
return False
|
|
return consumed is not None
|
|
|
|
|
|
def _authenticated_export_actor():
|
|
"""Reuse the normal Bearer auth/role guards for the non-signed path."""
|
|
return current_user()
|
|
|
|
|
|
_authenticated_export_actor = require_auth(
|
|
require_roles("admin")(_authenticated_export_actor)
|
|
)
|
|
|
|
|
|
def _signed_export_actor(token: str) -> tuple[dict, dict]:
|
|
token_actor = _verify_export_token(token)
|
|
if token_actor is None:
|
|
raise ApiError("invalid or expired export link", 403)
|
|
|
|
from ..storage.store import StoreError
|
|
|
|
stores = _stores()
|
|
token_user_id = token_actor.get("user_id")
|
|
token_org_id = token_actor.get("org_id")
|
|
global_super_admin = token_actor.get("role") == "super_admin" and token_org_id is None
|
|
if not _is_safe_export_identity(token_user_id) or (
|
|
not global_super_admin and not is_valid_tenant_id(token_org_id)
|
|
):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
try:
|
|
actor = stores["users"].get_user_or_none(token_user_id)
|
|
org = None if global_super_admin else stores["users"].get_org_or_none(token_org_id)
|
|
except (AuthError, OSError, StoreError, TypeError, UnicodeError, ValueError):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
if actor is not None and not isinstance(actor, dict):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
if org is not None and not isinstance(org, dict):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
actor_id = actor.get("id") if isinstance(actor, dict) else None
|
|
if (
|
|
not actor
|
|
or actor.get("active") is not True
|
|
or (
|
|
not global_super_admin
|
|
and not is_valid_tenant_id(actor.get("org_id"))
|
|
)
|
|
or (
|
|
global_super_admin
|
|
and actor.get("org_id") not in (None, "")
|
|
and not is_valid_tenant_id(actor.get("org_id"))
|
|
)
|
|
or token_actor.get("user_id") != actor_id
|
|
or token_actor.get("role") != actor.get("role")
|
|
):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
if global_super_admin:
|
|
if actor.get("role") != "super_admin":
|
|
raise ApiError("export link authorization is stale", 403)
|
|
elif (
|
|
not org
|
|
or org.get("active") is not True
|
|
or token_actor.get("org_id") != actor.get("org_id")
|
|
):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
if actor.get("role") not in {"admin", "super_admin"}:
|
|
raise ApiError("permission denied", 403)
|
|
|
|
if request.headers.get("Authorization"):
|
|
bearer_actor = _authenticated_export_actor()
|
|
if any(
|
|
bearer_actor.get(key) != actor.get(key)
|
|
for key in ("id", "org_id", "role")
|
|
):
|
|
raise ApiError("export link authorization is stale", 403)
|
|
g.user = actor
|
|
g.org_id = actor.get("org_id")
|
|
return actor, token_actor
|
|
|
|
|
|
@analytics_bp.get("/export/token")
|
|
@require_auth
|
|
@require_roles("admin")
|
|
def export_token():
|
|
"""Issue a short-lived signed download link for the CSV export."""
|
|
actor = current_user()
|
|
token = _sign_export_token(actor)
|
|
return jsonify({"token": token, "expires_in": _EXPORT_TTL, "url": f"/api/analytics/export?token={token}"})
|
|
|
|
|
|
def _parse_date_iso(value: str | None, *, end: bool = False) -> str | None:
|
|
"""Parse a YYYY-MM-DD into an ISO datetime bound for created_at filtering."""
|
|
if not value:
|
|
return None
|
|
try:
|
|
d = datetime.date.fromisoformat(value.strip())
|
|
except ValueError:
|
|
return None
|
|
if end:
|
|
# end-of-day bound (inclusive)
|
|
return datetime.datetime.combine(d, datetime.time(23, 59, 59, 999999), tzinfo=datetime.timezone.utc).isoformat()
|
|
return datetime.datetime.combine(d, datetime.time(0, 0, 0), tzinfo=datetime.timezone.utc).isoformat()
|
|
|
|
|
|
@analytics_bp.get("")
|
|
@require_auth
|
|
@require_roles("admin")
|
|
def analytics():
|
|
s = _stores()
|
|
actor = current_user()
|
|
|
|
# Date filter (optional) from ?from=YYYY-MM-DD&to=YYYY-MM-DD on created_at.
|
|
# Default window = last 30 days (today-30 .. today) when neither bound is given.
|
|
date_from = _parse_date_iso(request.args.get("from"))
|
|
date_to = _parse_date_iso(request.args.get("to"), end=True)
|
|
if date_from is None and date_to is None:
|
|
today = datetime.date.today()
|
|
date_from = datetime.datetime.combine(
|
|
today - datetime.timedelta(days=30), datetime.time(0, 0, 0),
|
|
tzinfo=datetime.timezone.utc,
|
|
).isoformat()
|
|
date_to = datetime.datetime.combine(
|
|
today, datetime.time(23, 59, 59, 999999), tzinfo=datetime.timezone.utc,
|
|
).isoformat()
|
|
if date_from:
|
|
date_from = date_from[:19]
|
|
if date_to:
|
|
date_to = date_to[:19]
|
|
|
|
def _in_window(sess) -> bool:
|
|
created = (sess.get("created_at") or "")[:19]
|
|
if not created:
|
|
return True
|
|
if date_from and created < date_from:
|
|
return False
|
|
if date_to and created > date_to:
|
|
return False
|
|
return True
|
|
|
|
max_scan_records = Config.ANALYTICS_EXPORT_MAX_SCAN_RECORDS
|
|
if (
|
|
isinstance(max_scan_records, bool)
|
|
or not isinstance(max_scan_records, int)
|
|
or max_scan_records <= 0
|
|
):
|
|
raise ApiError("analytics is unavailable", 503)
|
|
|
|
scanned_records = 0
|
|
|
|
def _bounded_records(collection: object) -> list[dict]:
|
|
nonlocal scanned_records
|
|
raw_iter = getattr(collection, "iter_all", None)
|
|
if not callable(raw_iter):
|
|
raise StoreError("analytics store lacks bounded iteration")
|
|
record_iter = cast(Callable[[], Iterator[dict]], raw_iter)
|
|
records: list[dict] = []
|
|
for record in record_iter():
|
|
scanned_records += 1
|
|
if scanned_records > max_scan_records:
|
|
raise ApiError("analytics scan limit exceeded", 413)
|
|
if isinstance(record, dict):
|
|
records.append(record)
|
|
return records
|
|
|
|
try:
|
|
groups = _bounded_records(s["groups"].groups)
|
|
users = _bounded_records(s["users"].users)
|
|
all_sessions = _bounded_records(s["sessions"].sessions)
|
|
except (OSError, StoreError, TypeError, ValueError) as exc:
|
|
raise ApiError("analytics unavailable", 503) from exc
|
|
|
|
shared_contexts = _shared_ready_contexts(groups)
|
|
if actor.get("role") == "super_admin":
|
|
trainee_users = [user for user in users if user.get("role") == "user"]
|
|
trainee_count = len(trainee_users)
|
|
user_org_by_id = {
|
|
user["id"]: user.get("org_id")
|
|
for user in trainee_users
|
|
if isinstance(user.get("id"), str)
|
|
}
|
|
sessions = [
|
|
x for x in all_sessions
|
|
if user_org_by_id.get(x.get("user_id")) == x.get("org_id")
|
|
and _is_legacy_trainee_mode(x.get("mode"))
|
|
and _session_matches_context(x, shared_contexts)
|
|
and _is_finished_outcome(x)
|
|
and _in_window(x)
|
|
]
|
|
else:
|
|
org_id = actor.get("org_id")
|
|
if not is_valid_tenant_id(org_id):
|
|
raise ApiError("permission denied", 403)
|
|
users = [user for user in users if user.get("org_id") == org_id]
|
|
trainee_users = [user for user in users if user.get("role") == "user"]
|
|
trainee_count = len(trainee_users)
|
|
user_ids = {
|
|
user["id"] for user in trainee_users if isinstance(user.get("id"), str)
|
|
}
|
|
group_ids = {
|
|
gid for gid, (group_org, _persona_ids) in shared_contexts.items()
|
|
if group_org == org_id
|
|
}
|
|
sessions = [
|
|
x for x in all_sessions
|
|
if x.get("user_id") in user_ids
|
|
and x.get("org_id") == org_id
|
|
and x.get("group_id") in group_ids
|
|
and _is_legacy_trainee_mode(x.get("mode"))
|
|
and _session_matches_context(x, shared_contexts, org_id=org_id)
|
|
and _is_finished_outcome(x)
|
|
and _in_window(x)
|
|
]
|
|
|
|
overall = {
|
|
"total_sessions": len(sessions),
|
|
"wins": sum(1 for x in sessions if x.get("outcome") == "won"),
|
|
"losses": sum(1 for x in sessions if x.get("outcome") == "lost"),
|
|
}
|
|
overall["close_rate"] = round(
|
|
overall["wins"] / overall["total_sessions"] * 100, 1
|
|
) if overall["total_sessions"] else 0
|
|
|
|
# average score
|
|
scores = [_debrief_score(x) for x in sessions]
|
|
overall["avg_score"] = round(sum(scores) / len(scores), 1) if scores else 0
|
|
|
|
# ── difficulty bucket lookup from group persona records ──────────────
|
|
# Map (group_id, persona_id) -> difficulty for every relevant group so each
|
|
# session's bucket can be read without per-row store lookups.
|
|
from ..services.trainee import analyze_team_weak_areas
|
|
|
|
persona_difficulty: dict[tuple[str, str], int] = {}
|
|
for group_candidate in groups:
|
|
if not isinstance(group_candidate, dict) or not isinstance(group_candidate.get("id"), str):
|
|
continue
|
|
gid = group_candidate["id"]
|
|
raw_personas = group_candidate.get("personas")
|
|
personas = raw_personas if isinstance(raw_personas, list) else []
|
|
for p in personas:
|
|
if not isinstance(p, dict) or not isinstance(p.get("id"), str):
|
|
continue
|
|
diff = p.get("difficulty")
|
|
persona_difficulty[(gid, p["id"])] = diff if isinstance(diff, int) else 5
|
|
|
|
def _bucket(diff: int) -> str:
|
|
if diff <= 2:
|
|
return "easy"
|
|
if diff == 3:
|
|
return "medium"
|
|
return "hard"
|
|
|
|
_BUCKET_META = {
|
|
"easy": {"label": "ง่าย", "range": "1-2"},
|
|
"medium": {"label": "กลาง", "range": "3"},
|
|
"hard": {"label": "ยาก", "range": "4-5"},
|
|
}
|
|
close_by_difficulty = {
|
|
key: {"plays": 0, "wins": 0, "losses": 0, "close_rate": 0, **meta}
|
|
for key, meta in _BUCKET_META.items()
|
|
}
|
|
for x in sessions:
|
|
gid_ = x.get("group_id")
|
|
pid_ = x.get("persona_id")
|
|
diff = persona_difficulty.get(
|
|
(gid_ if isinstance(gid_, str) else "", pid_ if isinstance(pid_, str) else ""), 5
|
|
)
|
|
rec = close_by_difficulty[_bucket(diff)]
|
|
rec["plays"] += 1
|
|
if x.get("outcome") == "won":
|
|
rec["wins"] += 1
|
|
elif x.get("outcome") == "lost":
|
|
rec["losses"] += 1
|
|
for rec in close_by_difficulty.values():
|
|
rec["close_rate"] = round(rec["wins"] / rec["plays"] * 100, 1) if rec["plays"] else 0
|
|
|
|
# ── active users ────────────────────────────────────────────────────
|
|
active_user_ids = {x.get("user_id") for x in sessions if isinstance(x.get("user_id"), str)}
|
|
active_users = len(active_user_ids)
|
|
|
|
# username map for the per-trainee table
|
|
username_of = {u.get("id"): u.get("username") or u.get("id")
|
|
for u in users if isinstance(u, dict) and isinstance(u.get("id"), str)}
|
|
|
|
# ── weekly trend (ISO week) ──────────────────────────────────────────
|
|
def _iso_week(created: object) -> str:
|
|
if not isinstance(created, str) or not created[:10]:
|
|
return "unknown"
|
|
try:
|
|
parsed = datetime.date.fromisoformat(created[:10])
|
|
return f"{parsed.isocalendar()[0]}-W{parsed.isocalendar()[1]:02d}"
|
|
except ValueError:
|
|
return "unknown"
|
|
|
|
weekly: dict[str, dict] = {}
|
|
for x in sessions:
|
|
wk = _iso_week(x.get("created_at"))
|
|
entry = weekly.setdefault(wk, {"week": wk, "sessions": 0, "wins": 0})
|
|
entry["sessions"] += 1
|
|
if x.get("outcome") == "won":
|
|
entry["wins"] += 1
|
|
weekly_trend = [
|
|
weekly[key] for key in sorted(weekly, key=lambda w: w.split("-W")[-1] if "-W" in w else w)
|
|
]
|
|
|
|
# ── per-trainee table ────────────────────────────────────────────────
|
|
user_sessions: dict[str, list[dict]] = {}
|
|
per_user: dict[str, dict] = {}
|
|
for x in sessions:
|
|
uid = x.get("user_id")
|
|
if not isinstance(uid, str):
|
|
continue
|
|
user_sessions.setdefault(uid, []).append(x)
|
|
rec = per_user.setdefault(
|
|
uid, {"username": username_of.get(uid, uid), "plays": 0, "wins": 0, "losses": 0, "scores": []}
|
|
)
|
|
rec["plays"] += 1
|
|
rec["scores"].append(_debrief_score(x))
|
|
if x.get("outcome") == "won":
|
|
rec["wins"] += 1
|
|
elif x.get("outcome") == "lost":
|
|
rec["losses"] += 1
|
|
trainee_table = []
|
|
for uid, rec in per_user.items():
|
|
rec["close_rate"] = round(rec["wins"] / rec["plays"] * 100, 1) if rec["plays"] else 0
|
|
rec["avg_score"] = round(sum(rec["scores"]) / len(rec["scores"]), 1) if rec["scores"] else 0
|
|
user_weak = _top_weak_area_label(user_sessions.get(uid, []))
|
|
trainee_table.append(
|
|
{
|
|
"username": rec["username"],
|
|
"plays": rec["plays"],
|
|
"wins": rec["wins"],
|
|
"losses": rec["losses"],
|
|
"close_rate": rec["close_rate"],
|
|
"avg_score": rec["avg_score"],
|
|
"top_weak_area": user_weak,
|
|
}
|
|
)
|
|
trainee_table.sort(key=lambda r: (-r["plays"], r["username"]))
|
|
|
|
# ── hardest personas = personas with most losses (lowest avg score) ──
|
|
by_persona: dict = {}
|
|
for x in sessions:
|
|
key = (x.get("group_id"), x.get("persona_id"), x.get("persona_name", "?"))
|
|
if key not in by_persona:
|
|
by_persona[key] = {"plays": 0, "losses": 0, "wins": 0, "scores": []}
|
|
rec = by_persona[key]
|
|
rec["plays"] += 1
|
|
rec["scores"].append(_debrief_score(x))
|
|
if x.get("outcome") == "won":
|
|
rec["wins"] += 1
|
|
elif x.get("outcome") == "lost":
|
|
rec["losses"] += 1
|
|
hardest = sorted(
|
|
(
|
|
{
|
|
"persona_name": k[2],
|
|
"plays": v["plays"],
|
|
"wins": v["wins"],
|
|
"losses": v["losses"],
|
|
"avg_score": round(sum(v["scores"]) / len(v["scores"]), 1) if v["scores"] else 0,
|
|
}
|
|
for k, v in by_persona.items()
|
|
),
|
|
key=lambda r: (-r["losses"], r["avg_score"]),
|
|
)[:10]
|
|
|
|
# ── team weak areas ──────────────────────────────────────────────────
|
|
team_org = actor.get("org_id") or "org-default"
|
|
team_weak = analyze_team_weak_areas(sessions, org_id=team_org)
|
|
|
|
return jsonify({
|
|
"overall": overall,
|
|
"trainee_count": trainee_count,
|
|
"active_users": active_users,
|
|
"weekly_trend": weekly_trend,
|
|
"close_by_difficulty": close_by_difficulty,
|
|
"trainee_table": trainee_table,
|
|
"hardest_personas": hardest,
|
|
"team_weak_areas": team_weak,
|
|
})
|
|
|
|
|
|
@require_auth
|
|
@require_roles("admin")
|
|
def _export_with_bearer():
|
|
return _export_csv_for_actor(current_user())
|
|
|
|
|
|
@analytics_bp.get("/export")
|
|
def export_csv():
|
|
"""Export via a one-time signed link or the normal Bearer JWT."""
|
|
token = request.args.get("token")
|
|
if token:
|
|
actor, token_actor = _signed_export_actor(token)
|
|
response = _export_csv_for_actor(actor, audit=False)
|
|
if not _consume_export_token(token_actor):
|
|
raise ApiError("invalid or expired export link", 403)
|
|
_log_audit(
|
|
"analytics.export",
|
|
actor.get("org_id") or "?",
|
|
detail={"signed": True},
|
|
)
|
|
return response
|
|
return _export_with_bearer()
|
|
|
|
|
|
def _export_csv_for_actor(actor: dict, *, audit: bool = True):
|
|
"""Export per-trainee finished session results as CSV (for HR/offline review)."""
|
|
import csv
|
|
from flask import Response, current_app
|
|
|
|
# Tenant: an admin exports only their own org's sessions; super_admin sees all.
|
|
is_super_admin = actor.get("role") == "super_admin"
|
|
export_org = None if is_super_admin else actor.get("org_id")
|
|
if not is_super_admin and not is_valid_tenant_id(export_org):
|
|
raise ApiError("permission denied", 403)
|
|
s = _stores()
|
|
max_rows = Config.ANALYTICS_EXPORT_MAX_ROWS
|
|
max_bytes = Config.ANALYTICS_EXPORT_MAX_BYTES
|
|
max_scan_records = Config.ANALYTICS_EXPORT_MAX_SCAN_RECORDS
|
|
if (
|
|
isinstance(max_rows, bool)
|
|
or not isinstance(max_rows, int)
|
|
or max_rows <= 0
|
|
or isinstance(max_bytes, bool)
|
|
or not isinstance(max_bytes, int)
|
|
or max_bytes <= 0
|
|
or isinstance(max_scan_records, bool)
|
|
or not isinstance(max_scan_records, int)
|
|
or max_scan_records <= 0
|
|
):
|
|
raise ApiError("analytics export is unavailable", 503)
|
|
scanned_records = 0
|
|
users: dict[str, dict[str, object]] = {}
|
|
sessions = []
|
|
try:
|
|
user_store = current_app.extensions.get("user_store")
|
|
grp_store = s["groups"]
|
|
sess_store = s["sessions"]
|
|
user_collection = getattr(user_store, "users", None)
|
|
group_collection = getattr(grp_store, "groups", grp_store)
|
|
session_collection = getattr(sess_store, "sessions", sess_store)
|
|
raw_user_iter = getattr(user_collection, "iter_all", None)
|
|
raw_group_iter = getattr(group_collection, "iter_all", None)
|
|
raw_session_iter = getattr(session_collection, "iter_all", None)
|
|
if not callable(raw_user_iter) or not callable(raw_group_iter) or not callable(raw_session_iter):
|
|
raise StoreError("analytics stores lack bounded iteration")
|
|
user_iter = cast(Callable[[], Iterator[dict]], raw_user_iter)
|
|
group_iter = cast(Callable[[], Iterator[dict]], raw_group_iter)
|
|
session_iter = cast(Callable[[], Iterator[dict]], raw_session_iter)
|
|
|
|
for rec in user_iter():
|
|
scanned_records += 1
|
|
if scanned_records > max_scan_records:
|
|
raise ApiError("analytics export scan limit exceeded", 413)
|
|
if not isinstance(rec, dict) or rec.get("role") != "user":
|
|
continue
|
|
if export_org is not None and rec.get("org_id") != export_org:
|
|
continue
|
|
user_id = rec.get("id") or rec.get("username")
|
|
if isinstance(user_id, str):
|
|
users[user_id] = {
|
|
"username": rec.get("username") or rec.get("id"),
|
|
"org_id": rec.get("org_id"),
|
|
}
|
|
|
|
shared_contexts: dict[str, tuple[str, set[str]]] = {}
|
|
for group in group_iter():
|
|
scanned_records += 1
|
|
if scanned_records > max_scan_records:
|
|
raise ApiError("analytics export scan limit exceeded", 413)
|
|
if (
|
|
isinstance(group, dict)
|
|
and isinstance(group.get("id"), str)
|
|
and (export_org is None or group.get("org_id") == export_org)
|
|
and _is_shared_group(group)
|
|
):
|
|
persona_ids = _canonical_persona_ids(group.get("personas"))
|
|
if persona_ids:
|
|
shared_contexts[group["id"]] = (group["org_id"], persona_ids)
|
|
|
|
for row in session_iter():
|
|
scanned_records += 1
|
|
if scanned_records > max_scan_records:
|
|
raise ApiError("analytics export scan limit exceeded", 413)
|
|
if not (
|
|
isinstance(row, dict)
|
|
and _is_legacy_trainee_mode(row.get("mode"))
|
|
and _session_matches_context(row, shared_contexts, org_id=export_org)
|
|
and row.get("status") == "finished"
|
|
and row.get("outcome") in {"won", "lost"}
|
|
and row.get("user_id") in users
|
|
and users[row["user_id"]].get("org_id") == row.get("org_id")
|
|
and (
|
|
export_org is None
|
|
or row.get("org_id") == export_org
|
|
)
|
|
):
|
|
continue
|
|
sessions.append(row)
|
|
if len(sessions) > max_rows:
|
|
raise ApiError("analytics export has too many rows", 413)
|
|
except (OSError, StoreError, TypeError, ValueError) as exc:
|
|
raise ApiError("analytics unavailable", 503) from exc
|
|
|
|
if audit:
|
|
_log_audit("analytics.export", actor.get("org_id") or "?", detail={"rows": len(sessions)})
|
|
|
|
class _BoundedStringIO(io.StringIO):
|
|
def __init__(self, byte_limit: int):
|
|
super().__init__()
|
|
self._byte_limit = byte_limit
|
|
self._bytes_written = 0
|
|
|
|
def write(self, value: str) -> int:
|
|
try:
|
|
byte_count = len(value.encode("utf-8"))
|
|
except UnicodeError as exc:
|
|
raise ApiError("analytics export contains invalid text", 400) from exc
|
|
if self._bytes_written + byte_count > self._byte_limit:
|
|
raise ApiError("analytics export is too large", 413)
|
|
self._bytes_written += byte_count
|
|
return super().write(value)
|
|
|
|
buf = _BoundedStringIO(max_bytes)
|
|
w = csv.writer(buf)
|
|
w.writerow(["username", "persona", "scenario", "outcome", "score", "created_at"])
|
|
for sess in sessions:
|
|
w.writerow([
|
|
_csv_cell(
|
|
users.get(sess.get("user_id"), {}).get(
|
|
"username", sess.get("user_id", "")
|
|
),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
_csv_cell(
|
|
sess.get("persona_name", ""),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
_csv_cell(
|
|
sess.get("scenario", ""),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
_csv_cell(
|
|
sess.get("outcome", ""),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
_csv_cell(
|
|
_debrief_score(sess),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
_csv_cell(
|
|
sess.get("created_at", ""),
|
|
max_chars=Config.ANALYTICS_EXPORT_MAX_CELL_CHARS,
|
|
),
|
|
])
|
|
return Response(
|
|
buf.getvalue(),
|
|
mimetype="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=sales-trainer-results.csv"},
|
|
)
|