gaoandQwen-Coder
2dc75d6a63
- Implement Claude Sonnet 4.5 based exam parser with multimodal understanding - Create enhanced dashboard with analytics, AI insights and knowledge tree - Add new UI components (AIAnalysisPanel, KnowledgeTreePanel, etc.) - Improve paper viewer with question extraction and preview panels - Add .env.example with configuration settings Co-authored-by: Qwen-Coder <[email protected]>
578 lines
23 KiB
Python
578 lines
23 KiB
Python
"""
|
||
增强版PDF试卷解析器 - 专门解决题号丢失问题
|
||
核心优化：
|
||
1. 改进题号识别正则表达式，支持更灵活的格式
|
||
2. 实现题号补全机制，自动修复遗漏的题目
|
||
3. 优化页眉页脚过滤，避免误删题目
|
||
4. 增强跨页题目处理能力
|
||
5. 添加详细的调试日志
|
||
"""
|
||
|
||
import re
|
||
import json
|
||
import math
|
||
import time
|
||
import os
|
||
import unicodedata
|
||
from collections import Counter, defaultdict
|
||
from typing import List, Dict, Optional, Tuple
|
||
import fitz # PyMuPDF
|
||
import pdfplumber
|
||
|
||
class PDFExamParserEnhancedV2:
|
||
def __init__(self):
|
||
# 全角转半角映射表
|
||
self.fullwidth_trans = str.maketrans(
|
||
"０１２３４５６７８９．（）ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ",
|
||
"0123456789.()ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
|
||
)
|
||
|
||
# 改进的题号正则表达式 - 更灵活的识别
|
||
# 支持格式：1. 1) 1、 1． 一、 （严格限制，避免匹配子题）
|
||
self.question_patterns = [
|
||
# 主要题号格式（行首）- 最高优先级，匹配数字后跟标点符号
|
||
re.compile(r'^\s*(?:第\s*)?([0-9]{1,3}|[一二三四五六七八九十]{1,3})\s*[.)．、]'),
|
||
# 特殊格式：数字后直接跟文字（如"1 解方程"）
|
||
re.compile(r'^\s*([0-9]{1,3})\s*(?:解方程:|计算：|化简：|综合与实践|如图|已知|求证|解答|证明)'),
|
||
# 大题号格式（一、二、等）- 高优先级，严格匹配行首
|
||
re.compile(r'^\s*([一二三四五六七八九十]{1,3})\s*[、.]'),
|
||
# 解方程等特殊格式 - 最低优先级
|
||
re.compile(r'^\s*(?:解方程：|计算：|化简：)([0-9]{1,2})\s*[.]?'),
|
||
]
|
||
|
||
# 页面标题正则（可能干扰题号识别的文本）
|
||
self.page_header_patterns = [
|
||
re.compile(r'试卷第\s*\d+\s*页[，,]\s*共\s*\d+\s*页'),
|
||
re.compile(r'\d{4}\s*年\s*\d{1,2}\s*月\s*\d{1,2}\s*日'),
|
||
re.compile(r'学校.*班级.*姓名.*考号'),
|
||
re.compile(r'密.*封.*线.*内.*不.*要.*答.*题'),
|
||
]
|
||
|
||
# 选项正则 - 支持更多格式
|
||
self.option_patterns = [
|
||
re.compile(r'^\s*([A-H])\s*[\.．、\)）]'), # A. xxx
|
||
re.compile(r'([A-H])\s*[\.．、\)）]\s*([^A-H]*)$'), # 行内选项
|
||
re.compile(r'\(([A-H])\)\s*([^A-H]*)'), # (A) xxx
|
||
]
|
||
|
||
# 子题正则
|
||
self.subquestion_pattern = re.compile(r'\(([0-9]+)\)')
|
||
|
||
# 调试开关
|
||
self.debug = True
|
||
|
||
def log(self, message: str, level: str = "INFO"):
|
||
"""调试日志"""
|
||
if self.debug:
|
||
timestamp = time.strftime("%H:%M:%S")
|
||
print(f"[{timestamp}] [{level}] {message}")
|
||
|
||
def normalize_text(self, text: str) -> str:
|
||
"""文本标准化处理"""
|
||
if not text:
|
||
return ""
|
||
|
||
# 全角转半角
|
||
text = text.translate(self.fullwidth_trans)
|
||
|
||
# Unicode规范化
|
||
text = unicodedata.normalize("NFKC", text)
|
||
|
||
# 替换特殊空格和标点
|
||
text = text.replace("\u00A0", " ")
|
||
text = text.replace("．", ".").replace("：", ":")
|
||
text = text.replace("（", "(").replace("）", ")")
|
||
|
||
# 清理多余空格
|
||
text = re.sub(r'\s+', ' ', text)
|
||
|
||
return text.strip()
|
||
|
||
def is_page_header(self, text: str) -> bool:
|
||
"""判断是否为页眉内容"""
|
||
for pattern in self.page_header_patterns:
|
||
if pattern.search(text):
|
||
return True
|
||
return False
|
||
|
||
def extract_question_number(self, text: str) -> Optional[str]:
|
||
"""从文本中提取题号"""
|
||
# 调试：检查特定的文本
|
||
if "1 解方程" in text:
|
||
self.log(f"调试：找到匹配文本 - '{text}'")
|
||
|
||
for i, pattern in enumerate(self.question_patterns):
|
||
match = pattern.search(text)
|
||
if match:
|
||
number = match.group(1)
|
||
self.log(f"题号识别成功: '{number}' (模式{i+1}) - 原文: '{text}'")
|
||
return number
|
||
return None
|
||
|
||
def extract_lines_smart(self, pdf_path: str) -> List[Dict]:
|
||
"""智能文本行提取 - 改进页眉页脚过滤"""
|
||
try:
|
||
doc = fitz.open(pdf_path)
|
||
except Exception as e:
|
||
self.log(f"无法打开PDF文件: {pdf_path}, 错误: {str(e)}", "ERROR")
|
||
return []
|
||
|
||
lines = []
|
||
all_headers = []
|
||
all_footers = []
|
||
|
||
self.log(f"开始处理PDF: {pdf_path}")
|
||
self.log(f"总页数: {len(doc)}")
|
||
|
||
# First, try the original method with blocks
|
||
try:
|
||
# 第一轮：收集页眉页脚候选
|
||
for page_num in range(len(doc)):
|
||
page = doc[page_num]
|
||
blocks = page.get_text("blocks")
|
||
page_height = page.rect.height
|
||
|
||
for block in blocks:
|
||
# PyMuPDF blocks format: (x0, y0, x1, y1, text, block_no, block_type)
|
||
# The text is at index 4
|
||
if len(block) >= 5:
|
||
x0, y0, x1, y1 = block[0], block[1], block[2], block[3]
|
||
text = block[4]
|
||
|
||
# Skip non-text blocks
|
||
if not isinstance(text, str) or not text.strip():
|
||
continue
|
||
|
||
text = self.normalize_text(text)
|
||
|
||
if not text or len(text) < 3:
|
||
continue
|
||
|
||
# 收集页眉候选（顶部15%）
|
||
if y0 <= page_height * 0.15:
|
||
all_headers.append(text)
|
||
|
||
# 收集页脚候选（底部10%）
|
||
if y1 >= page_height * 0.90:
|
||
all_footers.append(text)
|
||
|
||
# 分析页眉页脚
|
||
header_candidates = {text for text, count in Counter(all_headers).items() if count >= 2}
|
||
footer_candidates = {text for text, count in Counter(all_footers).items() if count >= 2}
|
||
|
||
self.log(f"识别页眉: {list(header_candidates)[:5]}") # Show first 5 only
|
||
self.log(f"识别页脚: {list(footer_candidates)[:5]}") # Show first 5 only
|
||
|
||
# 第二轮：提取有效文本行
|
||
for page_num in range(len(doc)):
|
||
page = doc[page_num]
|
||
blocks = page.get_text("blocks")
|
||
page_height = page.rect.height
|
||
|
||
for block in blocks:
|
||
# PyMuPDF blocks format: (x0, y0, x1, y1, text, block_no, block_type)
|
||
if len(block) >= 5:
|
||
x0, y0, x1, y1 = block[0], block[1], block[2], block[3]
|
||
text = block[4]
|
||
|
||
# Skip non-text blocks
|
||
if not isinstance(text, str) or not text.strip():
|
||
continue
|
||
|
||
text = self.normalize_text(text)
|
||
|
||
if not text:
|
||
continue
|
||
|
||
# 过滤页眉页脚
|
||
is_header = (y0 <= page_height * 0.15 and text in header_candidates)
|
||
is_footer = (y1 >= page_height * 0.90 and text in footer_candidates)
|
||
|
||
if is_header:
|
||
self.log(f"过滤页眉: {text[:30]}...")
|
||
continue
|
||
if is_footer:
|
||
self.log(f"过滤页脚: {text[:30]}...")
|
||
continue
|
||
|
||
# 按行处理 - split the block text into individual lines
|
||
for line in text.split('\n'):
|
||
line = line.strip()
|
||
if line and len(line) > 1:
|
||
lines.append({
|
||
"page": page_num + 1,
|
||
"y": y0,
|
||
"x": x0,
|
||
"text": line
|
||
})
|
||
|
||
except Exception as e:
|
||
self.log(f"按blocks提取失败: {str(e)}, 尝试按页面提取", "WARNING")
|
||
# If blocks method fails, try alternative method
|
||
for page_num in range(len(doc)):
|
||
page = doc[page_num]
|
||
try:
|
||
# Extract text for the page
|
||
text = page.get_text()
|
||
if text:
|
||
for line_num, line in enumerate(text.split('\n')):
|
||
line = self.normalize_text(line)
|
||
if line and len(line) > 1:
|
||
lines.append({
|
||
"page": page_num + 1,
|
||
"y": line_num * 10, # Approximate y position
|
||
"x": 0, # Approximate x position
|
||
"text": line
|
||
})
|
||
except Exception as page_error:
|
||
self.log(f"页面 {page_num + 1} 提取失败: {str(page_error)}", "WARNING")
|
||
continue
|
||
|
||
doc.close()
|
||
|
||
# 按阅读顺序排序
|
||
lines.sort(key=lambda l: (l["page"], round(l["y"], 1), round(l["x"], 1)))
|
||
|
||
self.log(f"提取完成: {len(lines)} 行文本")
|
||
return lines
|
||
|
||
def restore_missing_questions(self, questions: List[Dict], original_text: str) -> List[Dict]:
|
||
"""智能补全遗漏的题目"""
|
||
if not questions:
|
||
return questions
|
||
|
||
self.log("开始题号补全检查...")
|
||
|
||
# 提取所有已识别的题号
|
||
existing_numbers = set()
|
||
for q in questions:
|
||
if q.get("number"):
|
||
try:
|
||
# 处理中文数字
|
||
if q["number"] in ["一", "二", "三", "四", "五", "六", "七", "八", "九", "十"]:
|
||
num_map = {"一":1, "二":2, "三":3, "四":4, "五":5, "六":6, "七":7, "八":8, "九":9, "十":10}
|
||
existing_numbers.add(num_map[q["number"]])
|
||
else:
|
||
existing_numbers.add(int(q["number"]))
|
||
except (ValueError, KeyError):
|
||
pass
|
||
|
||
self.log(f"已识别题号: {sorted(existing_numbers)}")
|
||
|
||
# 查找缺失的题号
|
||
if existing_numbers:
|
||
max_number = max(existing_numbers)
|
||
missing_numbers = [i for i in range(1, max_number + 1) if i not in existing_numbers]
|
||
|
||
# 只有当确实缺失题号时才补全
|
||
if missing_numbers and len(missing_numbers) > 0:
|
||
self.log(f"发现缺失题号: {missing_numbers}")
|
||
|
||
# 如果缺失的题号数量过多，说明可能是识别问题，跳过补全
|
||
if len(missing_numbers) <= max_number * 0.5: # 最多补全一半的题号
|
||
# 尝试从原文中补全缺失的题目
|
||
for missing_num in missing_numbers:
|
||
# 查找该题号在原文中的位置
|
||
pattern = rf'{missing_num}\s*[.)．、]'
|
||
matches = list(re.finditer(pattern, original_text))
|
||
|
||
if matches:
|
||
# 找到第一个匹配位置
|
||
start_pos = matches[0].start()
|
||
|
||
# 查找题目结束位置（下一个题号或文件结束）
|
||
next_num_pattern = rf'{missing_num + 1}\s*[.)．、]'
|
||
next_match = re.search(next_num_pattern, original_text[start_pos + 10:])
|
||
|
||
if next_match:
|
||
end_pos = start_pos + 10 + next_match.start()
|
||
else:
|
||
end_pos = len(original_text)
|
||
|
||
# 提取题目文本
|
||
question_text = original_text[start_pos:end_pos].strip()
|
||
|
||
if len(question_text) > 10: # 确保有足够的题目内容
|
||
self.log(f"补全题目{missing_num}: {question_text[:50]}...")
|
||
|
||
# 提取选项和答案
|
||
stem, options = self.extract_options_and_stem(question_text, True)
|
||
|
||
new_question = {
|
||
"number": str(missing_num),
|
||
"type": "single_choice" if options else "essay",
|
||
"text": stem,
|
||
"options": {opt["label"]: opt["text"] for opt in options},
|
||
"answer": "",
|
||
"explanation": "",
|
||
"source": "restored",
|
||
"page": 1 # 默认第1页
|
||
}
|
||
questions.append(new_question)
|
||
else:
|
||
self.log(f"缺失题号过多，跳过补全: {missing_numbers}")
|
||
|
||
# 按题号重新排序
|
||
def get_question_number(q):
|
||
try:
|
||
return int(q["number"]) if q["number"].isdigit() else 999
|
||
except:
|
||
return 999
|
||
|
||
questions.sort(key=get_question_number)
|
||
|
||
self.log(f"补全后共 {len(questions)} 道题目")
|
||
|
||
# 调试：显示所有题目的题号
|
||
question_numbers = [q.get('number', 'unknown') for q in questions]
|
||
self.log(f"所有题目题号: {question_numbers}")
|
||
return questions
|
||
|
||
def extract_options_and_stem(self, question_text: str, prefer_choice: bool) -> Tuple[str, List[Dict]]:
|
||
"""提取题干和选项"""
|
||
lines = question_text.split('\n')
|
||
options = []
|
||
stem_lines = []
|
||
|
||
i = 0
|
||
while i < len(lines):
|
||
line = lines[i].strip()
|
||
|
||
if not line:
|
||
i += 1
|
||
continue
|
||
|
||
# 检查是否为选项行
|
||
option_found = False
|
||
for pattern in self.option_patterns:
|
||
match = pattern.match(line)
|
||
if match:
|
||
option_found = True
|
||
label = match.group(1)
|
||
text = match.group(2) if len(match.groups()) > 1 else ""
|
||
|
||
# 如果选项文本不完整，查看下一行
|
||
if len(text) < 5 and i + 1 < len(lines):
|
||
next_line = lines[i + 1].strip()
|
||
if not re.match(r'^[A-H]', next_line) and len(next_line) > 0:
|
||
text += " " + next_line
|
||
i += 1
|
||
|
||
options.append({"label": label, "text": text.strip()})
|
||
break
|
||
|
||
if not option_found:
|
||
stem_lines.append(line)
|
||
|
||
i += 1
|
||
|
||
stem = "\n".join(stem_lines).strip()
|
||
|
||
# 清理题干中的题号
|
||
stem = re.sub(r'^\d+\s*[.)．、]\s*', '', stem)
|
||
|
||
return stem, options
|
||
|
||
def process_pdf(self, pdf_path: str) -> Dict:
|
||
"""主处理函数"""
|
||
start_time = time.time()
|
||
self.log("=" * 50)
|
||
self.log("开始增强版PDF解析")
|
||
|
||
# Check if PDF file exists and is accessible
|
||
if not pdf_path:
|
||
return {"success": False, "error": "PDF路径为空"}
|
||
|
||
if not os.path.exists(pdf_path):
|
||
return {"success": False, "error": f"PDF文件不存在: {pdf_path}"}
|
||
|
||
if not os.path.isfile(pdf_path):
|
||
return {"success": False, "error": f"路径不是文件: {pdf_path}"}
|
||
|
||
try:
|
||
# 1. 提取文本行
|
||
lines = self.extract_lines_smart(pdf_path)
|
||
|
||
if not lines:
|
||
# Try one more method - extracting as plain text
|
||
self.log("尝试备用提取方法...")
|
||
try:
|
||
doc = fitz.open(pdf_path)
|
||
for page_num in range(len(doc)):
|
||
page = doc[page_num]
|
||
page_text = page.get_text()
|
||
if page_text.strip():
|
||
# Split page text into lines
|
||
for line_num, line_text in enumerate(page_text.split('\n')):
|
||
line_text = self.normalize_text(line_text.strip())
|
||
if line_text and len(line_text) > 1:
|
||
lines.append({
|
||
"page": page_num + 1,
|
||
"y": line_num * 10, # Approximate position
|
||
"x": 0,
|
||
"text": line_text
|
||
})
|
||
doc.close()
|
||
|
||
if lines:
|
||
self.log(f"备用方法成功提取 {len(lines)} 行文本")
|
||
else:
|
||
return {"success": False, "error": "PDF内容为空或无法读取文本"}
|
||
except Exception as backup_error:
|
||
self.log(f"备用方法失败: {str(backup_error)}", "ERROR")
|
||
return {"success": False, "error": "无法提取PDF文本"}
|
||
|
||
# 2. 提取原始文本用于题号补全
|
||
original_text = "\n".join([line["text"] for line in lines])
|
||
|
||
# 3. 识别题目
|
||
questions = []
|
||
current_question = None # 初始化为None
|
||
|
||
for line in lines:
|
||
text = line["text"]
|
||
|
||
# 检查是否为新题目开始
|
||
question_number = self.extract_question_number(text)
|
||
|
||
if question_number:
|
||
# 保存当前题目
|
||
if current_question and current_question.get("lines"):
|
||
question = self.parse_question_from_lines(current_question)
|
||
if question:
|
||
questions.append(question)
|
||
|
||
# 开始新题目
|
||
current_question = {
|
||
"number": question_number,
|
||
"lines": [text],
|
||
"page": line["page"],
|
||
"start_y": line["y"]
|
||
}
|
||
self.log(f"发现新题目: {question_number}")
|
||
else:
|
||
# 继续当前题目（如果有的话）
|
||
if current_question:
|
||
current_question["lines"].append(text)
|
||
|
||
# 处理最后一个题目
|
||
if current_question and current_question.get("lines"):
|
||
question = self.parse_question_from_lines(current_question)
|
||
if question:
|
||
questions.append(question)
|
||
|
||
# 4. 题号补全
|
||
questions = self.restore_missing_questions(questions, original_text)
|
||
|
||
# 5. 重新编号
|
||
questions = self.renumber_questions(questions)
|
||
|
||
end_time = time.time()
|
||
processing_time = end_time - start_time
|
||
|
||
self.log(f"解析完成: {len(questions)} 道题目，耗时 {processing_time:.2f}s")
|
||
|
||
return {
|
||
"success": True,
|
||
"questions": questions,
|
||
"stats": {
|
||
"total_questions": len(questions),
|
||
"processing_time": processing_time,
|
||
"method": "enhanced_v2",
|
||
"original_lines": len(lines)
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
self.log(f"解析失败: {str(e)}", "ERROR")
|
||
return {
|
||
"success": False,
|
||
"error": str(e),
|
||
"questions": [],
|
||
"stats": {"method": "enhanced_v2_failed"}
|
||
}
|
||
|
||
def parse_question_from_lines(self, question_data: Dict) -> Optional[Dict]:
|
||
"""从行数据解析题目"""
|
||
lines = question_data["lines"]
|
||
if not lines:
|
||
return None
|
||
|
||
# 合并所有文本
|
||
full_text = "\n".join(lines)
|
||
|
||
# 判断题目类型
|
||
has_options = any(re.search(r'[A-H]\s*[\.．、\)）]', line) for line in lines)
|
||
is_choice = has_options and len(lines) > 1
|
||
|
||
# 提取题干和选项
|
||
if is_choice:
|
||
stem, options = self.extract_options_and_stem(full_text, True)
|
||
return {
|
||
"number": question_data.get("number", ""),
|
||
"type": "single_choice",
|
||
"text": stem,
|
||
"options": {opt["label"]: opt["text"] for opt in options},
|
||
"answer": "",
|
||
"explanation": "",
|
||
"page": question_data.get("page", 1),
|
||
"source": "auto_extracted"
|
||
}
|
||
else:
|
||
return {
|
||
"number": question_data.get("number", ""),
|
||
"type": "essay" if len(full_text) > 20 else "fill_blank",
|
||
"text": full_text,
|
||
"options": {},
|
||
"answer": "",
|
||
"explanation": "",
|
||
"page": question_data.get("page", 1),
|
||
"source": "auto_extracted"
|
||
}
|
||
|
||
def renumber_questions(self, questions: List[Dict]) -> List[Dict]:
|
||
"""重新编号题目"""
|
||
for i, question in enumerate(questions):
|
||
question["number"] = str(i + 1)
|
||
return questions
|
||
|
||
# 测试函数
|
||
def test_enhanced_parser():
|
||
"""测试增强版解析器"""
|
||
parser = PDFExamParserEnhancedV2()
|
||
|
||
# 测试文件路径
|
||
test_pdf = "/Users/brunogao/work/codes/pingan/pager/backend/uploads/1/7c6df72c-3386-4909-8e6e-0471930d1284_2025年9月2日初中数学作业.pdf"
|
||
|
||
if not os.path.exists(test_pdf):
|
||
print(f"❌ 测试文件不存在: {test_pdf}")
|
||
return
|
||
|
||
print("🧪 开始测试增强版解析器...")
|
||
result = parser.process_pdf(test_pdf)
|
||
|
||
if result["success"]:
|
||
print(f"""
|
||
📊 测试结果:
|
||
- 成功: ✅
|
||
- 题目数量: {result["stats"]["total_questions"]}
|
||
- 处理时间: {result["stats"]["processing_time"]:.2f}s
|
||
- 处理方法: {result["stats"]["method"]}
|
||
""")
|
||
|
||
# 显示前3道题目
|
||
print("\n📝 前3道题目详情:")
|
||
for i, q in enumerate(result["questions"][:3]):
|
||
print(f"\n题目{q['number']}:")
|
||
print(f"类型: {q['type']}")
|
||
print(f"题干: {q['text'][:100]}...")
|
||
if q.get('options'):
|
||
print("选项:")
|
||
for label, text in q['options'].items():
|
||
print(f" {label}. {text}")
|
||
else:
|
||
print(f"❌ 测试失败: {result['error']}")
|
||
|
||
if __name__ == "__main__":
|
||
import os
|
||
test_enhanced_parser() |