""" 题目数据格式化工具 用于标准化题目数据的JSON输出格式 """ from typing import Dict, List, Any, Optional from datetime import datetime import json import re class QuestionFormatter: """题目数据格式化器""" def __init__(self): # 题目类型映射 self.type_mapping = { 'choice': 'single_choice', 'single_choice': 'single_choice', 'multiple_choice': 'multiple_choice', 'fill_blank': 'fill_blank', 'calculation': 'calculation', 'essay': 'essay', 'solve': 'solve', 'prove': 'prove', 'true_false': 'true_false', 'general': 'essay' } # 学科映射 self.subject_mapping = { 'math': '数学', 'chinese': '语文', 'english': '英语', 'physics': '物理', 'chemistry': '化学', 'biology': '生物', 'history': '历史', 'geography': '地理', 'politics': '政治' } # 难度等级映射 self.difficulty_mapping = { 1: '简单', 2: '较易', 3: '中等', 4: '较难', 5: '困难' } def format_question(self, raw_question: Dict, paper_info: Optional[Dict] = None) -> Dict: """ 格式化单个题目数据 Args: raw_question: 原始题目数据 paper_info: 试卷信息（可选） Returns: 标准化的题目数据 """ try: # 基础信息 formatted = { "id": raw_question.get("id"), "number": self._extract_question_number(raw_question), "type": self._normalize_question_type(raw_question), "subject": self._normalize_subject(raw_question, paper_info), "grade": raw_question.get("grade", ""), "chapter": raw_question.get("chapter", ""), "difficulty": self._normalize_difficulty(raw_question), "score": self._extract_score(raw_question), "estimated_time": self._estimate_time(raw_question) } # 题目内容 formatted.update({ "stem": self._clean_question_text(raw_question.get("question_content") or raw_question.get("text", "")), "options": self._format_options(raw_question.get("options", {})), "correct_answer": self._clean_answer(raw_question.get("correct_answer", "")), "explanation": self._clean_text(raw_question.get("explanation", "")), "analysis": self._clean_text(raw_question.get("analysis", "")), "hints": self._extract_hints(raw_question) }) # 知识点 formatted["knowledge_points"] = self._format_knowledge_points(raw_question.get("knowledge_points", [])) # 媒体资源 formatted.update({ "images": self._format_images(raw_question.get("images", [])), "formulas": self._format_formulas(raw_question.get("formula", [])), "audio": raw_question.get("audio", ""), "video": raw_question.get("video", "") }) # 来源信息 formatted.update({ "source": self._format_source(raw_question, paper_info), "metadata": self._extract_metadata(raw_question) }) # 统计信息 formatted["statistics"] = { "created_at": self._format_datetime(raw_question.get("created_at")), "updated_at": self._format_datetime(raw_question.get("updated_at")), "view_count": raw_question.get("view_count", 0), "attempt_count": raw_question.get("attempt_count", 0), "correct_count": raw_question.get("correct_count", 0), "accuracy_rate": self._calculate_accuracy(raw_question) } # 变体信息（子题） formatted["variants"] = self._extract_variants(raw_question) # 验证数据完整性 formatted["validation"] = self._validate_question(formatted) return formatted except Exception as e: print(f"格式化题目失败: {e}") return self._create_error_question(raw_question, str(e)) def format_paper_questions(self, paper_data: Dict) -> Dict: """ 格式化试卷的所有题目 Args: paper_data: 包含试卷信息和题目列表的数据 Returns: 标准化的试卷题目数据 """ questions = paper_data.get("questions", []) paper_info = { "id": paper_data.get("paper_id"), "title": paper_data.get("title", ""), "subject": paper_data.get("subject", ""), "grade": paper_data.get("grade", "") } formatted_questions = [] processing_errors = [] for i, raw_question in enumerate(questions): try: formatted = self.format_question(raw_question, paper_info) formatted_questions.append(formatted) except Exception as e: processing_errors.append({ "index": i, "question_id": raw_question.get("id"), "error": str(e), "raw_data": raw_question }) return { "paper_info": paper_info, "questions": formatted_questions, "statistics": self._calculate_paper_stats(formatted_questions), "processing_errors": processing_errors, "export_info": { "exported_at": datetime.now().isoformat(), "total_questions": len(formatted_questions), "error_count": len(processing_errors), "format_version": "2.0" } } def _extract_question_number(self, question: Dict) -> str: """提取题号""" number = question.get("question_number") or question.get("number") or question.get("index") if number: return str(number) # 尝试从题干中提取题号 text = question.get("question_content") or question.get("text", "") import re match = re.search(r'^(\d+)[.)．、]', text) if match: return match.group(1) return "" def _normalize_question_type(self, question: Dict) -> str: """标准化题目类型""" raw_type = question.get("question_type") or question.get("type") or "" # 如果有选项，默认为选择题 if question.get("options") and len(question.get("options", {})) > 0: return "single_choice" return self.type_mapping.get(raw_type, "essay") def _normalize_subject(self, question: Dict, paper_info: Optional[Dict] = None) -> str: """标准化学科""" subject = question.get("subject") if subject: return self.subject_mapping.get(subject.lower(), subject) if paper_info and paper_info.get("subject"): return self.subject_mapping.get(paper_info["subject"].lower(), paper_info["subject"]) return "未分类" def _normalize_difficulty(self, question: Dict) -> Dict: """标准化难度信息""" raw_difficulty = question.get("difficulty", 3) try: difficulty_value = int(raw_difficulty) except (ValueError, TypeError): difficulty_value = 3 return { "level": max(1, min(5, difficulty_value)), "label": self.difficulty_mapping.get(difficulty_value, "中等"), "stars": "★" * max(1, min(5, difficulty_value)) } def _extract_score(self, question: Dict) -> Dict: """提取分值信息""" text = question.get("question_content") or question.get("text", "") import re # 查找分值模式 match = re.search(r'(\d+\.?\d*)\s*分', text) if match: score_value = float(match.group(1)) return { "total": score_value, "formatted": f"{score_value}分", "extracted_from_text": True } # 查找其他分值模式 score_field = question.get("score") if score_field: try: score_value = float(score_field) return { "total": score_value, "formatted": f"{score_value}分", "extracted_from_text": False } except (ValueError, TypeError): pass return { "total": 0, "formatted": "未标分", "extracted_from_text": False } def _estimate_time(self, question: Dict) -> Dict: """估算答题时间""" question_type = self._normalize_question_type(question) text_length = len(question.get("question_content") or question.get("text", "")) options_count = len(question.get("options", {})) difficulty = self._normalize_difficulty(question)["level"] # 基础时间（分钟） base_times = { "single_choice": 1, "multiple_choice": 2, "fill_blank": 2, "calculation": 5, "essay": 10, "solve": 8, "prove": 12, "true_false": 0.5 } base_time = base_times.get(question_type, 3) # 根据难度调整 difficulty_multiplier = 0.5 + (difficulty * 0.3) # 根据文本长度调整 text_multiplier = 1 + (text_length / 500) estimated_time = base_time * difficulty_multiplier * text_multiplier return { "minutes": round(estimated_time, 1), "seconds": round(estimated_time * 60), "formatted": f"{round(estimated_time, 1)}分钟" } def _clean_question_text(self, text: str) -> str: """清理题目文本""" if not text: return "" # 移除题号前缀 import re text = re.sub(r'^\d+\s*[.)．、]\s*', '', text) # 移除分值信息 text = re.sub(r'^\d+\.?\d*\s*分[：:]\s*', '', text) # 清理多余空格 text = re.sub(r'\s+', ' ', text).strip() return text def _format_options(self, options: Dict) -> List[Dict]: """格式化选项""" if not options: return [] formatted_options = [] for key, value in options.items(): if not value or not value.strip(): continue formatted_options.append({ "label": key.upper(), "text": self._clean_text(value), "is_correct": False, # 需要根据答案判断 "explanation": "" }) # 按标签排序 formatted_options.sort(key=lambda x: x["label"]) return formatted_options def _clean_answer(self, answer: str) -> str: """清理答案""" if not answer: return "" # 移除多余空格 answer = " ".join(answer.split()) # 标准化多答案格式 if "," in answer: answers = [a.strip() for a in answer.split(",")] return ", ".join(sorted(answers)) return answer def _clean_text(self, text: str) -> str: """清理普通文本""" if not text: return "" # 清理多余空格和换行 text = re.sub(r'\s+', ' ', text).strip() return text def _extract_hints(self, question: Dict) -> List[str]: """提取提示信息""" hints = [] # 从解析中提取提示 explanation = question.get("explanation", "") if "提示" in explanation or "注意" in explanation: import re hint_matches = re.findall(r'[提示注意][:：]\s*([^。！？\n]*?)[。！？\n]', explanation) hints.extend(hint_matches) return hints def _format_knowledge_points(self, knowledge_points: List) -> List[Dict]: """格式化知识点""" if not knowledge_points: return [] formatted = [] for point in knowledge_points: if isinstance(point, str): formatted.append({ "name": point, "category": "", "weight": 1.0 }) elif isinstance(point, dict): formatted.append({ "name": point.get("name", ""), "category": point.get("category", ""), "weight": point.get("weight", 1.0) }) return formatted def _format_images(self, images: List) -> List[Dict]: """格式化图片信息""" if not images: return [] formatted = [] for i, image in enumerate(images): if isinstance(image, str): formatted.append({ "id": f"img_{i+1}", "url": image, "alt": f"题目图片{i+1}", "type": "image" }) elif isinstance(image, dict): formatted.append({ "id": image.get("id", f"img_{i+1}"), "url": image.get("url") or image.get("uri") or image.get("key", ""), "alt": image.get("alt", f"题目图片{i+1}"), "type": "image", "width": image.get("width"), "height": image.get("height"), "format": image.get("format", "unknown") }) return formatted def _format_formulas(self, formulas: List) -> List[Dict]: """格式化公式信息""" if not formulas: return [] formatted = [] for i, formula in enumerate(formulas): if isinstance(formula, dict): formatted.append({ "id": formula.get("id", f"formula_{i+1}"), "latex": formula.get("latex", ""), "position": formula.get("bbox", []), "type": "formula" }) return formatted def _format_source(self, question: Dict, paper_info: Optional[Dict] = None) -> Dict: """格式化来源信息""" source = { "type": "auto_extracted", "method": question.get("source", "unknown"), "paper": { "id": paper_info.get("id") if paper_info else None, "title": paper_info.get("title", "") if paper_info else "", "page": question.get("page", 1) }, "extraction_confidence": 0.8, # 提取置信度 "verified": False # 是否已人工验证 } return source def _extract_metadata(self, question: Dict) -> Dict: """提取元数据""" return { "extraction_method": question.get("auto_extracted", False) and "automatic" or "manual", "original_format": "pdf", "language": "zh-CN", "tags": question.get("tags", []), "notes": question.get("notes", ""), "page_spans": question.get("page_spans", []), "source_text": question.get("source_text", "") } def _format_datetime(self, dt) -> str: """格式化日期时间""" if not dt: return "" if isinstance(dt, str): return dt if hasattr(dt, 'isoformat'): return dt.isoformat() return str(dt) def _calculate_accuracy(self, question: Dict) -> float: """计算正确率""" attempt_count = question.get("attempt_count", 0) correct_count = question.get("correct_count", 0) if attempt_count == 0: return 0.0 return round(correct_count / attempt_count, 2) def _extract_variants(self, question: Dict) -> List[Dict]: """提取子题/变体""" variants = [] text = question.get("question_content") or question.get("text", "") # 查找子题模式 (1), (2), (3) 等 import re sub_matches = re.findall(r'\((\d+)\)\s*([^()]*?)(?=\(\d+\)|$)', text) for num, content in sub_matches: if content.strip(): variants.append({ "number": num, "type": "subquestion", "content": self._clean_text(content), "score": None # 可以尝试提取子题分值 }) return variants def _validate_question(self, question: Dict) -> Dict: """验证题目数据完整性""" validation = { "is_valid": True, "errors": [], "warnings": [], "completeness_score": 1.0 } # 必需字段检查 required_fields = ["stem", "type"] for field in required_fields: if not question.get(field): validation["errors"].append(f"缺少必需字段: {field}") validation["is_valid"] = False # 题目类型特定检查 if question["type"] in ["single_choice", "multiple_choice"]: if not question.get("options"): validation["errors"].append("选择题缺少选项") validation["is_valid"] = False elif not question.get("correct_answer"): validation["warnings"].append("选择题缺少正确答案") # 内容质量检查 stem_length = len(question.get("stem", "")) if stem_length < 5: validation["warnings"].append("题干内容过短") elif stem_length > 2000: validation["warnings"].append("题干内容过长") # 计算完整性分数 total_fields = 8 # 总共评估的字段数 filled_fields = sum(1 for field in ["stem", "options", "correct_answer", "explanation", "knowledge_points", "difficulty", "score", "images"] if question.get(field)) validation["completeness_score"] = filled_fields / total_fields return validation def _calculate_paper_stats(self, questions: List[Dict]) -> Dict: """计算试卷统计信息""" if not questions: return {} stats = { "total_questions": len(questions), "question_types": {}, "difficulty_distribution": {"简单": 0, "较易": 0, "中等": 0, "较难": 0, "困难": 0}, "total_score": 0, "estimated_time": 0, "knowledge_points": [], "has_images": 0, "average_accuracy": 0.0 } all_accuracy = [] all_knowledge_points = set() for question in questions: # 题目类型统计 q_type = question.get("type", "unknown") stats["question_types"][q_type] = stats["question_types"].get(q_type, 0) + 1 # 难度分布 difficulty = question.get("difficulty", {}).get("label", "中等") stats["difficulty_distribution"][difficulty] = stats["difficulty_distribution"].get(difficulty, 0) + 1 # 分值统计 score = question.get("score", {}).get("total", 0) stats["total_score"] += score # 时间估算 time_minutes = question.get("estimated_time", {}).get("minutes", 0) stats["estimated_time"] += time_minutes # 知识点统计 for kp in question.get("knowledge_points", []): kp_name = kp.get("name", kp) if isinstance(kp, dict) else kp all_knowledge_points.add(kp_name) # 图片统计 if question.get("images"): stats["has_images"] += 1 # 正确率统计 accuracy = question.get("statistics", {}).get("accuracy_rate", 0) if accuracy > 0: all_accuracy.append(accuracy) stats["knowledge_points"] = list(all_knowledge_points) if all_accuracy: stats["average_accuracy"] = round(sum(all_accuracy) / len(all_accuracy), 2) # 格式化时间 total_minutes = stats["estimated_time"] if total_minutes >= 60: hours = int(total_minutes // 60) minutes = int(total_minutes % 60) stats["estimated_time_formatted"] = f"{hours}小时{minutes}分钟" else: stats["estimated_time_formatted"] = f"{int(total_minutes)}分钟" return stats def _create_error_question(self, raw_question: Dict, error_msg: str) -> Dict: """创建错误题目对象""" return { "id": raw_question.get("id"), "number": self._extract_question_number(raw_question), "type": "error", "stem": f"[数据解析错误] {error_msg}", "options": [], "correct_answer": "", "explanation": "", "validation": { "is_valid": False, "errors": [error_msg], "warnings": [], "completeness_score": 0.0 }, "source": { "type": "error", "method": "error", "error_message": error_msg } } def export_to_json_file(self, data: Dict, file_path: str, indent: int = 2) -> bool: """导出到JSON文件""" try: with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=indent, default=str) return True except Exception as e: print(f"导出JSON文件失败: {e}") return False def export_to_json_string(self, data: Dict, indent: int = 2) -> str: """导出为JSON字符串""" try: return json.dumps(data, ensure_ascii=False, indent=indent, default=str) except Exception as e: print(f"导出JSON字符串失败: {e}") return json.dumps({"error": str(e)}, ensure_ascii=False, indent=indent)