# utils/api.py import os, json, requests from urllib3.util.retry import Retry from requests.adapters import HTTPAdapter # ---- Setup ---- BACKEND = (os.getenv("BACKEND_URL") or "").strip().rstrip("/") if not BACKEND: # Fail fast at import; Streamlit will surface this in the sidebar on first run raise RuntimeError("BACKEND_URL is not set in Space secrets.") # Accept either BACKEND_TOKEN or HF_TOKEN TOKEN = (os.getenv("BACKEND_TOKEN") or os.getenv("HF_TOKEN") or "").strip() DEFAULT_TIMEOUT = int(os.getenv("BACKEND_TIMEOUT", "30")) _session = requests.Session() # Light retry for transient network/server blips retry = Retry( total=3, connect=3, read=3, backoff_factor=0.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["GET", "POST", "PUT", "PATCH", "DELETE"]), ) _session.mount("https://", HTTPAdapter(max_retries=retry)) _session.mount("http://", HTTPAdapter(max_retries=retry)) # Default headers _session.headers.update({ "Accept": "application/json, */*;q=0.1", "User-Agent": "FinEdu-Frontend/1.0 (+spaces)", }) if TOKEN: _session.headers["Authorization"] = f"Bearer {TOKEN}" def _json_or_raise(resp: requests.Response): ctype = resp.headers.get("content-type", "") if "application/json" in ctype: return resp.json() # Try to parse anyway; show a helpful error if not JSON try: return resp.json() except Exception: snippet = (resp.text or "")[:300] raise RuntimeError(f"Expected JSON but got {ctype or 'unknown'}:\n{snippet}") def _req(method: str, path: str, **kw): if not path.startswith("/"): path = "/" + path url = f"{BACKEND}{path}" kw.setdefault("timeout", DEFAULT_TIMEOUT) try: r = _session.request(method, url, **kw) r.raise_for_status() except requests.HTTPError as e: body = "" try: body = r.text[:500] except Exception: pass status = getattr(r, "status_code", "?") # Give nicer hints for common auth misconfigs if status in (401, 403): raise RuntimeError( f"{method} {path} failed [{status}] – auth rejected. " f"Check BACKEND_TOKEN/HF_TOKEN permissions and that the backend Space is private/readable." ) from e raise RuntimeError(f"{method} {path} failed [{status}]: {body}") from e except requests.RequestException as e: raise RuntimeError(f"{method} {path} failed: {e.__class__.__name__}: {e}") from e return r # ---- Health ---- def health(): # Prefer /health but allow root fallback if you change the backend later try: return _json_or_raise(_req("GET", "/health")) except Exception: # best-effort fallback try: _req("GET", "/") return {"ok": True} except Exception: return {"ok": False} #---helpers # --- Optional API prefix (e.g., "/api" or "/v1") API_PREFIX_ENV = (os.getenv("BACKEND_API_PREFIX") or "").strip().rstrip("/") def _prefixes(): # Try configured prefix first, then common fallbacks seen, out = set(), [] for p in [API_PREFIX_ENV, "", "/api", "/v1", "/api/v1"]: p = (p or "").strip() p = "" if p == "" else ("/" + p.strip("/")) if p not in seen: out.append(p) seen.add(p) return out def _try_candidates(method: str, candidates: list[tuple[str, dict]]): """ candidates: list of (path, request_kwargs) where path starts with "/" and kwargs may include {'params':..., 'json':...}. Tries multiple prefixes (e.g., "", "/api", "/v1") and returns JSON for first 2xx. Auth errors (401/403) are raised immediately. """ tried = [] for pref in _prefixes(): for path, kw in candidates: url = f"{BACKEND}{pref}{path}" tried.append(f"{method} {url}") try: r = _session.request(method, url, timeout=DEFAULT_TIMEOUT, **kw) except requests.RequestException as e: # transient error: keep trying others continue if r.status_code in (401, 403): snippet = (r.text or "")[:200] raise RuntimeError(f"{method} {path} auth failed [{r.status_code}]: {snippet}") if 200 <= r.status_code < 300: return _json_or_raise(r) # 404/405/etc.: try next candidate raise RuntimeError("No matching endpoint for this operation. Tried:\n- " + "\n- ".join(tried)) #--helpers for student_db.py def user_stats(student_id: int): return _req("GET", f"/students/{student_id}/stats").json() def list_assignments_for_student(student_id: int): return _req("GET", f"/students/{student_id}/assignments").json() def student_quiz_average(student_id: int): d = _req("GET", f"/students/{student_id}/quiz_avg").json() # Normalize common shapes: {"avg": 82}, {"score_pct": "82"}, "82", 82 if isinstance(d, dict): for k in ("avg", "average", "score_pct", "score", "value"): if k in d: v = d[k] break else: # fallback: first numeric-ish value v = next((vv for vv in d.values() if isinstance(vv, (int, float, str))), 0) else: v = d try: # handle strings like "82" or "82%" return int(round(float(str(v).strip().rstrip("%")))) except Exception: return 0 def recent_lessons_for_student(student_id: int, limit: int = 5): return _req("GET", f"/students/{student_id}/recent", params={"limit": limit}).json() # # --- Teacher endpoints (backend Space) --- # def create_class(teacher_id: int, name: str): # return _json_or_raise(_req("POST", f"/teachers/{teacher_id}/classes", # json={"name": name})) # def teacher_tiles(teacher_id: int): # return _json_or_raise(_req("GET", f"/teachers/{teacher_id}/tiles")) # def list_classes_by_teacher(teacher_id: int): # return _json_or_raise(_req("GET", f"/teachers/{teacher_id}/classes")) # def class_student_metrics(class_id: int): # return _json_or_raise(_req("GET", f"/classes/{class_id}/student_metrics")) # def class_weekly_activity(class_id: int): # return _json_or_raise(_req("GET", f"/classes/{class_id}/weekly_activity")) # def class_progress_overview(class_id: int): # return _json_or_raise(_req("GET", f"/classes/{class_id}/progress_overview")) # def class_recent_activity(class_id: int, limit=6, days=30): # return _json_or_raise(_req("GET", f"/classes/{class_id}/recent_activity", # params={"limit": limit, "days": days})) # def list_students_in_class(class_id: int): # return _json_or_raise(_req("GET", f"/classes/{class_id}/students")) # Optional if you want to compute levels server-side def level_from_xp(xp: int): return _json_or_raise(_req("GET", "/levels/from_xp", params={"xp": xp}))["level"] #--teacherlink.py helpers def join_class_by_code(student_id: int, code: str): d = _json_or_raise(_req("POST", f"/students/{student_id}/classes/join", json={"code": code})) # backend may return {"class_id": ...} or full class object; both are fine return d.get("class_id", d) def list_classes_for_student(student_id: int): return _json_or_raise(_req("GET", f"/students/{student_id}/classes")) def class_content_counts(class_id: int): return _json_or_raise(_req("GET", f"/classes/{class_id}/counts")) def student_class_progress(student_id: int, class_id: int): return _json_or_raise(_req("GET", f"/classes/{class_id}/students/{student_id}/progress")) def leave_class(student_id: int, class_id: int): # could also be DELETE /classes/{class_id}/students/{student_id} _json_or_raise(_req("POST", f"/classes/{class_id}/leave", json={"student_id": student_id})) return True def student_assignments_for_class(student_id: int, class_id: int): return _json_or_raise(_req("GET", f"/classes/{class_id}/students/{student_id}/assignments")) # ---------- TEACHERS / CLASSES / CONTENT (BACKEND ROUTES THAT EXIST) ---------- # Classes def create_class(teacher_id: int, name: str): # Backend has POST /teachers/{teacher_id}/classes with body {name} return _try_candidates("POST", [ (f"/teachers/{teacher_id}/classes", {"json": {"name": name}}), # fallbacks if you ever rename: ("/classes", {"json": {"teacher_id": teacher_id, "name": name}}), ]) def list_classes_by_teacher(teacher_id: int): return _try_candidates("GET", [ (f"/teachers/{teacher_id}/classes", {}), ]) def list_students_in_class(class_id: int): # exact route in backend return _json_or_raise(_req("GET", f"/classes/{class_id}/students")) def class_content_counts(class_id: int): return _try_candidates("GET", [ (f"/classes/{class_id}/content_counts", {}), (f"/classes/{class_id}/counts", {}), ]) def list_class_assignments(class_id: int): return _json_or_raise(_req("GET", f"/classes/{class_id}/assignments")) def class_analytics(class_id: int): return _json_or_raise(_req("GET", f"/classes/{class_id}/analytics")) def teacher_tiles(teacher_id: int): return _json_or_raise(_req("GET", f"/teachers/{teacher_id}/tiles")) def class_student_metrics(class_id: int): # backend: /classes/{id}/students/metrics return _try_candidates("GET", [ (f"/classes/{class_id}/students/metrics", {}), # tolerant fallbacks: (f"/classes/{class_id}/student_metrics", {}), (f"/classes/{class_id}/students", {}), # older shape (list of students) ]) def class_weekly_activity(class_id: int): # backend: /classes/{id}/activity/weekly return _try_candidates("GET", [ (f"/classes/{class_id}/activity/weekly", {}), (f"/classes/{class_id}/weekly_activity", {}), ]) def class_progress_overview(class_id: int): # backend: /classes/{id}/progress return _try_candidates("GET", [ (f"/classes/{class_id}/progress", {}), (f"/classes/{class_id}/progress_overview", {}), ]) def class_recent_activity(class_id: int, limit=6, days=30): # backend: /classes/{id}/activity/recent return _try_candidates("GET", [ (f"/classes/{class_id}/activity/recent", {"params": {"limit": limit, "days": days}}), (f"/classes/{class_id}/recent_activity", {"params": {"limit": limit, "days": days}}), ]) # Lessons def list_lessons_by_teacher(teacher_id: int): return _json_or_raise(_req("GET", f"/teachers/{teacher_id}/lessons")) def create_lesson(teacher_id: int, title: str, description: str, subject: str, level: str, sections: list[dict]): payload = { "title": title, "description": description, "subject": subject, "level": level, "sections": sections, } # backend route: d = _try_candidates("POST", [ (f"/teachers/{teacher_id}/lessons", {"json": payload}), # fallback if you later add a flat /lessons route: ("/lessons", {"json": {"teacher_id": teacher_id, **payload}}), ]) # tolerate both {"lesson_id": N} or full object with id return d.get("lesson_id", d.get("id", d)) def get_lesson(lesson_id: int): return _json_or_raise(_req("GET", f"/lessons/{lesson_id}")) def update_lesson(lesson_id: int, teacher_id: int, title: str, description: str, subject: str, level: str, sections: list[dict]): d = _req("PUT", f"/lessons/{lesson_id}", json={ "teacher_id": teacher_id, "title": title, "description": description, "subject": subject, "level": level, "sections": sections }).json() return bool(d.get("ok", True)) def delete_lesson(lesson_id: int, teacher_id: int): d = _req("DELETE", f"/lessons/{lesson_id}", json={"teacher_id": teacher_id}).json() return bool(d.get("ok", True)), d.get("message", "") # Quizzes def list_quizzes_by_teacher(teacher_id: int): return _json_or_raise(_req("GET", f"/teachers/{teacher_id}/quizzes")) def create_quiz(lesson_id: int, title: str, items: list[dict], settings: dict): d = _req("POST", "/quizzes", json={ "lesson_id": lesson_id, "title": title, "items": items, "settings": settings }).json() return d.get("quiz_id", d.get("id", d)) def get_quiz(quiz_id: int): return _json_or_raise(_req("GET", f"/quizzes/{quiz_id}")) def get_quiz(quiz_id: int): # NEW wrapper that hits GET /quizzes/{quiz_id} return _req("GET", f"/quizzes/{quiz_id}") def submit_quiz(student_id, assignment_id, quiz_id, score, total, details): payload = {"student_id": student_id, "assignment_id": assignment_id, "quiz_id": quiz_id, "score": score, "total": total, "details": details} return _req("POST", "/quizzes/submit", json=payload) def update_quiz(quiz_id: int, teacher_id: int, title: str, items: list[dict], settings: dict): d = _req("PUT", f"/quizzes/{quiz_id}", json={ "teacher_id": teacher_id, "title": title, "items": items, "settings": settings }).json() return bool(d.get("ok", True)) def delete_quiz(quiz_id: int, teacher_id: int): d = _req("DELETE", f"/quizzes/{quiz_id}", json={"teacher_id": teacher_id}).json() return bool(d.get("ok", True)), d.get("message", "") def list_assigned_students_for_lesson(lesson_id: int): return _json_or_raise(_req("GET", f"/lessons/{lesson_id}/assignees")) def list_assigned_students_for_quiz(quiz_id: int): return _json_or_raise(_req("GET", f"/quizzes/{quiz_id}/assignees")) # Assignments def assign_to_class(lesson_id: int | None, quiz_id: int | None, class_id: int, teacher_id: int, due_at: str | None = None): # backend route name is /assign (not /assignments) d = _try_candidates("POST", [ ("/assign", {"json": { "lesson_id": lesson_id, "quiz_id": quiz_id, "class_id": class_id, "teacher_id": teacher_id, "due_at": due_at }}), ("/assignments", {"json": { "lesson_id": lesson_id, "quiz_id": quiz_id, "class_id": class_id, "teacher_id": teacher_id, "due_at": due_at }}), ]) return bool(d.get("ok", True)) # # ---- Classes / Teacher endpoints (tolerant) ---- # def create_class(teacher_id: int, name: str): # return _try_candidates("POST", [ # (f"/teachers/{teacher_id}/classes", {"json": {"name": name}}), # (f"/teachers/{teacher_id}/classrooms",{"json": {"name": name}}), # ("/classes", {"json": {"teacher_id": teacher_id, "name": name}}), # ("/classrooms", {"json": {"teacher_id": teacher_id, "name": name}}), # ]) # def list_classes_by_teacher(teacher_id: int): # return _try_candidates("GET", [ # (f"/teachers/{teacher_id}/classes", {}), # (f"/teachers/{teacher_id}/classrooms", {}), # (f"/classes/by-teacher/{teacher_id}", {}), # (f"/classrooms/by-teacher/{teacher_id}", {}), # ("/classes", {"params": {"teacher_id": teacher_id}}), # ("/classrooms", {"params": {"teacher_id": teacher_id}}), # ]) # def list_students_in_class(class_id: int): # return _try_candidates("GET", [ # (f"/classes/{class_id}/students", {}), # (f"/classrooms/{class_id}/students", {}), # ("/students", {"params": {"class_id": class_id}}), # ]) # def class_content_counts(class_id: int): # return _try_candidates("GET", [ # (f"/classes/{class_id}/content_counts", {}), # (f"/classrooms/{class_id}/content_counts", {}), # (f"/classes/{class_id}/counts", {}), # (f"/classrooms/{class_id}/counts", {}), # ]) # def list_class_assignments(class_id: int): # return _try_candidates("GET", [ # (f"/classes/{class_id}/assignments", {}), # (f"/classrooms/{class_id}/assignments", {}), # ("/assignments", {"params": {"class_id": class_id}}), # ]) # def class_analytics(class_id: int): # return _try_candidates("GET", [ # (f"/classes/{class_id}/analytics", {}), # (f"/classrooms/{class_id}/analytics", {}), # ]) # #--contentmanage.py helpers # # ---------- Teacher/content management endpoints (backend Space) ---------- # def list_classes_by_teacher(teacher_id: int): # return _req("GET", f"/teachers/{teacher_id}/classes").json() # def list_all_students_for_teacher(teacher_id: int): # return _req("GET", f"/teachers/{teacher_id}/students").json() # def list_lessons_by_teacher(teacher_id: int): # return _req("GET", f"/teachers/{teacher_id}/lessons").json() # def list_quizzes_by_teacher(teacher_id: int): # return _req("GET", f"/teachers/{teacher_id}/quizzes").json() # def create_lesson(teacher_id: int, title: str, description: str, subject: str, level: str, sections: list[dict]): # d = _req("POST", "/lessons", json={ # "teacher_id": teacher_id, "title": title, "description": description, # "subject": subject, "level": level, "sections": sections # }).json() # return d["lesson_id"] # def update_lesson(lesson_id: int, teacher_id: int, title: str, description: str, subject: str, level: str, sections: list[dict]): # d = _req("PUT", f"/lessons/{lesson_id}", json={ # "teacher_id": teacher_id, "title": title, "description": description, # "subject": subject, "level": level, "sections": sections # }).json() # return bool(d.get("ok", True)) # def delete_lesson(lesson_id: int, teacher_id: int): # d = _req("DELETE", f"/lessons/{lesson_id}", json={"teacher_id": teacher_id}).json() # return bool(d.get("ok", True)), d.get("message", "") # def get_lesson(lesson_id: int): # return _req("GET", f"/lessons/{lesson_id}").json() # {"lesson":{...}, "sections":[...]} # def create_quiz(lesson_id: int, title: str, items: list[dict], settings: dict): # d = _req("POST", "/quizzes", json={"lesson_id": lesson_id, "title": title, "items": items, "settings": settings}).json() # return d["quiz_id"] # def update_quiz(quiz_id: int, teacher_id: int, title: str, items: list[dict], settings: dict): # d = _req("PUT", f"/quizzes/{quiz_id}", json={"teacher_id": teacher_id, "title": title, "items": items, "settings": settings}).json() # return bool(d.get("ok", True)) # def delete_quiz(quiz_id: int, teacher_id: int): # d = _req("DELETE", f"/quizzes/{quiz_id}", json={"teacher_id": teacher_id}).json() # return bool(d.get("ok", True)), d.get("message", "") # def list_assigned_students_for_lesson(lesson_id: int): # return _req("GET", f"/lessons/{lesson_id}/assignees").json() # def list_assigned_students_for_quiz(quiz_id: int): # return _req("GET", f"/quizzes/{quiz_id}/assignees").json() # def assign_to_class(lesson_id: int | None, quiz_id: int | None, class_id: int, teacher_id: int): # d = _req("POST", "/assignments", json={ # "lesson_id": lesson_id, "quiz_id": quiz_id, "class_id": class_id, "teacher_id": teacher_id # }).json() # return bool(d.get("ok", True)) # #-- studentlist helpers # def list_classes_by_teacher(teacher_id: int): # return _req("GET", f"/teachers/{teacher_id}/classes").json() # def get_class(class_id: int): # return _req("GET", f"/classes/{class_id}").json() # def class_student_metrics(class_id: int): # # expected to return list of rows with fields used in the UI # return _req("GET", f"/classes/{class_id}/students").json() # def list_assignments_for_student(student_id: int): # return _req("GET", f"/students/{student_id}/assignments").json() # ---------- LLM-based quiz generation (backend uses GEN_MODEL) ---------- def generate_quiz_from_text(content: str, n_questions: int = 5, subject: str = "finance", level: str = "beginner"): """ Backend should read GEN_MODEL from env and use your chosen model (llama-3.1-8b-instruct). Return shape: {"items":[{"question":...,"options":[...],"answer_key":"A"}]} """ return _req("POST", "/quiz/generate", json={ "content": content, "n_questions": n_questions, "subject": subject, "level": level }).json() # ---- Legacy agent endpoints (keep) ---- def start_agent(student_id: int, lesson_id: int, level_slug: str): return _json_or_raise(_req("POST", "/agent/start", json={"student_id": student_id, "lesson_id": lesson_id, "level_slug": level_slug})) def get_quiz(student_id: int, lesson_id: int, level_slug: str): d = _json_or_raise(_req("POST", "/agent/quiz", json={"student_id": student_id, "lesson_id": lesson_id, "level_slug": level_slug})) return d["items"] def grade_quiz(student_id: int, lesson_id: int, level_slug: str, answers: list[str], assignment_id: int | None = None): d = _json_or_raise(_req("POST", "/agent/grade", json={"student_id": student_id, "lesson_id": lesson_id, "level_slug": level_slug, "answers": answers, "assignment_id": assignment_id})) return d["score"], d["total"] def next_step(student_id: int, lesson_id: int, level_slug: str, answers: list[str], assignment_id: int | None = None): return _json_or_raise(_req("POST", "/agent/coach_or_celebrate", json={"student_id": student_id, "lesson_id": lesson_id, "level_slug": level_slug, "answers": answers, "assignment_id": assignment_id})) # ---- Auth ---- def login(email: str, password: str): return _json_or_raise(_req("POST", "/auth/login", json={"email": email, "password": password})) def signup_student(name: str, email: str, password: str, level_label: str, country_label: str): payload_student = { "name": name, "email": email, "password": password, "level_label": level_label, "country_label": country_label } # Prefer dedicated route; fall back to /auth/register with role return _try_candidates("POST", [ ("/auth/signup/student", {"json": payload_student}), ("/auth/register", {"json": { "role": "student", "name": name, "email": email, "password": password, "level": level_label, "country": country_label }}), ]) def signup_teacher(title: str, name: str, email: str, password: str): payload_teacher = {"title": title, "name": name, "email": email, "password": password} return _try_candidates("POST", [ ("/auth/signup/teacher", {"json": payload_teacher}), ("/auth/register", {"json": { "role": "teacher", "title": title, "name": name, "email": email, "password": password }}), ]) # ---- New LangGraph-backed endpoints ---- def fetch_lesson_content(lesson: str, module: str, topic: str): r = _json_or_raise(_req("POST", "/lesson", json={"lesson": lesson, "module": module, "topic": topic})) return r["lesson_content"] def submit_lesson_quiz(lesson: str, module: str, topic: str, responses: dict): return _json_or_raise(_req("POST", "/lesson-quiz", json={"lesson": lesson, "module": module, "topic": topic, "responses": responses})) def submit_practice_quiz(lesson: str, responses: dict): return _json_or_raise(_req("POST", "/practice-quiz", json={"lesson": lesson, "responses": responses})) def send_to_chatbot(messages: list[dict]): return _json_or_raise(_req("POST", "/chatbot", json={"messages": messages})) # --- Game API helpers --- def record_money_match_play(user_id: int, target: int, total: int, elapsed_ms: int, matched: bool, gained_xp: int): payload = { "user_id": user_id, "target": target, "total": total, "elapsed_ms": elapsed_ms, "matched": matched, "gained_xp": gained_xp, } return _try_candidates("POST", [ ("/games/money_match/record", {"json": payload}), ]) def record_budget_builder_play(user_id: int, weekly_allowance: int, budget_score: int, elapsed_ms: int, allocations: list[dict], gained_xp: int | None): payload = { "user_id": user_id, "weekly_allowance": weekly_allowance, "budget_score": budget_score, "elapsed_ms": elapsed_ms, "allocations": allocations, "gained_xp": gained_xp, } return _try_candidates("POST", [ ("/games/budget_builder/record", {"json": payload}), ]) def record_debt_dilemma_play(user_id: int, loans_cleared: int, mistakes: int, elapsed_ms: int, gained_xp: int): payload = { "user_id": user_id, "loans_cleared": loans_cleared, "mistakes": mistakes, "elapsed_ms": elapsed_ms, "gained_xp": gained_xp, } return _try_candidates("POST", [ ("/games/debt_dilemma/record", {"json": payload}), ("/api/games/debt_dilemma/record", {"json": payload}), ("/api/v1/games/debt_dilemma/record", {"json": payload}), ]) def record_profit_puzzler_play(user_id: int, puzzles_solved: int, mistakes: int, elapsed_ms: int, gained_xp: int | None = None): payload = {"user_id": user_id, "puzzles_solved": puzzles_solved, "mistakes": mistakes, "elapsed_ms": elapsed_ms} if gained_xp is not None: payload["gained_xp"] = gained_xp return _try_candidates("POST", [("/games/profit_puzzler/record", {"json": payload})]) def generate_quiz(lesson_id: int, level_slug: str, lesson_title: str): r = requests.post(f"{BACKEND}/generate_quiz", json={ "lesson_id": lesson_id, "level_slug": level_slug, "lesson_title": lesson_title }, timeout=60) r.raise_for_status() return r.json()["quiz"] def submit_quiz(lesson_id: int, level_slug: str, user_answers: list[dict], original_quiz: list[dict]): r = requests.post(f"{BACKEND}/submit_quiz", json={ "lesson_id": lesson_id, "level_slug": level_slug, "user_answers": user_answers, "original_quiz": original_quiz }, timeout=90) r.raise_for_status() return r.json() def tutor_explain(lesson_id: int, level_slug: str, wrong: list[dict]): r = requests.post(f"{BACKEND}/tutor/explain", json={ "lesson_id": lesson_id, "level_slug": level_slug, "wrong": wrong }, timeout=60) r.raise_for_status() return r.json()["feedback"]