291 lines
9.8 KiB
Python
291 lines
9.8 KiB
Python
"""题库管理：题目列表、删除、按章节生成新题。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from database import REPO_ROOT
|
|
from dependencies import require_permission
|
|
from models import (
|
|
Chapter,
|
|
Knowledge,
|
|
LlmSetting,
|
|
Question,
|
|
QuestionKnowledge,
|
|
Textbook,
|
|
User,
|
|
)
|
|
from schemas import GenerateIn, QuestionBankItemOut, QuestionUpdate
|
|
from services.question_generator import generate_questions
|
|
from services.llm_config import get_default_channel
|
|
from services.knowledge_service import (
|
|
question_knowledge_names,
|
|
set_question_knowledge_names,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/question-bank", tags=["question-bank"])
|
|
|
|
IMAGE_UPLOAD_DIR = REPO_ROOT / "data" / "question_images"
|
|
|
|
|
|
def _store_generated_svg(svg: str) -> str:
|
|
"""把模型生成的 SVG 存成题目独立图片，返回可访问的相对 URL。"""
|
|
import re
|
|
|
|
clean = re.sub(r"<script[\s\S]*?</script>", "", svg, flags=re.I)
|
|
clean = re.sub(r"\son\w+=[\"'][^\"']*[\"']", "", clean, flags=re.I)
|
|
IMAGE_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
filename = f"generated_{uuid.uuid4().hex}.svg"
|
|
(IMAGE_UPLOAD_DIR / filename).write_text(clean, encoding="utf-8")
|
|
return f"/api/question-images/{filename}"
|
|
|
|
|
|
def _resolve_selected_knowledge(
|
|
db: Session,
|
|
knowledge_ids: list[int],
|
|
knowledge_names: list[str],
|
|
fallback: str,
|
|
) -> list[str]:
|
|
if knowledge_ids:
|
|
rows = (
|
|
db.query(Knowledge)
|
|
.filter(Knowledge.id.in_(knowledge_ids))
|
|
.all()
|
|
)
|
|
by_id = {row.id: row for row in rows}
|
|
missing = [kid for kid in knowledge_ids if kid not in by_id]
|
|
if missing:
|
|
raise HTTPException(
|
|
status_code=400, detail="部分所选知识点不存在，请刷新知识图谱后重试"
|
|
)
|
|
return [by_id[kid].name for kid in knowledge_ids]
|
|
names = [
|
|
name.strip() for name in knowledge_names if name.strip()
|
|
]
|
|
if names:
|
|
return names
|
|
return [fallback.strip()] if fallback.strip() else []
|
|
|
|
|
|
def _to_item(
|
|
db: Session,
|
|
question: Question,
|
|
chapter: Chapter | None,
|
|
textbook: Textbook | None,
|
|
):
|
|
knowledge_rows = (
|
|
db.query(Knowledge.id, Knowledge.name)
|
|
.join(
|
|
QuestionKnowledge,
|
|
QuestionKnowledge.knowledge_id == Knowledge.id,
|
|
)
|
|
.filter(QuestionKnowledge.question_id == question.id)
|
|
.order_by(QuestionKnowledge.id.asc())
|
|
.all()
|
|
)
|
|
if not knowledge_rows and question.knowledge_name:
|
|
knowledge_rows = [
|
|
(0, question.knowledge_name),
|
|
]
|
|
knowledge_names = [name for _, name in knowledge_rows]
|
|
return QuestionBankItemOut(
|
|
id=question.id,
|
|
chapter_id=question.chapter_id,
|
|
chapter_name=chapter.name if chapter else "",
|
|
textbook_name=textbook.name if textbook else "",
|
|
knowledge_name=question.knowledge_name,
|
|
stem=question.stem,
|
|
options=json.loads(question.options),
|
|
correct_index=question.correct_index,
|
|
explanation=question.explanation,
|
|
difficulty=question.difficulty,
|
|
is_generated=bool(question.is_generated),
|
|
used_count=question.used_count or 0,
|
|
knowledge_names=[str(name) for name in knowledge_names],
|
|
knowledge_ids=[kid for kid, _ in knowledge_rows if kid],
|
|
)
|
|
|
|
|
|
def _query(db: Session):
|
|
return (
|
|
db.query(Question, Chapter, Textbook)
|
|
.outerjoin(Chapter, Question.chapter_id == Chapter.id)
|
|
.outerjoin(Textbook, Chapter.textbook_id == Textbook.id)
|
|
)
|
|
|
|
|
|
@router.get("/questions", response_model=list[QuestionBankItemOut])
|
|
def list_questions(
|
|
textbook_id: int | None = None,
|
|
chapter_id: int | None = None,
|
|
knowledge_name: str = "",
|
|
generated: bool | None = None,
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(require_permission("question-bank:manage")),
|
|
):
|
|
query = _query(db)
|
|
if textbook_id:
|
|
query = query.filter(Textbook.id == textbook_id)
|
|
if chapter_id:
|
|
query = query.filter(Question.chapter_id == chapter_id)
|
|
if knowledge_name:
|
|
query = query.filter(Question.knowledge_name == knowledge_name)
|
|
if generated is not None:
|
|
query = query.filter(Question.is_generated == generated)
|
|
rows = query.order_by(Question.id.desc()).all()
|
|
return [_to_item(db, question, chapter, textbook) for question, chapter, textbook in rows]
|
|
|
|
|
|
@router.delete("/questions/{question_id}")
|
|
def delete_question(
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(require_permission("question-bank:manage")),
|
|
):
|
|
question = db.get(Question, question_id)
|
|
if question is None:
|
|
raise HTTPException(status_code=404, detail="题目不存在")
|
|
used_count = question.used_count or 0
|
|
if used_count > 0:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"该题目已被使用 {used_count} 次，不能删除；只能编辑题目内容",
|
|
)
|
|
db.delete(question)
|
|
db.commit()
|
|
return {"deleted": question_id}
|
|
|
|
|
|
@router.put("/questions/{question_id}", response_model=QuestionBankItemOut)
|
|
def update_question(
|
|
question_id: int,
|
|
payload: QuestionUpdate,
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(require_permission("question-bank:manage")),
|
|
):
|
|
question = db.get(Question, question_id)
|
|
if question is None:
|
|
raise HTTPException(status_code=404, detail="题目不存在")
|
|
if payload.chapter_id is not None:
|
|
chapter = db.get(Chapter, payload.chapter_id)
|
|
if chapter is None:
|
|
raise HTTPException(status_code=404, detail="章节不存在")
|
|
question.chapter_id = chapter.id
|
|
else:
|
|
question.chapter_id = None
|
|
options = [option.strip() for option in payload.options]
|
|
if any(not option for option in options) or len(set(options)) != 4:
|
|
raise HTTPException(status_code=400, detail="选项必须是 4 个非空且不重复的内容")
|
|
question.stem = payload.stem.strip()
|
|
question.options = json.dumps(options, ensure_ascii=False)
|
|
question.correct_index = payload.correct_index
|
|
question.explanation = payload.explanation.strip()
|
|
chosen = _resolve_selected_knowledge(
|
|
db, payload.knowledge_ids, payload.knowledge_names, payload.knowledge_name
|
|
)
|
|
if not chosen:
|
|
raise HTTPException(status_code=400, detail="请至少选择一个知识点")
|
|
question.knowledge_name = chosen[0]
|
|
set_question_knowledge_names(
|
|
db, question, chosen, attach_to_chapter=True
|
|
)
|
|
question.difficulty = payload.difficulty
|
|
db.commit()
|
|
chapter = (
|
|
db.query(Chapter).filter(Chapter.id == question.chapter_id).first()
|
|
if question.chapter_id is not None
|
|
else None
|
|
)
|
|
textbook = (
|
|
db.query(Textbook).filter(Textbook.id == chapter.textbook_id).first()
|
|
if chapter is not None
|
|
else None
|
|
)
|
|
return _to_item(db, question, chapter, textbook)
|
|
|
|
|
|
@router.post("/generate", response_model=list[QuestionBankItemOut])
|
|
def generate(
|
|
payload: GenerateIn,
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(require_permission("question-bank:manage")),
|
|
):
|
|
chapter = db.get(Chapter, payload.chapter_id)
|
|
if chapter is None:
|
|
raise HTTPException(status_code=404, detail="章节不存在")
|
|
textbook = db.get(Textbook, chapter.textbook_id)
|
|
if textbook is None:
|
|
raise HTTPException(status_code=404, detail="教材/课程不存在")
|
|
if payload.channel_id is not None:
|
|
setting = db.get(LlmSetting, payload.channel_id)
|
|
if setting is None:
|
|
raise HTTPException(status_code=404, detail="指定的模型通道不存在")
|
|
else:
|
|
setting = get_default_channel(db)
|
|
if setting is None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="尚未配置模型通道，请先到“模型配置”新增并设为默认",
|
|
)
|
|
|
|
try:
|
|
raw_items = generate_questions(
|
|
db=db,
|
|
setting=setting,
|
|
chapter=chapter,
|
|
textbook=textbook,
|
|
count=payload.count,
|
|
difficulty=payload.difficulty,
|
|
knowledge_name=(
|
|
"、".join(
|
|
name.strip()
|
|
for name in payload.knowledge_names
|
|
if name.strip()
|
|
)
|
|
or payload.knowledge_name.strip()
|
|
),
|
|
instructions=payload.instructions.strip(),
|
|
need_figure=payload.need_figure,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
inserted: list[Question] = []
|
|
for item in raw_items:
|
|
chosen = _resolve_selected_knowledge(
|
|
db,
|
|
payload.knowledge_ids,
|
|
payload.knowledge_names,
|
|
item["knowledge_name"],
|
|
)
|
|
if not chosen:
|
|
chosen = [item["knowledge_name"]]
|
|
stem = item["stem"]
|
|
if payload.need_figure and item.get("image_svg"):
|
|
image_url = _store_generated_svg(item["image_svg"])
|
|
stem = f"{stem}\n\n"
|
|
question = Question(
|
|
stem=stem,
|
|
options=json.dumps(item["options"], ensure_ascii=False),
|
|
correct_index=item["correct_index"],
|
|
knowledge_name=chosen[0],
|
|
explanation=item["explanation"],
|
|
chapter_id=chapter.id,
|
|
difficulty=item["difficulty"],
|
|
is_generated=True,
|
|
)
|
|
db.add(question)
|
|
inserted.append(question)
|
|
db.flush()
|
|
for question in inserted:
|
|
set_question_knowledge_names(db, question, chosen, attach_to_chapter=True)
|
|
|
|
result = [_to_item(db, question, chapter, textbook) for question in inserted]
|
|
db.commit()
|
|
return result
|