""" At-risk student analysis — called on-demand by NestJS. Flow: nhận room_ids + hours → fetch Redis → LLM detect signals per room → Python tính wait_minutes cho unanswered questions → merge theo student. """ import logging from datetime import datetime, timedelta, timezone from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from src.redis_client import redis_client from src.tools.base import get_llm logger = logging.getLogger(__name__) # ── Public output schemas (dùng cho api.py response_model) ──────────────────── class SignalItem(BaseModel): text: str room_id: str timestamp: str suggested_points: int class UnansweredQuestion(BaseModel): text: str room_id: str timestamp: str wait_minutes: int suggested_points: int class StudentSignals(BaseModel): stuck_phrases: list[SignalItem] unanswered_questions: list[UnansweredQuestion] frustration_phrases: list[SignalItem] class StudentResult(BaseModel): sender_id: str sender_name: str signals: StudentSignals class AnalysisResult(BaseModel): analyzed_at: str students: list[StudentResult] # ── Internal LLM output schemas ──────────────────────────────────────────────── class _Signal(BaseModel): sender_id: str = Field(description="sender_id chép nguyên văn từ tin nhắn") sender_name: str = Field(description="tên người gửi chép nguyên văn từ tin nhắn") text: str = Field(description="đoạn văn bản gốc từ tin nhắn") timestamp: str = Field(description="timestamp chép nguyên văn từ tin nhắn") suggested_points: int = Field(description="điểm mức độ nghiêm trọng") class _Unanswered(BaseModel): sender_id: str = Field(description="sender_id chép nguyên văn") sender_name: str = Field(description="tên người gửi chép nguyên văn") text: str = Field(description="câu hỏi gốc") timestamp: str = Field(description="timestamp chép nguyên văn") class _RoomAnalysis(BaseModel): stuck_phrases: list[_Signal] = Field(default_factory=list) unanswered_questions: list[_Unanswered] = Field(default_factory=list) frustration_phrases: list[_Signal] = Field(default_factory=list) # ── LLM Prompt ───────────────────────────────────────────────────────────────── _SYSTEM = """\ Bạn là chuyên gia phân tích hội thoại cho hệ thống giám sát học viên. Đọc tin nhắn từ một phòng học, xác định các tín hiệu nguy cơ theo từng học viên. Chỉ trả về JSON. Không giải thích. TÍN HIỆU 1 — stuck_phrases: Học viên bị kẹt hoặc không hiểu NỘI DUNG HỌC. Ví dụ: "không hiểu bài", "bí rồi", "không làm được bài tập", "chưa hiểu phần X". suggested_points: 2 (lúng túng nhẹ) đến 4 (bỏ cuộc, tuyệt vọng). TÍN HIỆU 2 — unanswered_questions: Câu hỏi học thuật KHÔNG có ai trả lời. Chỉ tính câu hỏi về nội dung học (bài tập, bài giảng, khái niệm kỹ thuật). Nếu có tin nhắn của người khác reply sau câu hỏi đó trong cuộc trò chuyện thì KHÔNG tính là chưa trả lời. Bỏ qua câu hỏi xã giao ("ai online?", "mọi người ăn chưa?"). Không cần trả về suggested_points — sẽ tính tự động theo thời gian chờ. TÍN HIỆU 3 — frustration_phrases: Chán nản, mệt mỏi liên quan đến học tập. Ví dụ: "chán quá", "học mãi không vô", "mệt rồi", "căng thẳng quá". suggested_points: 2 (nhẹ) đến 3 (rõ ràng). QUY TẮC: Mỗi tin nhắn chỉ thuộc MỘT loại tín hiệu. sender_id, sender_name, timestamp: chép NGUYÊN VĂN từ tin nhắn. Không tìm thấy tín hiệu nào thì trả danh sách rỗng. """ _HUMAN = """\ Room: {room_id} Tin nhắn (định dạng: [timestamp] sender_name (sender_id): nội dung): {messages} {format_instructions} """ # ── Helpers ──────────────────────────────────────────────────────────────────── # Tất cả field name có thể chứa username, theo thứ tự ưu tiên _NAME_FIELDS = ["sender_username", "username", "u_username", "name", "u_name", "senderName", "displayName", "display_name", "fullName", "sender_id"] _ID_FIELDS = ["sender_id", "u_id", "userId", "user_id", "sender_username", "username"] def _parse_ts(raw: str) -> datetime | None: try: dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) except Exception: return None def _get_field(m: dict, fields: list[str]) -> str: for f in fields: v = m.get(f) if v and str(v).strip(): return str(v).strip() return "" def _filter_recent(messages: list[dict], hours: int) -> list[dict]: cutoff = datetime.now(tz=timezone.utc) - timedelta(hours=hours) return [ m for m in messages if (dt := _parse_ts(m.get("created_at") or m.get("timestamp", ""))) and dt >= cutoff ] def _format_messages(messages: list[dict]) -> str: if messages: logger.debug("[AtRisk] Sample message keys: %s", list(messages[0].keys())) lines = [] for m in messages: ts = m.get("created_at") or m.get("timestamp", "") name = _get_field(m, _NAME_FIELDS) or "unknown" sid = _get_field(m, _ID_FIELDS) or name content = m.get("content") or m.get("text") or m.get("msg", "") lines.append(f"[{ts}] {name} ({sid}): {content}") return "\n".join(lines) def _sender_key(m: dict) -> str: return _get_field(m, _ID_FIELDS) def _wait_minutes(question_ts: str, question_sender: str, messages: list[dict]) -> int: """Tính số phút từ câu hỏi đến reply đầu tiên của người khác (hoặc đến now nếu chưa có).""" q_dt = _parse_ts(question_ts) if q_dt is None: return 0 for m in messages: if _sender_key(m) == question_sender: continue dt = _parse_ts(m.get("created_at") or m.get("timestamp", "")) if dt and dt > q_dt: return max(0, int((dt - q_dt).total_seconds() / 60)) return max(0, int((datetime.now(tz=timezone.utc) - q_dt).total_seconds() / 60)) def _points_from_wait(minutes: int) -> int: if minutes >= 120: return 4 if minutes >= 60: return 3 if minutes >= 30: return 2 if minutes >= 15: return 1 return 0 # < 15 phút → chưa đủ để tính # ── LLM call ───────────────────────────────────────────────────────────────── def _analyze_room(room_id: str, messages: list[dict]) -> _RoomAnalysis: parser = JsonOutputParser(pydantic_object=_RoomAnalysis) prompt = ChatPromptTemplate.from_messages([ ("system", _SYSTEM), ("human", _HUMAN), ]) result = (prompt | get_llm() | parser).invoke({ "room_id": room_id, "messages": _format_messages(messages), "format_instructions": parser.get_format_instructions(), }) return _RoomAnalysis(**result) # ── Accumulator ─────────────────────────────────────────────────────────────── def _add(store: dict, sender_id: str, sender_name: str, category: str, item: dict): if sender_id not in store: store[sender_id] = { "sender_name": sender_name, "stuck_phrases": [], "unanswered_questions": [], "frustration_phrases": [], } store[sender_id][category].append(item) # ── Public entry point ──────────────────────────────────────────────────────── def analyze_rooms(room_ids: list[str], hours: int = 24) -> AnalysisResult: """ Phân tích tín hiệu nguy cơ của học viên trong các room được chỉ định. Chạy đồng bộ — gọi từ thread pool trong FastAPI endpoint. """ store: dict[str, dict] = {} for room_id in room_ids: try: redis_room_id = room_id.removeprefix("room-") all_msgs = redis_client.get_room_messages(redis_room_id, limit=500) recent = _filter_recent(all_msgs, hours) if not recent: continue raw = _analyze_room(room_id, recent) for s in raw.stuck_phrases: _add(store, s.sender_id, s.sender_name, "stuck_phrases", { "text": s.text, "room_id": room_id, "timestamp": s.timestamp, "suggested_points": min(max(s.suggested_points, 2), 4), }) for q in raw.unanswered_questions: wait = _wait_minutes(q.timestamp, q.sender_id, recent) pts = _points_from_wait(wait) if pts == 0: continue # câu hỏi < 15 phút → bỏ qua _add(store, q.sender_id, q.sender_name, "unanswered_questions", { "text": q.text, "room_id": room_id, "timestamp": q.timestamp, "wait_minutes": wait, "suggested_points": pts, }) for f in raw.frustration_phrases: _add(store, f.sender_id, f.sender_name, "frustration_phrases", { "text": f.text, "room_id": room_id, "timestamp": f.timestamp, "suggested_points": min(max(f.suggested_points, 2), 3), }) except Exception: logger.exception("[AtRisk] Lỗi khi phân tích room '%s'", room_id) students = [ StudentResult( sender_id=sid, sender_name=data["sender_name"], signals=StudentSignals( stuck_phrases=[SignalItem(**x) for x in data["stuck_phrases"]], unanswered_questions=[UnansweredQuestion(**x) for x in data["unanswered_questions"]], frustration_phrases=[SignalItem(**x) for x in data["frustration_phrases"]], ), ) for sid, data in store.items() ] return AnalysisResult( analyzed_at=datetime.now(tz=timezone.utc).isoformat(), students=students, )