262 lines
8.5 KiB
Python
262 lines
8.5 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 测验题目管理 API（管理员）
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：管理后台的测验题目审批 API，统一 /api/admin/quiz 前缀。
|
|||
|
|
# 包含 4 个端点：
|
|||
|
|
# 1. POST /generate — 手动触发 Dify 生成题目
|
|||
|
|
# 2. GET /pending — 查看待审核题目列表（分页）
|
|||
|
|
# 3. POST /{id}/approve — 审批通过题目（is_active → True）
|
|||
|
|
# 4. DELETE /{id} — 删除质量差的题目
|
|||
|
|
#
|
|||
|
|
# 权限：所有端点需要管理员权限（Depends(require_admin)）
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
from typing import Any, Dict, Optional
|
|||
|
|
|
|||
|
|
from fastapi import APIRouter, Depends, Query
|
|||
|
|
from pydantic import BaseModel
|
|||
|
|
from sqlalchemy import select, func
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.api.agents import get_current_agent
|
|||
|
|
from app.database import get_db
|
|||
|
|
from app.models.agent import Agent
|
|||
|
|
from app.models.quiz import QuizQuestion
|
|||
|
|
from app.services.quiz_generation_service import get_quiz_generation_service
|
|||
|
|
from app.utils.response import AppException, success_response
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
router = APIRouter(prefix="/admin/quiz", tags=["测验题目管理"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 权限校验依赖（复用 admin_api.py 的模式）
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
async def require_admin(
|
|||
|
|
agent: Agent = Depends(get_current_agent),
|
|||
|
|
) -> Agent:
|
|||
|
|
"""管理员权限校验：仅 role='admin' 可访问。"""
|
|||
|
|
if agent.role != "admin":
|
|||
|
|
raise AppException(1004, "无管理权限")
|
|||
|
|
return agent
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 请求体定义
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
class GenerateRequest(BaseModel):
|
|||
|
|
"""手动触发生成题目请求。"""
|
|||
|
|
category: str # network/vpn/email/system/printer/security/office
|
|||
|
|
question_type: str = "knowledge" # knowledge / diagnostic
|
|||
|
|
count: int = 5
|
|||
|
|
problem_category: Optional[str] = None # 仅 question_type=diagnostic 时使用
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 1. POST /api/admin/quiz/generate — 手动触发 AI 生成题目
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
@router.post("/generate")
|
|||
|
|
async def generate_quiz_questions(
|
|||
|
|
body: GenerateRequest,
|
|||
|
|
admin: Agent = Depends(require_admin),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
):
|
|||
|
|
"""手动触发 AI 生成题目。
|
|||
|
|
|
|||
|
|
生成的题目 is_active=False，需通过 /approve 端点审批后激活。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
body: 生成请求（category, question_type, count, problem_category）
|
|||
|
|
admin: 管理员（权限校验）
|
|||
|
|
db: 数据库会话
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
生成结果摘要（成功/失败数量 + 题目列表）
|
|||
|
|
"""
|
|||
|
|
service = get_quiz_generation_service()
|
|||
|
|
|
|||
|
|
if body.question_type == "knowledge":
|
|||
|
|
result = await service.generate_knowledge_questions_batch(
|
|||
|
|
db=db,
|
|||
|
|
category=body.category,
|
|||
|
|
count=body.count,
|
|||
|
|
is_active=False, # 手动生成也需审批
|
|||
|
|
)
|
|||
|
|
elif body.question_type == "diagnostic":
|
|||
|
|
if not body.problem_category:
|
|||
|
|
raise AppException(1004, "diagnostic 类型必须提供 problem_category")
|
|||
|
|
result = await service.generate_diagnostic_questions_batch(
|
|||
|
|
db=db,
|
|||
|
|
problem_category=body.problem_category,
|
|||
|
|
count=body.count,
|
|||
|
|
is_active=False,
|
|||
|
|
)
|
|||
|
|
else:
|
|||
|
|
raise AppException(1004, f"不支持的题目类型: {body.question_type}")
|
|||
|
|
|
|||
|
|
await db.commit()
|
|||
|
|
|
|||
|
|
logger.info(
|
|||
|
|
f"管理员 {admin.name} 手动生成题目: "
|
|||
|
|
f"type={body.question_type}, category={body.category}, "
|
|||
|
|
f"成功={result['success_count']}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
return success_response(data=result)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 2. GET /api/admin/quiz/pending — 查看待审核题目列表
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
@router.get("/pending")
|
|||
|
|
async def list_pending_questions(
|
|||
|
|
category: Optional[str] = Query(None, description="按类别筛选"),
|
|||
|
|
page: int = Query(1, ge=1),
|
|||
|
|
page_size: int = Query(20, ge=1, le=100),
|
|||
|
|
admin: Agent = Depends(require_admin),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
):
|
|||
|
|
"""获取待审核题目列表（is_active=False）。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
category: 可选，按类别筛选
|
|||
|
|
page: 页码（从 1 开始）
|
|||
|
|
page_size: 每页数量（1-100）
|
|||
|
|
admin: 管理员（权限校验）
|
|||
|
|
db: 数据库会话
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
分页列表 {total, page, page_size, items}
|
|||
|
|
"""
|
|||
|
|
# 构建查询条件
|
|||
|
|
conditions = [QuizQuestion.is_active == False] # noqa: E712
|
|||
|
|
if category:
|
|||
|
|
conditions.append(QuizQuestion.category == category)
|
|||
|
|
|
|||
|
|
# 总数
|
|||
|
|
total = await db.scalar(
|
|||
|
|
select(func.count(QuizQuestion.id)).where(*conditions)
|
|||
|
|
)
|
|||
|
|
total = total or 0
|
|||
|
|
|
|||
|
|
# 分页查询
|
|||
|
|
offset = (page - 1) * page_size
|
|||
|
|
stmt = (
|
|||
|
|
select(QuizQuestion)
|
|||
|
|
.where(*conditions)
|
|||
|
|
.order_by(QuizQuestion.created_at.desc())
|
|||
|
|
.offset(offset)
|
|||
|
|
.limit(page_size)
|
|||
|
|
)
|
|||
|
|
result = await db.execute(stmt)
|
|||
|
|
questions = result.scalars().all()
|
|||
|
|
|
|||
|
|
items = [_question_to_dict(q) for q in questions]
|
|||
|
|
|
|||
|
|
return success_response(data={
|
|||
|
|
"total": total,
|
|||
|
|
"page": page,
|
|||
|
|
"page_size": page_size,
|
|||
|
|
"items": items,
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 3. POST /api/admin/quiz/{question_id}/approve — 审批通过题目
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
@router.post("/{question_id}/approve")
|
|||
|
|
async def approve_question(
|
|||
|
|
question_id: str,
|
|||
|
|
admin: Agent = Depends(require_admin),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
):
|
|||
|
|
"""审批通过一道待审核题目（is_active: False → True）。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
question_id: 题目ID
|
|||
|
|
admin: 管理员（权限校验）
|
|||
|
|
db: 数据库会话
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
更新后的题目信息
|
|||
|
|
"""
|
|||
|
|
result = await db.execute(
|
|||
|
|
select(QuizQuestion).where(QuizQuestion.id == question_id)
|
|||
|
|
)
|
|||
|
|
question = result.scalar_one_or_none()
|
|||
|
|
if not question:
|
|||
|
|
raise AppException(1004, "题目不存在")
|
|||
|
|
|
|||
|
|
if question.is_active:
|
|||
|
|
raise AppException(1004, "题目已激活，无需重复审批")
|
|||
|
|
|
|||
|
|
question.is_active = True
|
|||
|
|
await db.commit()
|
|||
|
|
|
|||
|
|
logger.info(f"管理员 {admin.name} 审批通过题目: {question_id}")
|
|||
|
|
|
|||
|
|
return success_response(data=_question_to_dict(question))
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 4. DELETE /api/admin/quiz/{question_id} — 删除题目
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
@router.delete("/{question_id}")
|
|||
|
|
async def delete_question(
|
|||
|
|
question_id: str,
|
|||
|
|
admin: Agent = Depends(require_admin),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
):
|
|||
|
|
"""删除一道题目（用于清理质量差的 AI 生成题）。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
question_id: 题目ID
|
|||
|
|
admin: 管理员（权限校验）
|
|||
|
|
db: 数据库会话
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
删除确认
|
|||
|
|
"""
|
|||
|
|
result = await db.execute(
|
|||
|
|
select(QuizQuestion).where(QuizQuestion.id == question_id)
|
|||
|
|
)
|
|||
|
|
question = result.scalar_one_or_none()
|
|||
|
|
if not question:
|
|||
|
|
raise AppException(1004, "题目不存在")
|
|||
|
|
|
|||
|
|
await db.delete(question)
|
|||
|
|
await db.commit()
|
|||
|
|
|
|||
|
|
logger.info(f"管理员 {admin.name} 删除题目: {question_id}")
|
|||
|
|
|
|||
|
|
return success_response(data={"deleted_id": question_id})
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ==========================================================================
|
|||
|
|
# 辅助函数
|
|||
|
|
# ==========================================================================
|
|||
|
|
|
|||
|
|
def _question_to_dict(q: QuizQuestion) -> Dict[str, Any]:
|
|||
|
|
"""将 QuizQuestion 对象转为字典。"""
|
|||
|
|
return {
|
|||
|
|
"id": q.id,
|
|||
|
|
"type": q.type,
|
|||
|
|
"category": q.category,
|
|||
|
|
"problem_category": q.problem_category,
|
|||
|
|
"difficulty": q.difficulty,
|
|||
|
|
"question": q.question,
|
|||
|
|
"options": q.options,
|
|||
|
|
"correct_index": q.correct_index,
|
|||
|
|
"explanation": q.explanation,
|
|||
|
|
"is_active": q.is_active,
|
|||
|
|
"created_at": q.created_at.isoformat() if q.created_at else None,
|
|||
|
|
}
|