318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""题库管理：题目列表、删除、按章节生成新题。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy import func
|
||
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 (
|
||
BankSummaryOut,
|
||
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.get("/summary", response_model=BankSummaryOut)
|
||
def summary(
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_permission("question-bank:manage")),
|
||
):
|
||
"题库侧栏计数：总数、未挂章节、AI 生成、已被使用，一次算完。"
|
||
|
||
def count(*conditions) -> int:
|
||
query = db.query(func.count(Question.id))
|
||
for condition in conditions:
|
||
query = query.filter(condition)
|
||
return int(query.scalar() or 0)
|
||
|
||
return BankSummaryOut(
|
||
total=count(),
|
||
unattached=count(Question.chapter_id.is_(None)),
|
||
generated=count(Question.is_generated.is_(True)),
|
||
used=count(Question.used_count > 0),
|
||
)
|
||
|
||
|
||
@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
|