Thêm logic scan tìm vấn đề học viên

api.py

@@ -165,6 +165,11 @@ class IndexPdfResponse(BaseModel):
     conversation_id: str
 
 
+class AtRiskRequest(BaseModel):
+    room_ids: list[str] = Field(..., description="Danh sách room ID cần quét")
+    hours:    int       = Field(24, ge=1, le=168, description="Cửa sổ thời gian nhìn lại (giờ), mặc định 24")
+
+
 # ── Helper ────────────────────────────────────────────────────────────────────
 
 def _request_id(request: Request) -> str:
@@ -583,6 +588,56 @@ async def summary_chart(request: Request, body: ChatRequest):
     )
 
 
+# ── At-risk endpoint ──────────────────────────────────────────────────────────
+
+@app.post(
+    "/api/v1/at-risk/analyze",
+    status_code=status.HTTP_200_OK,
+    summary="Phân tích tín hiệu nguy cơ học viên từ Redis",
+    tags=["At-Risk"],
+    responses={
+        400: {"model": ErrorDetail, "description": "room_ids rỗng"},
+        500: {"model": ErrorDetail, "description": "Lỗi xử lý nội bộ"},
+    },
+)
+async def at_risk_analyze(request: Request, body: AtRiskRequest):
+    """
+    Nhận danh sách room_ids và cửa sổ thời gian (hours), fetch tin nhắn từ Redis,
+    dùng LLM phát hiện tín hiệu nguy cơ per-student:
+      - stuck_phrases: học viên bị kẹt / không hiểu nội dung
+      - unanswered_questions: câu hỏi học thuật chưa ai trả lời (kèm wait_minutes và suggested_points)
+      - frustration_phrases: biểu đạt chán nản / burn out
+
+    suggested_points cho unanswered_questions được tính từ thời gian chờ:
+      15-30 phút → 1, 30-60 phút → 2, 1-2 giờ → 3, >2 giờ → 4.
+
+    NestJS tự cộng absence_points (từ Supabase profiles) để ra total_score.
+    """
+    if not body.room_ids:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="room_ids không được để trống.",
+        )
+
+    from src.at_risk import analyze_rooms, AnalysisResult
+    loop = get_running_loop()
+    try:
+        result: AnalysisResult = await loop.run_in_executor(
+            _executor,
+            lambda: analyze_rooms(body.room_ids, body.hours),
+        )
+    except Exception:
+        logger.exception(
+            "Unhandled error in POST /api/v1/at-risk/analyze [rid=%s]", _request_id(request)
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Lỗi phân tích. Xem log để biết thêm.",
+        )
+
+    return result
+
+
 # ── Exception handlers ────────────────────────────────────────────────────────
 
 @app.exception_handler(404)

at_risk.py

@@ -0,0 +1,280 @@
+"""
+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,
+    )