289 lines
9.5 KiB
Python
289 lines
9.5 KiB
Python
"""在原始网格里识别表：按空行切块，找表头，合并两级表头，去掉标题、说明和合计行
|
||
|
||
只用规则；判断不了的地方照常产出表，同时记一条警告，调用方据此提示用户
|
||
"""
|
||
|
||
import math
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from app.datafiles.base import MAX_TABLE_ROWS, brief, rows_exceeded, where
|
||
from app.datafiles.columns import as_text, classify
|
||
from app.datafiles.readers import Grid
|
||
from app.schemas.data_file import ParseWarning
|
||
|
||
# 只在每块的前几行里找表头
|
||
HEADER_SEARCH_ROWS = 10
|
||
|
||
# 首个或第二个非空单元格是这些词（去掉空白和冒号后）的行是合计行
|
||
TOTAL_LABELS = {
|
||
"合计",
|
||
"总计",
|
||
"小计",
|
||
"共计",
|
||
"总和",
|
||
"平均",
|
||
"平均值",
|
||
"total",
|
||
"subtotal",
|
||
"grandtotal",
|
||
"sum",
|
||
"average",
|
||
}
|
||
# 例如"华东小计""一季度合计"
|
||
TOTAL_SUFFIXES = ("小计", "合计", "总计")
|
||
|
||
# 警告里引用的单元格文本最长显示这么多字
|
||
QUOTE_CHARS = 30
|
||
|
||
|
||
@dataclass
|
||
class RawTable:
|
||
sheet: str | None
|
||
# 同一 sheet 切出多张表时的序号，从 1 开始；只有一张表时为空
|
||
part: int | None
|
||
header: list[Any]
|
||
# 数据行，宽度与表头相同
|
||
rows: list[list[Any]]
|
||
warnings: list[ParseWarning] = field(default_factory=list)
|
||
|
||
|
||
@dataclass
|
||
class _Piece:
|
||
columns: list[int]
|
||
# 表头所在行在网格中的下标
|
||
header_row: int
|
||
header: list[Any]
|
||
# 数据行在网格中的下标
|
||
data: list[int]
|
||
warnings: list[ParseWarning]
|
||
|
||
|
||
def split_tables(grid: Grid) -> tuple[list[RawTable], list[ParseWarning]]:
|
||
"""返回识别出的表和不属于任何表的警告（被丢弃的块）"""
|
||
rows = _normalize(grid.rows)
|
||
sheet = grid.sheet
|
||
pieces: list[_Piece] = []
|
||
dropped: list[ParseWarning] = []
|
||
for block in _blocks(rows):
|
||
columns = [
|
||
c for c in range(len(rows[0])) if any(rows[r][c] is not None for r in block)
|
||
]
|
||
last = pieces[-1] if pieces else None
|
||
# 表中间夹着空行时，后面的块以数据行开头、列也落在上一张表内，接在上一张表后面
|
||
if (
|
||
last
|
||
and not _is_label_row([rows[block[0]][c] for c in columns])
|
||
and set(columns) <= set(last.columns)
|
||
):
|
||
last.data.extend(block)
|
||
continue
|
||
if len(block) < 2:
|
||
dropped.append(
|
||
ParseWarning(
|
||
code="block_dropped",
|
||
message=f"{where(sheet)}第 {block[0] + 1} 行的{_quote(rows[block[0]])}不是表格，已跳过",
|
||
)
|
||
)
|
||
continue
|
||
pieces.append(_read_block(rows, block, columns, sheet))
|
||
|
||
tables = []
|
||
for piece in pieces:
|
||
table = _finish(rows, piece, sheet)
|
||
if table is None:
|
||
dropped.append(
|
||
ParseWarning(
|
||
code="block_dropped",
|
||
message=f"{where(sheet)}第 {piece.header_row + 1} 行开始的表只有表头没有数据，已跳过",
|
||
)
|
||
)
|
||
else:
|
||
tables.append(table)
|
||
if len(tables) > 1:
|
||
for part, table in enumerate(tables, start=1):
|
||
table.part = part
|
||
return tables, dropped
|
||
|
||
|
||
def _normalize(grid: list[list[Any]]) -> list[list[Any]]:
|
||
"""去掉字符串两端空白，空字符串记为 None，补齐成同样宽度"""
|
||
width = max((len(row) for row in grid), default=0)
|
||
rows = []
|
||
for row in grid:
|
||
cells = [_cell(value) for value in row]
|
||
cells.extend([None] * (width - len(cells)))
|
||
rows.append(cells)
|
||
return rows
|
||
|
||
|
||
def _cell(value: Any) -> Any:
|
||
if isinstance(value, str):
|
||
value = value.strip()
|
||
return value or None
|
||
return value
|
||
|
||
|
||
def _blocks(rows: list[list[Any]]) -> list[list[int]]:
|
||
blocks, current = [], []
|
||
for index, row in enumerate(rows):
|
||
if any(value is not None for value in row):
|
||
current.append(index)
|
||
elif current:
|
||
blocks.append(current)
|
||
current = []
|
||
if current:
|
||
blocks.append(current)
|
||
return blocks
|
||
|
||
|
||
def _read_block(
|
||
rows: list[list[Any]], block: list[int], columns: list[int], sheet: str | None
|
||
) -> _Piece:
|
||
cells = [[rows[r][c] for c in columns] for r in block]
|
||
need = max(1, math.ceil(len(columns) / 2))
|
||
dense = [sum(value is not None for value in row) >= need for row in cells]
|
||
search = range(min(HEADER_SEARCH_ROWS, len(cells)))
|
||
warnings = []
|
||
|
||
# 表头：前几行里第一个够满、并且全是文字的行；标题行通常只有一个单元格，不够满
|
||
header_at = next((i for i in search if dense[i] and _is_label_row(cells[i])), None)
|
||
confident = header_at is not None
|
||
if header_at is None:
|
||
header_at = next((i for i in search if dense[i]), 0)
|
||
warnings.append(
|
||
ParseWarning(
|
||
code="header_uncertain",
|
||
message=f"{where(sheet)}没有找到明确的表头，用第 {block[header_at] + 1} 行作表头，可能识别错误",
|
||
)
|
||
)
|
||
if header_at > 0:
|
||
skipped = "".join(
|
||
_quote(cells[i]) for i in range(header_at) if _quote(cells[i])
|
||
)
|
||
warnings.append(
|
||
ParseWarning(
|
||
code="title_skipped",
|
||
message=f"{where(sheet)}{_row_range(block[0], block[header_at - 1])}是标题，已跳过：{skipped}",
|
||
)
|
||
)
|
||
|
||
header = cells[header_at]
|
||
data_at = header_at + 1
|
||
if (
|
||
confident
|
||
and header_at + 2 < len(cells)
|
||
and _is_subheader(cells[header_at], cells[header_at + 1], cells[header_at + 2])
|
||
):
|
||
header = _merge_header(cells[header_at], cells[header_at + 1])
|
||
data_at = header_at + 2
|
||
warnings.append(
|
||
ParseWarning(
|
||
code="header_merged",
|
||
message=f"{where(sheet)}{_row_range(block[header_at], block[header_at + 1])}合并为表头",
|
||
)
|
||
)
|
||
return _Piece(columns, block[header_at], header, block[data_at:], warnings)
|
||
|
||
|
||
def _finish(rows: list[list[Any]], piece: _Piece, sheet: str | None) -> RawTable | None:
|
||
data = [(index, [rows[index][c] for c in piece.columns]) for index in piece.data]
|
||
warnings = piece.warnings
|
||
|
||
# 表格下方只有一个文字单元格的行是说明，例如"注：数据来源于……"
|
||
notes = []
|
||
while data and len(piece.columns) >= 3 and _is_note(data[-1][1]):
|
||
notes.append(data.pop())
|
||
if notes:
|
||
notes.reverse()
|
||
shown = "".join(_quote(row) for _, row in notes)
|
||
warnings.append(
|
||
ParseWarning(
|
||
code="note_skipped",
|
||
message=f"{where(sheet)}{_row_range(notes[0][0], notes[-1][0])}是说明，已跳过：{shown}",
|
||
)
|
||
)
|
||
|
||
# 合计、小计行不是明细数据，留着会让求和重复计算
|
||
totals = [index for index, row in data if _is_total(row)]
|
||
if totals:
|
||
data = [(index, row) for index, row in data if not _is_total(row)]
|
||
listed = brief([str(index + 1) for index in totals])
|
||
warnings.append(
|
||
ParseWarning(
|
||
code="total_rows_removed",
|
||
message=f"{where(sheet)}已排除 {len(totals)} 行合计、小计（第 {listed} 行）",
|
||
)
|
||
)
|
||
|
||
if not data:
|
||
return None
|
||
if len(data) > MAX_TABLE_ROWS:
|
||
raise rows_exceeded(sheet)
|
||
return RawTable(sheet, None, piece.header, [row for _, row in data], warnings)
|
||
|
||
|
||
def _is_label(value: Any) -> bool:
|
||
return isinstance(value, str) and classify(value)[0] == "text"
|
||
|
||
|
||
def _is_label_row(row: list[Any]) -> bool:
|
||
values = [value for value in row if value is not None]
|
||
return bool(values) and all(_is_label(value) for value in values)
|
||
|
||
|
||
def _is_subheader(top: list[Any], sub: list[Any], first_data: list[Any]) -> bool:
|
||
"""第二行也全是文字、紧接着是带数值的数据行，并且两行里有合并单元格留下的空格"""
|
||
return (
|
||
_is_label_row(sub)
|
||
and not _is_label_row(first_data)
|
||
and (None in top or None in sub)
|
||
)
|
||
|
||
|
||
def _merge_header(top: list[Any], sub: list[Any]) -> list[str]:
|
||
"""合并单元格只有左上角有值，第一行横向向前填充后与第二行用 - 连接，例如 销售额-一月"""
|
||
merged, carry = [], None
|
||
for upper, lower in zip(top, sub):
|
||
if upper is not None:
|
||
carry = upper
|
||
parts = [as_text(part) for part in (carry, lower) if part is not None]
|
||
merged.append("-".join(parts))
|
||
return merged
|
||
|
||
|
||
def _is_note(row: list[Any]) -> bool:
|
||
values = [value for value in row if value is not None]
|
||
return len(values) == 1 and _is_label(values[0])
|
||
|
||
|
||
def _is_total(row: list[Any]) -> bool:
|
||
checked = 0
|
||
for value in row:
|
||
if value is None:
|
||
continue
|
||
if isinstance(value, str):
|
||
key = re.sub(r"[\s:：]", "", value).lower()
|
||
if key in TOTAL_LABELS or key.endswith(TOTAL_SUFFIXES):
|
||
return True
|
||
checked += 1
|
||
if checked == 2:
|
||
return False
|
||
return False
|
||
|
||
|
||
def _quote(row: list[Any]) -> str:
|
||
"""一行里第一个非空单元格，截断后加引号"""
|
||
value = next((value for value in row if value is not None), None)
|
||
if value is None:
|
||
return ""
|
||
text = as_text(value)
|
||
if len(text) > QUOTE_CHARS:
|
||
text = text[:QUOTE_CHARS] + "…"
|
||
return f"「{text}」"
|
||
|
||
|
||
def _row_range(first: int, last: int) -> str:
|
||
return f"第 {first + 1} 行" if first == last else f"第 {first + 1}–{last + 1} 行"
|