at_risk.py

"""
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,
    )