"""Teacher dashboard routes backed by real server session state.""" from __future__ import annotations from datetime import datetime from typing import Annotated, Any from fastapi import APIRouter, Depends from pydantic import BaseModel, Field from .. import session_persistence from ..deps import Principal, Role, require_role from ..runtime_policy import require_runtime_fallback_allowed from ..store import InProcSession, store router = APIRouter(prefix="/teacher", tags=["teacher"]) TeacherPrincipal = Annotated[Principal, Depends(require_role(Role.TEACHER, Role.ADMIN))] class TeacherSessionSummary(BaseModel): session_id: str learner_id: str learner_label: str persona_code: str persona_name: str session_no: int status: str stage: str turn_count: int learner_turn_count: int client_turn_count: int started_at: str ended_at: str | None = None class TeacherGrowthPoint(BaseModel): session_id: str session_no: int persona_code: str stage: str started_at: str ended_at: str | None = None score: float | None = None rapport: float | None = None technique_count: int = 0 watch_count: int = 0 class TeacherLearnerGrowth(BaseModel): learner_id: str learner_label: str sessions: int ended_sessions: int latest_at: str first_score: float | None = None latest_score: float | None = None score_delta: float | None = None avg_score: float | None = None avg_rapport: float | None = None trend: str = "insufficient" top_techniques: list[str] = Field(default_factory=list) points: list[TeacherGrowthPoint] = Field(default_factory=list) class TeacherSafetyAlert(BaseModel): id: str session_id: str learner_id: str learner_label: str persona_code: str session_no: int trigger_type: str ko_risk_level: int escalated: bool created_at: str resource_title: str = "자살예방상담전화 109" resource_number: str = "109" class TeacherDashboardResponse(BaseModel): source: str = "in_memory" cohort_label: str = "현재 학습 기록" total_learners: int active_sessions: int ended_sessions: int safety_alerts: list[TeacherSafetyAlert] = Field(default_factory=list) learner_growth: list[TeacherLearnerGrowth] = Field(default_factory=list) pending_reviews: list[TeacherSessionSummary] = Field(default_factory=list) recent_sessions: list[TeacherSessionSummary] = Field(default_factory=list) message: str _APPROPRIATENESS_SCORE = { "neg": 0.0, "neutral": 0.5, "pos": 1.0, } def _iso(ts: float | None) -> str | None: if ts is None: return None return datetime.fromtimestamp(ts).isoformat(timespec="seconds") def _learner_label(learner_id: str) -> str: suffix = learner_id[-6:] if len(learner_id) > 6 else learner_id return f"학습자 {suffix}" def _safe_float(value: object) -> float | None: try: return float(value) # type: ignore[arg-type] except (TypeError, ValueError): return None def _avg(values: list[float]) -> float | None: if not values: return None return round(sum(values) / len(values), 3) def _turn_eval(turn: Any) -> dict[str, Any] | None: ev = getattr(turn, "evaluation", None) return ev if isinstance(ev, dict) else None def _turn_score(ev: dict[str, Any]) -> float | None: raw = str(ev.get("appropriateness") or "").strip().lower() return _APPROPRIATENESS_SCORE.get(raw) def _turn_rapport(ev: dict[str, Any]) -> float | None: value = _safe_float(ev.get("rapport_signal")) if value is None: return None return max(-1.0, min(1.0, value)) def _turn_techniques(ev: dict[str, Any]) -> list[str]: raw = ev.get("techniques") if not isinstance(raw, list): return [] labels: list[str] = [] for item in raw: if isinstance(item, dict): label = item.get("label") or item.get("name") or item.get("id") else: label = item if label: labels.append(str(label)) return labels def _session_growth_point(sess: InProcSession) -> TeacherGrowthPoint: scores: list[float] = [] rapports: list[float] = [] technique_count = 0 watch_count = 0 for turn in sess.turns: if turn.speaker != "counselor": continue ev = _turn_eval(turn) if ev is None: continue score = _turn_score(ev) if score is not None: scores.append(score) if score < 1.0: watch_count += 1 rapport = _turn_rapport(ev) if rapport is not None: rapports.append(rapport) technique_count += len(_turn_techniques(ev)) return TeacherGrowthPoint( session_id=sess.session_id, session_no=sess.session_no, persona_code=sess.persona_code, stage=sess.state.stage.value, started_at=_iso(sess.created_at) or "", ended_at=_iso(sess.ended_at), score=_avg(scores), rapport=_avg(rapports), technique_count=technique_count, watch_count=watch_count, ) def _build_learner_growth(sessions: list[InProcSession]) -> list[TeacherLearnerGrowth]: grouped: dict[str, list[InProcSession]] = {} for sess in sessions: grouped.setdefault(sess.learner_id, []).append(sess) result: list[TeacherLearnerGrowth] = [] for learner_id, learner_sessions in grouped.items(): ordered = sorted(learner_sessions, key=lambda sess: sess.created_at) points = [_session_growth_point(sess) for sess in ordered] scored = [point for point in points if point.score is not None] rapport_values = [point.rapport for point in points if point.rapport is not None] technique_counts: dict[str, int] = {} for sess in ordered: for turn in sess.turns: if turn.speaker != "counselor": continue ev = _turn_eval(turn) if ev is None: continue for label in _turn_techniques(ev): technique_counts[label] = technique_counts.get(label, 0) + 1 first_score = scored[0].score if scored else None latest_score = scored[-1].score if scored else None score_delta: float | None = None trend = "insufficient" if first_score is not None and latest_score is not None: score_delta = round(latest_score - first_score, 3) if len(scored) >= 2: if score_delta >= 0.1: trend = "up" elif score_delta <= -0.1: trend = "down" else: trend = "flat" latest_session = ordered[-1] top_techniques = [ label for label, _count in sorted( technique_counts.items(), key=lambda item: (-item[1], item[0]), )[:3] ] result.append( TeacherLearnerGrowth( learner_id=learner_id, learner_label=_learner_label(learner_id), sessions=len(ordered), ended_sessions=sum(1 for sess in ordered if sess.ended), latest_at=_iso(latest_session.ended_at or latest_session.created_at) or "", first_score=first_score, latest_score=latest_score, score_delta=score_delta, avg_score=_avg([point.score for point in scored if point.score is not None]), avg_rapport=_avg([value for value in rapport_values if value is not None]), trend=trend, top_techniques=top_techniques, points=points[-6:], ) ) return sorted(result, key=lambda item: item.latest_at, reverse=True)[:12] def _summary(sess: InProcSession) -> TeacherSessionSummary: learner_turns = sum(1 for turn in sess.turns if turn.speaker == "counselor") client_turns = sum(1 for turn in sess.turns if turn.speaker == "client") return TeacherSessionSummary( session_id=sess.session_id, learner_id=sess.learner_id, learner_label=_learner_label(sess.learner_id), persona_code=sess.persona_code, persona_name=sess.persona.display_name, session_no=sess.session_no, status="ended" if sess.ended else "active", stage=sess.state.stage.value, turn_count=len(sess.turns), learner_turn_count=learner_turns, client_turn_count=client_turns, started_at=_iso(sess.created_at) or "", ended_at=_iso(sess.ended_at), ) @router.get("/dashboard", response_model=TeacherDashboardResponse) async def teacher_dashboard(principal: TeacherPrincipal) -> TeacherDashboardResponse: """Return teacher-visible dashboard data from real sessions only.""" sessions, durable = await session_persistence.list_sessions( principal, include_turn_evaluation=True, ) if not durable: require_runtime_fallback_allowed("teacher dashboard") sessions = sorted(store.list(), key=lambda sess: sess.created_at, reverse=True) summaries = [_summary(sess) for sess in sessions] pending_reviews = [item for item in summaries if item.status == "ended"] learners = {sess.learner_id for sess in sessions} learner_growth = _build_learner_growth(sessions) safety_alerts: list[TeacherSafetyAlert] = [] if durable: raw_alerts, alerts_durable = await session_persistence.list_safety_alerts(principal) if alerts_durable: safety_alerts = [ TeacherSafetyAlert( id=str(item.get("id") or ""), session_id=str(item.get("session_id") or ""), learner_id=str(item.get("learner_id") or ""), learner_label=str(item.get("learner_label") or "학습자"), persona_code=str(item.get("persona_code") or ""), session_no=int(item.get("session_no") or 0), trigger_type=str(item.get("trigger_type") or "crisis"), ko_risk_level=int(item.get("ko_risk_level") or 0), escalated=bool(item.get("escalated")), created_at=str(item.get("created_at") or ""), resource_title=str( (item.get("detail") or {}).get("crisis_resource", {}).get( "title", "자살예방상담전화 109", ) ), resource_number=str( (item.get("detail") or {}).get("crisis_resource", {}).get("number", "109") ), ) for item in raw_alerts ] if sessions: message = "현재 기록된 실제 학습 세션만 표시합니다." else: message = "아직 표시할 실제 학습자 세션이 없습니다." return TeacherDashboardResponse( source="database" if durable else "runtime", total_learners=len(learners), active_sessions=sum(1 for sess in sessions if not sess.ended), ended_sessions=sum(1 for sess in sessions if sess.ended), safety_alerts=safety_alerts, learner_growth=learner_growth, pending_reviews=pending_reviews[:20], recent_sessions=summaries[:20], message=message, )