114 lines
5.4 KiB
Python
114 lines
5.4 KiB
Python
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from genesis.data_models import CellComment, ExcelTable, MixedParagraph, MixedSheet
|
||
from genesis.parsers.excel_reader import open_workbook, sheet_matrix
|
||
from genesis.parsers.sheet_detector import detect_sheet_type
|
||
from genesis.parsers.sheet_nature import SheetNature, classify_sheet, find_header_row
|
||
from genesis.parsers.merge_fill import forward_fill
|
||
from genesis.parsers.paragraph_splitter import split_paragraphs
|
||
from genesis.parsers.table_extractor import extract_table
|
||
from genesis.parsers.formatting_detector import cell_formatting, collect_comments
|
||
from genesis.parsers.free_text_extractor import build_free_text_table, extract_text_blocks
|
||
|
||
|
||
@dataclass
|
||
class ExcelParseResult:
|
||
file_name: str
|
||
tables: list[ExcelTable] = field(default_factory=list)
|
||
comments: list[CellComment] = field(default_factory=list)
|
||
skipped: list[str] = field(default_factory=list)
|
||
mixed: list[MixedSheet] = field(default_factory=list)
|
||
|
||
|
||
class ExcelParser:
|
||
"""要件定义 Excel 解析入口。"""
|
||
|
||
def parse(self, path: str | Path) -> ExcelParseResult:
|
||
wb = open_workbook(path)
|
||
file_name = Path(path).name
|
||
result = ExcelParseResult(file_name=file_name)
|
||
for ws in wb.worksheets:
|
||
matrix = sheet_matrix(ws)
|
||
if not matrix:
|
||
result.skipped.append(ws.title)
|
||
continue
|
||
detected_type = detect_sheet_type(ws.title, matrix)
|
||
nature = classify_sheet(matrix)
|
||
if nature == SheetNature.MIXED:
|
||
# 合并单元格：整 sheet 先填充再按段切片
|
||
merged = [
|
||
(r.min_row, r.min_col, r.max_row, r.max_col)
|
||
for r in ws.merged_cells.ranges
|
||
]
|
||
filled_all = forward_fill(matrix, merged) if merged else matrix
|
||
# 整 sheet 构建格式映射一次（按物理坐标）
|
||
fmt_map = {}
|
||
for row in ws.iter_rows():
|
||
for cell in row:
|
||
fmt = cell_formatting(cell)
|
||
if fmt is not None:
|
||
fmt_map[(cell.row - 1, cell.column - 1)] = fmt
|
||
mixed_sheet = MixedSheet(name=ws.title)
|
||
for (s, e) in split_paragraphs(matrix):
|
||
seg = filled_all[s:e + 1]
|
||
seg_nature = classify_sheet(seg)
|
||
if seg_nature == SheetNature.TABLE:
|
||
header_row = find_header_row(seg)
|
||
if header_row < 0: # pragma: no cover — classify==TABLE 时 find_header_row 恒 ≥0
|
||
header_row = 0
|
||
# extract_table 的 formatting_map 用段内矩阵坐标 (r,c)；
|
||
# 从整 sheet 物理 map 抽出段内子 map（物理行 pr → 段内 pr-s）
|
||
seg_fmt_map = {}
|
||
for (pr, pc), fmt in fmt_map.items():
|
||
if s <= pr <= e:
|
||
seg_fmt_map[(pr - s, pc)] = fmt
|
||
table = extract_table(
|
||
ws.title, seg, file_name, detected_type,
|
||
header_row=header_row, formatting_map=seg_fmt_map,
|
||
)
|
||
result.tables.append(table)
|
||
mixed_sheet.paragraphs.append(MixedParagraph(
|
||
kind="table", matrix=seg, table=table,
|
||
source_range=(s, e),
|
||
))
|
||
else:
|
||
blocks = extract_text_blocks(seg)
|
||
table = build_free_text_table(ws.title, blocks, file_name, detected_type)
|
||
result.tables.append(table)
|
||
mixed_sheet.paragraphs.append(MixedParagraph(
|
||
kind="free_text", matrix=seg,
|
||
text="\n".join(blocks), source_range=(s, e),
|
||
))
|
||
result.mixed.append(mixed_sheet)
|
||
elif nature == SheetNature.FREE_TEXT:
|
||
blocks = extract_text_blocks(matrix)
|
||
result.tables.append(
|
||
build_free_text_table(ws.title, blocks, file_name, detected_type)
|
||
)
|
||
else:
|
||
# 现有 TABLE 路径（MIXED 已独立，不折叠至此）
|
||
merged = [
|
||
(r.min_row, r.min_col, r.max_row, r.max_col)
|
||
for r in ws.merged_cells.ranges
|
||
]
|
||
filled = forward_fill(matrix, merged) if merged else matrix
|
||
header_row = find_header_row(filled)
|
||
if header_row < 0: # pragma: no cover — classify==TABLE 时 find_header_row 恒 ≥0
|
||
header_row = 0
|
||
fmt_map = {}
|
||
for row in ws.iter_rows():
|
||
for cell in row:
|
||
fmt = cell_formatting(cell)
|
||
if fmt is not None:
|
||
fmt_map[(cell.row - 1, cell.column - 1)] = fmt
|
||
result.tables.append(
|
||
extract_table(
|
||
ws.title, filled, file_name, detected_type,
|
||
header_row=header_row, formatting_map=fmt_map,
|
||
)
|
||
)
|
||
result.comments.extend(collect_comments(ws, file_name))
|
||
return result |