python-template/backend/utils/question_formatter.py

"""
题目数据格式化工具
用于标准化题目数据的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)