import asyncio
import base64
import html
import io
import json
import os
import re
import shutil
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
load_dotenv()
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, Response
from extractor import extract_page_range
app = FastAPI(title="Problem Extractor")
OUTPUT_DIR = Path("output")
OUTPUT_DIR.mkdir(exist_ok=True)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
SAVED_PDF = OUTPUT_DIR / "saved_textbook.pdf"
SAVED_PDF_NAME = OUTPUT_DIR / "saved_textbook_name.txt"
BUNDLED_PDF = (Path(__file__).parent / "../../Reference and Solution Manual/Modern Control Systems-12 Edition.pdf").resolve()
BUNDLED_PDF_NAME = "Modern Control Systems-12 Edition.pdf"
EDUVERSE_API_URL = os.getenv("EDUVERSE_API_URL", "https://eduverse-team-eduverse-backend.hf.space")
EDUVERSE_EMAIL = os.getenv("EDUVERSE_EMAIL", "")
EDUVERSE_PASSWORD = os.getenv("EDUVERSE_PASSWORD", "")
_eduverse_token: Optional[str] = None
def _active_pdf() -> Optional[Path]:
"""Return the PDF to use: user-uploaded first, then the bundled textbook."""
if SAVED_PDF.exists():
return SAVED_PDF
if BUNDLED_PDF.exists():
return BUNDLED_PDF
return None
async def _get_eduverse_token() -> str:
global _eduverse_token
if _eduverse_token:
return _eduverse_token
if not EDUVERSE_EMAIL or not EDUVERSE_PASSWORD:
raise HTTPException(status_code=500, detail="EDUVERSE_EMAIL and EDUVERSE_PASSWORD must be set in .env")
async with httpx.AsyncClient() as client:
r = await client.post(
f"{EDUVERSE_API_URL}/api/auth/login",
json={"email": EDUVERSE_EMAIL, "password": EDUVERSE_PASSWORD},
timeout=15,
)
if r.status_code != 200:
raise HTTPException(status_code=502, detail=f"EduVerse login failed: {r.text}")
data = r.json()
token = data.get("accessToken") or data.get("access_token") or data.get("token")
if not token:
raise HTTPException(status_code=502, detail=f"EduVerse login response missing token: {data}")
_eduverse_token = token
return token
GEMINI_PROMPT = """\
You are extracting problems from an academic textbook. The text was extracted via OCR from a scanned PDF and may contain artifacts, merged columns, and garbled spacing. Identify every distinct problem or question.
Return a JSON array where each object has exactly these fields:
- "question_number": string — the problem label as it appears (e.g. "P1.1", "Q3", "Exercise 4.2", "Problem 7")
- "question": string — complete question text, cleaned of OCR errors. Remove figure captions and page headers. Fix broken words from column wrapping. Format all math using LaTeX: inline with $...$ (e.g. $x_1$, $\\omega_n$, $K > 0$), display with $$...$$.
- "page": number — page where the problem starts (use the --- PAGE N --- markers)
- "question_type": string — one of exactly: "written", "mcq", "true_false". Use "mcq" only if the question lists labeled answer choices (A/B/C/D or similar). Use "true_false" only if the question is a direct true-or-false statement. Use "written" for everything else (derivations, calculations, short answer, design problems).
- "difficulty": string — one of exactly: "easy", "medium", "hard". Base this on cognitive load and prerequisite knowledge: easy = recall or single-step, medium = multi-step application, hard = synthesis, design, or proof.
- "bloom_level": string — one of exactly: "remembering", "understanding", "applying", "analyzing", "evaluating", "creating". Pick the highest Bloom's taxonomy level the question primarily demands.
- "has_diagram": boolean — true if the question references a Figure, asks to sketch a diagram or block diagram, or involves a circuit
- "figure_reference": string or null — specific figure label when present (e.g. "Figure 13.18", "Fig. P1.2"), otherwise null
- "exam_ready": boolean — true if suitable for a university exam. Mark false for: pure discussion/describe questions, problems entirely dependent on a figure students won't have, or questions too open-ended to grade objectively.
- "exam_notes": string or null — if exam_ready is false, a brief reason (e.g. "discussion only", "requires Figure P1.2", "open-ended design"). If exam_ready is true, use null.
- "expected_answer": string or null — for "written" and "true_false" question types only: a concise model answer (1–4 sentences or key steps). For "mcq" use null (the correct option already encodes the answer).
- "hints": string or null — a short hint (1–2 sentences) that nudges a student toward the solution without giving it away. Provide for all question types. Use null if no meaningful hint can be given.
Return ONLY a valid JSON array wrapped in ```json fences. No explanation.
--- TEXT START ---
"""
_HTML = """
Problem Extractor
Fetching questions...
Pulling from the question bank and checking for duplicates.
"""
@app.get("/latest-questions-data", response_class=HTMLResponse)
async def latest_questions_data(courseId: int, chapterId: Optional[int] = None, limit: int = 100):
"""Data fragment — fetched by the shell page via AJAX."""
from collections import defaultdict
from datetime import datetime as _dt, timezone as _tz
if limit < 1 or limit > 500:
limit = 100
PAGE_SIZE = 100
MAX_PAGES = 20
questions: list[dict] = []
total = 0
token = await _get_eduverse_token()
async with httpx.AsyncClient() as client:
for page in range(1, MAX_PAGES + 1):
params: dict = {"courseId": courseId, "page": page, "limit": PAGE_SIZE}
if chapterId is not None:
params["chapterId"] = chapterId
r = await client.get(
f"{EDUVERSE_API_URL}/api/question-bank/questions",
params=params,
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
if r.status_code == 401:
global _eduverse_token
_eduverse_token = None
token = await _get_eduverse_token()
r = await client.get(
f"{EDUVERSE_API_URL}/api/question-bank/questions",
params=params,
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
if r.status_code != 200:
raise HTTPException(status_code=502, detail=f"List failed page={page}: {r.text}")
body = r.json()
if isinstance(body, dict):
items = body.get("data") or body.get("items") or body.get("questions") or []
if page == 1:
total = int(body.get("total") or len(items))
elif isinstance(body, list):
items = body
if page == 1:
total = len(items)
else:
items = []
if not items:
break
questions.extend(items)
if len(questions) >= limit or len(items) < PAGE_SIZE:
break
questions.sort(
key=lambda q: (q.get("createdAt") or "", q.get("id") or 0),
reverse=True,
)
questions = questions[:limit]
# Duplicate detection
dup_groups: dict[str, list[dict]] = defaultdict(list)
for q in questions:
text = (q.get("questionText") or "").strip().lower()
if text:
dup_groups[text].append(q)
dups = {k: v for k, v in dup_groups.items() if len(v) > 1}
dup_ids: set[int] = {q.get("id") for items in dups.values() for q in items}
# Group by save-minute (YYYY-MM-DD HH:MM)
def minute_key(q):
raw = q.get("createdAt") or ""
if not raw:
return "~no-date"
try:
iso = raw[:-1] + "+00:00" if raw.endswith("Z") else raw
d = _dt.fromisoformat(iso)
return d.strftime("%Y-%m-%d %H:%M")
except Exception:
return "~no-date"
batches: dict[str, list[dict]] = defaultdict(list)
for q in questions:
batches[minute_key(q)].append(q)
# Sort batches newest first
sorted_batches = sorted(batches.items(), key=lambda kv: kv[0], reverse=True)
def fmt_date(s):
if not s:
return "—"
try:
iso = s[:-1] + "+00:00" if isinstance(s, str) and s.endswith("Z") else s
d = _dt.fromisoformat(iso) if isinstance(iso, str) else iso
return d.strftime("%Y-%m-%d %I:%M %p")
except Exception:
return str(s)
def fmt_minute(key):
if key == "~no-date":
return "Unknown time"
try:
d = _dt.strptime(key, "%Y-%m-%d %H:%M")
return d.strftime("%Y-%m-%d %I:%M %p") + " UTC"
except Exception:
return key
def trunc(s, n=240):
s = s or ""
return s if len(s) <= n else s[: n - 1] + "…"
def img_tag(q):
url = q.get("questionImageUrl")
if not url:
return '—'
esc = html.escape(url)
return f''
def type_badge(v):
v = str(v or "")
return f'{html.escape(v)}' if v else '—'
def diff_badge(v):
v = str(v or "")
c = {"easy": "badge-green", "medium": "badge-blue", "hard": "badge-red"}.get(v.lower(), "badge-gray")
return f'{html.escape(v)}' if v else '—'
def status_badge(v):
v = str(v or "")
c = {"approved": "badge-green", "draft": "badge-gray", "rejected": "badge-red"}.get(v.lower(), "badge-gray")
return f'{html.escape(v)}' if v else '—'
# Build batch HTML
batches_html_parts = []
for key, qs in sorted_batches:
def _card(q):
is_dup = q.get("id") in dup_ids
has_ans = bool((q.get("expectedAnswerText") or "").strip())
has_hint = bool((q.get("hints") or "").strip())
card_cls = "q-card dup-card-q" if is_dup else "q-card"
dup_marker = ' duplicate' if is_dup else ""
qtext = q.get("questionText") or ""
ans_text = html.escape(q.get("expectedAnswerText") or "")
hint_text = html.escape(q.get("hints") or "")
rid = q.get("id")
saved_fmt = html.escape(fmt_date(q.get("createdAt")))
img_url = q.get("questionImageUrl")
# Image block
if img_url:
esc_url = html.escape(img_url)
img_block = f'
'
else:
img_block = ""
# Answer + hint
if has_ans:
ans_block = f'
Answer
{ans_text}
'
else:
ans_block = ""
if has_hint:
hint_block = f'
Hint
{hint_text}
'
else:
hint_block = ""
body_layout = "card-body-split" if img_url else "card-body-single"
return (
f'
'
# Top bar
f'
'
f' #{rid}'
f' {type_badge(q.get("questionType"))}'
f' {diff_badge(q.get("difficulty"))}'
f' {html.escape(str(q.get("bloomLevel") or ""))}'
f' {status_badge(q.get("status"))}'
f' {dup_marker}'
f' {saved_fmt}'
f'
'
# Body
f'
'
f'
{html.escape(qtext)}
'
f' {img_block}'
f'
'
# Answer / hint
f'{ans_block}'
f'{hint_block}'
f'
'
)
q_cards = "".join(_card(q) for q in qs)
has_dups = any(q.get("id") in dup_ids for q in qs)
dup_warn = ' has duplicates' if has_dups else ""
batches_html_parts.append(f"""
{html.escape(fmt_minute(key))}—{len(qs)} question{'s' if len(qs) != 1 else ''}
{dup_warn}
{q_cards}
""")
batches_html = "".join(batches_html_parts)
# Duplicate cards
dup_html = ""
if dups:
parts = []
for text, items in sorted(dups.items(), key=lambda kv: -len(kv[1])):
sample = items[0].get("questionText") or ""
rows = "".join(
f'