cc4463d42c
Found by walking the permission model as a real learner account rather than reading the guards. The bank listing has always nulled the answer side for whoever does not write the question; this route, added for the full-page editor, returned the row whole behind nothing but get_current_user. Any signed-in account could ask for /questions/detail/3869 and be handed the correct option, the explanation, the per-option reasoning and the key points for a question it had never sat — 2,924 questions, one URL, no attempt required. Same rule as the listing now, via the same may_edit_question: the stem still reads, the answer side comes back null, and explanation figures — answer-side by definition — are filtered out with it. Nulled rather than refused, because the stem is legitimately readable. Three tests: a reader gets the stem and none of the answer, whoever writes it gets all of it, and an author counts as writing it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
1533 lines
66 KiB
Python
1533 lines
66 KiB
Python
"""Question bank — view, search, categorise, and create quizzes from individual questions."""
|
|
import csv
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import cast, String, or_, select, text as sa_text, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.utils.upload_access import validate_image_attachments, stored_upload_path
|
|
from app.utils.quiz_questions import validate_option_explanations, validate_key_points
|
|
from app.models.article import Article, QuestionArticleLink
|
|
from app.models.question import Question
|
|
from app.models.question_category import QuestionCategory, QuestionCategoryLink
|
|
from app.models.quiz import Quiz
|
|
from app.models.user import User
|
|
from app.models.favorite import Favorite
|
|
from app.models.folder import QuestionFolderQuestion
|
|
from app.services import article_service, file_intake, topic_claims, vision_service
|
|
from app.services.ai_service import get_configured_model
|
|
from app.services.search_service import hybrid_ids, hybrid_question_ids, rerank_ids
|
|
from app.services.question_figures import figures_for as _figures_for
|
|
from app.utils.quiz_access import may_edit_question
|
|
from app.services.prepared_session import prepare_session
|
|
from app.services.quiz_builder import (bank_query, bank_question_predicate, category_descendants,
|
|
filtered_bank_query, CreateFromBankRequest, GenerateTestRequest, TestOptions,
|
|
create_saved_test, exam_scope_predicate, generate_test)
|
|
from app.utils.auth import get_current_user, require_moderator
|
|
from app.utils.category_grants import (assert_can_manage_category, assert_user_can_manage,
|
|
is_question_manager, manageable_categories, question_in_scope, question_scope_predicate,
|
|
require_question_manager)
|
|
|
|
router = APIRouter()
|
|
|
|
# Ranked retrieval is capped so a broad query cannot pull the whole bank.
|
|
MAX_SEARCH_RESULTS = 500
|
|
# A matched test stays a study aid, not a dump of the bank.
|
|
MAX_MATCHED_QUESTIONS = 30
|
|
# A document here is a query, never content: it is read once to find matching
|
|
# questions in the bank and then discarded, so the cap is about how much text
|
|
# is worth reading rather than about storage. Two megabytes is a long syllabus
|
|
# or a score report; past that somebody is uploading a textbook.
|
|
MAX_UPLOAD_BYTES = 2 * 1024 * 1024
|
|
MAX_QUERY_CHARS = 6000
|
|
|
|
|
|
def parse_category_ids(value):
|
|
try:
|
|
ids = [int(part.strip()) for part in value.split(",")]
|
|
except ValueError:
|
|
raise HTTPException(400, "Category IDs must be comma-separated integers")
|
|
if any(cid <= 0 for cid in ids):
|
|
raise HTTPException(400, "Category IDs must be positive")
|
|
return ids
|
|
|
|
|
|
@router.delete("/{question_id}", status_code=204)
|
|
def delete_question(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Move a question to the trash. Restore with PATCH /{id}/restore."""
|
|
question = _question_for_delete(db, question_id, current_user)
|
|
question.deleted_at = datetime.utcnow()
|
|
db.commit()
|
|
|
|
|
|
def _question_for_delete(db: Session, question_id: int, current_user: User) -> Question:
|
|
"""The question, if this person is allowed to remove it."""
|
|
question = db.query(Question).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
is_mod = current_user.role in ("admin", "moderator")
|
|
scope = manageable_categories(db, current_user)
|
|
granted = scope is not None and bool(scope) and question_in_scope(db, scope, question)
|
|
if not is_mod and not granted:
|
|
if question.user_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not authorized to delete this question")
|
|
# Regular users can only delete questions they created (no source quiz)
|
|
if question.source_quiz_id is not None:
|
|
raise HTTPException(status_code=403, detail="Only moderators can delete extracted questions")
|
|
return question
|
|
|
|
|
|
@router.get("/trash")
|
|
def list_trashed_questions(
|
|
limit: int = 100,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""What this person has deleted and could still put back.
|
|
|
|
A moderator sees everything deleted; anyone else sees their own, and an
|
|
educator sees what falls inside the categories they were granted.
|
|
"""
|
|
query = db.query(Question).filter(Question.deleted_at.isnot(None))
|
|
if current_user.role not in ("admin", "moderator"):
|
|
scope = manageable_categories(db, current_user)
|
|
clause = Question.user_id == current_user.id
|
|
if scope:
|
|
clause = clause | Question.question_category_id.in_(scope)
|
|
query = query.filter(clause)
|
|
rows = query.order_by(Question.deleted_at.desc()).limit(min(limit, 500)).all()
|
|
categories = {
|
|
c.id: c.name for c in db.query(QuestionCategory).filter(
|
|
QuestionCategory.id.in_([r.question_category_id for r in rows if r.question_category_id])
|
|
).all()
|
|
} if rows else {}
|
|
return [{
|
|
"id": row.id,
|
|
"question_text": (row.question_text or "")[:400],
|
|
"question_type": row.question_type,
|
|
"category": categories.get(row.question_category_id),
|
|
"deleted_at": row.deleted_at,
|
|
} for row in rows]
|
|
|
|
|
|
@router.patch("/{question_id}/restore")
|
|
def restore_question(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Put a question back. Its id never changed, so everything that pointed at
|
|
it — attempts, quizzes, exams, media — still does."""
|
|
question = _question_for_delete(db, question_id, current_user)
|
|
if question.deleted_at is None:
|
|
raise HTTPException(400, "That question is not in the trash")
|
|
question.deleted_at = None
|
|
db.commit()
|
|
return {"id": question.id, "restored": True}
|
|
|
|
|
|
@router.delete("/{question_id}/permanent", status_code=204)
|
|
def delete_question_permanently(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Erase it. This takes its answers, its quiz membership and its media with
|
|
it, and no restore can bring any of that back — so only a moderator may,
|
|
and only something already in the trash."""
|
|
question = _question_for_delete(db, question_id, current_user)
|
|
if question.deleted_at is None:
|
|
raise HTTPException(400, "Move it to the trash first")
|
|
db.delete(question)
|
|
db.commit()
|
|
|
|
|
|
class QuestionEdit(BaseModel):
|
|
question_text: str | None = None
|
|
question_type: str | None = None
|
|
options: list[str] | None = None
|
|
correct_answer: str | None = None
|
|
explanation: str | None = None
|
|
question_category_id: int | None = None
|
|
additional_category_ids: list[int] | None = None # Full set of extra (non-primary) categories.
|
|
option_explanations: dict | None = None
|
|
key_points: list | None = None
|
|
attending_tip: str | None = None
|
|
difficulty: Literal['easy', 'medium', 'hard'] | None = None
|
|
image_path: str | None = None
|
|
explanation_image_path: str | None = None
|
|
|
|
|
|
@router.patch("/{question_id}")
|
|
def edit_question(
|
|
question_id: int,
|
|
data: QuestionEdit,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Edit a question — moderators, or an educator granted its category."""
|
|
scope = require_question_manager(db, current_user)
|
|
question = db.query(Question).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
assert_user_can_manage(db, current_user, [question_id])
|
|
_snapshot_question(db, question, current_user.id)
|
|
if "question_category_id" in data.model_fields_set:
|
|
# Moving a question out of your scope would put it beyond your reach.
|
|
assert_can_manage_category(scope, data.question_category_id)
|
|
for extra_id in (data.additional_category_ids or []):
|
|
assert_can_manage_category(scope, extra_id)
|
|
values = validate_image_attachments(db, current_user, data.model_dump(exclude_unset=True))
|
|
if values.get("question_category_id") is not None and not db.get(QuestionCategory, values["question_category_id"]):
|
|
raise HTTPException(400, "Category not found")
|
|
explanations = values.pop("option_explanations", None)
|
|
if explanations is not None:
|
|
values["option_explanations"] = validate_option_explanations(
|
|
values.get("options", question.options), explanations)
|
|
key_points = values.pop("key_points", None)
|
|
if key_points is not None:
|
|
values["key_points"] = validate_key_points(key_points, db)
|
|
extra_ids = values.pop("additional_category_ids", None)
|
|
if extra_ids is not None:
|
|
extra_ids = list(dict.fromkeys(extra_ids))
|
|
known = {row.id for row in db.query(QuestionCategory.id).filter(QuestionCategory.id.in_(extra_ids)).all()} if extra_ids else set()
|
|
if set(extra_ids) - known:
|
|
raise HTTPException(400, "Additional category not found")
|
|
primary = values.get("question_category_id", question.question_category_id)
|
|
extra_ids = [cid for cid in extra_ids if cid != primary]
|
|
db.query(QuestionCategoryLink).filter(QuestionCategoryLink.question_id == question.id).delete(synchronize_session=False)
|
|
for cid in extra_ids:
|
|
db.add(QuestionCategoryLink(question_id=question.id, category_id=cid))
|
|
for field, value in values.items():
|
|
setattr(question, field, value)
|
|
db.commit()
|
|
db.refresh(question)
|
|
return {
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"question_type": question.question_type,
|
|
"options": question.options,
|
|
"correct_answer": question.correct_answer,
|
|
"explanation": question.explanation,
|
|
"question_category_id": question.question_category_id,
|
|
"image_path": question.image_path,
|
|
"explanation_image_path": question.explanation_image_path,
|
|
}
|
|
|
|
|
|
@router.patch("/{question_id}/category")
|
|
def set_question_category(
|
|
question_id: int,
|
|
category_id: int | None = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Assign or remove a question category."""
|
|
question = db.query(Question).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
if category_id is not None:
|
|
cat = db.query(QuestionCategory).filter(QuestionCategory.id == category_id).first()
|
|
if not cat:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
question.question_category_id = category_id
|
|
db.commit()
|
|
# An article standing on this topic gains the question now, rather than
|
|
# when somebody next remembers to press the link button.
|
|
linked = topic_claims.link_new_question(db, question)
|
|
return {"question_id": question_id, "question_category_id": category_id,
|
|
"articles_linked": linked}
|
|
|
|
|
|
@router.get("/bank/ids")
|
|
def get_bank_ids(
|
|
q: str | None = Query(None),
|
|
quiz_id: int | None = Query(None),
|
|
category_id: int | None = Query(None),
|
|
category_ids: str | None = Query(None, description="Comma-separated category IDs (OR filter)"),
|
|
uncategorized: bool = Query(False),
|
|
favorites_only: bool = Query(False),
|
|
folder_id: int | None = Query(None, description="Only the questions in this folder"),
|
|
tag_ids: str | None = Query(None, description="Comma-separated tag IDs (AND filter)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Return just IDs for all matching questions (for server-side select-all)."""
|
|
query = bank_query(db, current_user).with_entities(Question.id)
|
|
if quiz_id:
|
|
query = query.filter(Question.source_quiz_id == quiz_id)
|
|
if folder_id is not None:
|
|
query = query.filter(Question.id.in_(select(QuestionFolderQuestion.question_id).where(
|
|
QuestionFolderQuestion.folder_id == folder_id)))
|
|
if category_ids:
|
|
cat_id_list = parse_category_ids(category_ids)
|
|
if cat_id_list:
|
|
query = query.filter(or_(
|
|
Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)),
|
|
Question.id.in_(select(QuestionCategoryLink.question_id).where(QuestionCategoryLink.category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)))),
|
|
))
|
|
elif category_id is not None:
|
|
query = query.filter(or_(
|
|
Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])),
|
|
Question.id.in_(select(QuestionCategoryLink.question_id).where(QuestionCategoryLink.category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])))),
|
|
))
|
|
if uncategorized:
|
|
query = query.filter(Question.question_category_id.is_(None))
|
|
if favorites_only:
|
|
favorite_ids = db.query(Favorite.question_id).filter(Favorite.user_id == current_user.id).all()
|
|
fav_ids = [f[0] for f in favorite_ids]
|
|
if not fav_ids:
|
|
return []
|
|
query = query.filter(Question.id.in_(fav_ids))
|
|
if q and q.strip():
|
|
phrase = q.strip()
|
|
query = query.filter(
|
|
or_(
|
|
Question.question_text.ilike(f"%{phrase}%"),
|
|
cast(Question.options, String).ilike(f"%{phrase}%"),
|
|
)
|
|
)
|
|
# Tag filter: questions must have ALL specified tags
|
|
if tag_ids:
|
|
tag_id_list = [int(t.strip()) for t in tag_ids.split(",") if t.strip().isdigit()]
|
|
if tag_id_list:
|
|
matching_ids = list(db.execute(sa_text("""
|
|
SELECT question_id FROM question_tag_links
|
|
WHERE tag_id = ANY(:tag_ids)
|
|
GROUP BY question_id
|
|
HAVING COUNT(DISTINCT tag_id) = :cnt
|
|
"""), {"tag_ids": tag_id_list, "cnt": len(tag_id_list)}).scalars())
|
|
if matching_ids:
|
|
query = query.filter(Question.id.in_(matching_ids))
|
|
else:
|
|
return []
|
|
return [row[0] for row in query.all()]
|
|
|
|
|
|
@router.get("/bank")
|
|
def get_question_bank(
|
|
q: str | None = Query(None),
|
|
quiz_id: int | None = Query(None),
|
|
category_id: int | None = Query(None),
|
|
category_ids: str | None = Query(None, description="Comma-separated category IDs (OR filter)"),
|
|
uncategorized: bool = Query(False),
|
|
needs: Literal["category", "explanation", "difficulty", "private"] | None = Query(
|
|
None, description="Editorial gap filter used by the question manager"),
|
|
favorites_only: bool = Query(False),
|
|
my_questions: bool = Query(False, description="Show only questions created by current user"),
|
|
difficulty: Literal["easy", "medium", "hard"] | None = Query(None),
|
|
article_ids: str | None = Query(None, description="Comma-separated article IDs (OR filter)"),
|
|
folder_id: int | None = Query(None, description="Only the questions in this folder"),
|
|
tag_ids: str | None = Query(None, description="Comma-separated tag IDs (AND filter)"),
|
|
limit: int = Query(50, le=200),
|
|
offset: int = Query(0),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""List all questions across all quizzes. Supports keyword filter and quiz filter."""
|
|
query = bank_query(db, current_user)
|
|
exam_filter = exam_scope_predicate(db, current_user)
|
|
if exam_filter is not None:
|
|
query = query.filter(exam_filter)
|
|
if my_questions:
|
|
query = query.filter(Question.user_id == current_user.id)
|
|
|
|
if difficulty:
|
|
query = query.filter(Question.difficulty == difficulty)
|
|
if article_ids:
|
|
article_list = [int(part) for part in article_ids.split(",") if part.strip().isdigit()]
|
|
if article_list:
|
|
query = query.filter(Question.id.in_(select(QuestionArticleLink.question_id).where(
|
|
QuestionArticleLink.article_id.in_(article_list))))
|
|
|
|
if quiz_id:
|
|
query = query.filter(Question.source_quiz_id == quiz_id)
|
|
if folder_id is not None:
|
|
# Not gated on holding the folder: this only ever narrows the bank the
|
|
# caller can already see, so at worst it shows them a subset of that.
|
|
query = query.filter(Question.id.in_(select(QuestionFolderQuestion.question_id).where(
|
|
QuestionFolderQuestion.folder_id == folder_id)))
|
|
|
|
if category_ids:
|
|
cat_id_list = parse_category_ids(category_ids)
|
|
if cat_id_list:
|
|
query = query.filter(or_(
|
|
Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)),
|
|
Question.id.in_(select(QuestionCategoryLink.question_id).where(QuestionCategoryLink.category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)))),
|
|
))
|
|
elif category_id is not None:
|
|
query = query.filter(or_(
|
|
Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])),
|
|
Question.id.in_(select(QuestionCategoryLink.question_id).where(QuestionCategoryLink.category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])))),
|
|
))
|
|
|
|
if uncategorized or needs == "category":
|
|
query = query.filter(Question.question_category_id.is_(None))
|
|
if needs == "explanation":
|
|
query = query.filter(or_(Question.explanation.is_(None), Question.explanation.in_(("", " "))))
|
|
elif needs == "difficulty":
|
|
query = query.filter(Question.difficulty.is_(None))
|
|
|
|
# Tag filter: questions must have ALL specified tags
|
|
if tag_ids:
|
|
import logging as _log
|
|
_log.getLogger(__name__).info(f"Tag filter: tag_ids={tag_ids!r}")
|
|
tag_id_list = [int(t.strip()) for t in tag_ids.split(",") if t.strip().isdigit()]
|
|
if tag_id_list:
|
|
matching_ids = list(db.execute(sa_text("""
|
|
SELECT question_id FROM question_tag_links
|
|
WHERE tag_id = ANY(:tag_ids)
|
|
GROUP BY question_id
|
|
HAVING COUNT(DISTINCT tag_id) = :cnt
|
|
"""), {"tag_ids": tag_id_list, "cnt": len(tag_id_list)}).scalars())
|
|
_log.getLogger(__name__).info(f"Tag filter matched {len(matching_ids)} questions for tags {tag_id_list}")
|
|
if matching_ids:
|
|
query = query.filter(Question.id.in_(matching_ids))
|
|
else:
|
|
return {"total": 0, "questions": []}
|
|
|
|
if favorites_only:
|
|
favorite_ids = db.query(Favorite.question_id).filter(Favorite.user_id == current_user.id).all()
|
|
fav_ids = [f[0] for f in favorite_ids]
|
|
if not fav_ids:
|
|
return {"total": 0, "questions": []}
|
|
query = query.filter(Question.id.in_(fav_ids))
|
|
|
|
# ── Hybrid retrieval: full text fused with embeddings ──────────
|
|
# Always both. Keyword-only silently drops the question that asks the same
|
|
# thing in different words, which is the one a concept search wants.
|
|
semantic_ids: set[int] = set()
|
|
if q and q.strip():
|
|
ranked_ids, semantic_ids = hybrid_question_ids(db, q.strip(), limit=MAX_SEARCH_RESULTS)
|
|
if not ranked_ids:
|
|
return {"total": 0, "questions": []}
|
|
query = query.filter(Question.id.in_(ranked_ids))
|
|
matched = query.all()
|
|
# A cross-encoder settles the front of the list, having read the query
|
|
# and the stem together. It sees only what survived the filters above,
|
|
# and it reorders rather than selects, so the count and the last page of
|
|
# results are the same whether or not it answers.
|
|
present = {question.id for question in matched}
|
|
reranked = rerank_ids(db, q.strip(), "question",
|
|
[question_id for question_id in ranked_ids if question_id in present])
|
|
rank_of = {question_id: position for position, question_id in enumerate(reranked)}
|
|
matched.sort(key=lambda question: rank_of.get(question.id, len(rank_of)))
|
|
total = len(matched)
|
|
questions = matched[offset:offset + limit]
|
|
else:
|
|
total = query.count()
|
|
questions = query.order_by(Question.source_quiz_id, Question.id).offset(offset).limit(limit).all()
|
|
|
|
quiz_cache: dict[int, str] = {}
|
|
cat_cache: dict[int, str] = {}
|
|
# Which of these the caller writes rather than sits. This listing used to
|
|
# hand every signed-in learner the correct option and the explanation for
|
|
# the whole bank — the answer key, one HTTP request away from the questions
|
|
# it answers. Editors still get it; everyone else gets the stem.
|
|
scope = question_scope_predicate(db, current_user)
|
|
if scope is None:
|
|
editable = {qu.id for qu in questions}
|
|
else:
|
|
editable = {row[0] for row in db.query(Question.id).filter(
|
|
Question.id.in_([qu.id for qu in questions]), scope).all()} if questions else set()
|
|
editable |= {qu.id for qu in questions
|
|
if qu.user_id is not None and qu.user_id == current_user.id}
|
|
link_rows = db.query(QuestionCategoryLink.question_id, QuestionCategoryLink.category_id).filter(
|
|
QuestionCategoryLink.question_id.in_([qu.id for qu in questions])).all() if questions else []
|
|
extra_map: dict[int, list[int]] = {}
|
|
for qid, cid in link_rows:
|
|
extra_map.setdefault(qid, []).append(cid)
|
|
result = []
|
|
for qu in questions:
|
|
src_id = qu.source_quiz_id
|
|
if src_id not in quiz_cache:
|
|
quiz = db.query(Quiz).filter(Quiz.id == src_id).first() if src_id else None
|
|
quiz_cache[src_id] = quiz.title if quiz else (f"Quiz {src_id}" if src_id else "Unknown")
|
|
cat_name = None
|
|
if qu.question_category_id:
|
|
if qu.question_category_id not in cat_cache:
|
|
cat = db.query(QuestionCategory).filter(QuestionCategory.id == qu.question_category_id).first()
|
|
cat_cache[qu.question_category_id] = cat.name if cat else None
|
|
cat_name = cat_cache.get(qu.question_category_id)
|
|
result.append({
|
|
"id": qu.id,
|
|
"quiz_id": qu.source_quiz_id,
|
|
"quiz_title": quiz_cache[src_id],
|
|
"question_category_id": qu.question_category_id,
|
|
"question_category_name": cat_name,
|
|
"category_ids": sorted(set([qu.question_category_id] if qu.question_category_id else []) | set(extra_map.get(qu.id, []))),
|
|
"question_text": qu.question_text,
|
|
"question_type": qu.question_type,
|
|
"options": qu.options,
|
|
# The answer side, for the people who write it.
|
|
"correct_answer": qu.correct_answer if qu.id in editable else None,
|
|
"explanation": qu.explanation if qu.id in editable else None,
|
|
"image_path": qu.image_path,
|
|
"explanation_image_path": qu.explanation_image_path if qu.id in editable else None,
|
|
"option_explanations": qu.option_explanations if qu.id in editable else None,
|
|
"key_points": qu.key_points if qu.id in editable else None,
|
|
"attending_tip": qu.attending_tip if qu.id in editable else None,
|
|
"difficulty": qu.difficulty,
|
|
"user_id": qu.user_id,
|
|
"match_source": "semantic" if qu.id in semantic_ids else "keyword",
|
|
})
|
|
|
|
return {"total": total, "questions": result}
|
|
|
|
|
|
def _strip_html(text: str) -> str:
|
|
"""Remove HTML tags and decode entities, returning plain text."""
|
|
if not text:
|
|
return text
|
|
# Remove tags
|
|
clean = re.sub(r"<[^>]+>", " ", text)
|
|
# Decode common entities
|
|
clean = clean.replace(" ", " ").replace("&", "&").replace("<", "<").replace(">", ">").replace(""", '"')
|
|
# Collapse whitespace
|
|
clean = re.sub(r"\s+", " ", clean).strip()
|
|
return clean
|
|
|
|
|
|
class ManualQuestionCreate(BaseModel):
|
|
question_text: str
|
|
question_type: str = "mcq"
|
|
options: list[str] | None = None
|
|
correct_answer: str
|
|
explanation: str | None = None
|
|
question_category_id: int | None = None
|
|
option_explanations: dict | None = None
|
|
key_points: list | None = None
|
|
attending_tip: str | None = None
|
|
difficulty: Literal['easy', 'medium', 'hard'] | None = None
|
|
image_path: str | None = None
|
|
explanation_image_path: str | None = None
|
|
|
|
|
|
@router.post("/create")
|
|
def create_question_manually(
|
|
data: ManualQuestionCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create a single question manually — moderators, or an educator in their categories."""
|
|
scope = require_question_manager(db, current_user)
|
|
assert_can_manage_category(scope, data.question_category_id)
|
|
q_text = data.question_text.strip()
|
|
if not q_text:
|
|
raise HTTPException(status_code=400, detail="Question text is required")
|
|
if not data.correct_answer.strip():
|
|
raise HTTPException(status_code=400, detail="Correct answer is required")
|
|
if data.question_type == "mcq" and (not data.options or len(data.options) < 2):
|
|
raise HTTPException(status_code=400, detail="MCQ questions need at least 2 options")
|
|
if data.question_type == "mcq" and data.correct_answer not in data.options:
|
|
raise HTTPException(status_code=400, detail="Correct answer must be one of the options")
|
|
|
|
images = validate_image_attachments(db, current_user, data.model_dump())
|
|
option_explanations = validate_option_explanations(data.options, data.option_explanations)
|
|
key_points = validate_key_points(data.key_points, db)
|
|
question = Question(
|
|
question_text=q_text,
|
|
question_type=data.question_type,
|
|
options=data.options,
|
|
correct_answer=data.correct_answer.strip(),
|
|
explanation=data.explanation,
|
|
question_category_id=data.question_category_id,
|
|
image_path=images["image_path"],
|
|
explanation_image_path=images["explanation_image_path"],
|
|
option_explanations=option_explanations,
|
|
key_points=key_points,
|
|
attending_tip=(data.attending_tip or None),
|
|
difficulty=data.difficulty,
|
|
user_id=current_user.id,
|
|
)
|
|
db.add(question)
|
|
db.commit()
|
|
db.refresh(question)
|
|
|
|
# Any article standing on this question's topic gains it immediately.
|
|
topic_claims.link_new_question(db, question)
|
|
|
|
# Generate embedding in background
|
|
try:
|
|
from app.services import embedding_service
|
|
embedding_service.embed_question(question)
|
|
db.commit()
|
|
except Exception:
|
|
logger.warning("Failed to generate embedding for question %d", question.id, exc_info=True)
|
|
|
|
return {
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"question_type": question.question_type,
|
|
"options": question.options,
|
|
"correct_answer": question.correct_answer,
|
|
"image_path": question.image_path,
|
|
}
|
|
|
|
|
|
@router.post("/upload-image")
|
|
def upload_question_image(
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Upload an image for use in questions. Returns the relative path."""
|
|
if not file.content_type or not file.content_type.startswith("image/"):
|
|
raise HTTPException(status_code=400, detail="File must be an image")
|
|
ext = file.filename.rsplit(".", 1)[-1].lower() if file.filename and "." in file.filename else "png"
|
|
if ext not in ("png", "jpg", "jpeg", "gif", "webp", "svg"):
|
|
raise HTTPException(status_code=400, detail="Unsupported image format")
|
|
filename = f"{uuid.uuid4().hex}.{ext}"
|
|
directory = os.path.join(settings.UPLOAD_DIR, "questions", str(current_user.id))
|
|
os.makedirs(directory, exist_ok=True)
|
|
filepath = os.path.join(directory, filename)
|
|
with open(filepath, "wb") as f:
|
|
f.write(file.file.read())
|
|
rel_path = f"questions/{current_user.id}/{filename}"
|
|
return {"image_path": rel_path, "url": f"/uploads/{rel_path}"}
|
|
|
|
|
|
@router.get("/images")
|
|
def list_question_images(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""List all unique question images for the image bank browser."""
|
|
stems = bank_query(db, current_user).with_entities(Question.image_path.label("path")).filter(
|
|
Question.image_path.isnot(None), Question.image_path != "")
|
|
explanations = bank_query(db, current_user).with_entities(Question.explanation_image_path.label("path")).filter(
|
|
Question.explanation_image_path.isnot(None), Question.explanation_image_path != "")
|
|
paths = sorted({stored_upload_path(row.path) or row.path
|
|
for row in stems.union(explanations).order_by("path").limit(200).all()})
|
|
return [{"image_path": path, "url": path if re.match(r"^https?://|^/uploads/", path, re.I) else f"/uploads/{path}"} for path in paths]
|
|
|
|
|
|
@router.get("/builder/count")
|
|
def count_builder_questions(
|
|
category_ids: list[int] = Query(default=[]),
|
|
state: Literal["all", "unused", "incorrect", "bookmarked"] = "all",
|
|
difficulty: Literal["easy", "medium", "hard"] | None = Query(None),
|
|
article_ids: str | None = Query(None, description="Comma-separated article IDs (OR filter)"),
|
|
tag_ids: str | None = Query(None, description="Comma-separated tag IDs (AND filter)"),
|
|
system_ids: str | None = Query(None, description="Comma-separated organ system IDs (OR filter)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
ids = [int(part) for part in (article_ids or "").split(",") if part.strip().isdigit()]
|
|
tag_list = [int(part) for part in (tag_ids or "").split(",") if part.strip().isdigit()]
|
|
systems = [int(part) for part in (system_ids or "").split(",") if part.strip().isdigit()]
|
|
scoped = filtered_bank_query(db, current_user, category_ids, state,
|
|
difficulty, ids, tag_list, systems)
|
|
# How many of the *other* filters' matches carry each difficulty, so the
|
|
# facet can say "Hard — none" instead of offering a choice that empties the
|
|
# bank. Counted without the difficulty filter applied, because a facet that
|
|
# counts only what it has already selected always reads zero for the rest.
|
|
by_difficulty = dict(filtered_bank_query(db, current_user, category_ids, state,
|
|
None, ids, tag_list, systems)
|
|
.with_entities(Question.difficulty, func.count(Question.id))
|
|
.group_by(Question.difficulty).all())
|
|
return {"count": scoped.count(),
|
|
"difficulties": {level: by_difficulty.get(level, 0)
|
|
for level in ("easy", "medium", "hard")},
|
|
"unrated": by_difficulty.get(None, 0)}
|
|
|
|
|
|
@router.post("/builder")
|
|
def create_builder_quiz(data: GenerateTestRequest, db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)):
|
|
return generate_test(db, current_user, data)
|
|
|
|
|
|
class PreparedSessionRequest(BaseModel):
|
|
"""Accepting a prepared session, optionally at a different length."""
|
|
|
|
count: int | None = Field(default=None, ge=1, le=200)
|
|
mode: Literal["timed", "learning"] = "learning"
|
|
title: str | None = Field(default=None, max_length=200)
|
|
|
|
|
|
@router.get("/builder/prepared")
|
|
def preview_prepared_session(
|
|
count: int | None = Query(default=None, ge=1, le=200),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""The session this learner would be given, before anything is written.
|
|
|
|
Returned without the question ids: the plan is for reading, and handing the
|
|
stems' identifiers to a page that only draws a summary is how a preview
|
|
becomes a way to enumerate the bank.
|
|
"""
|
|
plan = prepare_session(db, current_user, count)
|
|
return {key: value for key, value in plan.items() if key != "question_ids"}
|
|
|
|
|
|
@router.post("/builder/prepared")
|
|
def start_prepared_session(
|
|
data: PreparedSessionRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Commit a prepared session, and return the plan it was actually built from.
|
|
|
|
The plan is recomputed here rather than carried from the preview, and the
|
|
test is built from that plan's own question ids — so the account handed
|
|
back describes this session by construction, not by the two calls happening
|
|
to agree. A learner who changed the length gets the plan for the length
|
|
they chose, which is the honest answer and not the one they were shown.
|
|
"""
|
|
plan = prepare_session(db, current_user, data.count)
|
|
created = create_saved_test(
|
|
db, current_user,
|
|
TestOptions(title=(data.title or "Prepared session").strip() or "Prepared session",
|
|
mode=data.mode, time_limit_minutes=None),
|
|
plan["question_ids"])
|
|
return {**created, "plan": {key: value for key, value in plan.items() if key != "question_ids"}}
|
|
|
|
|
|
class DescribeRequest(BaseModel):
|
|
"""Free-text description of what the learner wants to study."""
|
|
|
|
text: str = Field(min_length=10, max_length=2000)
|
|
count: int = Field(default=20, ge=1, le=MAX_MATCHED_QUESTIONS)
|
|
mode: Literal["learning", "timed"] = "learning"
|
|
title: str | None = Field(default=None, max_length=200)
|
|
|
|
|
|
def _test_from_matches(db, current_user, matched_ids, count, mode, title):
|
|
"""Build a saved test from bank questions that already exist.
|
|
|
|
Deliberately a matcher, not a generator: these are the educator-reviewed
|
|
questions in the bank, ranked against the request. Nothing is invented.
|
|
"""
|
|
if not matched_ids:
|
|
raise HTTPException(400, "No questions in the bank match that closely enough")
|
|
chosen = matched_ids[:count]
|
|
return create_saved_test(
|
|
db, current_user,
|
|
GenerateTestRequest(title=title.strip(), count=len(chosen), mode=mode,
|
|
time_limit_minutes=None, is_shared=False),
|
|
chosen,
|
|
)
|
|
|
|
|
|
@router.post("/builder/describe")
|
|
def build_test_from_description(
|
|
data: DescribeRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Turn "what I want to study" into a test of matching bank questions."""
|
|
visible = {row[0] for row in bank_query(db, current_user).with_entities(Question.id).all()}
|
|
ranked, _ = hybrid_ids(db, data.text.strip(), "question", limit=MAX_SEARCH_RESULTS)
|
|
# Worth a round trip here in a way it is not for the upload below: this
|
|
# query is a sentence a learner wrote about what they want to study, which
|
|
# is what a cross-encoder is trained on, and the top of this list becomes
|
|
# the test rather than a page they can scroll past.
|
|
matched = rerank_ids(db, data.text.strip(), "question",
|
|
[question_id for question_id in ranked if question_id in visible])
|
|
title = data.title or f"Study: {data.text.strip()[:60]}"
|
|
created = _test_from_matches(db, current_user, matched, data.count, data.mode, title)
|
|
return {**created, "matched": len(matched)}
|
|
|
|
|
|
def _image_reader(db: Session):
|
|
"""How an uploaded image becomes words, or None if it cannot.
|
|
|
|
The tool model is the one an administrator chose for jobs like this. With
|
|
none configured there is no reader, and `file_intake` says so in a sentence
|
|
that names what to upload instead — rather than a 500 from a model call
|
|
that was never going to happen.
|
|
"""
|
|
chosen = get_configured_model(db, "tool")
|
|
if not chosen or not chosen[0]:
|
|
return None
|
|
model_id, api_key = chosen
|
|
|
|
def read_image(data: bytes, media_type: str) -> str:
|
|
image = vision_service.prepare(data, media_type, caption="an uploaded page")
|
|
if image is None:
|
|
raise file_intake.Rejected("That image could not be read.")
|
|
try:
|
|
return vision_service.describe([image], model_id, api_key)[0]
|
|
except vision_service.VisionUnavailable as unavailable:
|
|
raise file_intake.Rejected(str(unavailable))
|
|
|
|
return read_image
|
|
|
|
|
|
@router.post("/builder/from-upload")
|
|
def build_test_from_upload(
|
|
file: UploadFile = File(...),
|
|
count: int = Form(20),
|
|
mode: str = Form("learning"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Match an uploaded document against the bank and build a test from it.
|
|
|
|
The document is read in memory and never stored: it is a search query, not
|
|
a source of new questions, so there is nothing to keep or delete later.
|
|
"""
|
|
if mode not in ("learning", "timed"):
|
|
raise HTTPException(400, "Mode must be learning or timed")
|
|
if not 1 <= count <= MAX_MATCHED_QUESTIONS:
|
|
raise HTTPException(400, f"Choose between 1 and {MAX_MATCHED_QUESTIONS} questions")
|
|
|
|
# Size, then type, then text — and the type is sniffed from the bytes, not
|
|
# taken from the name. `handout.pdf` holding something else is not a PDF,
|
|
# and the old branch here decided by extension and fell through to
|
|
# "decode whatever it is as UTF-8" for everything else.
|
|
try:
|
|
raw = file_intake.read(file)
|
|
except file_intake.Rejected as refusal:
|
|
raise HTTPException(413 if "over" in str(refusal) else 400, str(refusal))
|
|
|
|
kind = file_intake.kind_of(raw)
|
|
if kind is None:
|
|
raise HTTPException(415, f"Upload {file_intake.ACCEPT_HUMAN}.")
|
|
|
|
try:
|
|
text = file_intake.text_from(raw, kind, MAX_QUERY_CHARS,
|
|
describe=_image_reader(db))
|
|
except file_intake.Rejected as refusal:
|
|
raise HTTPException(400, str(refusal))
|
|
|
|
if len(text.strip()) < 40:
|
|
raise HTTPException(400, "That file has too little text to match against")
|
|
|
|
visible = {row[0] for row in bank_query(db, current_user).with_entities(Question.id).all()}
|
|
ranked, _ = hybrid_ids(db, text.strip(), "question", limit=MAX_SEARCH_RESULTS)
|
|
# Not reranked. The "query" here is a whole lecture handout, and a
|
|
# cross-encoder trained on question-length queries reads the first 2000
|
|
# characters of it and judges every candidate against whatever that part
|
|
# happened to be about — confidently, and about the wrong thing.
|
|
matched = [question_id for question_id in ranked if question_id in visible]
|
|
title = f"From {file.filename or 'upload'}"[:200]
|
|
created = _test_from_matches(db, current_user, matched, count, mode, title)
|
|
return {**created, "matched": len(matched)}
|
|
|
|
|
|
@router.post("/from-bank")
|
|
def create_quiz_from_bank(data: CreateFromBankRequest, db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)):
|
|
return create_saved_test(db, current_user, data, data.question_ids)
|
|
|
|
|
|
class BulkCategoryRequest(BaseModel):
|
|
question_ids: list[int]
|
|
category_id: int | None = None
|
|
|
|
|
|
@router.post("/bulk-category")
|
|
def bulk_set_question_category(
|
|
data: BulkCategoryRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Assign or remove a category from multiple questions."""
|
|
if data.category_id is not None:
|
|
cat = db.query(QuestionCategory).filter(QuestionCategory.id == data.category_id).first()
|
|
if not cat:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
updated = db.query(Question).filter(Question.id.in_(data.question_ids)).update(
|
|
{"question_category_id": data.category_id}, synchronize_session=False
|
|
)
|
|
db.commit()
|
|
# Filing forty questions into a topic an article stands on links all forty,
|
|
# for the same reason filing one does.
|
|
linked = 0
|
|
if data.category_id is not None:
|
|
for question in db.query(Question).filter(Question.id.in_(data.question_ids)).all():
|
|
linked += topic_claims.link_new_question(db, question)
|
|
return {"updated": updated, "question_category_id": data.category_id,
|
|
"articles_linked": linked}
|
|
|
|
|
|
class BulkQuestionAction(BaseModel):
|
|
"""One editorial action applied to a checked set in the question manager."""
|
|
|
|
question_ids: list[int]
|
|
action: Literal["category", "difficulty", "delete"]
|
|
category_id: int | None = None
|
|
difficulty: Literal["easy", "medium", "hard"] | None = None
|
|
|
|
|
|
@router.post("/bulk")
|
|
def bulk_question_action(
|
|
data: BulkQuestionAction,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Apply one editorial action to up to 500 checked questions."""
|
|
scope = require_question_manager(db, current_user)
|
|
ids = list(dict.fromkeys(data.question_ids))
|
|
if not ids:
|
|
raise HTTPException(400, "No questions selected")
|
|
if len(ids) > 500:
|
|
raise HTTPException(400, "Select at most 500 questions per action")
|
|
assert_user_can_manage(db, current_user, ids)
|
|
if data.action == "category":
|
|
assert_can_manage_category(scope, data.category_id)
|
|
|
|
rows = db.query(Question).filter(Question.id.in_(ids))
|
|
if data.action == "category":
|
|
if data.category_id is not None and not db.get(QuestionCategory, data.category_id):
|
|
raise HTTPException(404, "Category not found")
|
|
updated = rows.update({"question_category_id": data.category_id}, synchronize_session=False)
|
|
elif data.action == "difficulty":
|
|
updated = rows.update({"difficulty": data.difficulty}, synchronize_session=False)
|
|
else: # delete
|
|
# Extra category links cascade with the question rows.
|
|
db.query(QuestionCategoryLink).filter(
|
|
QuestionCategoryLink.question_id.in_(ids)).delete(synchronize_session=False)
|
|
updated = rows.delete(synchronize_session=False)
|
|
db.commit()
|
|
return {"updated": updated, "action": data.action}
|
|
|
|
|
|
# Enough to undo a recent mistake without storing an unbounded history of full
|
|
# question bodies.
|
|
MAX_VERSIONS = 5
|
|
|
|
VERSIONED_FIELDS = ("question_text", "question_type", "options", "correct_answer",
|
|
"explanation", "option_explanations", "key_points", "attending_tip", "difficulty",
|
|
"question_category_id", "image_path", "explanation_image_path")
|
|
|
|
|
|
def _snapshot_question(db, question, user_id) -> None:
|
|
"""Store the question as it is now, then trim to the most recent MAX_VERSIONS."""
|
|
from app.models.question import QuestionVersion
|
|
|
|
db.add(QuestionVersion(
|
|
question_id=question.id,
|
|
snapshot={field: getattr(question, field, None) for field in VERSIONED_FIELDS},
|
|
edited_by=user_id,
|
|
))
|
|
db.flush()
|
|
keep = [row[0] for row in db.query(QuestionVersion.id).filter(
|
|
QuestionVersion.question_id == question.id
|
|
).order_by(QuestionVersion.created_at.desc(), QuestionVersion.id.desc()).limit(MAX_VERSIONS).all()]
|
|
if keep:
|
|
db.query(QuestionVersion).filter(
|
|
QuestionVersion.question_id == question.id,
|
|
~QuestionVersion.id.in_(keep),
|
|
).delete(synchronize_session=False)
|
|
|
|
|
|
class QuestionNoteIn(BaseModel):
|
|
content: str = Field(default="", max_length=8000)
|
|
|
|
|
|
@router.get("/detail/{question_id}/note")
|
|
def read_question_note(question_id: int, db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)):
|
|
"""This learner's own note on this question."""
|
|
from app.models.user_note import QuestionNote
|
|
|
|
row = db.query(QuestionNote).filter_by(
|
|
user_id=current_user.id, question_id=question_id).first()
|
|
return {"question_id": question_id, "content": row.content if row else "",
|
|
"updated_at": row.updated_at if row else None}
|
|
|
|
|
|
@router.put("/detail/{question_id}/note")
|
|
def write_question_note(question_id: int, data: QuestionNoteIn,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)):
|
|
"""Save or clear it. Empty means delete, so an emptied note leaves no trace."""
|
|
from app.models.user_note import QuestionNote
|
|
|
|
if not db.query(Question.id).filter(Question.id == question_id).first():
|
|
raise HTTPException(404, "Question not found")
|
|
row = db.query(QuestionNote).filter_by(
|
|
user_id=current_user.id, question_id=question_id).first()
|
|
content = data.content.strip()
|
|
if not content:
|
|
if row:
|
|
db.delete(row)
|
|
db.commit()
|
|
return {"question_id": question_id, "content": ""}
|
|
if row:
|
|
row.content = content
|
|
else:
|
|
db.add(QuestionNote(user_id=current_user.id, question_id=question_id, content=content))
|
|
db.commit()
|
|
return {"question_id": question_id, "content": content}
|
|
|
|
|
|
@router.get("/detail/{question_id}/versions")
|
|
def list_question_versions(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Recent snapshots of a question, newest first."""
|
|
from app.models.question import QuestionVersion
|
|
|
|
scope = require_question_manager(db, current_user)
|
|
assert_user_can_manage(db, current_user, [question_id])
|
|
rows = db.query(QuestionVersion).filter(
|
|
QuestionVersion.question_id == question_id
|
|
).order_by(QuestionVersion.created_at.desc(), QuestionVersion.id.desc()).limit(MAX_VERSIONS).all()
|
|
return [{
|
|
"id": row.id,
|
|
"created_at": row.created_at.isoformat() if row.created_at else None,
|
|
"edited_by": row.edited_by,
|
|
"question_text": (row.snapshot or {}).get("question_text"),
|
|
} for row in rows]
|
|
|
|
|
|
@router.post("/detail/{question_id}/versions/{version_id}/restore")
|
|
def restore_question_version(
|
|
question_id: int,
|
|
version_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Roll a question back, snapshotting the current state first so it is undoable too."""
|
|
from app.models.question import QuestionVersion
|
|
|
|
scope = require_question_manager(db, current_user)
|
|
assert_user_can_manage(db, current_user, [question_id])
|
|
question = db.query(Question).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(404, "Question not found")
|
|
version = db.query(QuestionVersion).filter_by(id=version_id, question_id=question_id).first()
|
|
if not version:
|
|
raise HTTPException(404, "Version not found")
|
|
|
|
_snapshot_question(db, question, current_user.id)
|
|
for field, value in (version.snapshot or {}).items():
|
|
if field in VERSIONED_FIELDS:
|
|
setattr(question, field, value)
|
|
db.commit()
|
|
db.refresh(question)
|
|
return {"id": question.id, "restored_from": version_id}
|
|
|
|
|
|
@router.get("/detail/{question_id}")
|
|
def get_question_detail(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""One question, for the full-page editor.
|
|
|
|
Path is /detail/{id} rather than /{id} so it cannot shadow the static
|
|
routes above it as more are added.
|
|
"""
|
|
question = bank_query(db, current_user).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(404, "Question not found")
|
|
extra = [row[0] for row in db.query(QuestionCategoryLink.category_id).filter(
|
|
QuestionCategoryLink.question_id == question.id).all()]
|
|
category = db.get(QuestionCategory, question.question_category_id) if question.question_category_id else None
|
|
# The answer side, for the people who write it — the same rule the bank
|
|
# listing applies, and this route did not. Being in the bank makes the stem
|
|
# yours to read; it has never made the answer yours. Anybody signed in
|
|
# could ask for /questions/detail/3869 and be handed the correct option,
|
|
# the explanation and the per-option reasoning for a question they had
|
|
# never sat. Nulled rather than refused, so the stem still reads.
|
|
mine = may_edit_question(db, question, current_user)
|
|
withheld = (lambda value: value if mine else None)
|
|
return {
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"question_type": question.question_type,
|
|
"options": question.options,
|
|
"correct_answer": withheld(question.correct_answer),
|
|
"explanation": withheld(question.explanation),
|
|
"option_explanations": withheld(question.option_explanations),
|
|
"key_points": withheld(question.key_points),
|
|
"attending_tip": withheld(question.attending_tip),
|
|
"figures": _figures_for(db, question.id) if mine else [
|
|
f for f in _figures_for(db, question.id) if f["role"] == "stem"],
|
|
"difficulty": question.difficulty,
|
|
"question_category_id": question.question_category_id,
|
|
"question_category_name": category.name if category else None,
|
|
"category_ids": sorted(set(extra) | ({question.question_category_id} if question.question_category_id else set())),
|
|
"image_path": question.image_path,
|
|
"explanation_image_path": withheld(question.explanation_image_path),
|
|
"user_id": question.user_id,
|
|
"source_quiz_id": question.source_quiz_id,
|
|
}
|
|
|
|
|
|
@router.get("/manage/summary")
|
|
def question_manager_summary(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Editorial health counters shown as one-click filters in the question manager.
|
|
|
|
A granted educator sees the counts for their own categories only.
|
|
"""
|
|
scope = require_question_manager(db, current_user)
|
|
base = db.query(func.count(Question.id))
|
|
if scope is not None:
|
|
base = base.filter(or_(
|
|
Question.question_category_id.in_(scope),
|
|
Question.id.in_(select(QuestionCategoryLink.question_id).where(
|
|
QuestionCategoryLink.category_id.in_(scope))),
|
|
))
|
|
blank = ("", " ")
|
|
return {
|
|
"total": base.scalar() or 0,
|
|
"uncategorized": base.filter(Question.question_category_id.is_(None)).scalar() or 0,
|
|
"no_difficulty": base.filter(Question.difficulty.is_(None)).scalar() or 0,
|
|
"no_explanation": base.filter(
|
|
or_(Question.explanation.is_(None), Question.explanation.in_(blank))).scalar() or 0,
|
|
"mine": base.filter(Question.user_id == current_user.id).scalar() or 0,
|
|
"scoped": scope is not None,
|
|
}
|
|
|
|
|
|
@router.get("/import/sample")
|
|
def download_sample_csv():
|
|
"""Download a sample CSV template for question import."""
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow(["question_text", "option_a", "option_b", "option_c", "option_d", "option_e", "correct_answer", "explanation", "category"])
|
|
writer.writerow([
|
|
"What is the most common cause of neonatal jaundice?",
|
|
"Physiological jaundice", "ABO incompatibility", "G6PD deficiency", "Breast milk jaundice", "",
|
|
"A",
|
|
"Physiological jaundice occurs in up to 60% of term neonates due to immature hepatic conjugation.",
|
|
"Neonatology",
|
|
])
|
|
writer.writerow([
|
|
"Which vaccine is given at birth?",
|
|
"BCG", "Hepatitis B", "Both BCG and Hepatitis B", "OPV", "",
|
|
"C",
|
|
"Both BCG and Hepatitis B are given at birth per the WHO immunization schedule.",
|
|
"Immunology",
|
|
])
|
|
buf.seek(0)
|
|
return StreamingResponse(
|
|
iter([buf.getvalue()]),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=question_import_sample.csv"},
|
|
)
|
|
|
|
|
|
@router.post("/import/upload")
|
|
def upload_questions_csv(
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Import questions from CSV or XLSX file.
|
|
|
|
Expected columns: question_text, option_a, option_b, option_c, option_d, option_e,
|
|
correct_answer (A/B/C/D/E or full text), explanation, category
|
|
"""
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="No file provided")
|
|
|
|
ext = file.filename.rsplit(".", 1)[-1].lower() if "." in file.filename else ""
|
|
raw = file.file.read()
|
|
|
|
rows = []
|
|
if ext in ("xlsx", "xls"):
|
|
try:
|
|
import openpyxl
|
|
wb = openpyxl.load_workbook(io.BytesIO(raw), read_only=True)
|
|
ws = wb.active
|
|
header = None
|
|
for row in ws.iter_rows(values_only=True):
|
|
if header is None:
|
|
header = [str(c or "").strip().lower().replace(" ", "_") for c in row]
|
|
continue
|
|
rows.append(dict(zip(header, [str(c) if c is not None else "" for c in row])))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Failed to parse XLSX: {e}")
|
|
elif ext == "csv":
|
|
try:
|
|
text = raw.decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
for row in reader:
|
|
rows.append({k.strip().lower().replace(" ", "_"): (v or "").strip() for k, v in row.items()})
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Failed to parse CSV: {e}")
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Unsupported file type. Use .csv or .xlsx")
|
|
|
|
if not rows:
|
|
raise HTTPException(status_code=400, detail="File is empty or has no data rows")
|
|
|
|
# Cache category lookups
|
|
cat_cache = {}
|
|
for cat in db.query(QuestionCategory).all():
|
|
cat_cache[cat.name.lower()] = cat.id
|
|
|
|
LETTER_MAP = {"a": 0, "b": 1, "c": 2, "d": 3, "e": 4}
|
|
created = []
|
|
errors = []
|
|
|
|
for i, row in enumerate(rows, start=2): # row 2 = first data row
|
|
q_text = row.get("question_text", "").strip()
|
|
if not q_text:
|
|
errors.append(f"Row {i}: missing question_text")
|
|
continue
|
|
|
|
options = []
|
|
for key in ("option_a", "option_b", "option_c", "option_d", "option_e"):
|
|
val = row.get(key, "").strip()
|
|
if val:
|
|
options.append(val)
|
|
|
|
correct_raw = row.get("correct_answer", "").strip()
|
|
correct_answer = ""
|
|
if correct_raw.lower() in LETTER_MAP and LETTER_MAP[correct_raw.lower()] < len(options):
|
|
correct_answer = options[LETTER_MAP[correct_raw.lower()]]
|
|
elif correct_raw in options:
|
|
correct_answer = correct_raw
|
|
else:
|
|
errors.append(f"Row {i}: invalid correct_answer '{correct_raw}'")
|
|
continue
|
|
|
|
explanation = row.get("explanation", "").strip() or None
|
|
cat_name = row.get("category", "").strip()
|
|
cat_id = None
|
|
if cat_name:
|
|
cat_lower = cat_name.lower()
|
|
if cat_lower not in cat_cache:
|
|
new_cat = QuestionCategory(name=cat_name, user_id=current_user.id)
|
|
db.add(new_cat)
|
|
db.flush()
|
|
cat_cache[cat_lower] = new_cat.id
|
|
cat_id = cat_cache[cat_lower]
|
|
|
|
question = Question(
|
|
question_text=q_text,
|
|
question_type="mcq" if len(options) >= 2 else "short_answer",
|
|
options=options if len(options) >= 2 else None,
|
|
correct_answer=correct_answer,
|
|
explanation=explanation,
|
|
question_category_id=cat_id,
|
|
)
|
|
db.add(question)
|
|
created.append(question)
|
|
|
|
db.commit()
|
|
|
|
# Embed in background
|
|
for q in created:
|
|
db.refresh(q)
|
|
try:
|
|
from app.services import embedding_service
|
|
embedding_service.embed_question(q)
|
|
except Exception:
|
|
logger.warning("Failed to generate embedding for imported question %d", q.id, exc_info=True)
|
|
if created:
|
|
db.commit()
|
|
|
|
return {
|
|
"imported": len(created),
|
|
"errors": errors[:20], # cap error list
|
|
"total_rows": len(rows),
|
|
}
|
|
|
|
|
|
# ── QTI Export / Import ──────────────────────────────────────────────
|
|
|
|
def _escape_xml(text: str) -> str:
|
|
"""Escape text for XML content."""
|
|
if not text:
|
|
return ""
|
|
return (text.replace("&", "&").replace("<", "<")
|
|
.replace(">", ">").replace('"', """).replace("'", "'"))
|
|
|
|
|
|
@router.get("/export/qti")
|
|
def export_qti(
|
|
question_ids: str = Query(None, description="Comma-separated question IDs (omit for all shared)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Export questions as QTI 2.1 XML."""
|
|
if question_ids:
|
|
ids = [int(x.strip()) for x in question_ids.split(",") if x.strip().isdigit()]
|
|
questions = bank_query(db, current_user).filter(Question.id.in_(ids)).all()
|
|
if len(questions) != len(set(ids)):
|
|
raise HTTPException(400, "Some questions are missing, private, or unavailable")
|
|
else:
|
|
questions = bank_query(db, current_user).limit(500).all()
|
|
|
|
items_xml = []
|
|
for q in questions:
|
|
options = q.options or []
|
|
correct = q.correct_answer or ""
|
|
correct_id = None
|
|
|
|
choices_xml = []
|
|
for i, opt in enumerate(options):
|
|
choice_id = f"choice_{i}"
|
|
if opt.strip().lower() == correct.strip().lower():
|
|
correct_id = choice_id
|
|
choices_xml.append(
|
|
f' <simpleChoice identifier="{choice_id}">{_escape_xml(opt)}</simpleChoice>'
|
|
)
|
|
|
|
if not correct_id:
|
|
correct_id = "choice_0"
|
|
|
|
explanation_xml = ""
|
|
if q.explanation:
|
|
explanation_xml = f"""
|
|
<modalFeedback outcomeIdentifier="FEEDBACK" showHide="show">
|
|
{_escape_xml(q.explanation)}
|
|
</modalFeedback>"""
|
|
|
|
item = f""" <assessmentItem identifier="q_{q.id}" title="{_escape_xml(q.question_text[:80])}" adaptive="false" timeDependent="false">
|
|
<responseDeclaration identifier="RESPONSE" cardinality="single" baseType="identifier">
|
|
<correctResponse>
|
|
<value>{correct_id}</value>
|
|
</correctResponse>
|
|
</responseDeclaration>
|
|
<outcomeDeclaration identifier="SCORE" cardinality="single" baseType="float">
|
|
<defaultValue><value>0</value></defaultValue>
|
|
</outcomeDeclaration>
|
|
<itemBody>
|
|
<choiceInteraction responseIdentifier="RESPONSE" shuffle="false" maxChoices="1">
|
|
<prompt>{_escape_xml(q.question_text)}</prompt>
|
|
{chr(10).join(choices_xml)}
|
|
</choiceInteraction>
|
|
</itemBody>
|
|
<responseProcessing template="http://www.imsglobal.org/question/qti_v2p1/rptemplates/match_correct"/>{explanation_xml}
|
|
</assessmentItem>"""
|
|
items_xml.append(item)
|
|
|
|
xml = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<assessmentTest xmlns="http://www.imsglobal.org/xsd/imsqti_v2p1"
|
|
identifier="pedshub_export" title="PedsHub Question Export">
|
|
{chr(10).join(items_xml)}
|
|
</assessmentTest>"""
|
|
|
|
return StreamingResponse(
|
|
io.BytesIO(xml.encode("utf-8")),
|
|
media_type="application/xml",
|
|
headers={"Content-Disposition": "attachment; filename=questions_qti.xml"},
|
|
)
|
|
|
|
|
|
@router.post("/import/qti")
|
|
def import_qti(
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Import questions from a QTI 2.1 XML file."""
|
|
import xml.etree.ElementTree as ET
|
|
|
|
if not file.filename or not file.filename.lower().endswith(".xml"):
|
|
raise HTTPException(status_code=400, detail="Please upload a QTI XML file")
|
|
|
|
contents = file.file.read()
|
|
try:
|
|
root = ET.fromstring(contents)
|
|
except ET.ParseError:
|
|
raise HTTPException(status_code=400, detail="Invalid XML file")
|
|
|
|
# Handle namespaces
|
|
ns_match = re.match(r"\{(.+?)\}", root.tag)
|
|
ns = ns_match.group(1) if ns_match else ""
|
|
prefix = f"{{{ns}}}" if ns else ""
|
|
|
|
created = []
|
|
errors = []
|
|
|
|
# Find all assessmentItem elements (could be nested or flat)
|
|
items = root.findall(f".//{prefix}assessmentItem")
|
|
if not items:
|
|
items = root.findall(f".//assessmentItem") # try without namespace
|
|
|
|
for idx, item in enumerate(items):
|
|
try:
|
|
# Extract prompt/question text
|
|
prompt_el = item.find(f".//{prefix}prompt")
|
|
if prompt_el is None:
|
|
prompt_el = item.find(f".//prompt")
|
|
question_text = (prompt_el.text or "").strip() if prompt_el is not None else item.get("title", "")
|
|
|
|
if not question_text:
|
|
errors.append(f"Item {idx + 1}: No question text found")
|
|
continue
|
|
|
|
# Extract choices
|
|
options = []
|
|
choice_map = {} # identifier -> text
|
|
for choice in item.findall(f".//{prefix}simpleChoice") or item.findall(f".//simpleChoice"):
|
|
text = (choice.text or "").strip()
|
|
identifier = choice.get("identifier", "")
|
|
options.append(text)
|
|
choice_map[identifier] = text
|
|
|
|
# Extract correct answer
|
|
correct_answer = ""
|
|
correct_el = item.find(f".//{prefix}correctResponse/{prefix}value")
|
|
if correct_el is None:
|
|
correct_el = item.find(f".//correctResponse/value")
|
|
if correct_el is not None and correct_el.text:
|
|
correct_id = correct_el.text.strip()
|
|
correct_answer = choice_map.get(correct_id, "")
|
|
|
|
# Extract explanation from modalFeedback
|
|
explanation = ""
|
|
feedback_el = item.find(f".//{prefix}modalFeedback")
|
|
if feedback_el is None:
|
|
feedback_el = item.find(f".//modalFeedback")
|
|
if feedback_el is not None:
|
|
explanation = (feedback_el.text or "").strip()
|
|
|
|
q = Question(
|
|
question_text=question_text,
|
|
question_type="multiple_choice",
|
|
options=options if options else None,
|
|
correct_answer=correct_answer,
|
|
explanation=explanation or None,
|
|
user_id=current_user.id,
|
|
)
|
|
db.add(q)
|
|
created.append(q)
|
|
except Exception as e:
|
|
errors.append(f"Item {idx + 1}: {str(e)[:80]}")
|
|
|
|
if created:
|
|
db.commit()
|
|
|
|
return {
|
|
"imported": len(created),
|
|
"errors": errors[:20],
|
|
"total_items": len(items),
|
|
}
|
|
|
|
|
|
# ── Figures ───────────────────────────────────────────────────────────────────
|
|
|
|
class FigureIn(BaseModel):
|
|
media_id: int
|
|
role: Literal["stem", "explanation"] = "stem"
|
|
label: str | None = Field(default=None, max_length=80)
|
|
caption: str | None = None
|
|
|
|
|
|
class FigureUpdate(BaseModel):
|
|
label: str | None = Field(default=None, max_length=80)
|
|
caption: str | None = None
|
|
role: Literal["stem", "explanation"] | None = None
|
|
position: int | None = None
|
|
|
|
|
|
@router.get("/detail/{question_id}/figures")
|
|
def list_figures(question_id: int, db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)):
|
|
return _figures_for(db, question_id)
|
|
|
|
|
|
@router.post("/detail/{question_id}/figures", status_code=201)
|
|
def add_figure(question_id: int, data: FigureIn, db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator)):
|
|
from app.models.media import MediaAsset
|
|
from app.models.question_media import QuestionMedia
|
|
|
|
if not db.query(Question.id).filter(Question.id == question_id).first():
|
|
raise HTTPException(404, "Question not found")
|
|
if not db.get(MediaAsset, data.media_id):
|
|
raise HTTPException(404, "Image not found")
|
|
if db.query(QuestionMedia.id).filter_by(
|
|
question_id=question_id, media_id=data.media_id, role=data.role).first():
|
|
raise HTTPException(409, "That image is already on this question")
|
|
|
|
position = db.query(QuestionMedia).filter_by(question_id=question_id, role=data.role).count()
|
|
link = QuestionMedia(question_id=question_id, media_id=data.media_id, role=data.role,
|
|
label=data.label, caption=data.caption, position=position)
|
|
db.add(link)
|
|
db.commit()
|
|
return {"figures": _figures_for(db, question_id)}
|
|
|
|
|
|
@router.patch("/figures/{figure_id}")
|
|
def update_figure(figure_id: int, data: FigureUpdate, db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator)):
|
|
from app.models.question_media import QuestionMedia
|
|
|
|
link = db.get(QuestionMedia, figure_id)
|
|
if not link:
|
|
raise HTTPException(404, "Figure not found")
|
|
for field, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(link, field, value)
|
|
db.commit()
|
|
return {"figures": _figures_for(db, link.question_id)}
|
|
|
|
|
|
@router.delete("/figures/{figure_id}", status_code=204)
|
|
def remove_figure(figure_id: int, db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator)):
|
|
"""Take a figure off a question. The image itself stays in the bank."""
|
|
from app.models.question_media import QuestionMedia
|
|
|
|
link = db.get(QuestionMedia, figure_id)
|
|
if not link:
|
|
raise HTTPException(404, "Figure not found")
|
|
question_id, role, position = link.question_id, link.role, link.position
|
|
db.delete(link)
|
|
db.flush()
|
|
# Close the gap so labels that fall back to a number stay sequential.
|
|
for other in db.query(QuestionMedia).filter(
|
|
QuestionMedia.question_id == question_id, QuestionMedia.role == role,
|
|
QuestionMedia.position > position).all():
|
|
other.position -= 1
|
|
db.commit()
|
|
|
|
|
|
@router.get("/{question_id}/articles")
|
|
def question_articles(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""The reading linked to one question: the mirror of `GET /articles/{id}/questions`.
|
|
|
|
Both sides answer with the same fields for the link itself — `section_id`
|
|
and `section_title` — so the relationship reads the same whichever end you
|
|
are standing at. Only cards are returned, never article bodies: this used
|
|
to be `GET /articles/linked`, which sent the whole prose of every linked
|
|
article to a page that draws a list of titles.
|
|
"""
|
|
if not db.query(Question.id).filter(
|
|
Question.id == question_id, bank_question_predicate(current_user)).first():
|
|
raise HTTPException(404, "Question not found")
|
|
rows = db.query(QuestionArticleLink, Article).join(
|
|
Article, Article.id == QuestionArticleLink.article_id,
|
|
).filter(QuestionArticleLink.question_id == question_id).all()
|
|
articles = []
|
|
for link, article in rows:
|
|
# Drafts are editorial work in progress; a learner following a link into
|
|
# one would be reading something nobody has checked.
|
|
if article.status != "published" and not current_user.is_moderator:
|
|
continue
|
|
articles.append({
|
|
"article_id": article.id, "title": article.title, "slug": article.slug,
|
|
"summary": article.summary, "status": article.status,
|
|
"section_id": link.section_id,
|
|
"section_title": article_service.section_title(article, link.section_id),
|
|
})
|
|
return articles
|