"""Admin analytics: aggregate trainee results (supports date-range filter).""" from __future__ import annotations import datetime from flask import Blueprint, jsonify, request from .helpers import ApiError, current_user, 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 _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 _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 date_from = _parse_date_iso(request.args.get("from")) date_to = _parse_date_iso(request.args.get("to"), end=True) def _in_window(sess) -> bool: created = (sess.get("created_at") or "")[:19] if not created: return True if date_from and created < date_from[:19]: return False if date_to and created > date_to[:19]: return False return True if actor.get("role") == "super_admin": users = s["users"].list_users() sessions = [x for x in s["sessions"].sessions.all() if _in_window(x)] else: org_id = actor.get("org_id") users = s["users"].list_users(org_id=org_id) user_ids = {u["id"] for u in users} sessions = [ x for x in s["sessions"].sessions.all() if x.get("user_id") in user_ids 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 = [ (x.get("debrief") or {}).get("score", 0) for x in sessions if x.get("outcome") ] overall["avg_score"] = round(sum(scores) / len(scores), 1) if scores else 0 # 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((x.get("debrief") or {}).get("score", 0)) 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] return jsonify({ "overall": overall, "trainee_count": len(users), "hardest_personas": hardest, }) @analytics_bp.get("/export") @require_auth @require_roles("admin") def export_csv(): """Export per-trainee finished session results as CSV (for HR/offline review).""" import csv import io from flask import Response, current_app s = _stores() actor = current_user() # Tenant: an admin exports only their own org's sessions; super_admin sees all. export_org = actor.get("org_id") if actor.get("role") != "super_admin" else None users = {} try: user_store = current_app.extensions.get("user_store") if user_store and hasattr(user_store, "list_users"): for rec in user_store.list_users(org_id=export_org): users[rec.get("id") or rec.get("username")] = rec.get("username") or rec.get("id") except Exception: pass sessions = [] try: grp_store = s["groups"] sess_store = s["sessions"] if hasattr(grp_store, "groups"): group_ids = [g.get("id") for g in grp_store.groups.all() if export_org is None or g.get("org_id") == export_org] elif hasattr(grp_store, "all"): group_ids = [g.get("id") for g in grp_store.all() if export_org is None or g.get("org_id") == export_org] else: group_ids = [] for gid in group_ids: try: ss = sess_store.sessions.where( lambda r, _gid=gid: r.get("group_id") == _gid and r.get("outcome") ) sessions.extend(ss) except Exception: pass except Exception: pass _log_audit("analytics.export", actor.get("org_id") or "?", detail={"rows": len(sessions)}) buf = io.StringIO() w = csv.writer(buf) w.writerow(["username", "persona", "scenario", "outcome", "score", "created_at"]) for sess in sessions: w.writerow([ users.get(sess.get("user_id"), sess.get("user_id", "")), sess.get("persona_name", ""), sess.get("scenario", ""), sess.get("outcome", ""), (sess.get("debrief") or {}).get("score", ""), sess.get("created_at", ""), ]) return Response( buf.getvalue(), mimetype="text/csv", headers={"Content-Disposition": "attachment; filename=sales-trainer-results.csv"}, )