# phase/Student_view/chatbot.py import os import datetime import traceback import streamlit as st # --- use our backend client (utils/api.py) --- try: from utils import api as backend except ModuleNotFoundError: # fallback if running from a different CWD import sys, pathlib ROOT = pathlib.Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from utils import api as backend TUTOR_WELCOME = "Hi! I'm your AI Financial Tutor. What would you like to learn today?" # ------------------------------- # History helpers # ------------------------------- def add_message(text: str, sender: str): if "messages" not in st.session_state: st.session_state.messages = [] st.session_state.messages.append( { "id": str(datetime.datetime.now().timestamp()), "text": (text or "").strip(), "sender": sender, "timestamp": datetime.datetime.now(), } ) def _coerce_ts(ts): if isinstance(ts, datetime.datetime): return ts if isinstance(ts, (int, float)): try: return datetime.datetime.fromtimestamp(ts) except Exception: return None if isinstance(ts, str): for parser in (datetime.datetime.fromisoformat, lambda s: datetime.datetime.fromtimestamp(float(s))): try: return parser(ts) except Exception: pass return None def _normalize_messages(): msgs = st.session_state.get("messages", []) normed = [] now = datetime.datetime.now() for m in msgs: text = (m.get("text") or "").strip() sender = m.get("sender") or "user" ts = _coerce_ts(m.get("timestamp")) or now normed.append({**m, "text": text, "sender": sender, "timestamp": ts}) st.session_state.messages = normed def _history_for_backend(): """Convert our local history into [{role, content}] for the backend.""" hist = [] for m in st.session_state.get("messages", []): text = (m.get("text") or "").strip() if not text: continue role = "assistant" if (m.get("sender") == "assistant") else "user" hist.append({"role": role, "content": text}) return hist # ------------------------------- # Reply via backend (/chat) # ------------------------------- def _reply_via_backend(user_text: str) -> str: # Defaults: use selected lesson/level if present lesson_id = st.session_state.get("current_lesson_id") or 0 level_slug = (st.session_state.get("user", {}).get("level") or "beginner").strip().lower() try: answer = backend.chat_ai( query=user_text, lesson_id=lesson_id, level_slug=level_slug, history=_history_for_backend(), ) return (answer or "").strip() except Exception as e: err_text = "".join(traceback.format_exception_only(type(e), e)).strip() return f"⚠️ Chat failed: {err_text}" # ------------------------------- # Streamlit page # ------------------------------- def show_page(): st.title("🤖 AI Financial Tutor") st.caption("Get personalized help with your financial questions") if "messages" not in st.session_state: st.session_state.messages = [{ "id": "1", "text": TUTOR_WELCOME, "sender": "assistant", "timestamp": datetime.datetime.now() }] if "is_typing" not in st.session_state: st.session_state.is_typing = False _normalize_messages() chat_container = st.container() with chat_container: for msg in st.session_state.messages: t = msg["timestamp"].strftime("%H:%M") if msg.get("sender") == "assistant": bubble = ( "