8137d5b20c
- Fully decouple course quizzes from main quiz system (hidden from dashboard stats, history, search, results page) - Course quiz results show "Back to Course" instead of retake/delete - Add allow_review toggle for course creators to control answer review - Show quiz title on course page, hide pool size from students - Add course thumbnails to browse cards - Replace passlib with bcrypt directly (compatible with existing hashes) - Add HIBP breached password warnings on register/reset/change password - Add CLI management tools (reset-password, set-role, stats, etc.) - Fix quiz PATCH endpoint: ownership check instead of moderator-only - Add max_length validation on course/module/lesson titles - Fix score display bug on results page (0 of N when review disabled) - Fix question count on course quiz start (show per-attempt, not pool) - Improve suspend warning for timed course quizzes with max attempts - Clean up validation error messages (show "Invalid email" not Pydantic dump) - Add DDL migration for allow_review column Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
658 lines
26 KiB
Python
658 lines
26 KiB
Python
"""Question bank — view, search, categorise, and create quizzes from individual questions."""
|
|
import csv
|
|
import io
|
|
import os
|
|
import re
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import cast, String, or_, text as sa_text, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models.question import Question
|
|
from app.models.question_category import QuestionCategory
|
|
from app.models.quiz import Quiz
|
|
from app.models.user import User
|
|
from app.models.favorite import Favorite
|
|
from app.utils.auth import get_current_user, require_moderator
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@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),
|
|
):
|
|
"""Delete a question. Moderators can delete any; regular users can delete their own."""
|
|
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")
|
|
if not is_mod:
|
|
# 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")
|
|
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
|
|
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. Owner or moderator can edit."""
|
|
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")
|
|
if question.user_id != current_user.id and not is_mod:
|
|
raise HTTPException(status_code=403, detail="Not authorized to edit this question")
|
|
for field, value in data.model_dump(exclude_unset=True).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,
|
|
}
|
|
|
|
|
|
@router.patch("/{question_id}/share")
|
|
def toggle_question_share(
|
|
question_id: int,
|
|
shared: int = Query(..., description="1 = shared, 0 = private"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Toggle question sharing. Owner or admin can change."""
|
|
question = db.query(Question).filter(Question.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
is_admin = current_user.role in ("admin", "moderator")
|
|
if question.user_id != current_user.id and not is_admin:
|
|
raise HTTPException(status_code=403, detail="Not your question")
|
|
question.is_shared = 1 if shared else 0
|
|
db.commit()
|
|
return {"id": question_id, "is_shared": question.is_shared}
|
|
|
|
|
|
@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()
|
|
return {"question_id": question_id, "question_category_id": category_id}
|
|
|
|
|
|
@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),
|
|
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 = db.query(Question.id)
|
|
if quiz_id:
|
|
query = query.filter(Question.source_quiz_id == quiz_id)
|
|
if category_ids:
|
|
cat_id_list = [int(c.strip()) for c in category_ids.split(",") if c.strip().isdigit()]
|
|
if cat_id_list:
|
|
query = query.filter(Question.question_category_id.in_(cat_id_list))
|
|
elif category_id is not None:
|
|
query = query.filter(Question.question_category_id == 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),
|
|
favorites_only: bool = Query(False),
|
|
my_questions: bool = Query(False, description="Show only questions created by current user"),
|
|
tag_ids: str | None = Query(None, description="Comma-separated tag IDs (AND filter)"),
|
|
search_mode: str = Query("hybrid"), # "keyword" | "semantic" | "hybrid"
|
|
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."""
|
|
if my_questions:
|
|
query = db.query(Question).filter(Question.user_id == current_user.id)
|
|
else:
|
|
query = db.query(Question).filter(
|
|
or_(
|
|
Question.is_shared == 1,
|
|
Question.is_shared.is_(None), # legacy questions without is_shared
|
|
Question.user_id == current_user.id, # always show own questions
|
|
)
|
|
)
|
|
|
|
if quiz_id:
|
|
query = query.filter(Question.source_quiz_id == quiz_id)
|
|
|
|
if category_ids:
|
|
cat_id_list = [int(c.strip()) for c in category_ids.split(",") if c.strip().isdigit()]
|
|
if cat_id_list:
|
|
query = query.filter(Question.question_category_id.in_(cat_id_list))
|
|
elif category_id is not None:
|
|
query = query.filter(Question.question_category_id == category_id)
|
|
|
|
if uncategorized:
|
|
query = query.filter(Question.question_category_id.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))
|
|
|
|
# ── Semantic search (pgvector) ─────────────────────────────────
|
|
semantic_ids_ordered: list[int] = []
|
|
if q and q.strip() and search_mode in ("semantic", "hybrid"):
|
|
from app.services.embedding_service import generate_embedding
|
|
emb = generate_embedding(q.strip())
|
|
if emb:
|
|
# Validate all values are finite floats before interpolating into SQL
|
|
emb_literal = "[" + ",".join(str(float(x)) for x in emb) + "]"
|
|
rows = db.execute(sa_text("""
|
|
SELECT id, 1 - (embedding <=> CAST(:vec AS vector)) AS sim
|
|
FROM questions
|
|
WHERE embedding IS NOT NULL
|
|
ORDER BY embedding <=> CAST(:vec AS vector)
|
|
LIMIT :lim
|
|
"""), {"vec": emb_literal, "lim": limit * 2}).fetchall()
|
|
semantic_ids_ordered = [r.id for r in rows if float(r.sim) >= 0.55]
|
|
|
|
# ── Keyword filter ─────────────────────────────────────────────
|
|
if q and q.strip() and search_mode in ("keyword", "hybrid"):
|
|
phrase = q.strip()
|
|
query = query.filter(
|
|
or_(
|
|
Question.question_text.ilike(f"%{phrase}%"),
|
|
cast(Question.options, String).ilike(f"%{phrase}%"),
|
|
)
|
|
)
|
|
|
|
# Apply semantic ID filter if semantic-only mode
|
|
if q and q.strip() and search_mode == "semantic" and semantic_ids_ordered:
|
|
query = query.filter(Question.id.in_(semantic_ids_ordered))
|
|
|
|
total = query.count()
|
|
questions = query.order_by(Question.source_quiz_id, Question.id).offset(offset).limit(limit).all()
|
|
|
|
# If hybrid: merge semantic first then keyword remainder
|
|
if semantic_ids_ordered and search_mode == "hybrid":
|
|
sem_set = set(semantic_ids_ordered)
|
|
sem_qs = [qu for qu in questions if qu.id in sem_set]
|
|
kw_qs = [qu for qu in questions if qu.id not in sem_set]
|
|
# Sort semantic by original similarity order
|
|
sem_order = {qid: i for i, qid in enumerate(semantic_ids_ordered)}
|
|
sem_qs.sort(key=lambda qu: sem_order.get(qu.id, 999))
|
|
questions = sem_qs + kw_qs
|
|
|
|
quiz_cache: dict[int, str] = {}
|
|
cat_cache: dict[int, str] = {}
|
|
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,
|
|
"question_text": qu.question_text,
|
|
"question_type": qu.question_type,
|
|
"options": qu.options,
|
|
"correct_answer": qu.correct_answer,
|
|
"explanation": qu.explanation,
|
|
"image_path": qu.image_path,
|
|
"user_id": qu.user_id,
|
|
"is_shared": qu.is_shared if qu.is_shared is not None else 1,
|
|
})
|
|
|
|
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
|
|
|
|
|
|
UPLOAD_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "..", "uploads", "questions")
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
|
|
|
|
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
|
|
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 (not from PDF extraction)."""
|
|
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")
|
|
|
|
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=data.image_path,
|
|
user_id=current_user.id,
|
|
is_shared=1,
|
|
)
|
|
db.add(question)
|
|
db.commit()
|
|
db.refresh(question)
|
|
|
|
# Generate embedding in background
|
|
try:
|
|
from app.services import embedding_service
|
|
embedding_service.embed_question(question)
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
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[:12]}.{ext}"
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
with open(filepath, "wb") as f:
|
|
f.write(file.file.read())
|
|
rel_path = f"questions/{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."""
|
|
images = (
|
|
db.query(Question.image_path)
|
|
.filter(Question.image_path.isnot(None), Question.image_path != "")
|
|
.distinct()
|
|
.limit(200)
|
|
.all()
|
|
)
|
|
return [{"image_path": img[0], "url": f"/uploads/{img[0]}"} for img in images]
|
|
|
|
|
|
class CreateFromBankRequest(BaseModel):
|
|
title: str
|
|
question_ids: list[int]
|
|
mode: str = "timed"
|
|
time_limit_minutes: int | None = None
|
|
|
|
|
|
@router.post("/from-bank")
|
|
def create_quiz_from_bank(
|
|
data: CreateFromBankRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create a new quiz referencing existing bank questions (no copying — edits propagate)."""
|
|
if not data.title.strip():
|
|
raise HTTPException(status_code=400, detail="Title is required")
|
|
if not data.question_ids:
|
|
raise HTTPException(status_code=400, detail="Select at least one question")
|
|
if data.mode not in ("timed", "learning"):
|
|
raise HTTPException(status_code=400, detail="Mode must be timed or learning")
|
|
|
|
source_questions = db.query(Question).filter(Question.id.in_(data.question_ids)).all()
|
|
if not source_questions:
|
|
raise HTTPException(status_code=404, detail="No valid questions found")
|
|
# Preserve caller's requested order
|
|
id_order = {qid: i for i, qid in enumerate(data.question_ids)}
|
|
source_questions.sort(key=lambda q: id_order.get(q.id, len(data.question_ids)))
|
|
|
|
# Find a section_id from the source questions' origin quiz
|
|
first_src_quiz_id = source_questions[0].source_quiz_id
|
|
first_quiz = db.query(Quiz).filter(Quiz.id == first_src_quiz_id).first() if first_src_quiz_id else None
|
|
section_id = first_quiz.section_id if first_quiz else None
|
|
|
|
is_mod = current_user.role in ("admin", "moderator")
|
|
|
|
new_quiz = Quiz(
|
|
section_id=section_id,
|
|
user_id=current_user.id,
|
|
title=data.title.strip(),
|
|
questions_count=len(source_questions),
|
|
mode=data.mode,
|
|
time_limit_minutes=data.time_limit_minutes,
|
|
is_published=1 if is_mod else 0,
|
|
is_shared=1 if is_mod else 0,
|
|
)
|
|
db.add(new_quiz)
|
|
db.flush()
|
|
|
|
# Reference existing questions via junction (no copies)
|
|
from app.models.quiz_question_link import QuizQuestionLink
|
|
for pos, sq in enumerate(source_questions):
|
|
db.add(QuizQuestionLink(quiz_id=new_quiz.id, question_id=sq.id, position=pos))
|
|
|
|
db.commit()
|
|
db.refresh(new_quiz)
|
|
return {"id": new_quiz.id, "title": new_quiz.title, "questions_count": new_quiz.questions_count}
|
|
|
|
|
|
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()
|
|
return {"updated": updated, "question_category_id": data.category_id}
|
|
|
|
|
|
@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:
|
|
pass
|
|
if created:
|
|
db.commit()
|
|
|
|
return {
|
|
"imported": len(created),
|
|
"errors": errors[:20], # cap error list
|
|
"total_rows": len(rows),
|
|
}
|