From b6631c626b4653ade4d17e7c162cec9a6a46f8b7 Mon Sep 17 00:00:00 2001 From: lacerate551 <128470311+lacerate551@users.noreply.github.com> Date: Fri, 5 Jun 2026 12:06:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(exam):=20=E4=BF=AE=E5=A4=8D=E5=87=BA?= =?UTF-8?q?=E9=A2=98=E6=89=B9=E5=8D=B7=20API=20=E5=A5=91=E7=BA=A6=E4=B8=80?= =?UTF-8?q?=E8=87=B4=E6=80=A7=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0: - 修复 V2 降级路径响应格式，fallback 输出经过 enrich + dedup + balance - 补充主观题 format example，要求 LLM 生成 scoring_points 评分标准 P1: - 补充多选题、判断题 format example，明确 answer 为列表格式 - 统一批卷响应格式，所有题型增加 grading_status 字段 - 出题数量不足时返回 requested_types/actual_types/warnings - 修复 _call_llm 未配置时抛异常而非静默返回 0 分 P2: - API 层增加 question_types 和 difficulty 入参校验 - 修复 generate_questions_from_content 方法名和参数类型 bug --- exam_pkg/api.py | 70 +- exam_pkg/generator.py | 106 ++- exam_pkg/grader.py | 69 +- exam_pkg/manager.py | 1505 +++++++++++++++++++++-------------------- 4 files changed, 957 insertions(+), 793 deletions(-) diff --git a/exam_pkg/api.py b/exam_pkg/api.py index c8827ef..48313e8 100644 --- a/exam_pkg/api.py +++ b/exam_pkg/api.py @@ -32,6 +32,35 @@ from auth.gateway import ( from core.status_codes import EXAM_SUCCESS, GRADE_SUCCESS, EXAM_ERROR, GRADE_ERROR, BAD_REQUEST, NO_CONTENT, LLM_ERROR from api.response_utils import success_response, error_response +# 合法题型 +VALID_QUESTION_TYPES = {'single_choice', 'multiple_choice', 'true_false', 'fill_blank', 'subjective'} + + +def validate_question_types(question_types: dict) -> str: + """ + 校验 question_types 参数 + + Returns: + None 表示校验通过，字符串表示错误信息 + """ + if not isinstance(question_types, dict): + return "question_types 必须为对象" + + for q_type, count in question_types.items(): + if q_type not in VALID_QUESTION_TYPES: + return f"不支持的题型: {q_type}，合法值: {', '.join(sorted(VALID_QUESTION_TYPES))}" + if not isinstance(count, int) or count < 0: + return f"题型 {q_type} 的数量必须为非负整数，当前值: {count}" + + return None + + +def validate_difficulty(difficulty) -> str: + """校验 difficulty 参数""" + if not isinstance(difficulty, int) or difficulty < 1 or difficulty > 5: + return "difficulty 必须为 1-5 的整数" + return None + # 创建蓝图 exam_bp = Blueprint('exam', __name__) @@ -69,12 +98,16 @@ def api_generate_questions(): "request_id": "uuid", "questions": [ { - "metadata": {"question_id": "...", "question_type": "...", ...}, - "source_trace": {"document_name": "...", "sources": [...], ...}, - "content": {"stem": "...", "data": {...}, "answer": "...", ...} + "question_type": "single_choice", + "difficulty": 3, + "content": {"stem": "...", "data": {...}, "answer": "...", "explanation": "..."}, + "source_trace": {"document_name": "...", "chunk_ids": [...], "sources": [...]} } ], "total": 10, + "requested_types": {"single_choice": 3, "fill_blank": 2}, + "actual_types": {"single_choice": 3, "fill_blank": 2}, + "warnings": ["fill_blank: 请求 3 道，实际生成 2 道"], "source_chunks_used": 15 } """ @@ -91,6 +124,17 @@ def api_generate_questions(): if not question_types: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 question_types 参数", http_status=400) + # 校验题型和数量 + type_error = validate_question_types(question_types) + if type_error: + return error_response("INVALID_PARAMS", BAD_REQUEST, type_error, http_status=400) + + # 校验难度 + difficulty = data.get('difficulty', 3) + diff_error = validate_difficulty(difficulty) + if diff_error: + return error_response("INVALID_PARAMS", BAD_REQUEST, diff_error, http_status=400) + # 获取当前用户 user = get_current_user() if not user: @@ -149,6 +193,12 @@ def api_generate_smart(): if not file_path or not collection: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 file_path 或 collection 参数", http_status=400) + # 校验难度 + difficulty = data.get('difficulty', 3) + diff_error = validate_difficulty(difficulty) + if diff_error: + return error_response("INVALID_PARAMS", BAD_REQUEST, diff_error, http_status=400) + # 获取当前用户 user = get_current_user() if not user: @@ -237,11 +287,23 @@ def api_grade_answers(): { "success": true, "request_id": "uuid", - "results": [...], + "results": [ + { + "question_id": "uuid", + "score": 2, + "max_score": 2, + "grading_status": "success", + "details": {"correct": true, "student_answer": "A", "correct_answer": "B", "feedback": "..."} + } + ], "total_score": 10.5, "total_max_score": 16.0, "score_rate": 65.6 } + + grading_status 取值： + - "success": 评分成功 + - "failed": 评分失败（LLM 超时/解析失败等），score 为 0 """ try: data = request.json diff --git a/exam_pkg/generator.py b/exam_pkg/generator.py index 96950ef..f8e419d 100644 --- a/exam_pkg/generator.py +++ b/exam_pkg/generator.py @@ -322,13 +322,17 @@ class QuestionGenerator: ) -> List[Dict]: """ 降级出题方法：当知识点提取失败时，直接使用 chunks 出题 + + 输出格式与正常路径一致（嵌套结构），确保 API 响应格式统一 """ source_content = build_source_context(chunks) total_questions = sum(question_types.values()) if not source_content or total_questions == 0: return [] - # 构造出题 prompt + q_type_str = ", ".join(f"{t}:{n}道" for t, n in question_types.items()) + + # 构造出题 prompt —— 使用与正常路径相同的嵌套格式 prompt = f"""请根据以下参考资料生成题目。 ## 参考资料 @@ -337,41 +341,41 @@ class QuestionGenerator: ## 要求 -- 题型及数量：{json.dumps(question_types, ensure_ascii=False)} +- 题型及数量：{q_type_str} - 难度等级：{difficulty}/5 -- 每道题必须包含：题干、答案、解析 - 答案必须基于参考资料 -## 输出格式 +## 输出格式（JSON 数组） +{self._get_format_examples()} -返回 JSON 数组，每个元素格式： -{{"stem": "题干", "type": "single_choice/multiple_choice/true_false/fill_blank/subjective", "answer": "答案", "explanation": "解析", "options": [{{"key": "A", "content": "选项内容"}}]}} - -只输出 JSON 数组，不要其他内容。""" +请直接输出 JSON 数组：""" try: - # 使用 llm_utils 统一调用 - messages = [{"role": "user", "content": prompt}] + messages = [ + {"role": "system", "content": "你是一个专业的出题专家，必须严格按照JSON格式输出。"}, + {"role": "user", "content": prompt} + ] content = call_llm( client=self.client, prompt=prompt, model=self.model, temperature=0.7, - max_tokens=2000, + max_tokens=4000, messages=messages ) if not content: return [] # 解析 JSON - questions = parse_json_list_from_response(content) - if questions: - # 添加元数据 - for q in questions: - q["source_trace"] = { - "document_name": document_name, - "chunks_count": len(chunks) - } - return questions + raw_questions = parse_json_list_from_response(content) + if not raw_questions: + return [] + + # Schema 校验 + validated = validate_questions_schema(raw_questions) + + # 补充溯源信息（与正常路径格式统一） + enriched = self._enrich_with_source_trace(validated, chunks, document_name) + return enriched except Exception as e: logger.error(f"[降级出题失败] {e}") @@ -510,7 +514,7 @@ class QuestionGenerator: return result def _get_format_examples(self) -> str: - """返回各题型格式示例""" + """返回各题型格式示例（覆盖全部 5 种题型）""" return ''' ### 单选题示例 { @@ -524,6 +528,30 @@ class QuestionGenerator: "referenced_chunk_ids": ["chunk_001"] } +### 多选题示例 +{ + "type": "multiple_choice", + "content": { + "stem": "题干内容", + "data": {"options": [{"key": "A", "content": "选项A"}, {"key": "B", "content": "选项B"}, {"key": "C", "content": "选项C"}, {"key": "D", "content": "选项D"}]}, + "answer": ["A", "C"], + "explanation": "解析..." + }, + "referenced_chunk_ids": ["chunk_001"] +} + +### 判断题示例 +{ + "type": "true_false", + "content": { + "stem": "判断：某个陈述内容", + "data": {}, + "answer": "对", + "explanation": "解析..." + }, + "referenced_chunk_ids": ["chunk_002"] +} + ### 填空题示例 { "type": "fill_blank", @@ -535,6 +563,24 @@ class QuestionGenerator: }, "referenced_chunk_ids": ["chunk_003"] } + +### 主观题示例 +{ + "type": "subjective", + "content": { + "stem": "请简述某个概念的含义及其应用场景。", + "data": { + "scoring_points": [ + {"point": "要点1：概念定义", "weight": 0.4}, + {"point": "要点2：应用场景", "weight": 0.3}, + {"point": "要点3：举例说明", "weight": 0.3} + ] + }, + "answer": "参考答案范文...", + "explanation": "解析..." + }, + "referenced_chunk_ids": ["chunk_004"] +} ''' def _extract_knowledge_points( @@ -862,9 +908,14 @@ def generate_questions_from_content( question_types: Dict[str, int], difficulty: int = 3 ) -> List[Dict]: - """便捷函数：从内容生成题目""" + """便捷函数：从文本内容生成题目（将纯文本包装为单 chunk 后调用结构化出题）""" + if not source_content or not source_content.strip(): + return [] + + # 将纯文本包装为一个 chunk + chunks = [{"content": source_content, "section": "", "page": None, "chunk_id": "content_0"}] generator = QuestionGenerator() - return generator.generate_questions(source_content, document_name, question_types, difficulty) + return generator.generate_questions_structured(chunks, document_name, question_types, difficulty) def analyze_document_for_exam(chunks: List[Dict]) -> Dict[str, Any]: @@ -1114,9 +1165,14 @@ def generate_questions_structured_v2( logger.info(f" 全局唯一知识点: {len(all_knowledge_points)}") if not all_knowledge_points: - # 降级：直接使用 chunks 出题 + # 降级：直接使用 chunks 出题（格式已由 fallback 内部统一） logger.warning("[v2] 知识点提取失败，降级使用 chunks 出题") - return generator._generate_questions_fallback(chunks, document_name, question_types, difficulty) + fallback_questions = generator._generate_questions_fallback(chunks, document_name, question_types, difficulty) + # 降级路径也执行去重和数量平衡 + deduped = _deduplicate_questions(fallback_questions) + final = _balance_question_types(deduped, question_types) + logger.info(f" [降级] 最终题目数: {len(final)}") + return final # AI 分配：哪些知识点出什么题型 assignments = _ai_assign_question_types(all_knowledge_points, question_types) diff --git a/exam_pkg/grader.py b/exam_pkg/grader.py index 1334064..1052108 100644 --- a/exam_pkg/grader.py +++ b/exam_pkg/grader.py @@ -97,8 +97,13 @@ def grade_objective(answer: Dict) -> Dict: "question_id": answer.get('question_id'), "score": max_score if correct else 0, "max_score": max_score, - "correct": correct, - "feedback": f"正确答案: {correct_answer}" if not correct else "正确！" + "grading_status": "success", + "details": { + "correct": correct, + "student_answer": student_answer, + "correct_answer": correct_answer, + "feedback": f"正确答案: {correct_answer}" if not correct else "正确！" + } } @@ -119,7 +124,8 @@ def grade_fill_blank(answer: Dict) -> Dict: "question_id": answer.get('question_id'), "score": 0, "max_score": max_score, - "details": {"error": "答案格式错误"} + "grading_status": "failed", + "details": {"error": "答案格式错误，correct_answers 或 student_answer 为空"} } # 计算每空分数 @@ -149,9 +155,11 @@ def grade_fill_blank(answer: Dict) -> Dict: "question_id": answer.get('question_id'), "score": round(total_score, 1), "max_score": max_score, + "grading_status": "success", "details": { "blank_scores": blank_scores, - "correct_answers": correct_answers + "total_blanks": len(correct_answers), + "correct_blanks": sum(1 for s in blank_scores if s > 0) } } @@ -243,22 +251,38 @@ class AnswerGrader: } # 收集结果 - for future in as_completed(future_to_qid, timeout=60): - qid = future_to_qid[future] - try: - result = future.result(timeout=15) - results_map[qid] = result - except Exception as e: - # 失败时返回默认结果 - results_map[qid] = { - "question_id": qid, - "score": 0, - "max_score": next( - (a.get('max_score', 10) for a in questions if a.get('question_id') == qid), - 10 - ), - "error": str(e) - } + try: + for future in as_completed(future_to_qid, timeout=60): + qid = future_to_qid[future] + try: + result = future.result(timeout=15) + results_map[qid] = result + except Exception as e: + # 失败时返回默认结果 + results_map[qid] = { + "question_id": qid, + "score": 0, + "max_score": next( + (a.get('max_score', 10) for a in questions if a.get('question_id') == qid), + 10 + ), + "grading_status": "failed", + "details": {"error": str(e)} + } + except TimeoutError: + # 超时：为未完成的任务设置失败状态 + for future, qid in future_to_qid.items(): + if qid not in results_map: + results_map[qid] = { + "question_id": qid, + "score": 0, + "max_score": next( + (a.get('max_score', 10) for a in questions if a.get('question_id') == qid), + 10 + ), + "grading_status": "failed", + "details": {"error": "批阅超时"} + } @retry(times=2, delay=1) def _grade_subjective(self, answer: Dict) -> Dict: @@ -315,8 +339,7 @@ class AnswerGrader: def _call_llm(self, prompt: str) -> str: """调用本地 LLM""" if not self.client: - # 无 LLM 客户端，返回默认评分 - return json.dumps({"score": 0, "overall_feedback": "LLM 未配置"}) + raise ValueError("LLM 未配置，无法进行主观题评分") messages = [ {"role": "system", "content": "你是一个专业的阅卷老师，请严格按照JSON格式输出评分结果。"}, @@ -347,6 +370,7 @@ class AnswerGrader: "question_id": answer.get('question_id'), "score": score, "max_score": max_score, + "grading_status": "success", "details": { "scoring_breakdown": result.get('scoring_breakdown', []), "highlights": result.get('highlights', []), @@ -360,6 +384,7 @@ class AnswerGrader: "question_id": answer.get('question_id'), "score": 0, "max_score": max_score, + "grading_status": "failed", "details": {"error": "评分结果解析失败"} } diff --git a/exam_pkg/manager.py b/exam_pkg/manager.py index 3c7752e..2bc33a1 100644 --- a/exam_pkg/manager.py +++ b/exam_pkg/manager.py @@ -1,742 +1,763 @@ -""" -出题与批题系统管理器 - -核心功能： -1. 题目生成（调用 generator.py） -2. 答案批阅（调用 grader.py） -3. 题库管理（保存/加载/搜索） - -职责边界： -- RAG 服务负责：生成题目 + 批阅答案 -- 后端服务负责：审核入库 + 状态管理 - -使用方式： - from exam_pkg.manager import generate_questions_from_file, grade_answers -""" -import json -import os -import re - -import uuid -import hashlib -from datetime import datetime -from typing import Optional, List, Dict, Any -from concurrent.futures import ThreadPoolExecutor, as_completed -import logging - -logger = logging.getLogger(__name__) - -# 状态常量 -EXAM_STATUS_DRAFT = "draft" -EXAM_STATUS_APPROVED = "approved" - -# 导入新的生成器和批题器 -from exam_pkg.generator import ( - QuestionGenerator, - build_semantic_query, - build_source_context, - safe_parse_questions, - validate_questions_schema -) -from exam_pkg.grader import ( - AnswerGrader, - grade_answers as grade_answers_v2, - grade_objective, - grade_fill_blank -) - -# MinerU 解析器（可选） -try: - from parsers import parse_document, MINERU_AVAILABLE - if MINERU_AVAILABLE: - from parsers.mineru_parser import MinerUChunk - PARSE_AVAILABLE = MINERU_AVAILABLE -except ImportError: - PARSE_AVAILABLE = False - -# 导入配置 - -# 题库目录 -QUESTION_BANK_DIR = "./题库" -DRAFT_DIR = "./题库/草稿" - - -# ==================== 新版出题接口 ==================== - -def generate_questions_from_file( - file_path: str, - collection: str, - question_types: Dict[str, int], - difficulty: int = 3, - options: Dict = None, - request_id: str = None -) -> Dict: - """ - 从文件生成题目（结构化出题 v2） - - 🔥 改进版 Structure-aware RAG 出题架构： - - Phase 1: 文档结构分析 - - Phase 2: 知识点规划（全局唯一） - - Phase 3: 精准出题（每个知识点单独检索） - - Phase 4: 质量校验 - - Args: - file_path: 文件路径 - collection: 向量库名称 - question_types: 题型及数量 {"single_choice": 3, "fill_blank": 2, ...} - difficulty: 难度 1-5 - options: 可选配置 - - max_source_chunks: 最大切片数（默认 30） - request_id: 请求 ID（幂等性支持） - - Returns: - { - "success": True, - "request_id": "...", - "questions": [...], - "total": 10, - "source_chunks_used": 15 - } - """ - options = options or {} - - # 1. 检索文件的所有切片 - chunks = retrieve_file_chunks( - file_path=file_path, - collection=collection, - question_types=question_types, - top_k=options.get('max_source_chunks', 50) - ) - logger.debug(f"generate_questions_from_file: chunks 数量 = {len(chunks)}") - - # 2. 结构化出题 v2：全局唯一知识点 + 精准出题 - from exam_pkg.generator import generate_questions_structured_v2 - questions = generate_questions_structured_v2( - chunks=chunks, - document_name=file_path, - question_types=question_types, - difficulty=difficulty - ) - - return { - "success": True, - "request_id": request_id, - "questions": questions, - "total": len(questions), - "source_chunks_used": len(chunks) - } - - -def analyze_file_for_exam( - file_path: str, - collection: str, - top_k: int = 50 -) -> Dict[str, Any]: - """ - 分析文件内容，返回 AI 推荐的题型和数量 - - Args: - file_path: 文件路径 - collection: 向量库名称 - top_k: 最大切片数（默认 50，用于分析） - - Returns: - { - "total_knowledge_points": int, - "suitable_types": List[str], - "question_types": Dict[str, int], - "reason": str - } - """ - from exam_pkg.generator import analyze_document_for_exam - - # 1. 检索文件切片（用于分析） - # 使用通用 query 检索，获取尽可能多的内容 - query = "重点内容 关键概念 定义 流程 步骤" - chunks = retrieve_file_chunks_for_analysis( - file_path=file_path, - collection=collection, - query=query, - top_k=top_k - ) - - if not chunks: - return { - "total_knowledge_points": 0, - "suitable_types": [], - "question_types": {}, - "reason": "未找到文件内容，请检查文件路径和向量库" - } - - # 2. 调用 AI 分析 - return analyze_document_for_exam(chunks) - - -def retrieve_file_chunks_for_analysis( - file_path: str, - collection, - query: str, - top_k: int = 50 -) -> List[Dict]: - """ - 检索文件切片（用于分析，不需要 question_types） - - Args: - file_path: 文件路径 - collection: 向量库名称 - query: 检索 query - top_k: 最大切片数 - - Returns: - 切片列表 - """ - try: - # 获取 RAG 引擎 - from core.engine import get_engine - engine = get_engine() - if not engine: - logger.error("RAG 引擎未初始化") - return [] - - # 准备 collection 列表 - collections = [collection] if isinstance(collection, str) else collection - - # 提取文件名（向量库存储的是文件名，不含路径前缀） - filename = os.path.basename(file_path) - - # 向量检索（尝试文件名和完整路径） - results = None - for source_filter in [filename, file_path]: - results = engine.search_knowledge( - query=query, - collections=collections, - source_filter=source_filter, - top_k=top_k - ) - if results.get('documents') and results['documents'][0]: - break - - if not results or not results.get('documents') or not results['documents'][0]: - logger.warning(f"未检索到相关切片: file={file_path}, collection={collection}") - return [] - - # 整理结果（search_knowledge 返回格式: {ids: [[...]], documents: [[...]], metadatas: [[...]], distances: [[...]]}) - chunks = [] - for i, (doc, meta, score) in enumerate(zip( - results['documents'][0], - results['metadatas'][0], - results['distances'][0] - )): - chunks.append({ - "chunk_id": results['ids'][0][i], - "content": doc, - "source": meta.get('source', ''), - "page": meta.get('page'), - "section": meta.get('section', ''), - "score": score - }) - - return chunks - - except Exception as e: - logger.error(f"检索文件切片失败: {e}") - return [] - - -def retrieve_file_chunks( - file_path: str, - collection, # 支持字符串或列表 - question_types: Dict[str, int], - top_k: int = None -) -> List[Dict]: - """ - 检索文件相关切片 - - Args: - file_path: 文件路径 - collection: 向量库名称（支持字符串或列表，列表时按优先级顺序检索） - question_types: 题型及数量 - top_k: 最大切片数 - - Returns: - 切片列表 - """ - # 动态计算 top_k：题型数量 × 3，上限 50 - if top_k is None: - total_questions = sum(question_types.values()) - top_k = min(50, total_questions * 3) - - # 根据题型构建语义 query - query = build_semantic_query(question_types) - - # 提取文件名（向量库存储的是文件名，不含路径前缀） - filename = os.path.basename(file_path) - - # 统一处理为列表 - if isinstance(collection, str): - collections = [collection] - else: - collections = list(collection) if collection else [] - - if not collections: - logger.error("[ERROR] 未指定向量库") - return [] - - try: - from core.engine import get_engine - engine = get_engine() - - # 按优先级遍历 collections，找到文件即停止 - for coll in collections: - # 尝试两种格式：文件名和完整路径 - for source_filter in [filename, file_path]: - results = engine.search_knowledge( - query=query, - collections=[coll], - source_filter=source_filter, - top_k=top_k - ) - - # 检查是否有结果 - if results.get('documents') and results['documents'][0]: - logger.debug(f"在向量库 [{coll}] 中找到文件 [{filename}]") - break - else: - continue - break # 外层循环跳出 - - chunks = [] - if results.get('documents') and results['documents'][0]: - for i, (doc, meta, score) in enumerate(zip( - results['documents'][0], - results['metadatas'][0], - results['distances'][0] - )): - chunks.append({ - "chunk_id": results['ids'][0][i], - "content": doc, - "source": meta.get('source'), - "page": meta.get('page'), - "section": meta.get('section', ''), - "score": score - }) - - logger.debug(f"retrieve_file_chunks: 检索到 {len(chunks)} 个切片") - return chunks - - except Exception as e: - logger.error(f"检索切片失败: {e}") - return [] - - -# ==================== 新版批题接口 ==================== - -def grade_answers(answers: List[Dict], request_id: str = None) -> Dict: - """ - 批阅答案入口函数 - - 🔥 改进： - - 本地批阅选择题/判断题 - - 填空题模糊匹配 - - 主观题 LLM 评分 - - 并发 + 限流 + 顺序保持 - """ - return grade_answers_v2(answers, request_id) - - -def get_source_chapters_for_question( - file_path: str, - question_topic: str, - top_k: int = 3 -) -> List[Dict[str, Any]]: - """ - 获取与问题主题相关的章节内容（用于出题或批阅参考） - - Args: - file_path: 文档文件路径（支持 PDF/DOCX/PPTX） - question_topic: 问题主题或关键词 - top_k: 返回的章节数量 - - Returns: - [ - { - "title": "章节标题", - "content": "完整内容", - "section_path": "章节路径", - "page_range": "1-2" - } - ] - """ - if not PARSE_AVAILABLE: - raise ImportError("MinerU 解析器不可用") - - result = parse_document( - file_path, - output_base=".data/mineru_temp", - images_output=".data/images", - cleanup_after_image_move=True - ) - chunks = result.get('chunks', []) - - # 构建章节内容列表 - chapter_contents = [] - for chunk in chunks: - content = chunk.content if hasattr(chunk, 'content') else str(chunk) - if not content.strip(): - continue - - chapter_contents.append({ - "title": chunk.title if hasattr(chunk, 'title') else '', - "content": content.strip(), - "section_path": chunk.section_path if hasattr(chunk, 'section_path') else '', - "page_range": f"{chunk.page_start}-{chunk.page_end}" if hasattr(chunk, 'page_start') else '', - "source_file": chunk.source_file if hasattr(chunk, 'source_file') else os.path.basename(file_path) - }) - - # 使用关键词检索相关章节 - keywords = question_topic.split() if question_topic else [] - - # 简单的关键词匹配筛选 - relevant = [] - for c in chapter_contents: - content_lower = c['content'].lower() - if any(kw.lower() in content_lower for kw in keywords): - relevant.append(c) - - return relevant[:top_k] - - -def sanitize_filename(name: str) -> str: - # 移除或替换非法字符 - illegal_chars = r'[<>:"/\\|?*\x00-\x1f]' - safe_name = re.sub(illegal_chars, '_', name) - # 移除首尾空格和点 - safe_name = safe_name.strip('. ') - # 限制长度 - if len(safe_name) > 100: - safe_name = safe_name[:100] - # 如果清理后为空，使用时间戳 - if not safe_name: - safe_name = datetime.now().strftime('%Y%m%d_%H%M%S') - return safe_name - - -def save_exam(exam: dict, name: str = None) -> str: - """ - 保存试卷到对应目录 - - - draft 状态：保存到 题库/草稿/ 目录 - - approved 状态：保存到 题库/ 目录 - - Args: - exam: 试卷JSON - name: 文件名(不含扩展名)，默认使用试卷名称 - - Returns: - 保存的文件路径 - """ - # 根据状态确定保存目录 - status = exam.get("status", EXAM_STATUS_DRAFT) - if status == EXAM_STATUS_APPROVED: - save_dir = QUESTION_BANK_DIR - else: - save_dir = DRAFT_DIR - - # 确保目录存在 - os.makedirs(save_dir, exist_ok=True) - # 确保草稿目录存在 - os.makedirs(DRAFT_DIR, exist_ok=True) - - # 确定文件名：优先使用试卷名称 - if name is None: - # 使用试卷名称作为文件名 - exam_name = exam.get("name") or exam.get("topic", "未命名试卷") - name = sanitize_filename(exam_name) - - # 检查是否已存在同名文件，如果存在则添加时间戳 - filepath = os.path.join(save_dir, f"{name}.json") - if os.path.exists(filepath): - timestamp = datetime.now().strftime('_%Y%m%d_%H%M%S') - name = f"{name}{timestamp}" - - filepath = os.path.join(save_dir, f"{name}.json") - - with open(filepath, 'w', encoding='utf-8') as f: - json.dump(exam, f, ensure_ascii=False, indent=2) - - logger.info(f"试卷已保存: {filepath} (状态: {status})") - return filepath - - -def load_exam(filepath: str) -> dict: - """加载试卷""" - with open(filepath, 'r', encoding='utf-8') as f: - return json.load(f) - - -# ==================== 题库管理函数 ==================== - -def list_exams(status: str = None, page: int = 1, limit: int = 20) -> dict: - """ - 获取试卷列表 - - Args: - status: 状态过滤（draft/pending_review/approved/rejected） - page: 页码 - limit: 每页数量 - - Returns: - { - "exams": [...], - "total": int, - "page": int - } - """ - # 确保目录存在 - os.makedirs(QUESTION_BANK_DIR, exist_ok=True) - os.makedirs(DRAFT_DIR, exist_ok=True) - - exams = [] - - # 从题库目录和草稿目录加载试卷 - for directory in [QUESTION_BANK_DIR, DRAFT_DIR]: - if not os.path.exists(directory): - continue - for filename in os.listdir(directory): - if not filename.endswith('.json'): - continue - - filepath = os.path.join(directory, filename) - try: - exam = load_exam(filepath) - # 状态过滤 - if status and exam.get("status") != status: - continue - # 只返回摘要信息 - exams.append({ - "exam_id": exam.get("exam_id", filename[:-5]), - "name": exam.get("name") or exam.get("topic", "未命名试卷"), - "topic": exam.get("topic", ""), - "status": exam.get("status", "approved"), - "total_count": exam.get("total_count", 0), - "total_score": exam.get("total_score", 0), - "created_at": exam.get("created_at", ""), - "created_by": exam.get("created_by", ""), - "filename": filename, - "filepath": filepath - }) - except Exception as e: - logger.error(f"加载试卷失败 {filename}: {e}") - continue - - # 按创建时间倒序 - exams.sort(key=lambda x: x.get("created_at", ""), reverse=True) - - # 分页 - total = len(exams) - start = (page - 1) * limit - end = start + limit - - return { - "exams": exams[start:end], - "total": total, - "page": page - } - - -def _find_exam_filepath(exam_id: str) -> Optional[str]: - """ - 查找试卷文件路径（在题库和草稿目录中查找） - - Args: - exam_id: 试卷ID - - Returns: - 文件路径，如果不存在返回None - """ - # 确保目录存在 - os.makedirs(QUESTION_BANK_DIR, exist_ok=True) - os.makedirs(DRAFT_DIR, exist_ok=True) - - # 先检查题库目录 - for directory in [DRAFT_DIR, QUESTION_BANK_DIR]: - for filename in os.listdir(directory): - if not filename.endswith('.json'): - continue - filepath = os.path.join(directory, filename) - try: - exam = load_exam(filepath) - if exam.get("exam_id") == exam_id: - return filepath - except (json.JSONDecodeError, OSError, KeyError): - continue - return None - - -def get_exam_by_id(exam_id: str) -> Optional[dict]: - """ - 根据 ID 获取试卷 - - Args: - exam_id: 试卷ID - - Returns: - 试卷JSON，如果不存在返回None - """ - filepath = _find_exam_filepath(exam_id) - if filepath: - return load_exam(filepath) - return None - - -def update_exam(exam_id: str, exam_data: dict) -> Optional[dict]: - """ - 更新试卷（合并更新，保留未传字段） - - Args: - exam_id: 试卷ID - exam_data: 新的试卷数据（部分字段） - - Returns: - 更新后的试卷，如果不存在返回None - """ - # 查找试卷文件路径 - filepath = _find_exam_filepath(exam_id) - if not filepath: - return None - - # 加载现有试卷数据 - existing_exam = load_exam(filepath) - - # 合并数据：现有数据 + 新数据（新数据覆盖同名字段） - merged = {**existing_exam, **exam_data} - - # 保留不可修改的字段 - merged["exam_id"] = exam_id - merged["created_at"] = existing_exam.get("created_at") - merged["created_by"] = existing_exam.get("created_by") - - # 保留状态（除非明确传入新状态） - if "status" not in exam_data: - merged["status"] = existing_exam.get("status", EXAM_STATUS_DRAFT) - - # 更新时间戳 - merged["updated_at"] = datetime.now().isoformat() - - # 重新计算统计 - merged["total_count"] = ( - len(merged.get("choice_questions", [])) + - len(merged.get("blank_questions", [])) + - len(merged.get("short_answer_questions", [])) - ) - - # 保存到原位置 - with open(filepath, 'w', encoding='utf-8') as f: - json.dump(merged, f, ensure_ascii=False, indent=2) - - return merged - - -def delete_exam(exam_id: str) -> bool: - """ - 删除试卷 - - Args: - exam_id: 试卷ID - - Returns: - 是否删除成功 - """ - filepath = _find_exam_filepath(exam_id) - if filepath and os.path.exists(filepath): - os.remove(filepath) - return True - - return False - - -# ==================== 题目搜索 ==================== - -def search_questions(keyword: str, question_type: str = None, - difficulty: int = None, limit: int = 50) -> dict: - """ - 搜索题目 - - Args: - keyword: 搜索关键词 - question_type: 题型过滤（choice/blank/short_answer） - difficulty: 难度过滤 - limit: 返回数量限制 - - Returns: - { - "questions": [...], - "total": int - } - """ - os.makedirs(QUESTION_BANK_DIR, exist_ok=True) - - results = [] - keyword_lower = keyword.lower() - - for filename in os.listdir(QUESTION_BANK_DIR): - if not filename.endswith('.json'): - continue - - filepath = os.path.join(QUESTION_BANK_DIR, filename) - try: - exam = load_exam(filepath) - exam_id = exam.get("exam_id", filename[:-5]) - exam_name = exam.get("topic", filename[:-5]) - - # 搜索选择题 - if not question_type or question_type == "choice": - for q in exam.get("choice_questions", []): - if difficulty and q.get("difficulty") != difficulty: - continue - content = q.get("content", "").lower() - if keyword_lower in content: - results.append({ - "exam_id": exam_id, - "exam_name": exam_name, - "type": "choice", - "question": q - }) - - # 搜索填空题 - if not question_type or question_type == "blank": - for q in exam.get("blank_questions", []): - if difficulty and q.get("difficulty") != difficulty: - continue - content = q.get("content", "").lower() - if keyword_lower in content: - results.append({ - "exam_id": exam_id, - "exam_name": exam_name, - "type": "blank", - "question": q - }) - - # 搜索简答题 - if not question_type or question_type == "short_answer": - for q in exam.get("short_answer_questions", []): - if difficulty and q.get("difficulty") != difficulty: - continue - content = q.get("content", "").lower() - if keyword_lower in content: - results.append({ - "exam_id": exam_id, - "exam_name": exam_name, - "type": "short_answer", - "question": q - }) - - except Exception as e: - logger.error(f"搜索试卷失败 {filename}: {e}") - continue - - return { - "questions": results[:limit], - "total": len(results) - } - +""" +出题与批题系统管理器 + +核心功能： +1. 题目生成（调用 generator.py） +2. 答案批阅（调用 grader.py） +3. 题库管理（保存/加载/搜索） + +职责边界： +- RAG 服务负责：生成题目 + 批阅答案 +- 后端服务负责：审核入库 + 状态管理 + +使用方式： + from exam_pkg.manager import generate_questions_from_file, grade_answers +""" +import json +import os +import re + +import uuid +import hashlib +from datetime import datetime +from typing import Optional, List, Dict, Any +from concurrent.futures import ThreadPoolExecutor, as_completed +import logging + +logger = logging.getLogger(__name__) + +# 状态常量 +EXAM_STATUS_DRAFT = "draft" +EXAM_STATUS_APPROVED = "approved" + +# 导入新的生成器和批题器 +from exam_pkg.generator import ( + QuestionGenerator, + build_semantic_query, + build_source_context, + safe_parse_questions, + validate_questions_schema +) +from exam_pkg.grader import ( + AnswerGrader, + grade_answers as grade_answers_v2, + grade_objective, + grade_fill_blank +) + +# MinerU 解析器（可选） +try: + from parsers import parse_document, MINERU_AVAILABLE + if MINERU_AVAILABLE: + from parsers.mineru_parser import MinerUChunk + PARSE_AVAILABLE = MINERU_AVAILABLE +except ImportError: + PARSE_AVAILABLE = False + +# 导入配置 + +# 题库目录 +QUESTION_BANK_DIR = "./题库" +DRAFT_DIR = "./题库/草稿" + + +# ==================== 新版出题接口 ==================== + +def generate_questions_from_file( + file_path: str, + collection: str, + question_types: Dict[str, int], + difficulty: int = 3, + options: Dict = None, + request_id: str = None +) -> Dict: + """ + 从文件生成题目（结构化出题 v2） + + 🔥 改进版 Structure-aware RAG 出题架构： + - Phase 1: 文档结构分析 + - Phase 2: 知识点规划（全局唯一） + - Phase 3: 精准出题（每个知识点单独检索） + - Phase 4: 质量校验 + + Args: + file_path: 文件路径 + collection: 向量库名称 + question_types: 题型及数量 {"single_choice": 3, "fill_blank": 2, ...} + difficulty: 难度 1-5 + options: 可选配置 + - max_source_chunks: 最大切片数（默认 30） + request_id: 请求 ID（幂等性支持） + + Returns: + { + "success": True, + "request_id": "...", + "questions": [...], + "total": 10, + "source_chunks_used": 15 + } + """ + options = options or {} + + # 1. 检索文件的所有切片 + chunks = retrieve_file_chunks( + file_path=file_path, + collection=collection, + question_types=question_types, + top_k=options.get('max_source_chunks', 50) + ) + logger.debug(f"generate_questions_from_file: chunks 数量 = {len(chunks)}") + + # 2. 结构化出题 v2：全局唯一知识点 + 精准出题 + from exam_pkg.generator import generate_questions_structured_v2 + questions = generate_questions_structured_v2( + chunks=chunks, + document_name=file_path, + question_types=question_types, + difficulty=difficulty + ) + + # 3. 统计实际出题数量，与请求数量对比 + from collections import defaultdict + actual_types = defaultdict(int) + for q in questions: + q_type = q.get('question_type', 'unknown') + actual_types[q_type] += 1 + + # 构建警告信息 + warnings = [] + for q_type, requested_count in question_types.items(): + actual_count = actual_types.get(q_type, 0) + if actual_count < requested_count: + warnings.append(f"{q_type}: 请求 {requested_count} 道，实际生成 {actual_count} 道") + + result = { + "success": True, + "request_id": request_id, + "questions": questions, + "total": len(questions), + "requested_types": question_types, + "actual_types": dict(actual_types), + "source_chunks_used": len(chunks) + } + + if warnings: + result["warnings"] = warnings + + return result + + +def analyze_file_for_exam( + file_path: str, + collection: str, + top_k: int = 50 +) -> Dict[str, Any]: + """ + 分析文件内容，返回 AI 推荐的题型和数量 + + Args: + file_path: 文件路径 + collection: 向量库名称 + top_k: 最大切片数（默认 50，用于分析） + + Returns: + { + "total_knowledge_points": int, + "suitable_types": List[str], + "question_types": Dict[str, int], + "reason": str + } + """ + from exam_pkg.generator import analyze_document_for_exam + + # 1. 检索文件切片（用于分析） + # 使用通用 query 检索，获取尽可能多的内容 + query = "重点内容 关键概念 定义 流程 步骤" + chunks = retrieve_file_chunks_for_analysis( + file_path=file_path, + collection=collection, + query=query, + top_k=top_k + ) + + if not chunks: + return { + "total_knowledge_points": 0, + "suitable_types": [], + "question_types": {}, + "reason": "未找到文件内容，请检查文件路径和向量库" + } + + # 2. 调用 AI 分析 + return analyze_document_for_exam(chunks) + + +def retrieve_file_chunks_for_analysis( + file_path: str, + collection, + query: str, + top_k: int = 50 +) -> List[Dict]: + """ + 检索文件切片（用于分析，不需要 question_types） + + Args: + file_path: 文件路径 + collection: 向量库名称 + query: 检索 query + top_k: 最大切片数 + + Returns: + 切片列表 + """ + try: + # 获取 RAG 引擎 + from core.engine import get_engine + engine = get_engine() + if not engine: + logger.error("RAG 引擎未初始化") + return [] + + # 准备 collection 列表 + collections = [collection] if isinstance(collection, str) else collection + + # 提取文件名（向量库存储的是文件名，不含路径前缀） + filename = os.path.basename(file_path) + + # 向量检索（尝试文件名和完整路径） + results = None + for source_filter in [filename, file_path]: + results = engine.search_knowledge( + query=query, + collections=collections, + source_filter=source_filter, + top_k=top_k + ) + if results.get('documents') and results['documents'][0]: + break + + if not results or not results.get('documents') or not results['documents'][0]: + logger.warning(f"未检索到相关切片: file={file_path}, collection={collection}") + return [] + + # 整理结果（search_knowledge 返回格式: {ids: [[...]], documents: [[...]], metadatas: [[...]], distances: [[...]]}) + chunks = [] + for i, (doc, meta, score) in enumerate(zip( + results['documents'][0], + results['metadatas'][0], + results['distances'][0] + )): + chunks.append({ + "chunk_id": results['ids'][0][i], + "content": doc, + "source": meta.get('source', ''), + "page": meta.get('page'), + "section": meta.get('section', ''), + "score": score + }) + + return chunks + + except Exception as e: + logger.error(f"检索文件切片失败: {e}") + return [] + + +def retrieve_file_chunks( + file_path: str, + collection, # 支持字符串或列表 + question_types: Dict[str, int], + top_k: int = None +) -> List[Dict]: + """ + 检索文件相关切片 + + Args: + file_path: 文件路径 + collection: 向量库名称（支持字符串或列表，列表时按优先级顺序检索） + question_types: 题型及数量 + top_k: 最大切片数 + + Returns: + 切片列表 + """ + # 动态计算 top_k：题型数量 × 3，上限 50 + if top_k is None: + total_questions = sum(question_types.values()) + top_k = min(50, total_questions * 3) + + # 根据题型构建语义 query + query = build_semantic_query(question_types) + + # 提取文件名（向量库存储的是文件名，不含路径前缀） + filename = os.path.basename(file_path) + + # 统一处理为列表 + if isinstance(collection, str): + collections = [collection] + else: + collections = list(collection) if collection else [] + + if not collections: + logger.error("[ERROR] 未指定向量库") + return [] + + try: + from core.engine import get_engine + engine = get_engine() + + # 按优先级遍历 collections，找到文件即停止 + for coll in collections: + # 尝试两种格式：文件名和完整路径 + for source_filter in [filename, file_path]: + results = engine.search_knowledge( + query=query, + collections=[coll], + source_filter=source_filter, + top_k=top_k + ) + + # 检查是否有结果 + if results.get('documents') and results['documents'][0]: + logger.debug(f"在向量库 [{coll}] 中找到文件 [{filename}]") + break + else: + continue + break # 外层循环跳出 + + chunks = [] + if results.get('documents') and results['documents'][0]: + for i, (doc, meta, score) in enumerate(zip( + results['documents'][0], + results['metadatas'][0], + results['distances'][0] + )): + chunks.append({ + "chunk_id": results['ids'][0][i], + "content": doc, + "source": meta.get('source'), + "page": meta.get('page'), + "section": meta.get('section', ''), + "score": score + }) + + logger.debug(f"retrieve_file_chunks: 检索到 {len(chunks)} 个切片") + return chunks + + except Exception as e: + logger.error(f"检索切片失败: {e}") + return [] + + +# ==================== 新版批题接口 ==================== + +def grade_answers(answers: List[Dict], request_id: str = None) -> Dict: + """ + 批阅答案入口函数 + + 🔥 改进： + - 本地批阅选择题/判断题 + - 填空题模糊匹配 + - 主观题 LLM 评分 + - 并发 + 限流 + 顺序保持 + """ + return grade_answers_v2(answers, request_id) + + +def get_source_chapters_for_question( + file_path: str, + question_topic: str, + top_k: int = 3 +) -> List[Dict[str, Any]]: + """ + 获取与问题主题相关的章节内容（用于出题或批阅参考） + + Args: + file_path: 文档文件路径（支持 PDF/DOCX/PPTX） + question_topic: 问题主题或关键词 + top_k: 返回的章节数量 + + Returns: + [ + { + "title": "章节标题", + "content": "完整内容", + "section_path": "章节路径", + "page_range": "1-2" + } + ] + """ + if not PARSE_AVAILABLE: + raise ImportError("MinerU 解析器不可用") + + result = parse_document( + file_path, + output_base=".data/mineru_temp", + images_output=".data/images", + cleanup_after_image_move=True + ) + chunks = result.get('chunks', []) + + # 构建章节内容列表 + chapter_contents = [] + for chunk in chunks: + content = chunk.content if hasattr(chunk, 'content') else str(chunk) + if not content.strip(): + continue + + chapter_contents.append({ + "title": chunk.title if hasattr(chunk, 'title') else '', + "content": content.strip(), + "section_path": chunk.section_path if hasattr(chunk, 'section_path') else '', + "page_range": f"{chunk.page_start}-{chunk.page_end}" if hasattr(chunk, 'page_start') else '', + "source_file": chunk.source_file if hasattr(chunk, 'source_file') else os.path.basename(file_path) + }) + + # 使用关键词检索相关章节 + keywords = question_topic.split() if question_topic else [] + + # 简单的关键词匹配筛选 + relevant = [] + for c in chapter_contents: + content_lower = c['content'].lower() + if any(kw.lower() in content_lower for kw in keywords): + relevant.append(c) + + return relevant[:top_k] + + +def sanitize_filename(name: str) -> str: + # 移除或替换非法字符 + illegal_chars = r'[<>:"/\\|?*\x00-\x1f]' + safe_name = re.sub(illegal_chars, '_', name) + # 移除首尾空格和点 + safe_name = safe_name.strip('. ') + # 限制长度 + if len(safe_name) > 100: + safe_name = safe_name[:100] + # 如果清理后为空，使用时间戳 + if not safe_name: + safe_name = datetime.now().strftime('%Y%m%d_%H%M%S') + return safe_name + + +def save_exam(exam: dict, name: str = None) -> str: + """ + 保存试卷到对应目录 + + - draft 状态：保存到 题库/草稿/ 目录 + - approved 状态：保存到 题库/ 目录 + + Args: + exam: 试卷JSON + name: 文件名(不含扩展名)，默认使用试卷名称 + + Returns: + 保存的文件路径 + """ + # 根据状态确定保存目录 + status = exam.get("status", EXAM_STATUS_DRAFT) + if status == EXAM_STATUS_APPROVED: + save_dir = QUESTION_BANK_DIR + else: + save_dir = DRAFT_DIR + + # 确保目录存在 + os.makedirs(save_dir, exist_ok=True) + # 确保草稿目录存在 + os.makedirs(DRAFT_DIR, exist_ok=True) + + # 确定文件名：优先使用试卷名称 + if name is None: + # 使用试卷名称作为文件名 + exam_name = exam.get("name") or exam.get("topic", "未命名试卷") + name = sanitize_filename(exam_name) + + # 检查是否已存在同名文件，如果存在则添加时间戳 + filepath = os.path.join(save_dir, f"{name}.json") + if os.path.exists(filepath): + timestamp = datetime.now().strftime('_%Y%m%d_%H%M%S') + name = f"{name}{timestamp}" + + filepath = os.path.join(save_dir, f"{name}.json") + + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(exam, f, ensure_ascii=False, indent=2) + + logger.info(f"试卷已保存: {filepath} (状态: {status})") + return filepath + + +def load_exam(filepath: str) -> dict: + """加载试卷""" + with open(filepath, 'r', encoding='utf-8') as f: + return json.load(f) + + +# ==================== 题库管理函数 ==================== + +def list_exams(status: str = None, page: int = 1, limit: int = 20) -> dict: + """ + 获取试卷列表 + + Args: + status: 状态过滤（draft/pending_review/approved/rejected） + page: 页码 + limit: 每页数量 + + Returns: + { + "exams": [...], + "total": int, + "page": int + } + """ + # 确保目录存在 + os.makedirs(QUESTION_BANK_DIR, exist_ok=True) + os.makedirs(DRAFT_DIR, exist_ok=True) + + exams = [] + + # 从题库目录和草稿目录加载试卷 + for directory in [QUESTION_BANK_DIR, DRAFT_DIR]: + if not os.path.exists(directory): + continue + for filename in os.listdir(directory): + if not filename.endswith('.json'): + continue + + filepath = os.path.join(directory, filename) + try: + exam = load_exam(filepath) + # 状态过滤 + if status and exam.get("status") != status: + continue + # 只返回摘要信息 + exams.append({ + "exam_id": exam.get("exam_id", filename[:-5]), + "name": exam.get("name") or exam.get("topic", "未命名试卷"), + "topic": exam.get("topic", ""), + "status": exam.get("status", "approved"), + "total_count": exam.get("total_count", 0), + "total_score": exam.get("total_score", 0), + "created_at": exam.get("created_at", ""), + "created_by": exam.get("created_by", ""), + "filename": filename, + "filepath": filepath + }) + except Exception as e: + logger.error(f"加载试卷失败 {filename}: {e}") + continue + + # 按创建时间倒序 + exams.sort(key=lambda x: x.get("created_at", ""), reverse=True) + + # 分页 + total = len(exams) + start = (page - 1) * limit + end = start + limit + + return { + "exams": exams[start:end], + "total": total, + "page": page + } + + +def _find_exam_filepath(exam_id: str) -> Optional[str]: + """ + 查找试卷文件路径（在题库和草稿目录中查找） + + Args: + exam_id: 试卷ID + + Returns: + 文件路径，如果不存在返回None + """ + # 确保目录存在 + os.makedirs(QUESTION_BANK_DIR, exist_ok=True) + os.makedirs(DRAFT_DIR, exist_ok=True) + + # 先检查题库目录 + for directory in [DRAFT_DIR, QUESTION_BANK_DIR]: + for filename in os.listdir(directory): + if not filename.endswith('.json'): + continue + filepath = os.path.join(directory, filename) + try: + exam = load_exam(filepath) + if exam.get("exam_id") == exam_id: + return filepath + except (json.JSONDecodeError, OSError, KeyError): + continue + return None + + +def get_exam_by_id(exam_id: str) -> Optional[dict]: + """ + 根据 ID 获取试卷 + + Args: + exam_id: 试卷ID + + Returns: + 试卷JSON，如果不存在返回None + """ + filepath = _find_exam_filepath(exam_id) + if filepath: + return load_exam(filepath) + return None + + +def update_exam(exam_id: str, exam_data: dict) -> Optional[dict]: + """ + 更新试卷（合并更新，保留未传字段） + + Args: + exam_id: 试卷ID + exam_data: 新的试卷数据（部分字段） + + Returns: + 更新后的试卷，如果不存在返回None + """ + # 查找试卷文件路径 + filepath = _find_exam_filepath(exam_id) + if not filepath: + return None + + # 加载现有试卷数据 + existing_exam = load_exam(filepath) + + # 合并数据：现有数据 + 新数据（新数据覆盖同名字段） + merged = {**existing_exam, **exam_data} + + # 保留不可修改的字段 + merged["exam_id"] = exam_id + merged["created_at"] = existing_exam.get("created_at") + merged["created_by"] = existing_exam.get("created_by") + + # 保留状态（除非明确传入新状态） + if "status" not in exam_data: + merged["status"] = existing_exam.get("status", EXAM_STATUS_DRAFT) + + # 更新时间戳 + merged["updated_at"] = datetime.now().isoformat() + + # 重新计算统计 + merged["total_count"] = ( + len(merged.get("choice_questions", [])) + + len(merged.get("blank_questions", [])) + + len(merged.get("short_answer_questions", [])) + ) + + # 保存到原位置 + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(merged, f, ensure_ascii=False, indent=2) + + return merged + + +def delete_exam(exam_id: str) -> bool: + """ + 删除试卷 + + Args: + exam_id: 试卷ID + + Returns: + 是否删除成功 + """ + filepath = _find_exam_filepath(exam_id) + if filepath and os.path.exists(filepath): + os.remove(filepath) + return True + + return False + + +# ==================== 题目搜索 ==================== + +def search_questions(keyword: str, question_type: str = None, + difficulty: int = None, limit: int = 50) -> dict: + """ + 搜索题目 + + Args: + keyword: 搜索关键词 + question_type: 题型过滤（choice/blank/short_answer） + difficulty: 难度过滤 + limit: 返回数量限制 + + Returns: + { + "questions": [...], + "total": int + } + """ + os.makedirs(QUESTION_BANK_DIR, exist_ok=True) + + results = [] + keyword_lower = keyword.lower() + + for filename in os.listdir(QUESTION_BANK_DIR): + if not filename.endswith('.json'): + continue + + filepath = os.path.join(QUESTION_BANK_DIR, filename) + try: + exam = load_exam(filepath) + exam_id = exam.get("exam_id", filename[:-5]) + exam_name = exam.get("topic", filename[:-5]) + + # 搜索选择题 + if not question_type or question_type == "choice": + for q in exam.get("choice_questions", []): + if difficulty and q.get("difficulty") != difficulty: + continue + content = q.get("content", "").lower() + if keyword_lower in content: + results.append({ + "exam_id": exam_id, + "exam_name": exam_name, + "type": "choice", + "question": q + }) + + # 搜索填空题 + if not question_type or question_type == "blank": + for q in exam.get("blank_questions", []): + if difficulty and q.get("difficulty") != difficulty: + continue + content = q.get("content", "").lower() + if keyword_lower in content: + results.append({ + "exam_id": exam_id, + "exam_name": exam_name, + "type": "blank", + "question": q + }) + + # 搜索简答题 + if not question_type or question_type == "short_answer": + for q in exam.get("short_answer_questions", []): + if difficulty and q.get("difficulty") != difficulty: + continue + content = q.get("content", "").lower() + if keyword_lower in content: + results.append({ + "exam_id": exam_id, + "exam_name": exam_name, + "type": "short_answer", + "question": q + }) + + except Exception as e: + logger.error(f"搜索试卷失败 {filename}: {e}") + continue + + return { + "questions": results[:limit], + "total": len(results) + } +