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
from utils.logger import logger

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 "- 未提供有效题型"

    ppt_content = (data.pptContent or "").strip()
    if ppt_content:
        max_chars = 12000
        if len(ppt_content) > max_chars:
            head_len = max_chars // 2
            tail_len = max_chars - head_len
            ppt_content = (
                ppt_content[:head_len]
                + "\n\n（出题依据内容过长，已截断，以下为结尾片段）\n\n"
                + ppt_content[-tail_len:]
            )
            logger.info(
                f"[exam/build_prompt] pptContent truncated: original_len={len(data.pptContent)} kept_len={len(ppt_content)}"
            )

    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"出题依据内容：{ppt_content or '无'}\n"
        "出题依据内容是本次试卷的核心来源，所有题目必须围绕该内容中的知识点、术语、流程、规范要求和场景展开。\n"
        "如果出题依据内容中出现了章节、条款、培训主题或专业术语，题目必须优先考查这些内容，不能偏离到无关知识。\n"
        "单选题、多选题、判断题和简答题的题干、选项、答案解析都要与出题依据内容直接相关，不能泛泛而谈。\n"
        "请结合出题依据内容、工程类型和题型要求，生成有具体内容、具体选项、具体答案、具体解析的试卷。\n"
        "凡是题型配置中 count 大于 0 的题型，必须返回对应数量的非空题目，不能返回空数组，不能少题。\n"
        "即使出题依据内容较短，也要优先围绕已有内容中的关键词、术语、场景和要求组织出题，不能因为信息少而返回空题目。\n"
        "如果某题型要求生成 3 道题，就必须生成 3 道完整可作答的题目，少于要求数量视为不合格。\n"
        "禁止输出“选项A”“题目1”“桥梁工程相关单选题1”“题目内容”“解析内容”这类占位内容，所有题目必须是可直接展示和作答的真实内容。\n"
        "下面的 JSON 结构示例只用于说明字段格式，示例中的字符串不能原样照抄到最终结果中，最终返回的每个字符串都必须替换成结合出题依据生成的具体内容。\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": "<选项A具体内容>"}, {"key": "B", "text": "<选项B具体内容>"}, {"key": "C", "text": "<选项C具体内容>"}, {"key": "D", "text": "<选项D具体内容>"}], "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": "<选项A具体内容>"}, {"key": "B", "text": "<选项B具体内容>"}, {"key": "C", "text": "<选项C具体内容>"}, {"key": "D", "text": "<选项D具体内容>"}], "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
        }
    }