3f1980ba78
- grader: 多策略 JSON 提取 + 3 次重试 + prompt 优化，修复 LLM 输出不稳定问题 - grader: question_content → content 统一字段名，与出题接口一致 - api: 新增入参校验（题型枚举、difficulty、总题数上限 20、grade question_type） - api/generator/manager: 新增 exclude_stems 参数，支持跨调用去重
767 lines
22 KiB
Python
767 lines
22 KiB
Python
"""
|
||
出题与批题系统管理器
|
||
|
||
核心功能：
|
||
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,
|
||
exclude_stems: List[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（幂等性支持）
|
||
exclude_stems: 需要排除的已有题目题干列表（跨调用去重）
|
||
|
||
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,
|
||
exclude_stems=exclude_stems
|
||
)
|
||
|
||
# 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)
|
||
}
|
||
|