532 lines
17 KiB
Python
532 lines
17 KiB
Python
"""
|
||
chapter_segmenter.py
|
||
对外接口：segment_chapters(data) -> List[Chapter]
|
||
|
||
纯规则章节切分，不依赖大模型，不依赖 extract_util / entity_registry。
|
||
输入：content_list JSON（list of dict，每条含 type/text/text_level/page_idx/bbox 等字段）
|
||
输出：Chapter 列表，每个 Chapter 包含标题信息、所属 records、以及二级子章节。
|
||
|
||
多信号融合标题识别（v8）：
|
||
- 两遍扫描：先扫描所有候选标题 → 全局校验编号连续性 → 再切分
|
||
- 置信度融合：text_level + 编号格式 + 文本长度 + 标点密度
|
||
- 全局校验：编号跳跃过大的假标题自动剔除
|
||
"""
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import List, Dict, Optional, Set
|
||
|
||
|
||
# ==============================
|
||
# 配置（可在调用前修改模块级变量）
|
||
# ==============================
|
||
|
||
# 一级标题含以下关键词时整章跳过
|
||
FILTER_KEYWORDS: List[str] = [
|
||
"维修工作卡", "维修项目索引",
|
||
"常见故障", "维修项目表"
|
||
]
|
||
|
||
# 二级标题含以下关键词时跳过该二级节
|
||
FILTER_L2_KEYWORDS: List[str] = [
|
||
"常见故障表", "维修项目表", "维修工作卡","操作程序"
|
||
]
|
||
|
||
# 特殊章节关键词：一级标题含以下任一时，不作为整体实体，二级标题各自独立
|
||
SPECIAL_CHAPTER_KEYWORDS: List[str] = [
|
||
"组成、技术特征及接口",
|
||
"组成和技术特征",
|
||
"组成与技术特征",
|
||
"组成、技术特征",
|
||
]
|
||
|
||
# 目录关键词（去空格后匹配，兼容"目 次"/"目 录"等 OCR 带空格的写法）
|
||
TOC_KEYWORDS: List[str] = ["目次", "目录", "contents"]
|
||
|
||
# 置信度阈值：低于此值的不认为是标题
|
||
HEADING_CONFIDENCE_THRESHOLD: float = 0.5
|
||
|
||
# 一级标题编号最大允许跳跃数（连续性校验）
|
||
MAX_L1_GAP: int = 3
|
||
|
||
|
||
# ==============================
|
||
# 数据结构
|
||
# ==============================
|
||
|
||
@dataclass
|
||
class L2Section:
|
||
"""二级子章节"""
|
||
title: str # 去编号后的标题名
|
||
numbering: str # 编号，如 "3.2"
|
||
records: List[dict] = field(default_factory=list) # 该二级节的所有 records
|
||
|
||
|
||
@dataclass
|
||
class Chapter:
|
||
"""一级章节（含所有二级子节）"""
|
||
title: str # 去编号后的标题名（已应用 L1_TITLE_MAPPING 等由调用方处理）
|
||
raw_title: str # 原始文本（含编号）
|
||
numbering: str # 编号，如 "3"
|
||
is_special: bool # 是否特殊章节（二级各自独立）
|
||
is_skipped: bool # 是否被 FILTER_KEYWORDS 过滤
|
||
records: List[dict] = field(default_factory=list) # 本章所有 records（含标题行）
|
||
loose_records: List[dict] = field(default_factory=list) # 一级标题到第一个二级标题之间的 records
|
||
l2_sections: List[L2Section] = field(default_factory=list) # 二级子章节列表
|
||
|
||
|
||
@dataclass
|
||
class _HeadingNode:
|
||
"""内部用：统一的标题节点"""
|
||
raw_text: str
|
||
clean_title: str
|
||
numbering: str
|
||
level: int
|
||
index_in_data: int
|
||
confidence: float = 1.0
|
||
|
||
|
||
# ==============================
|
||
# 正则
|
||
# ==============================
|
||
|
||
_CHAPTER_NUM_RE = re.compile(r'^(\d{1,2}(?:\.\d{1,2})*\.?)')
|
||
_LIST_ITEM_RE = re.compile(r'^[\d]+[）\)]\s*')
|
||
_ALPHA_LIST_RE = re.compile(r'^[a-zA-Z][）\)\.]\s*')
|
||
_TOC_ENTRY_SUFFIX_RE = re.compile(r'([\s.]+\d{1,4}|\.{2,})\s*$')
|
||
|
||
|
||
# ==============================
|
||
# 标题工具函数
|
||
# ==============================
|
||
|
||
def _is_real_chapter_title(text: str) -> bool:
|
||
text = text.strip()
|
||
if _LIST_ITEM_RE.match(text):
|
||
return False
|
||
if _ALPHA_LIST_RE.match(text):
|
||
return False
|
||
if re.search(r'[。！？：]\s*$', text):
|
||
return False
|
||
chapter_num = r'^\d{1,2}(?:\.\d{1,2})*'
|
||
if re.match(chapter_num + r'\s+\S', text):
|
||
return True
|
||
if re.match(chapter_num + r'\.\s+\S', text):
|
||
return True
|
||
if re.match(chapter_num + r'\.?[\u4e00-\u9fff]', text):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _is_toc_entry_line(text: str) -> bool:
|
||
if '\n' in text:
|
||
return False
|
||
return bool(_TOC_ENTRY_SUFFIX_RE.search(text))
|
||
|
||
|
||
def _get_chapter_level(text: str) -> int:
|
||
match = _CHAPTER_NUM_RE.match(text.strip())
|
||
if match:
|
||
num = match.group(1).rstrip('.')
|
||
return num.count('.') + 1
|
||
return 0
|
||
|
||
|
||
def _get_chapter_numbering(text: str) -> str:
|
||
match = _CHAPTER_NUM_RE.match(text.strip())
|
||
return match.group(1).rstrip('.') if match else ""
|
||
|
||
|
||
def _clean_title(text: str) -> str:
|
||
return re.sub(r'^\d{1,2}(?:\.\d{1,2})*\.?\s*', '', text.strip()).strip()
|
||
|
||
|
||
def _should_skip_l2(title_clean: str) -> bool:
|
||
return any(kw in title_clean for kw in FILTER_L2_KEYWORDS)
|
||
|
||
|
||
def _is_special_chapter(title: str) -> bool:
|
||
return any(kw in title for kw in SPECIAL_CHAPTER_KEYWORDS)
|
||
|
||
|
||
# ==============================
|
||
# 目录页检测
|
||
# ==============================
|
||
|
||
def _is_toc_content(ins: dict) -> bool:
|
||
"""
|
||
判断是否是目录页内容（三层判断）。
|
||
|
||
层1：text_level>=1 且含目录关键词 → 直接判定
|
||
层2：严格目录条目格式（"文字..页码" ≥2次）
|
||
层3：宽松兜底：换行多 + '..'密集 + 排除正文列表/技术参数
|
||
"""
|
||
text = ins.get("text", "")
|
||
text_level = ins.get("text_level", 0)
|
||
|
||
text_nospace = text.replace(" ", "").replace("\u3000", "")
|
||
if text_level and any(kw in text_nospace.lower() for kw in TOC_KEYWORDS):
|
||
return True
|
||
|
||
toc_entry_re = re.compile(
|
||
r'[\u4e00-\u9fffa-zA-Z）\)]\s*\.{2,}\s*\.?\s*(\d{1,4})\s*$',
|
||
re.MULTILINE
|
||
)
|
||
if len(toc_entry_re.findall(text)) >= 2:
|
||
return True
|
||
|
||
if text.count('\n') > 3 and '..' in text:
|
||
has_list_markers = bool(re.search(
|
||
r'(?:^|\n)\s*(?:\d+[）\)]|[a-zA-Z][）\).])', text
|
||
))
|
||
if has_list_markers:
|
||
return False
|
||
range_pattern_count = len(re.findall(r'[\d℃%]\.\.[+\-\d]', text))
|
||
dot_count = len(re.findall(r'\.{2,}', text))
|
||
if range_pattern_count > 0 and range_pattern_count >= dot_count * 0.5:
|
||
return False
|
||
line_count = text.count('\n') + 1
|
||
if dot_count >= max(3, line_count * 0.3):
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def _get_toc_pages(data: list) -> Set[int]:
|
||
"""找出所有目录页的 page_idx，支持多页目录，并向后延伸。"""
|
||
toc_pages: Set[int] = set()
|
||
for ins in data:
|
||
if _is_toc_content(ins):
|
||
toc_pages.add(ins.get("page_idx", -1))
|
||
text = ins.get("text", "")
|
||
text_level = ins.get("text_level", 0)
|
||
text_nospace = text.replace(" ", "").replace("\u3000", "")
|
||
if text_level and any(kw in text_nospace.lower() for kw in TOC_KEYWORDS):
|
||
toc_pages.add(ins.get("page_idx", -1))
|
||
toc_pages.discard(-1)
|
||
|
||
if not toc_pages:
|
||
return toc_pages
|
||
|
||
pages_records: Dict[int, list] = {}
|
||
for ins in data:
|
||
p = ins.get("page_idx", -1)
|
||
if p < 0:
|
||
continue
|
||
pages_records.setdefault(p, []).append(ins)
|
||
|
||
max_toc = max(toc_pages)
|
||
for next_page in range(max_toc + 1, max_toc + 10):
|
||
if next_page not in pages_records:
|
||
break
|
||
records = pages_records[next_page]
|
||
text_records = [
|
||
r for r in records
|
||
if r.get("type") == "text" and (r.get("text") or "").strip()
|
||
]
|
||
if not text_records:
|
||
toc_pages.add(next_page)
|
||
continue
|
||
toc_like = sum(
|
||
1 for r in text_records
|
||
if _is_toc_content(r) or _is_toc_entry_line((r.get("text") or ""))
|
||
)
|
||
if toc_like == len(text_records):
|
||
toc_pages.add(next_page)
|
||
else:
|
||
break
|
||
|
||
return toc_pages
|
||
|
||
|
||
# ==============================
|
||
# 多信号融合标题识别
|
||
# ==============================
|
||
|
||
def _compute_heading_confidence(ins: dict, text: str, level: int) -> float:
|
||
"""
|
||
4个信号融合（0~1）：
|
||
text_level>=1（0.35）+ 编号格式规范（0.30）+ 长度合理（0.20）+ 标点密度低（0.15）
|
||
"""
|
||
score = 0.0
|
||
text_level = ins.get("text_level", 0)
|
||
|
||
if text_level >= 1:
|
||
score += 0.35
|
||
|
||
normalized = re.sub(r'\s+', ' ', text.replace('\u3000', ' '))
|
||
if re.match(r'^\d{1,2}(?:\.\d{1,2})*\.?\s+[\u4e00-\u9fffa-zA-Z]', normalized):
|
||
score += 0.30
|
||
elif re.match(r'^\d{1,2}(?:\.\d{1,2})*[\u4e00-\u9fff]', normalized):
|
||
score += 0.20
|
||
|
||
clean = _clean_title(text)
|
||
if 1 <= len(clean) <= 20:
|
||
score += 0.20
|
||
elif len(clean) <= 35:
|
||
score += 0.10
|
||
|
||
punct_count = len(re.findall(r'[，。；：、！？,.;:!?]', clean))
|
||
if punct_count <= 1:
|
||
score += 0.15
|
||
elif punct_count <= 3:
|
||
score += 0.05
|
||
|
||
return score
|
||
|
||
|
||
def _scan_all_headings(data: list, toc_pages: Set[int] = None) -> List[_HeadingNode]:
|
||
"""第一遍：识别所有置信度足够的标题节点。"""
|
||
headings = []
|
||
debug_rejected = []
|
||
|
||
for i, ins in enumerate(data):
|
||
if ins.get("type") != "text":
|
||
continue
|
||
if toc_pages and ins.get("page_idx") in toc_pages:
|
||
continue
|
||
|
||
text = ins.get("text", "").strip()
|
||
if not text:
|
||
continue
|
||
if _is_toc_content(ins):
|
||
debug_rejected.append(f" [i={i}] 目录页跳过: {text[:40]!r}")
|
||
continue
|
||
if _is_toc_entry_line(text):
|
||
debug_rejected.append(f" [i={i}] 目录条目行: {text[:60]!r}")
|
||
continue
|
||
if not _is_real_chapter_title(text):
|
||
first_char = text[0] if text else ''
|
||
if first_char.isdigit() and len(debug_rejected) < 50:
|
||
debug_rejected.append(f" [i={i}] 非章节标题格式: {text[:60]!r}")
|
||
continue
|
||
|
||
level = _get_chapter_level(text)
|
||
if level == 0:
|
||
continue
|
||
|
||
numbering = _get_chapter_numbering(text)
|
||
clean = _clean_title(text)
|
||
confidence = _compute_heading_confidence(ins, text, level)
|
||
|
||
if confidence >= HEADING_CONFIDENCE_THRESHOLD:
|
||
headings.append(_HeadingNode(
|
||
raw_text=text,
|
||
clean_title=clean,
|
||
numbering=numbering,
|
||
level=level,
|
||
index_in_data=i,
|
||
confidence=confidence,
|
||
))
|
||
else:
|
||
debug_rejected.append(
|
||
f" [i={i}] 置信度不足({confidence:.2f}): {text[:60]!r} "
|
||
f"(text_level={ins.get('text_level', 0)})"
|
||
)
|
||
|
||
print(f"\n [DEBUG] data总条数={len(data)}, "
|
||
f"text类型条数={sum(1 for x in data if x.get('type')=='text')}")
|
||
print(f" [DEBUG] 候选标题={len(headings)}, 被拒绝={len(debug_rejected)}")
|
||
if debug_rejected:
|
||
print(" [DEBUG] 被拒绝样本（最多20条）:")
|
||
for msg in debug_rejected[:20]:
|
||
print(msg)
|
||
|
||
return headings
|
||
|
||
|
||
def _validate_heading_sequence(headings: List[_HeadingNode]) -> List[_HeadingNode]:
|
||
"""第二遍：全局校验编号连续性，去除可疑标题并去重。"""
|
||
if not headings:
|
||
return headings
|
||
|
||
level1_nums = sorted(set(
|
||
int(h.numbering) for h in headings if h.level == 1
|
||
))
|
||
|
||
valid_level1: Set[int] = set()
|
||
if level1_nums:
|
||
for i, num in enumerate(level1_nums):
|
||
if i == 0:
|
||
valid_level1.add(num)
|
||
continue
|
||
gap = num - level1_nums[i - 1]
|
||
if gap <= MAX_L1_GAP:
|
||
valid_level1.add(num)
|
||
else:
|
||
print(f" ⚠️ 一级标题编号跳跃过大: {level1_nums[i-1]} → {num}，丢弃 {num}")
|
||
|
||
validated = []
|
||
for h in headings:
|
||
if h.level == 1:
|
||
if int(h.numbering) in valid_level1:
|
||
validated.append(h)
|
||
elif h.level >= 2:
|
||
parent_num = int(h.numbering.split('.')[0])
|
||
if parent_num in valid_level1:
|
||
validated.append(h)
|
||
else:
|
||
validated.append(h)
|
||
|
||
def _has_lower_l2_after(h: _HeadingNode) -> bool:
|
||
if h.level != 1:
|
||
return False
|
||
h_num = int(h.numbering)
|
||
after_l2 = [x for x in validated if x.index_in_data > h.index_in_data and x.level == 2]
|
||
for x in after_l2[:8]:
|
||
parent = int(x.numbering.split('.')[0])
|
||
if parent < h_num:
|
||
return True
|
||
if parent >= h_num:
|
||
break
|
||
return False
|
||
|
||
from collections import defaultdict
|
||
by_num: Dict[str, list] = defaultdict(list)
|
||
for h in validated:
|
||
by_num[h.numbering].append(h)
|
||
|
||
best: Dict[str, _HeadingNode] = {}
|
||
for numbering, candidates in by_num.items():
|
||
if len(candidates) == 1:
|
||
best[numbering] = candidates[0]
|
||
else:
|
||
ok = [c for c in candidates if not _has_lower_l2_after(c)]
|
||
chosen = ok[0] if ok else candidates[-1]
|
||
for c in candidates:
|
||
if c is not chosen:
|
||
print(f" ⚠️ 编号重复，丢弃误识别: {c.numbering} 【{c.clean_title}】")
|
||
best[numbering] = chosen
|
||
|
||
return [h for h in validated if best.get(h.numbering) is h]
|
||
|
||
|
||
def _build_heading_index(data: list) -> Dict[int, _HeadingNode]:
|
||
"""构建 data索引 → HeadingNode 的映射。"""
|
||
toc_pages = _get_toc_pages(data)
|
||
if toc_pages:
|
||
print(f" [TOC页] 检测到目录页 page_idx={sorted(toc_pages)}，跳过这些页的标题识别")
|
||
|
||
headings = _scan_all_headings(data, toc_pages)
|
||
headings = _validate_heading_sequence(headings)
|
||
|
||
if headings:
|
||
print(f" [标题识别] 共识别 {len(headings)} 个有效标题：")
|
||
for h in headings:
|
||
indent = " " * h.level
|
||
conf_str = f" (conf={h.confidence:.2f})" if h.confidence < 1.0 else ""
|
||
print(f" {indent}{h.numbering} {h.clean_title}{conf_str}")
|
||
|
||
return {h.index_in_data: h for h in headings}
|
||
|
||
|
||
# ==============================
|
||
# 对外接口
|
||
# ==============================
|
||
|
||
def segment_chapters(data: list) -> List[Chapter]:
|
||
"""
|
||
对外唯一接口。
|
||
|
||
将 content_list JSON 按章节结构切分，返回 Chapter 列表。
|
||
每个 Chapter 包含：
|
||
- title / raw_title / numbering：标题信息
|
||
- is_special：是否特殊章节（二级节各自独立，不合并）
|
||
- is_skipped：是否被 FILTER_KEYWORDS 过滤（内容为空）
|
||
- records：本章全部 records（含标题行，不含目录页记录）
|
||
- loose_records：一级标题到第一个二级标题之间的 records
|
||
- l2_sections：List[L2Section]，每个二级节含 title/numbering/records
|
||
|
||
:param data: content_list JSON，list of dict
|
||
:return: List[Chapter]，按文档顺序排列
|
||
"""
|
||
toc_pages = _get_toc_pages(data)
|
||
heading_index = _build_heading_index(data)
|
||
|
||
chapters: List[Chapter] = []
|
||
|
||
# 当前章节状态
|
||
cur: Optional[Chapter] = None
|
||
cur_l2: Optional[L2Section] = None
|
||
has_met_l2 = False
|
||
|
||
def _flush_l2():
|
||
nonlocal cur_l2
|
||
if cur_l2 is not None and cur is not None:
|
||
if not _should_skip_l2(cur_l2.title):
|
||
cur.l2_sections.append(cur_l2)
|
||
cur_l2 = None
|
||
|
||
def _flush_chapter():
|
||
nonlocal cur, cur_l2, has_met_l2
|
||
if cur is None:
|
||
return
|
||
_flush_l2()
|
||
chapters.append(cur)
|
||
cur = None
|
||
cur_l2 = None
|
||
has_met_l2 = False
|
||
|
||
for i, ins in enumerate(data):
|
||
# 跳过目录页
|
||
if toc_pages and ins.get("page_idx") in toc_pages:
|
||
continue
|
||
if _is_toc_content(ins):
|
||
continue
|
||
# discarded 记录：仍然归入当前章节（保留位置信息供调用方使用）
|
||
if ins.get("type") == "discarded":
|
||
if cur is not None and not cur.is_skipped:
|
||
cur.records.append(ins)
|
||
continue
|
||
|
||
heading = heading_index.get(i)
|
||
|
||
if heading is not None and heading.level == 1:
|
||
# 新的一级标题 → 刷出上一章
|
||
_flush_chapter()
|
||
is_skipped = any(kw in heading.raw_text for kw in FILTER_KEYWORDS)
|
||
cur = Chapter(
|
||
title=heading.clean_title,
|
||
raw_title=heading.raw_text,
|
||
numbering=heading.numbering,
|
||
is_special=_is_special_chapter(heading.raw_text),
|
||
is_skipped=is_skipped,
|
||
)
|
||
has_met_l2 = False
|
||
if not is_skipped:
|
||
cur.records.append(ins)
|
||
tag = " [特殊]" if cur.is_special else ""
|
||
print(f"\n[一级标题] {heading.raw_text}{tag} (conf={heading.confidence:.2f})")
|
||
continue
|
||
|
||
if cur is None or cur.is_skipped:
|
||
continue
|
||
|
||
if heading is not None and heading.level == 2:
|
||
# 二级标题
|
||
_flush_l2()
|
||
l2_clean = heading.clean_title
|
||
if not _should_skip_l2(l2_clean):
|
||
cur_l2 = L2Section(
|
||
title=l2_clean,
|
||
numbering=heading.numbering,
|
||
records=[ins],
|
||
)
|
||
has_met_l2 = True
|
||
cur.records.append(ins)
|
||
continue
|
||
|
||
# 普通内容记录
|
||
cur.records.append(ins)
|
||
if has_met_l2:
|
||
if cur_l2 is not None:
|
||
cur_l2.records.append(ins)
|
||
else:
|
||
cur.loose_records.append(ins)
|
||
|
||
_flush_chapter()
|
||
|
||
print(f"\n[chapter_segmenter] 共切分章节: {len(chapters)} 个")
|
||
return chapters
|
||
|