Claude CodeandClaude
fc41dd359f
• 新增增强版PDF解析器，基于ChatGPT5策略使用文本层解析 • 实现混合PDF拆分器，优先文本层解析，OCR作为备份 • 添加数学符号选项修复功能，解决选择题空选项问题 • 性能提升：0.01秒vs29秒(2900倍)，22+题目vs6题目(3.5倍) • 更新前端拆分界面，增加错误处理和用户体验优化 • 扩展数据库架构支持OCR相关字段 • 统一前后端配置管理，修复硬编码URL问题 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
510 lines
19 KiB
Python
510 lines
19 KiB
Python
"""
|
||
增强版PDF试卷解析器 - 基于ChatGPT5策略
|
||
核心改进：
|
||
1. 页眉页脚自动过滤
|
||
2. 智能选项识别（行内、行首、续写）
|
||
3. 文本标准化处理
|
||
4. 跨页题目处理
|
||
"""
|
||
|
||
import re
|
||
import json
|
||
import math
|
||
import time
|
||
import unicodedata
|
||
from collections import Counter, defaultdict
|
||
from typing import List, Dict, Optional, Tuple
|
||
import fitz # PyMuPDF
|
||
import pdfplumber
|
||
|
||
class PDFExamParserEnhanced:
|
||
def __init__(self):
|
||
# 全角转半角映射表
|
||
self.fullwidth_trans = str.maketrans(
|
||
"０１２３４５６７８９．（）ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ",
|
||
"0123456789.()ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
|
||
)
|
||
|
||
# 大题分区正则
|
||
self.heading_pattern = re.compile(r"^\s*[（(]?\s*([一二三四五六七八九十]{1,3})\s*[、.)]\s*(\S*题)")
|
||
|
||
# 题号正则
|
||
self.question_start_pattern = re.compile(r"^\s*(?:第\s*)?([0-9]{1,3}|[一二三四五六七八九十]{1,3})\s*[.)]")
|
||
|
||
# 选项正则
|
||
self.option_label = r"([A-H])\s*[\.．、\)）]"
|
||
self.option_line_begin = re.compile(r"^\s*" + self.option_label + r"\s*(.*)")
|
||
self.option_inline_finder = re.compile(self.option_label)
|
||
|
||
def normalize_text(self, text: str) -> str:
|
||
"""
|
||
文本标准化：全角转半角、Unicode规范化等
|
||
"""
|
||
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 = "\n".join([line.strip() for line in text.splitlines() if line.strip()])
|
||
|
||
return text
|
||
|
||
def extract_lines_with_pymupdf(self, pdf_path: str) -> List[Dict]:
|
||
"""
|
||
使用PyMuPDF提取文本行，包含页眉页脚过滤
|
||
"""
|
||
doc = fitz.open(pdf_path)
|
||
pages_data = []
|
||
header_candidates = []
|
||
footer_candidates = []
|
||
|
||
# 第一遍：收集所有文本块和页眉页脚候选
|
||
for page_num, page in enumerate(doc):
|
||
blocks = page.get_text("blocks")
|
||
page_height = page.rect.height
|
||
page_width = page.rect.width
|
||
|
||
local_blocks = []
|
||
|
||
for block in blocks:
|
||
x0, y0, x1, y1, text = block[0], block[1], block[2], block[3], block[4]
|
||
text = self.normalize_text(text)
|
||
|
||
if not text:
|
||
continue
|
||
|
||
block_info = {
|
||
"page": page_num + 1,
|
||
"x0": x0, "y0": y0, "x1": x1, "y1": y1,
|
||
"page_width": page_width, "page_height": page_height,
|
||
"text": text
|
||
}
|
||
local_blocks.append(block_info)
|
||
|
||
# 收集页眉页脚候选（顶部12%和底部10%）
|
||
if y0 <= page_height * 0.12 and len(text) <= 80:
|
||
header_candidates.append(text.strip())
|
||
if y1 >= page_height * 0.90 and len(text) <= 80:
|
||
footer_candidates.append(text.strip())
|
||
|
||
# 按阅读顺序排序（从上到下，从左到右）
|
||
local_blocks.sort(key=lambda b: (round(b["y0"], 1), round(b["x0"], 1)))
|
||
pages_data.append(local_blocks)
|
||
|
||
doc.close()
|
||
|
||
# 识别重复的页眉页脚
|
||
n_pages = len(pages_data)
|
||
min_occurrences = max(2, math.ceil(n_pages * 0.6))
|
||
|
||
header_common = {
|
||
text for text, count in Counter(header_candidates).items()
|
||
if count >= min_occurrences
|
||
}
|
||
footer_common = {
|
||
text for text, count in Counter(footer_candidates).items()
|
||
if count >= min_occurrences
|
||
}
|
||
|
||
print(f"🔍 识别页眉: {header_common}")
|
||
print(f"🔍 识别页脚: {footer_common}")
|
||
|
||
# 第二遍：提取文本行，过滤页眉页脚
|
||
lines = []
|
||
for page_blocks in pages_data:
|
||
for block in page_blocks:
|
||
text = block["text"]
|
||
|
||
# 过滤页眉页脚
|
||
is_header = (block["y0"] <= block["page_height"] * 0.12 and
|
||
text.strip() in header_common)
|
||
is_footer = (block["y1"] >= block["page_height"] * 0.90 and
|
||
text.strip() in footer_common)
|
||
|
||
if is_header or is_footer:
|
||
continue
|
||
|
||
# 将文本块拆分为行
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if line:
|
||
lines.append({
|
||
"page": block["page"],
|
||
"y": block["y0"],
|
||
"x": block["x0"],
|
||
"text": line
|
||
})
|
||
|
||
# 按页面和位置排序
|
||
lines.sort(key=lambda l: (l["page"], round(l["y"], 1), round(l["x"], 1)))
|
||
|
||
print(f"✅ PyMuPDF提取完成: {len(lines)}行")
|
||
return lines
|
||
|
||
def extract_lines_with_pdfplumber(self, pdf_path: str) -> List[Dict]:
|
||
"""
|
||
使用pdfplumber提取文本行（备用方案）
|
||
"""
|
||
lines = []
|
||
|
||
with pdfplumber.open(pdf_path) as pdf:
|
||
n_pages = len(pdf.pages)
|
||
top_texts = []
|
||
bottom_texts = []
|
||
|
||
# 收集页眉页脚候选
|
||
for i, page in enumerate(pdf.pages, start=1):
|
||
words = page.extract_words(x_tolerance=2, y_tolerance=2, keep_blank_chars=False)
|
||
if not words:
|
||
continue
|
||
|
||
# 按位置排序
|
||
words.sort(key=lambda w: (round(w["top"], 1), round(w["x0"], 1)))
|
||
|
||
# 按行分组
|
||
line_map = defaultdict(list)
|
||
for word in words:
|
||
line_map[round(word["top"], 1)].append(word)
|
||
|
||
sorted_tops = sorted(line_map.keys())
|
||
if sorted_tops:
|
||
# 第一行和最后一行
|
||
first_line = " ".join([
|
||
w["text"] for w in sorted(line_map[sorted_tops[0]], key=lambda x: x["x0"])
|
||
]).strip()
|
||
last_line = " ".join([
|
||
w["text"] for w in sorted(line_map[sorted_tops[-1]], key=lambda x: x["x0"])
|
||
]).strip()
|
||
|
||
if first_line:
|
||
top_texts.append(self.normalize_text(first_line))
|
||
if last_line:
|
||
bottom_texts.append(self.normalize_text(last_line))
|
||
|
||
# 识别重复的页眉页脚
|
||
min_occurrences = max(2, math.ceil(n_pages * 0.6))
|
||
header_common = {
|
||
text for text, count in Counter(top_texts).items()
|
||
if count >= min_occurrences
|
||
}
|
||
footer_common = {
|
||
text for text, count in Counter(bottom_texts).items()
|
||
if count >= min_occurrences
|
||
}
|
||
|
||
# 提取所有行，过滤页眉页脚
|
||
for i, page in enumerate(pdf.pages, start=1):
|
||
words = page.extract_words(x_tolerance=2, y_tolerance=2, keep_blank_chars=False)
|
||
if not words:
|
||
continue
|
||
|
||
# 按行分组
|
||
line_map = defaultdict(list)
|
||
for word in words:
|
||
line_map[round(word["top"], 1)].append(word)
|
||
|
||
for top in sorted(line_map.keys()):
|
||
word_list = sorted(line_map[top], key=lambda x: x["x0"])
|
||
text = self.normalize_text(" ".join(w["text"] for w in word_list))
|
||
|
||
if not text:
|
||
continue
|
||
|
||
# 过滤页眉页脚
|
||
if text in header_common or text in footer_common:
|
||
continue
|
||
|
||
lines.append({
|
||
"page": i,
|
||
"y": top,
|
||
"x": word_list[0]["x0"],
|
||
"text": text
|
||
})
|
||
|
||
lines.sort(key=lambda l: (l["page"], round(l["y"], 1), round(l["x"], 1)))
|
||
|
||
print(f"✅ pdfplumber提取完成: {len(lines)}行")
|
||
return lines
|
||
|
||
def extract_options_and_stem(self, question_text: str, prefer_choice: bool) -> Tuple[str, List[Dict]]:
|
||
"""
|
||
提取题干和选项，支持多种选项格式
|
||
"""
|
||
lines = question_text.splitlines()
|
||
options = []
|
||
current_option = None
|
||
stem_buffer = []
|
||
|
||
def save_current_option():
|
||
nonlocal current_option, options
|
||
if current_option is not None:
|
||
label, text = current_option
|
||
options.append({"label": label, "text": text.strip()})
|
||
current_option = None
|
||
|
||
for line in lines:
|
||
# 检测行内多选项格式：A. xxx；B. yyy；C. zzz
|
||
inline_matches = list(self.option_inline_finder.finditer(line))
|
||
if len(inline_matches) >= 2:
|
||
save_current_option()
|
||
for idx, match in enumerate(inline_matches):
|
||
label = match.group(1)
|
||
start_pos = match.end()
|
||
end_pos = inline_matches[idx + 1].start() if idx + 1 < len(inline_matches) else len(line)
|
||
option_text = line[start_pos:end_pos].strip(" ；;，, ")
|
||
if option_text:
|
||
options.append({"label": label, "text": option_text})
|
||
continue
|
||
|
||
# 检测行首选项格式：A. xxx
|
||
line_begin_match = self.option_line_begin.match(line)
|
||
if line_begin_match:
|
||
save_current_option()
|
||
label = line_begin_match.group(1)
|
||
text = line_begin_match.group(2).strip()
|
||
current_option = (label, text)
|
||
continue
|
||
|
||
# 选项续行
|
||
if current_option is not None:
|
||
current_option = (current_option[0], (current_option[1] + " " + line).strip())
|
||
else:
|
||
stem_buffer.append(line)
|
||
|
||
save_current_option()
|
||
|
||
stem = "\n".join(stem_buffer).strip()
|
||
|
||
# 如果不是选择题，保持原文
|
||
if not prefer_choice:
|
||
return (question_text.strip(), [])
|
||
|
||
# 选择题但没找到选项的情况，尝试贪婪提取
|
||
if prefer_choice and not options:
|
||
def extract_option_text(label: str, text: str) -> Optional[str]:
|
||
pattern = label + r"\s*[\.．、\)）]\s*(.*?)(?=(?:[；;]\s*)?[A-H]\s*[\.．、\)）]|$)"
|
||
match = re.search(pattern, text, re.S)
|
||
return match.group(1).strip() if match else None
|
||
|
||
for label in "ABCDEFGH":
|
||
option_text = extract_option_text(label, question_text)
|
||
if option_text:
|
||
options.append({"label": label, "text": re.sub(r"\s+", " ", option_text)})
|
||
|
||
# 只保留A-D选项
|
||
options = [opt for opt in options if opt["label"] in "ABCD" and opt["text"]]
|
||
|
||
if options:
|
||
# 从原文中清除选项部分，提取纯题干
|
||
clean_pattern = r"[A-D]\s*[\.．、\)）]\s*.*?(?=(?:[；;]\s*)?[A-D]\s*[\.．、\)）]|$)"
|
||
stem_clean = re.sub(clean_pattern, "", question_text, flags=re.S).strip()
|
||
stem = re.sub(r"[；;]+", " ", stem_clean).strip()
|
||
|
||
return (stem if stem else question_text.strip(), options)
|
||
|
||
def split_questions(self, lines: List[Dict]) -> List[Dict]:
|
||
"""
|
||
拆分题目
|
||
"""
|
||
questions = []
|
||
current_section = None
|
||
buffer = []
|
||
start_page = None
|
||
|
||
def flush_current_question():
|
||
nonlocal buffer, start_page, current_section, questions
|
||
if not buffer:
|
||
return
|
||
|
||
question_text = "\n".join([item["text"] for item in buffer]).strip()
|
||
first_line = buffer[0]["text"]
|
||
|
||
# 提取题号
|
||
match = self.question_start_pattern.search(first_line)
|
||
question_number = match.group(1) if match else None
|
||
|
||
# 判断是否为选择题
|
||
prefer_choice = bool(current_section and "选择" in current_section)
|
||
|
||
# 提取题干和选项
|
||
stem, options = self.extract_options_and_stem(question_text, prefer_choice)
|
||
|
||
# 推断题目类型
|
||
question_type = "subjective"
|
||
if current_section and "选择" in current_section:
|
||
question_type = "single_choice"
|
||
elif current_section and "填空" in current_section:
|
||
question_type = "fill_blank"
|
||
|
||
end_page = buffer[-1]["page"]
|
||
|
||
question_item = {
|
||
"section": current_section,
|
||
"qno": question_number,
|
||
"type": question_type,
|
||
"stem": stem,
|
||
"options": options,
|
||
"page_start": start_page,
|
||
"page_end": end_page
|
||
}
|
||
|
||
questions.append(question_item)
|
||
|
||
# 重置缓冲区
|
||
buffer = []
|
||
start_page = None
|
||
|
||
# 逐行处理
|
||
for line in lines:
|
||
text = line["text"]
|
||
|
||
# 检测大题分区
|
||
heading_match = self.heading_pattern.match(text)
|
||
if heading_match:
|
||
flush_current_question()
|
||
current_section = f"{heading_match.group(1)}、{heading_match.group(2)}"
|
||
print(f"📚 发现分区: {current_section}")
|
||
continue
|
||
|
||
# 检测题号
|
||
question_match = self.question_start_pattern.match(text)
|
||
if question_match:
|
||
flush_current_question()
|
||
buffer = [line]
|
||
start_page = line["page"]
|
||
print(f"📝 发现题目: {question_match.group(1)}")
|
||
else:
|
||
if buffer:
|
||
buffer.append(line)
|
||
# 跳过序言等内容
|
||
|
||
# 处理最后一题
|
||
flush_current_question()
|
||
|
||
# 清理无效题目
|
||
valid_questions = []
|
||
for question in questions:
|
||
# 去除题号前缀，检查是否有实际内容
|
||
content = re.sub(r"^\s*(\d+|[一二三四五六七八九十]+)\s*[.)]\s*", "", question["stem"] or "").strip()
|
||
if content:
|
||
valid_questions.append(question)
|
||
|
||
return valid_questions
|
||
|
||
def process_pdf(self, pdf_path: str) -> Dict:
|
||
"""
|
||
主处理函数
|
||
"""
|
||
print(f"🎯 开始增强PDF解析: {pdf_path}")
|
||
start_time = time.time()
|
||
|
||
try:
|
||
# 优先使用PyMuPDF
|
||
lines = self.extract_lines_with_pymupdf(pdf_path)
|
||
|
||
if not lines:
|
||
print("⚠️ PyMuPDF失败，尝试pdfplumber")
|
||
lines = self.extract_lines_with_pdfplumber(pdf_path)
|
||
|
||
if not lines:
|
||
return {"success": False, "error": "无法提取PDF文本"}
|
||
|
||
# 拆分题目
|
||
questions = self.split_questions(lines)
|
||
|
||
# 转换为API格式
|
||
formatted_questions = []
|
||
for q in questions:
|
||
# 转换选项格式
|
||
options_dict = {}
|
||
if q.get("options"):
|
||
for opt in q["options"]:
|
||
options_dict[opt["label"]] = opt["text"]
|
||
|
||
formatted_questions.append({
|
||
'number': q['qno'],
|
||
'type': q['type'],
|
||
'text': q['stem'],
|
||
'options': options_dict if options_dict else None,
|
||
'page': q['page_start'],
|
||
'bbox': [],
|
||
'confidence': 1.0,
|
||
'page_spans': list(range(q['page_start'], q['page_end'] + 1)),
|
||
'source_text': q['stem'],
|
||
'images': [],
|
||
'auto_extracted': True,
|
||
'section': q['section']
|
||
})
|
||
|
||
total_time = time.time() - start_time
|
||
print(f"🎉 增强解析完成！总耗时: {total_time:.2f}秒")
|
||
|
||
return {
|
||
"success": True,
|
||
"questions": formatted_questions,
|
||
"total_time": total_time,
|
||
"pages_processed": len(set(line["page"] for line in lines)),
|
||
"questions_found": len(formatted_questions),
|
||
"stats": {
|
||
"total_time": total_time,
|
||
"pages_processed": len(set(line["page"] for line in lines)),
|
||
"questions_found": len(formatted_questions),
|
||
"avg_time_per_page": total_time / len(set(line["page"] for line in lines)) if lines else 0,
|
||
"method": "enhanced_text_parsing"
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"❌ 增强解析失败: {str(e)}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# 测试函数
|
||
def test_enhanced_parser():
|
||
"""
|
||
测试增强解析器
|
||
"""
|
||
import os
|
||
import time
|
||
|
||
pdf_path = "/Users/brunogao/Downloads/深圳市八年级（上）期中数学试卷1.pdf"
|
||
|
||
if not os.path.exists(pdf_path):
|
||
print(f"❌ 测试文件不存在: {pdf_path}")
|
||
return
|
||
|
||
parser = PDFExamParserEnhanced()
|
||
result = parser.process_pdf(pdf_path)
|
||
|
||
if result["success"]:
|
||
print(f"""
|
||
📊 增强解析测试结果:
|
||
- 总时间: {result['total_time']:.2f}秒
|
||
- 页数: {result['pages_processed']}
|
||
- 题目: {result['questions_found']}道
|
||
- 方法: {result['stats']['method']}
|
||
""")
|
||
|
||
print("前8道题预览:")
|
||
for i, q in enumerate(result['questions'][:8]):
|
||
options_str = ""
|
||
if q.get('options'):
|
||
options_str = f" 选项: {', '.join([f'{k}. {v[:20]}...' for k, v in q['options'].items()])}"
|
||
|
||
print(f"题目{q['number']} ({q.get('type', 'unknown')}): {q['text'][:60]}...{options_str}")
|
||
else:
|
||
print(f"❌ 测试失败: {result['error']}")
|
||
|
||
if __name__ == "__main__":
|
||
import time
|
||
test_enhanced_parser() |