""" 基于 Claude Sonnet 4.5 的智能试卷解析器 支持多学科（数学、物理、化学等）试卷的智能解析 支持多格式（PDF、Word、图片）文件处理 能够识别题目、图片、数学公式等多种内容 """ import os import io import json import base64 import tempfile import requests import logging from typing import List, Dict, Optional, Tuple, Union from pathlib import Path import fitz # PyMuPDF from PIL import Image from docx import Document from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P from docx.table import _Cell, Table from docx.text.paragraph import Paragraph # 配置日志 logger = logging.getLogger(__name__) class ClaudeExamParser: """ 使用 Claude Sonnet 4.5 的智能试卷解析器 核心优势： 1. 多模态理解：支持文字、图片、公式混合内容 2. 学科识别：自动识别数学、物理、化学等学科 3. 智能分割：自动识别题目边界和题号 4. 公式识别：理解 LaTeX、图片公式等多种格式 5. 上下文理解：保持题目完整性，不会截断 """ def __init__(self, api_key: Optional[str] = None, model: Optional[str] = None): """ 初始化解析器 Args: api_key: OpenRouter API密钥，如不提供则从环境变量 OPENROUTER_API_KEY 读取 model: 模型名称，可选： - anthropic/claude-sonnet-4.5 (默认，推荐) - google/gemini-3-pro-preview (Gemini 3 Pro，数学能力强) - openai/gpt-4o - openai/gpt-4-turbo """ self.api_key = api_key or os.getenv("OPENROUTER_API_KEY") if not self.api_key: raise ValueError("需要提供 OpenRouter API Key，或在环境变量中设置 OPENROUTER_API_KEY") self.api_url = "https://openrouter.ai/api/v1/chat/completions" self.model = model or os.getenv("AI_MODEL", "anthropic/claude-sonnet-4.5") # 检测是否使用 Gemini 模型 self.is_gemini = "gemini" in self.model.lower() logger.info(f"🤖 使用模型: {self.model}") if self.is_gemini: logger.info(f"💎 检测到 Gemini 模型，将启用数学优化模式") def _get_optimized_prompt(self, subject: Optional[str], grade: Optional[str], total_pages: int) -> str: """ 根据学科和模型类型生成优化的 prompt Args: subject: 学科类型 grade: 年级 total_pages: 总页数 Returns: 优化的 prompt 文本 """ is_math = subject and "数学" in subject # 基础 prompt base_prompt = f"""你是一个专业的试卷解析助手。请分析这份试卷（共{total_pages}页），并提取所有题目信息。""" # Gemini 数学专项优化 if self.is_gemini and is_math: return base_prompt + """ **🎯 数学试卷专项解析要求（Gemini 优化版）** 1. **题目识别**（高精度） - 准确识别所有题型：选择题、填空题、解答题、证明题、计算题、应用题 - 精确提取题号（支持：1、2、3 或 一、二、三 或 (1)、(2) 等多种格式） - 识别大题下的小题（如：第18题(1)(2)(3)） 2. **数学公式识别**（核心能力） - **LaTeX 转换**：所有数学表达式必须转换为标准 LaTeX 格式 - **内联公式**：用 $...$ 包裹，如 $x^2 + y^2 = r^2$ - **块级公式**：用 $$...$$ 包裹，如 $$\\int_a^b f(x)dx$$ - **复杂公式**： * 分数：$\\frac{a}{b}$ * 根式：$\\sqrt{x}$、$\\sqrt[n]{x}$ * 极限：$\\lim_{x \\to \\infty}$ * 求和：$\\sum_{i=1}^{n}$ * 积分：$\\int$、$\\iint$、$\\iiint$ * 矩阵：$\\begin{pmatrix} a & b \\\\ c & d \\end{pmatrix}$ * 向量：$\\vec{a}$、$\\overrightarrow{AB}$ * 三角函数：$\\sin$、$\\cos$、$\\tan$ - **特殊符号**： * 不等式：$\\leq$、$\\geq$、$\\neq$ * 集合：$\\in$、$\\subset$、$\\cup$、$\\cap$、$\\varnothing$ * 逻辑：$\\forall$、$\\exists$、$\\therefore$ * 箭头：$\\rightarrow$、$\\Rightarrow$、$\\Leftrightarrow$ 3. **几何图形识别** - 详细描述图形类型（三角形、圆、抛物线、函数图像等） - 标注关键点、线段、角度标记 - 记录坐标系信息（坐标轴、刻度、原点） - 描述图形中的文字标注和数值 4. **解题步骤提取**（如有答案） - 识别每个步骤的推导过程 - 提取关键公式和定理引用 - 记录计算过程中的中间结果 5. **知识点标注**（数学专项） - 代数：方程、不等式、函数、数列 - 几何：平面几何、立体几何、解析几何 - 三角：三角函数、三角恒等式 - 微积分：导数、积分、极限 - 概率统计：概率、统计、随机变量 - 向量与矩阵：向量运算、矩阵运算 6. **难度评估**（1-5级） - 1级：基础概念和简单计算 - 2级：常规题型和标准解法 - 3级：综合应用和中等难度 - 4级：较难题型和复杂推导 - 5级：压轴题和创新题型 **输出格式：** 返回严格的 JSON 格式： ```json { "subject": "数学", "grade": "年级（如果能识别）", "exam_type": "试卷类型（期中/期末/月考/模拟等）", "total_questions": 20, "total_score": 150, "questions": [ { "number": 1, "sub_number": null, "type": "choice/fill_blank/calculation/essay/proof", "question": "题目完整内容（包含LaTeX公式）", "options": ["A. $x=1$", "B. $x=2$", "C. $x=3$", "D. $x=4$"], "answer": "C", "solution": "详细解答步骤（如果试卷提供）", "analysis": "解题思路和方法总结", "difficulty": 3, "score": 5, "knowledge_points": ["一元二次方程", "因式分解"], "has_image": true, "image_description": "直角坐标系中的抛物线y=x^2，顶点在原点，开口向上", "page": 1, "formulas": ["$x^2 - 3x + 2 = 0$", "$x = \\frac{-b \\pm \\sqrt{b^2-4ac}}{2a}$"], "key_concepts": ["二次函数性质", "求根公式"] } ], "stats": { "choice_count": 10, "fill_blank_count": 5, "calculation_count": 3, "essay_count": 2, "difficulty_distribution": {"1": 2, "2": 5, "3": 8, "4": 4, "5": 1} } } ``` **特别注意事项：** - ✅ 所有数学符号必须用 LaTeX 表示，不要用纯文本 - ✅ 公式要完整准确，不能有语法错误 - ✅ 图形描述要详尽，包含所有关键信息 - ✅ 如果题目跨页，需要合并完整内容 - ✅ 保持题目原文，不要改写或简化 - ✅ 选项中的公式也要用 LaTeX 格式 - ❌ 不要遗漏任何题目或步骤 - ❌ 不要混淆题号和小题号 """ # 通用 prompt（适用于其他学科或模型） else: return base_prompt + """ **解析要求：** 1. **题目识别**：识别所有题目，包括选择题、填空题、计算题、解答题等 2. **题号提取**：准确提取题目编号（如：1、2、3 或 一、二、三） 3. **内容完整**：保留题目的完整内容，包括文字、公式、图片描述 4. **公式识别**：将数学公式转换为 LaTeX 格式，例如：$x^2 + y^2 = r^2$ 5. **图片描述**：详细描述题目中的图片内容（如几何图形、实验装置、化学结构式等） 6. **选项提取**：对于选择题，提取所有选项（A、B、C、D等） 7. **学科识别**：自动识别学科（数学、物理、化学、生物、英语等） 8. **难度评估**：根据题目内容评估难度（1-5） 9. **知识点标注**：标注涉及的主要知识点 **输出格式：** 返回 JSON 格式，结构如下： ```json { "subject": "数学/物理/化学/生物/英语等", "grade": "年级（如果能识别）", "total_questions": 20, "questions": [ { "number": 1, "type": "choice/fill_blank/calculation/essay", "question": "题目完整内容（包含LaTeX公式）", "options": ["A. 选项1", "B. 选项2", "C. 选项3", "D. 选项4"], "answer": "C", "analysis": "解析说明（如果试卷中提供）", "difficulty": 3, "knowledge_points": ["知识点1", "知识点2"], "has_image": true, "image_description": "图片详细描述", "page": 1, "formulas": ["$x^2 + y^2 = r^2$"] } ], "stats": { "choice_count": 10, "fill_blank_count": 5, "calculation_count": 3, "essay_count": 2 } } ``` **注意事项：** - 如果题目跨页，需要合并完整内容 - 公式务必使用 LaTeX 格式 - 图片要详细描述，包括关键信息 - 题号要准确，即使格式不统一 - 保持题目原文，不要改写 """ if subject: return base_prompt + f"\n**预设学科：** {subject}" if grade: return base_prompt + f"\n**预设年级：** {grade}" return base_prompt def parse_exam_file( self, file_path: str, subject: Optional[str] = None, grade: Optional[str] = None ) -> Dict: """ 解析试卷文件（支持 PDF、Word、图片） Args: file_path: 文件路径 subject: 学科（可选，AI会自动识别） grade: 年级（可选） Returns: 解析结果，包含题目列表和元数据 """ file_ext = Path(file_path).suffix.lower() if file_ext == '.pdf': return self._parse_pdf(file_path, subject, grade) elif file_ext in ['.doc', '.docx']: return self._parse_word(file_path, subject, grade) elif file_ext in ['.jpg', '.jpeg', '.png', '.bmp']: return self._parse_image(file_path, subject, grade) else: raise ValueError(f"不支持的文件格式: {file_ext}") def _parse_pdf( self, pdf_path: str, subject: Optional[str] = None, grade: Optional[str] = None ) -> Dict: """ 解析 PDF 试卷 策略：将每页转为高清图片，使用 Claude 的视觉能力解析 """ doc = fitz.open(pdf_path) total_pages = len(doc) # 存储所有页面的图片 page_images = [] logger.info(f"📄 开始解析 PDF，共 {total_pages} 页...") for page_num in range(total_pages): page = doc[page_num] # 使用高分辨率渲染（300 DPI） mat = fitz.Matrix(300/72, 300/72) pix = page.get_pixmap(matrix=mat) # 转换为 PIL Image img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) # 转为 base64 buffer = io.BytesIO() img.save(buffer, format='PNG', optimize=True, quality=95) img_base64 = base64.standard_b64encode(buffer.getvalue()).decode() page_images.append({ "page": page_num + 1, "image": img_base64 }) logger.info(f" ✓ 第 {page_num + 1}/{total_pages} 页已转换") doc.close() # 使用 Claude 解析所有页面 return self._parse_with_claude(page_images, subject, grade, total_pages) def _parse_word( self, word_path: str, subject: Optional[str] = None, grade: Optional[str] = None ) -> Dict: """ 解析 Word 试卷 策略：提取文本和图片，组合后发送给 Claude """ doc = Document(word_path) # 提取内容（文本 + 图片） content_blocks = [] logger.info(f"📝 开始解析 Word 文档...") for element in doc.element.body: if isinstance(element, CT_P): # 段落 para = Paragraph(element, doc) text = para.text.strip() if text: content_blocks.append({ "type": "text", "content": text }) # 检查段落中的图片 for run in para.runs: for drawing in run.element.findall('.//a:blip', namespaces={'a': 'http://schemas.openxmlformats.org/drawingml/2006/main'}): image_id = drawing.get('{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed') image_part = doc.part.related_parts[image_id] image_bytes = image_part.blob # 转为 base64 img_base64 = base64.standard_b64encode(image_bytes).decode() content_blocks.append({ "type": "image", "content": img_base64 }) logger.info(f" ✓ 提取了 {len(content_blocks)} 个内容块") # 使用 Claude 解析 return self._parse_word_content(content_blocks, subject, grade) def _parse_image( self, image_path: str, subject: Optional[str] = None, grade: Optional[str] = None ) -> Dict: """ 解析图片试卷 """ logger.info(f"🖼️ 开始解析图片试卷...") # 读取图片并转为 base64 with open(image_path, 'rb') as f: img_bytes = f.read() img_base64 = base64.standard_b64encode(img_bytes).decode() page_images = [{ "page": 1, "image": img_base64 }] return self._parse_with_claude(page_images, subject, grade, 1) def _parse_with_claude( self, page_images: List[Dict], subject: Optional[str], grade: Optional[str], total_pages: int ) -> Dict: """ 使用 Claude API 解析试卷图片 """ model_name = "Gemini 3 Pro" if self.is_gemini else "Claude Sonnet 4.5" logger.info(f"\n🤖 正在调用 {model_name} 进行智能解析...") if self.is_gemini and subject and "数学" in subject: logger.info(f"💡 启用数学专项优化模式...") # 构建消息内容 content = [] # 使用优化的 prompt instruction = self._get_optimized_prompt(subject, grade, total_pages) content.append({ "type": "text", "text": instruction }) # 添加所有页面图片 for page_data in page_images: content.append({ "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": page_data["image"] } }) # 构建 OpenRouter 格式的消息 # OpenRouter 使用 OpenAI 格式的 API messages = [] # 将内容转换为 OpenRouter 格式 user_content = [] for item in content: if item["type"] == "text": user_content.append({ "type": "text", "text": item["text"] }) elif item["type"] == "image": user_content.append({ "type": "image_url", "image_url": { "url": f"data:{item['source']['media_type']};base64,{item['source']['data']}" } }) messages.append({ "role": "user", "content": user_content }) # 调用 OpenRouter API try: headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "HTTP-Referer": "https://github.com/your-repo", # 可选，用于排名 "X-Title": "PaperExamParser" # 可选，显示在排名中 } payload = { "model": self.model, "messages": messages, "max_tokens": 16000, "temperature": 0 } logger.info(f"📤 发送请求到 OpenRouter...") logger.info(f" URL: {self.api_url}") logger.info(f" 模型: {self.model}") logger.info(f" 图片数量: {len(page_images)}") response = requests.post( self.api_url, headers=headers, json=payload, timeout=300 # 5分钟超时 ) logger.info(f"📥 收到响应: HTTP {response.status_code}") response.raise_for_status() result_data = response.json() # 解析响应 response_text = result_data["choices"][0]["message"]["content"] logger.info(f"📄 响应内容长度: {len(response_text)} 字符") # 提取 JSON（可能被包裹在 markdown 代码块中） json_match = response_text.strip() # 移除 markdown 代码块标记 if json_match.startswith("```json"): json_match = json_match[7:] # 移除 ```json elif json_match.startswith("```"): json_match = json_match[3:] # 移除 ``` if json_match.endswith("```"): json_match = json_match[:-3] # 移除结尾的 ``` json_match = json_match.strip() logger.info(f"📝 提取的JSON长度: {len(json_match)} 字符") logger.info(f"📝 JSON前100字符: {json_match[:100]}") logger.info(f"📝 JSON后100字符: {json_match[-100:]}") # 修复JSON中的转义问题（LaTeX公式中的反斜杠、引号等） import re def fix_json_escaping(json_str): r""" 修复JSON字符串中的转义问题 主要处理LaTeX公式中的反斜杠（如 \frac, \sqrt 等） 策略：在字符串值中，转义所有反斜杠，但保留已知的有效JSON转义序列 """ result = [] i = 0 in_string = False while i < len(json_str): char = json_str[i] # 检测字符串的开始和结束（非转义的引号） if char == '"': # 检查前面是否有反斜杠 num_backslashes = 0 j = i - 1 while j >= 0 and json_str[j] == '\\': num_backslashes += 1 j -= 1 # 偶数个反斜杠（包括0）表示引号未被转义 if num_backslashes % 2 == 0: in_string = not in_string result.append(char) i += 1 continue # 在字符串内部处理反斜杠 if in_string and char == '\\': if i + 1 < len(json_str): next_char = json_str[i + 1] # 已经正确转义的反斜杠 \\ if next_char == '\\': result.append('\\\\') i += 2 continue # 转义的引号 \" if next_char == '"': result.append('\\"') i += 2 continue # 转义的斜杠 \/ if next_char == '/': result.append('\\/') i += 2 continue # Unicode转义 \uXXXX if next_char == 'u' and i + 5 < len(json_str): hex_chars = json_str[i+2:i+6] if len(hex_chars) == 4 and all(c in '0123456789abcdefABCDEF' for c in hex_chars): # 有效的Unicode转义，保持原样 result.append('\\u') result.append(hex_chars) i += 6 continue # 其他所有情况：这是LaTeX命令，需要转义反斜杠 # 包括 \frac, \sqrt, \ne, \theta, \begin, 等等 # 也包括 \b, \f, \n, \r, \t (在数学公式中是LaTeX，不是JSON转义) result.append('\\\\') result.append(next_char) i += 2 else: # 字符串末尾的反斜杠，转义它 result.append('\\\\') i += 1 else: result.append(char) i += 1 return ''.join(result) # 使用更宽松的JSON解析 try: result = json.loads(json_match) except json.JSONDecodeError as e: logger.warning(f"⚠️ 标准JSON解析失败，尝试修复...") logger.info(f" 错误位置: 行{e.lineno}, 列{e.colno}") logger.info(f" 错误信息: {str(e)}") # 尝试修复JSON转义问题 try: fixed_json = fix_json_escaping(json_match) logger.info(f" 已修复JSON转义，重新尝试解析...") result = json.loads(fixed_json) except json.JSONDecodeError as e2: logger.error(f" 修复后仍然失败: {str(e2)}") logger.error(f" 尝试使用宽松模式...") # 保存修复后的内容用于调试 import tempfile with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f: f.write(fixed_json) debug_file = f.name logger.error(f" 修复后的JSON已保存到: {debug_file}") # Also save original for comparison with tempfile.NamedTemporaryFile(mode='w', suffix='_original.json', delete=False, encoding='utf-8') as f: f.write(json_match) orig_file = f.name logger.error(f" 原始JSON已保存到: {orig_file}") raise e2 logger.info(f"✅ 解析成功！共识别 {result.get('total_questions', 0)} 道题目") logger.info(f" 学科: {result.get('subject', '未识别')}") logger.info(f" 统计: {result.get('stats', {})}") return { "success": True, "result": result, "raw_response": response_text } except requests.exceptions.Timeout as e: logger.error(f"❌ API 请求超时: {e}") logger.error(f"请求URL: {self.api_url}") logger.error(f"模型: {self.model}") return { "success": False, "error": f"API 请求超时: {str(e)}" } except requests.exceptions.ConnectionError as e: logger.error(f"❌ 网络连接错误: {e}") logger.error(f"请求URL: {self.api_url}") return { "success": False, "error": f"无法连接到OpenRouter服务: {str(e)}" } except requests.exceptions.HTTPError as e: logger.error(f"❌ HTTP错误: {e}") logger.error(f"响应状态码: {e.response.status_code if e.response else 'N/A'}") logger.error(f"响应内容: {e.response.text if e.response else 'N/A'}") return { "success": False, "error": f"API返回错误 (HTTP {e.response.status_code if e.response else 'N/A'}): {e.response.text if e.response else str(e)}" } except requests.exceptions.RequestException as e: logger.error(f"❌ API 请求错误: {e}") return { "success": False, "error": f"API 请求失败: {str(e)}" } except json.JSONDecodeError as e: logger.error(f"❌ JSON 解析错误: {e}") logger.error(f"原始响应: {response_text}") return { "success": False, "error": "返回的格式无法解析", "raw_response": response_text } except Exception as e: logger.error(f"❌ 未知错误: {e}", exc_info=True) return { "success": False, "error": str(e) } def _parse_word_content( self, content_blocks: List[Dict], subject: Optional[str], grade: Optional[str] ) -> Dict: """ 解析 Word 文档内容 """ logger.info(f"\n🤖 正在调用 Claude Sonnet 4.5 进行智能解析...") # 构建消息内容 content = [] # 添加系统提示（同上） instruction = f"""你是一个专业的试卷解析助手。请分析这份 Word 试卷，并提取所有题目信息。 **解析要求：**（同上，省略...） 返回 JSON 格式的解析结果。 """ if subject: instruction += f"\n**预设学科：** {subject}" if grade: instruction += f"\n**预设年级：** {grade}" content.append({ "type": "text", "text": instruction }) # 添加所有内容块 for block in content_blocks: if block["type"] == "text": content.append({ "type": "text", "text": block["content"] }) elif block["type"] == "image": content.append({ "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": block["content"] } }) # 使用相同的 OpenRouter API 调用逻辑 # 构建 OpenRouter 格式的消息 messages = [] user_content = [] for item in content: if item["type"] == "text": user_content.append({ "type": "text", "text": item["text"] }) elif item["type"] == "image": user_content.append({ "type": "image_url", "image_url": { "url": f"data:{item['source']['media_type']};base64,{item['source']['data']}" } }) messages.append({ "role": "user", "content": user_content }) # 调用 OpenRouter API try: headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "HTTP-Referer": "https://github.com/your-repo", "X-Title": "PaperExamParser" } payload = { "model": self.model, "messages": messages, "max_tokens": 16000, "temperature": 0 } response = requests.post( self.api_url, headers=headers, json=payload, timeout=300 ) response.raise_for_status() result_data = response.json() response_text = result_data["choices"][0]["message"]["content"] # 提取 JSON json_match = response_text if "```json" in response_text: json_match = response_text.split("```json")[1].split("```")[0].strip() elif "```" in response_text: json_match = response_text.split("```")[1].split("```")[0].strip() result = json.loads(json_match) logger.info(f"✅ 解析成功！共识别 {result.get('total_questions', 0)} 道题目") return { "success": True, "result": result, "raw_response": response_text } except Exception as e: logger.info(f"❌ 解析错误: {e}") return { "success": False, "error": str(e) } # 便捷函数 def parse_exam( file_path: str, api_key: Optional[str] = None, subject: Optional[str] = None, grade: Optional[str] = None ) -> Dict: """ 便捷的试卷解析函数 Args: file_path: 试卷文件路径 api_key: Claude API Key subject: 学科 grade: 年级 Returns: 解析结果 Example: >>> result = parse_exam("math_exam.pdf", subject="数学", grade="高一") >>> logger.info(result["result"]["total_questions"]) """ parser = ClaudeExamParser(api_key=api_key) return parser.parse_exam_file(file_path, subject, grade) if __name__ == "__main__": # 测试代码 import sys if len(sys.argv) < 2: logger.info("用法: python claude_exam_parser.py <试卷文件路径> [学科] [年级]") sys.exit(1) file_path = sys.argv[1] subject = sys.argv[2] if len(sys.argv) > 2 else None grade = sys.argv[3] if len(sys.argv) > 3 else None result = parse_exam(file_path, subject=subject, grade=grade) if result["success"]: logger.info("\n" + "="*60) logger.info("解析结果：") logger.info(json.dumps(result["result"], ensure_ascii=False, indent=2)) else: logger.info(f"\n❌ 解析失败: {result['error']}")