import asyncio
import json
import logging
import re
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from uuid import uuid4
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from app.auth import get_current_user
from app.config import get_settings
from app.meeting import _send_email, STAFF_EMAILS
from app.safety.safeguard import (
safeguard_check,
CRISIS, ESCALATION, RESOLVED, UNKNOWN,
FALLING_GRADES, MISSED_CLASSES, FINANCIAL_AID, DEGREE_PLAN, PANIC,
ADVISOR_REVIEW_PROFILE, ADVISOR_DRAFT_MESSAGE, ADVISOR_APPOINTMENT,
STUDENT_INTENTS, ADVISOR_INTENTS,
)
from app.agent.chain import generate_reply
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/chatbot", tags=["chatbot"])
# timezone aware utc timestamp in iso format
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
# Config shortcuts
def _cfg():
return get_settings()
# Session storage (file-based)
_DATA_DIR: Optional[Path] = None
def _data_dir() -> Path:
global _DATA_DIR
if _DATA_DIR is None:
_DATA_DIR = Path(_cfg().data_dir)
_DATA_DIR.mkdir(parents=True, exist_ok=True)
return _DATA_DIR
def _load_session(session_id: str) -> Optional[Dict]:
path = _data_dir() / f"{session_id}.json"
if not path.exists():
return None
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("Could not load session %s: %s", session_id, exc)
return None
def _save_session(session: Dict) -> None:
path = _data_dir() / f"{session['session_id']}.json"
try:
path.write_text(json.dumps(session, indent=2, ensure_ascii=False), encoding="utf-8")
except Exception as exc:
# log and continue, a failed write should not crash the request
logger.error("Could not save session %s: %s", session.get("session_id"), exc)
# Notification storage (file-based)
_NOTIFICATIONS_DIR: Optional[Path] = None
def _notifications_dir() -> Path:
global _NOTIFICATIONS_DIR
if _NOTIFICATIONS_DIR is None:
_NOTIFICATIONS_DIR = Path(_cfg().data_dir).parent / "notifications"
_NOTIFICATIONS_DIR.mkdir(parents=True, exist_ok=True)
return _NOTIFICATIONS_DIR
def _save_notification(notification: Dict) -> None:
"""Save a crisis notification to file."""
path = _notifications_dir() / f"{notification['notification_id']}.json"
path.write_text(json.dumps(notification, indent=2, ensure_ascii=False), encoding="utf-8")
def _load_all_notifications() -> List[Dict]:
"""Load all crisis notifications."""
notifications: List[Dict] = []
try:
for p in _notifications_dir().glob("*.json"):
try:
n = json.loads(p.read_text(encoding="utf-8"))
notifications.append(n)
except Exception:
continue
except Exception as exc:
logger.error("Error loading notifications: %s", exc)
return notifications
def _get_staff_alert_emails() -> List[str]:
return [email for email in STAFF_EMAILS.values() if email]
def _send_crisis_alert(current_user: Dict[str, Any], user_message: str) -> None:
notification_id = str(uuid4())
timestamp = _now_iso()
student_name = current_user.get("first_name") or current_user.get("username") or "A student"
student_id = current_user.get("student_id", "")
student_username = current_user.get("username", "")
# Save notification to dashboard
notification = {
"notification_id": notification_id,
"student_name": student_name,
"student_id": student_id,
"student_username": student_username,
"detected_message": user_message,
"timestamp": timestamp,
"read": False,
"alert_type": "crisis",
}
_save_notification(notification)
logger.info("Saved crisis notification %s for student %s", notification_id, student_name)
# Send email alert
emails = _get_staff_alert_emails()
if not emails:
logger.debug("No staff alert email addresses configured — skipping email alert.")
else:
subject = f"[MyACC Virtual Assistant Alert] Crisis language detected for {student_name}"
body = f"""
Urgent MyACC Virtual Assistant Alert
Dear staff,
The chatbot detected possible self-harm/crisis language in a student chat session.
| Student | {student_name} |
| Student ID | {student_id} |
| Detected message | {user_message} |
Please review the student immediately in the MyACC Virtual Assistant dashboard.
This is an automated alert from MyACC Virtual Assistant.
""".strip()
for email in emails:
try:
_send_email(email, subject, body)
logger.info("Sent crisis alert email to %s for student %s (%s)", email, student_name, student_id)
except Exception as exc:
logger.error("Failed to send crisis alert to %s: %s", email, exc)
# MySQL support
_mysql_ready = False
ChatSession = None # SQLAlchemy model (populated below if MySQL)
SessionLocal = None
if get_settings().db_type == "mysql":
try:
from sqlalchemy import create_engine, Column, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session as SQLSession
_engine = create_engine(get_settings().mysql_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=_engine)
_Base = declarative_base()
class ChatSession(_Base): # type: ignore[no-redef]
__tablename__ = "chat_sessions"
session_id = Column(String(64), primary_key=True, index=True)
student_id = Column(String(128), index=True)
messages = Column(Text)
status = Column(String(32))
created = Column(DateTime)
last_updated = Column(DateTime)
_Base.metadata.create_all(bind=_engine)
_mysql_ready = True
logger.info("MySQL session storage ready.")
except Exception as exc:
logger.error("MySQL setup failed — falling back to file storage: %s", exc)
from sqlalchemy.orm import Session # noqa: E402 (may not be installed if file mode)
def get_db():
if _mysql_ready and SessionLocal:
db = SessionLocal()
try:
yield db
finally:
db.close()
else:
yield None
# Pydantic models
class ChatMessage(BaseModel):
session_id: str
message: str
class StartChatRequest(BaseModel):
student_id: str
# #fallback_response_engine
def compose_reply(latest_text: str = "", prev_messages: list = [], known_intent: str = "") -> dict:
"""
Fallback response handler. Used when the LLM is unavailable
or returns an unparseable response.
"""
if _cfg().debug:
logger.debug("compose_reply: text=%r intent=%r", latest_text[:60], known_intent)
text = (latest_text or "").lower().strip()
crisis_patterns = [
r"\bkill myself\b", r"\bsuicide\b", r"\bwant to die\b", r"\bending it all\b",
r"\bhurt myself\b", r"\bself[- ]harm\b", r"\bnever been born\b",
r"\bhadn't been born\b", r"\bbetter off dead\b", r"\bkilling myself\b",
r"\bend it all\b", r"\ball be over\b", r"\bspiraling down\b",
r"\blosing control\b", r"\bwrong with me\b", r"\bfeeling down\b",
r"\bjust die\b", r"\bsuicidal\b", r"\beverything to be over\b",
]
for pat in crisis_patterns:
if known_intent == CRISIS or re.search(pat, text):
return {
"reply": (
"I'm really sorry you're feeling this way. "
"If you are in immediate danger, please call or text 988 for the Suicide & Crisis Lifeline. "
"Would you like me to connect you with your advisor to discuss this sensitive issue?"
),
"intent": CRISIS, "resolved": False, "escalate": True,
}
affirmative = [r"\b(yes|yeah|yep|sure|ok(?:ay)?|that works|sounds good|please|do it|would be great|that'd be great|i'd like that|that would be preferable)\b"]
dismissive = [r"\b(no|nah|nope|not really|don't think so|never mind|forget it|ill be fine|i'll be fine|i'll be okay|i'm okay|im okay|its fine|it's fine|won't be necessary)\b"]
last_bot = None
for m in reversed(prev_messages[:-1]):
if m.get("sender") == "MyACC Virtual Assistant":
last_bot = m
break
if last_bot and last_bot.get("intent") in STUDENT_INTENTS:
intent = last_bot.get("intent")
for pat in dismissive:
if re.search(pat, text):
if intent == CRISIS:
return {"reply": "Understood. If you need help, then I hope you get the help you need. You're not alone.", "intent": RESOLVED, "resolved": True, "escalate": True}
if intent == ESCALATION:
return {"reply": "Alright. Sorry I couldn't be of more help. Have a good rest of your day!", "intent": RESOLVED, "resolved": True, "escalate": False}
if intent == FALLING_GRADES:
return {"reply": "Got it, and I hope my advice helps. Have a great rest of your day!", "intent": RESOLVED, "resolved": True, "escalate": False}
if intent == PANIC:
return {"reply": "I see. Would you like to talk to your advisor about this as soon as possible?", "intent": ESCALATION, "resolved": False, "escalate": False}
for pat in affirmative:
if re.search(pat, text):
if intent == CRISIS:
return {"reply": "I understand. Your advisor will contact you as soon as possible. Please know you're not alone.", "intent": RESOLVED, "resolved": True, "escalate": True}
if intent == ESCALATION:
return {"reply": "Understood. I've put in a request for your advisor to take a look and get back to you. Have a good day!", "intent": RESOLVED, "resolved": True, "escalate": True}
if intent == FALLING_GRADES:
return {"reply": "I understand. I've contacted your advisor, and they'll get back to you as soon as they can. Thank you for reaching out!", "intent": RESOLVED, "resolved": True, "escalate": True}
if intent == PANIC:
return {"reply": "Good, I'm glad to hear that! I hope you have a good rest of your day!", "intent": RESOLVED, "resolved": True, "escalate": False}
elif last_bot and last_bot.get("intent") in ADVISOR_INTENTS:
return {"reply": "Sorry, but my capabilities are a bit limited right now. Please try again later.", "intent": RESOLVED, "resolved": True, "escalate": False}
if known_intent == RESOLVED or re.search(r"\b(stop talking|leave me alone|don't want to talk|not now|it's fine|im fine|i'm fine|resolved|done)\b", text):
return {"reply": "Okay, I understand. If you'd like to talk later or want some help, I'm here for you.", "intent": RESOLVED, "resolved": True, "escalate": False}
if known_intent == ESCALATION or re.search(r"\b(advisor|human|talk to someone|counselor|tutor|someone help|someone to help|someone else|anyone else|escalate)\b", text):
return {"reply": "I'm sorry I couldn't be more useful. I have escalated your issue to your advisor.", "intent": ESCALATION, "resolved": False, "escalate": True}
if known_intent == FALLING_GRADES or re.search(r"\b(falling grades|improve|how.*grade|grade.*falling|grade.*down|raise.*grade(?:s)?|better grades|help.*grade(?:s)?|what should i do|procrastinat|can't keep up|time management|no time|overwhelmed|too much|dropping out)\b", text):
return {
"reply": (
"Here are some steps you can take:
"
"1. Make a list of all your current tasks for the next two weeks and sort them by due date.
"
"2. Prioritize working on these tasks in order.
"
"3. If you have issues completing the assignment, feel free to reach out to the instructors.
"
"Would you like to talk to your advisor about this instead?"
),
"intent": FALLING_GRADES, "resolved": False, "escalate": False,
}
if re.search(r"\b(missing assignment(?:s)?|missed assignment(?:s)?|failed|failing|zero|incomplete|deadline(?:s)?)\b", text):
return {
"reply": (
"Missing assignments can often be fixed by contacting the instructor and explaining your situation. "
"Most instructors are willing to work with you — their goal is to see you succeed! "
"If your instructor cannot help you, you can always talk to your advisor instead. "
"Would you like for me to contact your advisor at this time?"
),
"intent": FALLING_GRADES, "resolved": False, "escalate": False,
}
if known_intent == PANIC or re.search(r"\b(stress|stressed|anxious|anxiety|depressed|panic)\b", text):
return {
"reply": (
"I'm sorry you're feeling this way. If you're feeling overwhelmed, take a small 5-minute break — "
"you might surprise yourself how much it helps! "
"If you're having a panic attack, force yourself to take slow, deep breaths. "
"Did this advice help you?"
),
"intent": PANIC, "resolved": False, "escalate": False,
}
if re.search(r"\breview (this )?case\b|\breview student\b|\bcase details\b", text):
return {"reply": "Sure. Which student would you like to review? Your current students are: Ashley.", "intent": ADVISOR_REVIEW_PROFILE, "resolved": False, "escalate": False}
if re.search(r"\bcontact student\b|\breach out\b|\bemail student\b", text):
return {"reply": "Normally, I'd be able to help you draft a message, but my capabilities are a bit limited right now. Please try again later.", "intent": ADVISOR_DRAFT_MESSAGE, "resolved": True, "escalate": False}
if re.search(r"\bappointment\b|\bmeet with\b|\bbook a(n)?\b", text):
return {"reply": "Sure, what student would you like to set up an appointment with?", "intent": ADVISOR_APPOINTMENT, "resolved": False, "escalate": False}
if len(text.split()) <= 3:
return {"reply": "I'm not sure I understood your intent. Could you tell me in greater detail?", "intent": UNKNOWN, "resolved": False, "escalate": False}
return {"reply": "I'm sorry, I either couldn't understand what you said or what you asked for isn't within my capabilities. Please clearly rephrase your question.", "intent": UNKNOWN, "resolved": False, "escalate": False}
# Helper: list open sessions
def list_open_sessions(db: Optional[Session], current_user: Dict) -> List[Dict]:
sessions: List[Dict] = []
username = current_user.get("username")
if not username:
return sessions
if _mysql_ready and db:
try:
rows = db.query(ChatSession).filter(
ChatSession.student_id == username,
ChatSession.status != RESOLVED,
ChatSession.status != "provisional",
).order_by(ChatSession.last_updated.desc()).all()
for r in rows:
msgs = json.loads(r.messages) if r.messages else []
preview = (msgs[-1].get("text", "") if msgs else "")[:30]
sessions.append({
"session_id": r.session_id,
"preview": preview,
"status": r.status,
"last_updated": r.last_updated.isoformat() if r.last_updated else None,
})
except Exception as exc:
logger.error("list_open_sessions MySQL error: %s", exc)
return sessions
for p in _data_dir().glob("*.json"):
try:
s = json.loads(p.read_text(encoding="utf-8"))
except Exception:
continue
if s.get("student_id") != username:
continue
if s.get("status") in (RESOLVED, "provisional"):
continue
msgs = s.get("messages", [])
preview = ((msgs[-1].get("text", "") if msgs else "") or "")[:30]
sessions.append({
"session_id": s.get("session_id"),
"preview": preview,
"status": s.get("status", "in_progress"),
"last_updated": s.get("last_updated"),
})
sessions.sort(key=lambda it: it.get("last_updated") or "", reverse=True)
return sessions
# Endpoints
@router.post("/start")
async def start_chat(
intent: Optional[str] = Query(None),
db: Optional[Session] = Depends(get_db),
current_user: Dict = Depends(get_current_user),
):
"""Start or resume a chat session."""
session_id = None
# Reuse an existing provisional session
for p in _data_dir().glob("*.json"):
try:
s = json.loads(p.read_text(encoding="utf-8"))
except Exception:
continue
if s.get("status") == "provisional":
session_id = s.get("session_id")
break
if session_id is None:
session_id = str(uuid4())
now = _now_iso()
role = current_user.get("role", "student")
# Build opening bot message
bot_text = ""
intent_found = True
if role == "student":
student_name = (current_user.get("username") or "Student")
student_name = student_name.replace("-", " ").replace("_", " ").title()
bot_text = f"Hello, {student_name}. "
if intent == FALLING_GRADES:
bot_text += "I wanted to check in with you because I noticed you haven't been very active in Blackboard lately. Just wanted to see how things were going."
elif intent == MISSED_CLASSES:
bot_text += "I've noticed you've missed several classes and wanted to check in with you."
elif intent == FINANCIAL_AID:
bot_text += "I see your FAFSA application has been received, but it's showing as incomplete. You're missing two documents: proof of household size and a signed verification worksheet."
elif intent == DEGREE_PLAN:
bot_text += "I wanted to talk about your course schedule. A few classes you're enrolled in don't directly count toward your degree plan. How are you feeling about your program so far?"
else:
intent_found = False
else:
intent_found = False
names = {"advisor": "John Doe", "financial-aid": "Joey", "counselor": "Joanne"}
bot_text = f"Hello, {names.get(role, 'there')}. "
if not intent_found:
bot_text += "What can I help you with today?"
initial_messages = [{
"sender": "MyACC Virtual Assistant",
"text": bot_text,
"timestamp": now,
"intent": intent or None,
}]
session = {
"session_id": session_id,
"student_id": current_user.get("username"),
"messages": initial_messages,
"status": "provisional",
"created": now,
"last_updated": now,
}
_save_session(session)
return {"session_id": session_id, "messages": initial_messages}
@router.get("/advisor/sessions")
async def advisor_sessions(
student_id: Optional[str] = Query(None),
db: Optional[Session] = Depends(get_db),
current_user: Dict = Depends(get_current_user),
):
"""Return chat sessions for advisor review."""
sessions: List[Dict] = []
username = current_user.get("username")
role = current_user.get("role")
def _append(sid, stud, msgs, status, created, last_updated):
msgs = msgs or []
last_msg = msgs[-1] if msgs else None
preview = ((last_msg.get("text") or "")[:120]).strip() if last_msg else ""
last_intent = last_msg.get("intent") if last_msg else None
escalate_flag = status == "escalate" or any(
m.get("intent") in (CRISIS, ESCALATION) for m in msgs
)
sessions.append({
"session_id": sid, "student_id": stud, "status": status,
"created": created, "last_updated": last_updated,
"messages_count": len(msgs), "last_message_preview": preview,
"last_intent": last_intent, "escalate": escalate_flag,
})
for p in _data_dir().glob("*.json"):
try:
s = json.loads(p.read_text(encoding="utf-8"))
except Exception:
continue
sid = s.get("session_id")
stud = s.get("student_id")
status = s.get("status", "in_progress")
if status == "provisional":
continue
if role == "student":
if stud != username:
continue
else:
if stud == username or (student_id and stud != student_id):
continue
_append(sid, stud, s.get("messages", []), status, s.get("created"), s.get("last_updated"))
sessions.sort(key=lambda it: it.get("last_updated") or it.get("created") or "", reverse=True)
return {"sessions": sessions}
@router.get("/session/{session_id}")
async def get_session(
session_id: str,
db: Optional[Session] = Depends(get_db),
current_user: Dict = Depends(get_current_user),
):
def _allowed(owner, user):
if not user:
return False
if owner and owner == user.get("username"):
return True
return user.get("role") != "student"
session = _load_session(session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found.")
if not _allowed(session.get("student_id"), current_user):
raise HTTPException(status_code=403, detail="Not authorised to view this session.")
return {
"session_id": session.get("session_id"),
"messages": session.get("messages", []),
"status": session.get("status", "in_progress"),
"created": session.get("created"),
"last_updated": session.get("last_updated"),
}
@router.post("/session/{session_id}/resolve")
async def resolve_session(
session_id: str,
db: Optional[Session] = Depends(get_db),
current_user: Dict = Depends(get_current_user),
):
def _allowed(owner, user):
if not user:
return False
if owner and owner == user.get("username"):
return True
return user.get("role") != "student"
session = _load_session(session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found.")
if not _allowed(session.get("student_id"), current_user):
raise HTTPException(status_code=403, detail="Not authorised.")
msgs = session.get("messages", []) or []
msgs.append({
"sender": "MyACC Virtual Assistant",
"text": "Conversation closed by user.",
"intent": RESOLVED,
"timestamp": _now_iso(),
})
session["messages"] = msgs
session["status"] = RESOLVED
session["last_updated"] = _now_iso()
_save_session(session)
logger.info("Session %s resolved by user", session_id)
return {"session_id": session["session_id"], "status": session["status"], "messages": msgs}
@router.post("/message")
async def chat_message(
msg: ChatMessage,
db: Optional[Session] = Depends(get_db),
current_user: Dict = Depends(get_current_user),
):
"""
Main chat endpoint.
Order of operations:
1. Load session
2. Append user message
3. Run safeguard (crisis patterns) - if triggered, skip LLM call
4. Call LangChain + RAG + OpenAI for intelligent response
5. Fall back to rule-based compose_reply() if LLM fails
6. Append bot reply, save session, return
"""
start_time = time.monotonic()
cfg = _cfg()
session = _load_session(msg.session_id)
if not session:
raise HTTPException(status_code=400, detail="Invalid session ID")
messages = session.get("messages", [])
status = session.get("status", "in_progress")
if status == RESOLVED:
reply = "This conversation has already ended. Please start a new chat session if you'd like to talk again."
messages.append({"sender": "MyACC Virtual Assistant", "text": reply, "intent": RESOLVED, "timestamp": _now_iso()})
return {"messages": messages, "status": status}
role = current_user.get("role", "student")
messages.append({
"sender": role,
"text": msg.message,
"intent": UNKNOWN,
"timestamp": _now_iso(),
})
# Step 3: Safeguard (always first)
reply_data = safeguard_check(msg.message, use_llm=True, history=messages)
if reply_data:
if cfg.debug:
logger.debug("Safeguard triggered: intent=%s", reply_data["intent"])
else:
# Step 4: LangChain + RAG + OpenAI
try:
# Pass the AUTHENTICATED student's id (from session, never chat text)
# so the router can fetch THIS student's own personal data.
sid = current_user.get("student_id") or current_user.get("username")
llm_result = await generate_reply(messages, role=role, student_id=sid)
if llm_result:
reply_data = llm_result
if cfg.debug:
logger.debug("LLM reply: intent=%s resolved=%s", llm_result["intent"], llm_result["resolved"])
else:
logger.warning("LLM returned no result — using fallback.")
except Exception as exc:
logger.error("LLM call raised exception: %s", exc)
# Step 5: Rule-based fallback
if not reply_data:
reply_data = compose_reply(msg.message, messages)
if cfg.debug:
logger.debug("Fallback used: intent=%s", reply_data["intent"])
if reply_data.get("intent") == CRISIS:
_send_crisis_alert(current_user, msg.message)
# Step 6: Append reply and save
messages.append({
"sender": "MyACC Virtual Assistant",
"text": reply_data["reply"],
"intent": reply_data.get("intent", UNKNOWN),
"timestamp": _now_iso(),
})
if status == "provisional":
status = "in_progress"
if reply_data.get("escalate"):
status = "escalate"
if reply_data.get("resolved"):
status = RESOLVED
session["messages"] = messages
session["status"] = status
session["last_updated"] = _now_iso()
_save_session(session)
# Respect minimum response delay
elapsed = time.monotonic() - start_time
if elapsed < cfg.min_response_seconds:
await asyncio.sleep(cfg.min_response_seconds - elapsed)
return {"messages": messages, "status": status}
@router.get("/notifications/crisis")
async def get_crisis_notifications(
current_user: Dict = Depends(get_current_user),
):
"""Fetch all crisis notifications for advisor dashboard."""
role = current_user.get("role")
if role == "student":
raise HTTPException(status_code=403, detail="Not authorized to view notifications.")
notifications = _load_all_notifications()
notifications.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
return {"notifications": notifications}
@router.post("/notifications/crisis/{notification_id}/read")
async def mark_notification_read(
notification_id: str,
current_user: Dict = Depends(get_current_user),
):
"""Mark a crisis notification as read."""
role = current_user.get("role")
if role == "student":
raise HTTPException(status_code=403, detail="Not authorized to mark notifications.")
path = _notifications_dir() / f"{notification_id}.json"
if not path.exists():
raise HTTPException(status_code=404, detail="Notification not found.")
try:
notification = json.loads(path.read_text(encoding="utf-8"))
notification["read"] = True
path.write_text(json.dumps(notification, indent=2, ensure_ascii=False), encoding="utf-8")
logger.info("Marked notification %s as read", notification_id)
return {"notification_id": notification_id, "read": True}
except HTTPException:
raise
except Exception as exc:
logger.error("Error marking notification as read: %s", exc)
raise HTTPException(status_code=500, detail="Error updating notification.")