Andrey Epifancev 0015532019 fix(parser): детектор заголовка — обязательны обороты и конец, ИНН без ведущего нуля (T4.1–T4.4, T4.6)
- T4.1 (R-12): строка групп обязана содержать «обороты» и «сальдо на конец»; строка
  только с «началом» и «концом» (преамбула) пропускается, нет подходящей — InvalidFormatError
  с недостающими группами лучшего кандидата.
- T4.2: сторона в заголовке в одну строку — после пунктуации («Сальдо на начало, Дт», «(Кт)»).
- T4.3: inn_from_cell — общий хелпер ИНН (10/12 цифр); число из 9/11 цифр дополняется
  ведущим нулём; переиспользуется классификатором (Task 8).
- T4.4: колонка ИНН по значениям — не меньше INN_MIN_VALUES = 3 значений-ИНН.
- T4.6: у слитого (в том числе только по вертикали) названия группы поиск подзаголовков
  ограничен его диапазоном.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
2026-10-05 13:03:15 +04:00

475 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Детекторы структуры листа по содержимому (design D-2; `05-domain-rules.md` §1.0 п. 1–4, §1.2).
Чистые функции над `Sheet`: вид листа (ОСВ / форма ББ-ОФР / прочее), блок заголовка таблицы, шесть колонок сумм,
колонки текста строки и ИНН, дополнительные колонки, повтор заголовка внутри данных. Результат — индексы колонок и
номера строк, вычисленные из текста ячеек; ни одна позиция не задана заранее. Буквы колонок появляются только в
диагностике (`ColumnMap.as_diagnostics` через `grid.column_letter`). Синонимы — в `rules/*.yaml`.
"""
import re
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, fields
from decimal import Decimal
from enum import StrEnum
from functools import lru_cache
from typing import Literal
from src.parser.errors import InvalidFormatError
from src.parser.grid import Cell, Sheet, column_letter
from src.parser.rules import Pattern, Rules, SumGroup, normalize_text
__all__ = [
"ColumnMap",
"SheetKind",
"TableHeader",
"classify_sheet",
"find_table_header",
"first_data_row",
"inn_from_cell",
"is_header_repeat",
]
CollapsedGroup = Literal["opening", "closing"]
Side = Literal["debit", "credit"]
# Признак ОСВ / формы ищется в начале листа: шапка ведомости всегда над таблицей
SHEET_MARKER_ROWS = 30
# Блок заголовка таблицы — от 1 до 4 строк: строка групп и до трёх строк под ней (подзаголовки, метки группировок)
HEADER_MAX_ROWS = 4
# Сколько строк данных смотреть, чтобы найти колонку текста строки и колонку ИНН по значениям
TEXT_COL_SAMPLE_ROWS = 20
INN_SAMPLE_ROWS = 200
# Колонка без заголовка «ИНН» считается колонкой ИНН, если в ней не меньше 3 непустых значений и не меньше 80 % из
# них — ИНН (10 или 12 цифр): одно случайное десятизначное число на всю колонку — не признак
INN_SHARE_PERCENT = 80
PERCENT = 100
INN_MIN_VALUES = 3
INN_LENGTHS = frozenset({10, 12})
# ИНН регионов 01–09, записанный числом, теряет ведущий ноль: 9 или 11 цифр дополняются нулём слева
_INN_LENGTHS_WITHOUT_LEADING_ZERO = frozenset(length - 1 for length in INN_LENGTHS)
# Строка групп сумм обязана содержать «обороты» и «сальдо на конец»; «сальдо на начало» может отсутствовать
# (spec «Допустимы: отсутствие группы «начало»», ruling R-12)
REQUIRED_GROUPS: frozenset[SumGroup] = frozenset({"turnover", "closing"})
# Повтор заголовка внутри данных: строка, где встретились хотя бы две группы сумм
MIN_GROUPS = 2
# Знаки между названием группы и стороной в заголовке в одну строку: «Сальдо на начало, Дт», «Обороты: Кт», «(Дт)»
_LEADING_PUNCTUATION = re.compile(r"^[\W_]+")
# Кэш нормализованного текста ячеек: каждая ячейка нормализуется один раз на все паттерны
_NORMALIZE_CACHE_SIZE = 16_384
GROUP_ORDER: tuple[SumGroup, ...] = ("opening", "turnover", "closing")
# Названия групп в сообщениях об ошибках (spec: INVALID_FORMAT «с названием ненайденной группы»)
GROUP_TITLES: dict[SumGroup, str] = {
"opening": "сальдо на начало",
"turnover": "обороты",
"closing": "сальдо на конец",
}
SIDE_TITLES: dict[Side, str] = {"debit": "Дебет", "credit": "Кредит"}
class SheetKind(StrEnum):
OSV = "osv"
STATEMENT = "statement"
OTHER = "other"
@dataclass(frozen=True)
class ColumnMap:
"""Шесть колонок сумм (1-based индексы; `None` — группы нет).
Свёрнутое сальдо (одна колонка в группе без «Дебет/Кредит») кладётся в `*_debit`, `*_credit = None`, группа
попадает в `collapsed`; раскладку значения по знаку делает классификатор строк (ruling R-1).
"""
opening_debit: int | None
opening_credit: int | None
turnover_debit: int | None
turnover_credit: int | None
closing_debit: int | None
closing_credit: int | None
collapsed: frozenset[CollapsedGroup] = frozenset()
def indexes(self) -> dict[str, int]:
"""Заданные колонки сумм по именам полей (`opening_debit`, …), в порядке полей."""
found: dict[str, int] = {}
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, int):
found[field.name] = value
return found
def as_diagnostics(self) -> dict[str, str]:
"""Диагностика для `columns` контракта: `{"opening_debit": "E", …}` — только найденные колонки."""
return {name: column_letter(index) for name, index in self.indexes().items()}
@dataclass(frozen=True)
class TableHeader:
"""Блок заголовка таблицы ОСВ и колонки, определённые по нему и по первым строкам данных."""
first_row: int # строка групп сумм
last_row: int # последняя строка блока (включительно): подзаголовки, их merged-продолжение, метки группировок
columns: ColumnMap
text_col: int
inn_col: int | None
right_edge: int # последняя колонка таблицы (правый край последней колонки сумм); правее — не таблица
extra_cols: dict[int, str] # колонка → заголовок в блоке (или буква колонки, если заголовка нет)
group_label_rows: list[int] # строки блока, где в text_col начинается текст группировки («Счет», «Контрагенты»)
# --- Текст ячеек ---
@lru_cache(maxsize=_NORMALIZE_CACHE_SIZE)
def _normalized(text: str) -> str:
return normalize_text(text)
def _search(patterns: Iterable[Pattern], normalized: str) -> re.Match[str] | None:
for pattern in patterns:
found = pattern.search_normalized(normalized)
if found is not None:
return found
return None
def _texts(sheet: Sheet, row: int) -> Iterator[tuple[int, str]]:
"""`(col, нормализованный текст)` непустых ячеек строки (merged — только origin)."""
for col, cell in sheet.row_cells(row):
yield col, _normalized(cell.text)
# --- Вид листа ---
def classify_sheet(sheet: Sheet, rules: Rules) -> SheetKind:
"""ОСВ — если в первых строках есть признак ведомости; иначе форма ББ/ОФР по её признакам; иначе прочее."""
head = [text for row in range(1, min(sheet.n_rows, SHEET_MARKER_ROWS) + 1) for _, text in _texts(sheet, row)]
if any(_search(rules.headers.sheet_markers, text) for text in head):
return SheetKind.OSV
if any(_search(rules.headers.statement_markers, text) for text in head):
return SheetKind.STATEMENT
return SheetKind.OTHER
# --- Заголовок таблицы ---
@dataclass(frozen=True)
class _GroupCell:
"""Ячейка строки групп, совпавшая с группой сумм; `side` — «Дт/Кт» в той же ячейке (заголовок в одну строку)."""
group: SumGroup
col: int
side: Side | None
@dataclass(frozen=True)
class _SumCell:
"""Найденная колонка суммы: индекс и строки/правый край ячейки заголовка (с учётом merged)."""
col: int
top: int
bottom: int
right: int
def _side(normalized: str, rules: Rules) -> Side | None:
if _search(rules.headers.debit, normalized):
return "debit"
if _search(rules.headers.credit, normalized):
return "credit"
return None
def _group_cells(sheet: Sheet, row: int, rules: Rules) -> list[_GroupCell]:
"""Ячейки строки, где найдена группа сумм; у каждой ячейки — не больше одной группы (порядок групп — приоритет)."""
found: list[_GroupCell] = []
for col, text in _texts(sheet, row):
for group in GROUP_ORDER:
match = _search(rules.headers.groups[group], text)
if match is not None:
remainder = _LEADING_PUNCTUATION.sub("", text[match.end() :])
found.append(_GroupCell(group, col, _side(remainder, rules)))
break
return found
def _sum_cell(sheet: Sheet, row: int, col: int) -> _SumCell:
span = sheet.merged_span(row, col)
if span is None:
return _SumCell(col, row, row, col)
return _SumCell(col, span.top, span.bottom, span.right)
def _missing_groups_message(sheet: Sheet, found: set[SumGroup]) -> str:
missing = ", ".join(f"«{GROUP_TITLES[group]}»" for group in GROUP_ORDER if group not in found)
return f"лист «{sheet.name}»: не найден заголовок таблицы ОСВ — нет групп сумм {missing}"
def _find_group_row(sheet: Sheet, rules: Rules) -> tuple[int, list[_GroupCell]]:
"""Первая сверху строка с «оборотами» и «сальдо на конец» («начало» — по возможности).
Строка, где нет одной из обязательных групп (например, преамбула «Сальдо на начало: …» | «Сальдо на конец: …»),
пропускается. Не нашлось — `InvalidFormatError` с группами, недостающими в лучшей строке-кандидате.
"""
best: set[SumGroup] = set()
for row in range(1, sheet.n_rows + 1):
cells = _group_cells(sheet, row, rules)
groups = {cell.group for cell in cells}
if REQUIRED_GROUPS <= groups:
return row, cells
if len(groups) > len(best):
best = groups
raise InvalidFormatError(_missing_groups_message(sheet, best))
@dataclass(frozen=True)
class _GroupColumns:
debit: _SumCell | None
credit: _SumCell | None
collapsed: _SumCell | None
def _columns_of_group(
sheet: Sheet, group_row: int, group_cells: list[_GroupCell], group: SumGroup, rules: Rules
) -> _GroupColumns:
own = [cell for cell in group_cells if cell.group == group]
sided = [cell for cell in own if cell.side is not None]
if sided:
# Заголовок в одну строку: «Нач. сальдо Дт» | «Нач. сальдо Кт»
debit = next((_sum_cell(sheet, group_row, c.col) for c in sided if c.side == "debit"), None)
credit = next((_sum_cell(sheet, group_row, c.col) for c in sided if c.side == "credit"), None)
return _check_sides(sheet, group, debit, credit)
title = own[0]
title_span = sheet.merged_span(group_row, title.col)
if title_span is not None:
# Слитое название (в том числе только по вертикали): подзаголовки — строго под его диапазоном
right = title_span.right
else:
following = [cell.col for cell in group_cells if cell.col > title.col]
right = min(following) - 1 if following else sheet.n_cols
first_sub_row = group_row + 1 if title_span is None else title_span.bottom + 1
debit = credit = None
for row in range(first_sub_row, group_row + HEADER_MAX_ROWS):
for col, text in _texts(sheet, row):
if not title.col <= col <= right:
continue
side = _side(text, rules)
if side == "debit" and debit is None:
debit = _sum_cell(sheet, row, col)
elif side == "credit" and credit is None:
credit = _sum_cell(sheet, row, col)
if debit is None and credit is None:
if group == "turnover":
raise InvalidFormatError(
f"лист «{sheet.name}»: группа «{GROUP_TITLES[group]}» без колонок «Дебет»/«Кредит» — "
"свёрнутыми бывают только сальдо"
)
return _GroupColumns(None, None, _sum_cell(sheet, group_row, title.col))
return _check_sides(sheet, group, debit, credit)
def _check_sides(sheet: Sheet, group: SumGroup, debit: _SumCell | None, credit: _SumCell | None) -> _GroupColumns:
if debit is None or credit is None:
present: Side = "credit" if debit is None else "debit"
absent: Side = "debit" if debit is None else "credit"
raise InvalidFormatError(
f"лист «{sheet.name}»: группа «{GROUP_TITLES[group]}»: найден «{SIDE_TITLES[present]}», "
f"нет «{SIDE_TITLES[absent]}»"
)
return _GroupColumns(debit, credit, None)
def _is_label_row(sheet: Sheet, row: int, sums: range, rules: Rules) -> bool:
"""Строка-метка блока заголовка: левее сумм — текст группировки, «ИНН» или «Субконто», в колонках сумм — пусто.
Пустые колонки сумм отличают метку «Контрагенты» от строки данных с контрагентом, чьё имя так начинается.
"""
label_patterns = [
*(pattern for pattern, _ in rules.groups.mapping),
*rules.headers.service_columns["inn"],
*rules.headers.header_markers["subkonto"],
]
texts = list(_texts(sheet, row))
if any(col in sums for col, _ in texts):
return False
return any(col < sums.start and _search(label_patterns, text) for col, text in texts)
def _block_last_row(sheet: Sheet, group_row: int, sum_cells: list[_SumCell], rules: Rules) -> int:
"""Последняя строка блока: подзаголовки (с merged-продолжением) и подряд идущие строки-метки под строкой групп."""
covered = {row for cell in sum_cells for row in range(cell.top, cell.bottom + 1)}
sums = range(min(cell.col for cell in sum_cells), max(cell.right for cell in sum_cells) + 1)
last = group_row
for row in range(group_row + 1, group_row + HEADER_MAX_ROWS):
if row in covered or _is_label_row(sheet, row, sums, rules):
last = row
elif not sheet.is_row_empty(row):
break # первая строка данных
return last
def _data_rows(sheet: Sheet, after: int, limit: int | None = None) -> Iterator[int]:
"""Непустые строки после `after` (не больше `limit`)."""
taken = 0
for row in range(after + 1, sheet.n_rows + 1):
if limit is not None and taken >= limit:
return
if not sheet.is_row_empty(row):
taken += 1
yield row
def _text_col(sheet: Sheet, first_row: int, last_row: int, first_sum: int) -> int:
"""Первая непустая колонка левее сумм по первым строкам данных; без строк данных — по меткам блока."""
data = list(_data_rows(sheet, last_row, TEXT_COL_SAMPLE_ROWS))
rows = data if data else list(range(first_row, last_row + 1))
candidates = [col for row in rows for col, _ in sheet.row_cells(row) if col < first_sum]
if not candidates:
raise InvalidFormatError(f"лист «{sheet.name}»: не найдена колонка текста строк слева от колонок сумм")
return min(candidates)
def inn_from_cell(cell: Cell) -> str | None:
"""ИНН из ячейки колонки ИНН: 10 или 12 цифр, иначе `None` (единое правило для детектора и классификатора).
Текст — как есть (после trim), только цифры. Целое число — без дробной части и экспоненты; если при записи числом
потерян ведущий ноль (ИНН регионов 01–09: 9 или 11 цифр), он восстанавливается. Текст из 9/11 цифр не дополняется:
в тексте ноль не теряется.
"""
value = cell.value
if isinstance(value, Decimal):
if not value.is_finite() or value != value.to_integral_value() or value < 0:
return None
digits = str(int(value))
if len(digits) in _INN_LENGTHS_WITHOUT_LEADING_ZERO:
digits = digits.zfill(len(digits) + 1)
else:
digits = cell.text
return digits if digits.isdigit() and len(digits) in INN_LENGTHS else None
def _inn_col(sheet: Sheet, block: range, text_col: int, first_sum: int, rules: Rules) -> int | None:
"""Колонка между текстом и суммами: по заголовку «ИНН» в блоке, иначе по доле значений из 10/12 цифр."""
inn_patterns = rules.headers.service_columns["inn"]
for row in block:
for col, text in _texts(sheet, row):
if text_col < col < first_sum and _search(inn_patterns, text):
return col
values: dict[int, list[Cell]] = {}
for row in _data_rows(sheet, block[-1], INN_SAMPLE_ROWS):
for col, cell in sheet.row_cells(row):
if text_col < col < first_sum:
values.setdefault(col, []).append(cell)
for col in sorted(values):
cells = values[col]
inn_count = sum(1 for cell in cells if inn_from_cell(cell) is not None)
if inn_count >= INN_MIN_VALUES and inn_count * PERCENT >= INN_SHARE_PERCENT * len(cells):
return col
return None
def _block_title(sheet: Sheet, first_row: int, last_row: int, col: int) -> str | None:
"""Заголовок колонки в блоке: текст самой нижней непустой ячейки блока ровно в этой колонке."""
for row in range(last_row, first_row - 1, -1):
text = sheet.cell(row, col).text
if text:
return text
return None
def _group_label_rows(sheet: Sheet, first_row: int, last_row: int, text_col: int) -> list[int]:
labels: list[int] = []
for row in range(first_row, last_row + 1):
origin_row, origin_col = sheet.merged_origin(row, text_col)
if origin_row == row and sheet.cell(origin_row, origin_col).text:
labels.append(row)
return labels
def find_table_header(sheet: Sheet, rules: Rules) -> TableHeader:
"""Блок заголовка таблицы и колонки; не найдено — `InvalidFormatError` с названием ненайденного."""
group_row, group_cells = _find_group_row(sheet, rules)
found = {cell.group for cell in group_cells}
by_group = {group: _columns_of_group(sheet, group_row, group_cells, group, rules) for group in found}
def index(group: SumGroup, side: Side) -> int | None:
columns = by_group.get(group)
if columns is None:
return None
if columns.collapsed is not None:
return columns.collapsed.col if side == "debit" else None
cell = columns.debit if side == "debit" else columns.credit
return None if cell is None else cell.col
collapsed: frozenset[CollapsedGroup] = frozenset(
group for group in ("opening", "closing") if group in by_group and by_group[group].collapsed is not None
)
columns = ColumnMap(
opening_debit=index("opening", "debit"),
opening_credit=index("opening", "credit"),
turnover_debit=index("turnover", "debit"),
turnover_credit=index("turnover", "credit"),
closing_debit=index("closing", "debit"),
closing_credit=index("closing", "credit"),
collapsed=collapsed,
)
sum_cells = [
cell
for group_columns in by_group.values()
for cell in (group_columns.debit, group_columns.credit, group_columns.collapsed)
if cell is not None
]
sum_cols = set(columns.indexes().values())
first_sum = min(sum_cols)
right_edge = max(cell.right for cell in sum_cells)
last_row = _block_last_row(sheet, group_row, sum_cells, rules)
text_col = _text_col(sheet, group_row, last_row, first_sum)
inn_col = _inn_col(sheet, range(group_row, last_row + 1), text_col, first_sum, rules)
partial = TableHeader(group_row, last_row, columns, text_col, inn_col, right_edge, {}, [])
occupied = {*sum_cols, text_col} | ({inn_col} if inn_col is not None else set())
extra_cols: dict[int, str] = {}
for row in _data_rows(sheet, last_row):
if is_header_repeat(sheet, row, partial, rules):
continue
for col, _ in sheet.row_cells(row):
if text_col < col <= right_edge and col not in occupied and col not in extra_cols:
extra_cols[col] = _block_title(sheet, group_row, last_row, col) or column_letter(col)
return TableHeader(
first_row=group_row,
last_row=last_row,
columns=columns,
text_col=text_col,
inn_col=inn_col,
right_edge=right_edge,
extra_cols=dict(sorted(extra_cols.items())),
group_label_rows=_group_label_rows(sheet, group_row, last_row, text_col),
)
def first_data_row(sheet: Sheet, header: TableHeader) -> int:
"""Первая непустая строка после блока заголовка; таблица без строк данных — `n_rows + 1`."""
return next(_data_rows(sheet, header.last_row), sheet.n_rows + 1)
def _signature(sheet: Sheet, row: int, right_edge: int) -> tuple[tuple[int, str], ...]:
return tuple((col, text) for col, text in _texts(sheet, row) if col <= right_edge)
def is_header_repeat(sheet: Sheet, row: int, header: TableHeader, rules: Rules) -> bool:
"""Строка данных — повтор заголовка (печать «по страницам»): две группы сумм или копия строки блока заголовка."""
if sheet.is_row_empty(row):
return False
if len({cell.group for cell in _group_cells(sheet, row, rules)}) >= MIN_GROUPS:
return True
signature = _signature(sheet, row, header.right_edge)
if not signature:
return False # только ячейки правее таблицы: не заголовок (и не данные таблицы)
return any(
signature == _signature(sheet, block_row, header.right_edge)
for block_row in range(header.first_row, header.last_row + 1)
)