shudao-main/shudao-chat-py/routers/exam.py

from fastapi import APIRouter, Depends, Request
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field
from typing import Optional
from database import get_db
from models.chat import AIMessage
from services.qwen_service import qwen_service

router = APIRouter()


class QuestionTypeItem(BaseModel):
    questionType: str = ""
    name: str = ""
    count: int = 0
    questionCount: int = 0
    scorePerQuestion: int = 0
    romanNumeral: str = ""


class BuildPromptRequest(BaseModel):
    mode: str = ""
    client: str = ""
    projectType: str = ""
    examTitle: str = ""
    totalScore: int = 0
    questionTypes: list[QuestionTypeItem] = Field(default_factory=list)
    pptContent: str = ""


@router.post("/exam/build_prompt")
async def build_exam_prompt(
    request: Request,
    data: BuildPromptRequest,
    db: Session = Depends(get_db)
):
    """根据前端考试工坊参数生成提示词"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    question_desc = []
    total_count = 0
    for item in data.questionTypes:
        count = item.count or item.questionCount or 0
        score = item.scorePerQuestion or 0
        qtype = item.questionType or item.name or "未命名题型"
        total_count += count
        question_desc.append(f"{qtype}{count}道，每道{score}分")

    question_text = "；".join(question_desc) if question_desc else "题型未提供"
    question_schema_lines = []
    for item in data.questionTypes:
        count = item.count or item.questionCount or 0
        score = item.scorePerQuestion or 0
        qtype = item.questionType or item.name or "未命名题型"
        if count <= 0:
            continue
        question_schema_lines.append(f"- {qtype}: {count}道，每道{score}分")

    question_schema = "\n".join(
        question_schema_lines) if question_schema_lines else "- 未提供有效题型"

    prompt = (
        "请根据以下要求直接生成一份完整试卷，并严格返回纯 JSON，不要输出 markdown 代码块、解释说明或额外文字。\n"
        f"生成模式：{data.mode or '未指定'}\n"
        f"客户端：{data.client or '未指定'}\n"
        f"项目类型：{data.projectType or '未指定'}\n"
        f"考试标题：{data.examTitle or '未命名考试'}\n"
        f"总分：{data.totalScore or 0}\n"
        f"总题量：{total_count}\n"
        f"题型要求：{question_text}\n"
        f"出题依据内容：{data.pptContent or '无'}\n"
        "出题依据内容是本次试卷的核心来源，所有题目必须围绕该内容中的知识点、术语、流程、规范要求和场景展开。\n"
        "如果出题依据内容中出现了章节、条款、培训主题或专业术语，题目必须优先考查这些内容，不能偏离到无关知识。\n"
        "单选题、多选题、判断题和简答题的题干、选项、答案解析都要与出题依据内容直接相关，不能泛泛而谈。\n"
        "请结合出题依据内容、工程类型和题型要求，生成有具体内容、具体选项、具体答案、具体解析的试卷。\n"
        "禁止输出“选项A”“题目1”这类占位内容，所有题目必须是可直接展示和作答的真实内容。\n"
        "JSON 输出结构必须符合以下格式：\n"
        "{\n"
        '  "title": "试卷标题",\n'
        '  "totalScore": 100,\n'
        '  "totalQuestions": 10,\n'
        '  "singleChoice": {"scorePerQuestion": 2, "totalScore": 20, "count": 10, "questions": [{"text": "题目内容", "options": [{"key": "A", "text": "具体选项内容"}, {"key": "B", "text": "具体选项内容"}, {"key": "C", "text": "具体选项内容"}, {"key": "D", "text": "具体选项内容"}], "answer": "A", "analysis": "解析内容"}]},\n'
        '  "judge": {"scorePerQuestion": 2, "totalScore": 0, "count": 0, "questions": [{"text": "题目内容", "answer": "正确", "analysis": "解析内容"}]},\n'
        '  "multiple": {"scorePerQuestion": 3, "totalScore": 0, "count": 0, "questions": [{"text": "题目内容", "options": [{"key": "A", "text": "具体选项内容"}, {"key": "B", "text": "具体选项内容"}, {"key": "C", "text": "具体选项内容"}, {"key": "D", "text": "具体选项内容"}], "answers": ["A", "C"], "analysis": "解析内容"}]},\n'
        '  "short": {"scorePerQuestion": 10, "totalScore": 0, "count": 0, "questions": [{"text": "题目内容", "outline": {"keyFactors": "答题要点", "measures": "参考措施"}}]}\n'
        "}\n"
        "请按下面的题型配置生成对应数量的题目，没有的题型 count 返回 0、questions 返回空数组：\n"
        f"{question_schema}"
    )

    return {
        "statusCode": 200,
        "msg": "success",
        "data": {"prompt": prompt}
    }


class BuildSinglePromptRequest(BaseModel):
    question_type: str
    topic: str
    difficulty: str


@router.post("/exam/build_single_prompt")
async def build_single_question_prompt(
    request: Request,
    data: BuildSinglePromptRequest,
    db: Session = Depends(get_db)
):
    """生成单题提示词 - 对齐Go版本函数名"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    prompt = f"""请生成1道关于{data.topic}的{data.question_type}，难度为{data.difficulty}。"""
    return {
        "statusCode": 200,
        "msg": "success",
        "data": {"prompt": prompt}
    }


class ModifyQuestionRequest(BaseModel):
    ai_conversation_id: int
    content: str


@router.post("/re_modify_question")
async def re_modify_question(
    request: Request,
    data: ModifyQuestionRequest,
    db: Session = Depends(get_db)
):
    """修改考试题目 - 实际修改ai_message表"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    # 修改ai_message表中type='ai'的content
    result = db.query(AIMessage).filter(
        AIMessage.ai_conversation_id == data.ai_conversation_id,
        AIMessage.type == 'ai'
    ).update({"content": data.content})

    if result == 0:
        return {"statusCode": 404, "msg": "消息不存在"}

    db.commit()
    return {"statusCode": 200, "msg": "success"}


class ReproduceSingleQuestionRequest(BaseModel):
    message: str = ""
    ai_conversation_id: Optional[int] = None
    regenerate_reason: str = ""


@router.post("/re_produce_single_question")
async def re_produce_single_question(
    request: Request,
    data: ReproduceSingleQuestionRequest,
    db: Session = Depends(get_db)
):
    """重新生成单题"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    prompt = (data.message or "").strip()

    # 兼容旧版调用：未传 message 时，尝试根据会话和重生成原因构造提示词。
    if not prompt and data.ai_conversation_id:
        message = db.query(AIMessage).filter(
            AIMessage.ai_conversation_id == data.ai_conversation_id,
            AIMessage.type == 'ai'
        ).first()

        if not message:
            return {"statusCode": 404, "msg": "消息不存在"}

        prompt = (message.content or "").strip()
        if data.regenerate_reason:
            prompt = f"{prompt}\n\n请根据以下要求重新生成：{data.regenerate_reason}"

    if not prompt:
        return {"statusCode": 400, "msg": "缺少生成内容"}

    try:
        new_question = await qwen_service.chat([
            {"role": "user", "content": prompt}
        ])
    except Exception as e:
        return {"statusCode": 500, "msg": f"AI生成失败: {str(e)}"}

    return {
        "statusCode": 200,
        "msg": "success",
        "data": {
            "ai_conversation_id": data.ai_conversation_id,
            "new_question": new_question,
            "reply": new_question,
            "content": new_question,
            "message": new_question
        }
    }