feat: шаг 2 — столбец кодов с защитой от блока «Коды», нумерация, строка заголовка, роли строк
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev
parent
f73f386378
commit
f4533f45e1
306
arm_statements/layout.py
Normal file
306
arm_statements/layout.py
Normal file
@ -0,0 +1,306 @@
|
||||
"""Шаг 2: столбец кодов, нумерация, строка заголовка, роли строк, столбец наименования."""
|
||||
|
||||
import re
|
||||
from collections import Counter
|
||||
from collections.abc import Sequence
|
||||
from decimal import Decimal
|
||||
|
||||
from arm_statements.codes import code_parts, is_code_cell, parse_code, section_of
|
||||
from arm_statements.dates import parse_period_label
|
||||
from arm_statements.loader import CodeSection, Rules
|
||||
from arm_statements.model import Cell, DocumentType, PeriodKind, RowRole
|
||||
from arm_statements.text import fix_lookalikes, normalize_caption
|
||||
|
||||
Grid = Sequence[Sequence[Cell | None]]
|
||||
|
||||
_MIN_NUMBERING_CELLS = 3
|
||||
_SHORT_NOTE = 3 # «Пояснения» без подписи: значения не длиннее 3 символов
|
||||
_SKIP_STEP = 2 # скачок через одну пропущенную цифру нумерации
|
||||
_NUMBER = re.compile(r"[0-9]+")
|
||||
_FULL_CONFIDENCE = 100.0
|
||||
|
||||
|
||||
def _text(cell: Cell | None) -> str:
|
||||
return "" if cell is None else cell.text.strip()
|
||||
|
||||
|
||||
def cell_at(rows: Grid, row: int, col: int) -> Cell | None:
|
||||
"""Ячейка сетки; вне строки или сетки — `None`."""
|
||||
if row < 0 or row >= len(rows) or col < 0:
|
||||
return None
|
||||
line = rows[row]
|
||||
return line[col] if col < len(line) else None
|
||||
|
||||
|
||||
def _width(rows: Grid, first: int, last: int) -> int:
|
||||
return max((len(rows[r]) for r in range(max(first, 0), min(last, len(rows) - 1) + 1)), default=0)
|
||||
|
||||
|
||||
def _relaxed_count(rows: Grid, span: range, col: int, section: CodeSection, rules: Rules) -> int:
|
||||
count = 0
|
||||
for row in span:
|
||||
text = _text(cell_at(rows, row, col))
|
||||
if not text or not is_code_cell(text, rules):
|
||||
continue
|
||||
codes = [parse_code(part, rules).code for part in code_parts(text)]
|
||||
if all(code is not None and section_of(code, rules) is section for code in codes):
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
def _share_ok(codes: int, nonempty: int, min_share: Decimal) -> bool:
|
||||
return nonempty > 0 and Decimal(codes) >= min_share * Decimal(nonempty)
|
||||
|
||||
|
||||
def _tally(rows: Grid, start: int, last: int, col: int, rules: Rules) -> tuple[int, int]:
|
||||
"""(кодов, непустых ячеек) столбца в строках `start..last`."""
|
||||
codes = nonempty = 0
|
||||
for row in range(start, last + 1):
|
||||
text = _text(cell_at(rows, row, col))
|
||||
if text:
|
||||
nonempty += 1
|
||||
if is_code_cell(text, rules):
|
||||
codes += 1
|
||||
return codes, nonempty
|
||||
|
||||
|
||||
def _best(candidates: dict[int, int]) -> int | None:
|
||||
"""Столбец с наибольшим числом кодов; ничья — левый."""
|
||||
best: int | None = None
|
||||
for col in sorted(candidates):
|
||||
if best is None or candidates[col] > candidates[best]:
|
||||
best = col
|
||||
return best
|
||||
|
||||
|
||||
def _caption_candidates(rows: Grid, first: int, last: int, width: int, rules: Rules) -> dict[int, int]:
|
||||
"""Правило (а): столбец с подписью кода; под подписью — ячейки ниже неё до `last`."""
|
||||
ranges = rules.code_ranges
|
||||
found: dict[int, int] = {}
|
||||
for col in range(width):
|
||||
first_code = next(
|
||||
(r for r in range(first, last + 1) if is_code_cell(_text(cell_at(rows, r, col)), rules)), last + 1
|
||||
)
|
||||
captions = [
|
||||
r
|
||||
for r in range(first, min(first_code, last + 1))
|
||||
if rules.headers.code_caption.search(normalize_caption(_text(cell_at(rows, r, col))))
|
||||
]
|
||||
if not captions:
|
||||
continue
|
||||
codes, nonempty = _tally(rows, captions[-1] + 1, last, col, rules)
|
||||
if codes > 0 and (codes >= ranges.min_codes or _share_ok(codes, nonempty, ranges.min_share)):
|
||||
found[col] = codes
|
||||
return found
|
||||
|
||||
|
||||
def find_code_column(
|
||||
rows: Grid, first: int, last: int, rules: Rules, *, relaxed_section: CodeSection | None = None
|
||||
) -> int | None:
|
||||
"""Столбец кодов по правилу (а)/(б) шага 2; `relaxed_section` — ослабленный поиск продолжения."""
|
||||
width = _width(rows, first, last)
|
||||
if relaxed_section is not None:
|
||||
counts = {
|
||||
col: _relaxed_count(rows, range(first, last + 1), col, relaxed_section, rules) for col in range(width)
|
||||
}
|
||||
counts = {col: n for col, n in counts.items() if n >= rules.code_ranges.min_codes_continuation}
|
||||
return _best(counts)
|
||||
by_caption = _best(_caption_candidates(rows, first, last, width, rules))
|
||||
if by_caption is not None:
|
||||
return by_caption
|
||||
ranges = rules.code_ranges
|
||||
scored: dict[int, int] = {}
|
||||
for col in range(width):
|
||||
codes, nonempty = _tally(rows, first, last, col, rules)
|
||||
if codes >= ranges.min_codes and _share_ok(codes, nonempty, ranges.min_share):
|
||||
scored[col] = codes
|
||||
return _best(scored)
|
||||
|
||||
|
||||
def code_column(rows: Grid, rules: Rules, *, relaxed: DocumentType | None = None) -> int | None:
|
||||
"""Столбец кодов всей сетки — то же правило, что у шага 2 (для адаптеров)."""
|
||||
section: CodeSection | None = None
|
||||
if relaxed is DocumentType.DISCLOSURE:
|
||||
raise ValueError("relaxed=DISCLOSURE не поддерживается: у пояснений несколько отрезков кодов")
|
||||
if relaxed is DocumentType.BALANCE:
|
||||
section = CodeSection.BALANCE
|
||||
elif relaxed is DocumentType.INCOME:
|
||||
section = CodeSection.INCOME
|
||||
if not rows:
|
||||
return None
|
||||
return find_code_column(rows, 0, len(rows) - 1, rules, relaxed_section=section)
|
||||
|
||||
|
||||
def first_code_row(rows: Grid, first: int, last: int, code_col: int, rules: Rules) -> int | None:
|
||||
for row in range(first, last + 1):
|
||||
if is_code_cell(_text(cell_at(rows, row, code_col)), rules):
|
||||
return row
|
||||
return None
|
||||
|
||||
|
||||
def is_numbering_row(cells: Sequence[Cell | None], rules: Rules) -> bool:
|
||||
"""≥ 3 непустых ячеек-целых, слева направо +1, начало 1 или 2, не больше одного пропуска."""
|
||||
numbers: list[int] = []
|
||||
for cell in cells:
|
||||
text = _text(cell)
|
||||
if not text:
|
||||
continue
|
||||
fixed = fix_lookalikes(text, rules.lookalikes, anywhere=True).strip()
|
||||
if _NUMBER.fullmatch(fixed) is None:
|
||||
return False
|
||||
numbers.append(int(fixed))
|
||||
if len(numbers) < _MIN_NUMBERING_CELLS or numbers[0] not in (1, 2):
|
||||
return False
|
||||
skips = 0
|
||||
for prev, current in zip(numbers, numbers[1:], strict=False):
|
||||
step = current - prev
|
||||
if step == 1:
|
||||
continue
|
||||
if step == _SKIP_STEP and skips == 0:
|
||||
skips += 1
|
||||
continue
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _is_period(text: str, ref: str, rules: Rules) -> bool:
|
||||
cell = Cell(text, None, _FULL_CONFIDENCE, ref)
|
||||
return parse_period_label(cell, rules).kind is not PeriodKind.NONE
|
||||
|
||||
|
||||
def _has_header_periods(rows: Grid, row: int, code_col: int, rules: Rules) -> bool:
|
||||
"""Две и более подписи-периода правее кода, либо одна — не голый год."""
|
||||
labels = [
|
||||
parse_period_label(cell, rules)
|
||||
for col, cell in enumerate(rows[row])
|
||||
if col > code_col and cell is not None and cell.text.strip()
|
||||
]
|
||||
periods = [label for label in labels if label.kind is not PeriodKind.NONE]
|
||||
strong = [label for label in periods if not label.bare_year]
|
||||
return bool(strong) or len(periods) >= 2 # noqa: PLR2004 - две подписи подряд — заголовок, одна «2025» — титул
|
||||
|
||||
|
||||
def _glued_with_upper(rows: Grid, upper: int, lower: int, code_col: int, rules: Rules) -> bool:
|
||||
"""P1-9: подпись верхней строки отдельно — не период, а «верхняя + нижняя» — период."""
|
||||
for col, cell in enumerate(rows[upper]):
|
||||
if col <= code_col:
|
||||
continue
|
||||
top = _text(cell)
|
||||
bottom = _text(cell_at(rows, lower, col))
|
||||
if not top or not bottom:
|
||||
continue
|
||||
if not _is_period(top, "", rules) and _is_period(f"{top} {bottom}", "", rules):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _upper_candidate(rows: Grid, row: int, first: int, rules: Rules) -> int | None:
|
||||
above = row - 1
|
||||
while above >= first and is_numbering_row(rows[above], rules):
|
||||
above -= 1
|
||||
return above if above >= first else None
|
||||
|
||||
|
||||
def find_header_rows(rows: Grid, first: int, first_code: int, code_col: int, rules: Rules) -> tuple[int, ...]:
|
||||
"""Строка заголовка (одна или две, склеенные); не найдена — `()`."""
|
||||
candidates = [r for r in range(first, first_code) if not is_numbering_row(rows[r], rules)]
|
||||
captioned = [
|
||||
r for r in candidates if rules.headers.code_caption.search(normalize_caption(_text(cell_at(rows, r, code_col))))
|
||||
]
|
||||
header: int | None = captioned[-1] if captioned else None
|
||||
if header is None:
|
||||
header = next((r for r in reversed(candidates) if _has_header_periods(rows, r, code_col, rules)), None)
|
||||
if header is None:
|
||||
return ()
|
||||
upper = _upper_candidate(rows, header, first, rules)
|
||||
if upper is not None and _glued_with_upper(rows, upper, header, code_col, rules):
|
||||
return (upper, header)
|
||||
return (header,)
|
||||
|
||||
|
||||
def segment_roles(rows: Grid, first: int, last: int, header_rows: tuple[int, ...], rules: Rules) -> list[RowRole]:
|
||||
"""Роли строк `first..last`: выше заголовка — TITLE (только при найденном заголовке)."""
|
||||
roles: list[RowRole] = []
|
||||
top = min(header_rows) if header_rows else None
|
||||
for row in range(first, last + 1):
|
||||
if row in header_rows:
|
||||
roles.append(RowRole.HEADER)
|
||||
elif top is not None and row < top:
|
||||
roles.append(RowRole.TITLE)
|
||||
elif is_numbering_row(rows[row], rules):
|
||||
roles.append(RowRole.NUMBERING)
|
||||
else:
|
||||
roles.append(RowRole.DATA)
|
||||
return roles
|
||||
|
||||
|
||||
def _caption_column(
|
||||
rows: Grid, header_rows: tuple[int, ...], code_col: int, pattern: re.Pattern[str], *, left_of_code: bool
|
||||
) -> int | None:
|
||||
for row in header_rows:
|
||||
for col, cell in enumerate(rows[row]):
|
||||
if col == code_col or (left_of_code and col > code_col):
|
||||
continue
|
||||
if pattern.search(normalize_caption(_text(cell))):
|
||||
return col
|
||||
return None
|
||||
|
||||
|
||||
def _code_rows(rows: Grid, span: range, skip: tuple[int, ...], code_col: int, rules: Rules) -> list[int]:
|
||||
return [r for r in span if r not in skip and is_code_cell(_text(cell_at(rows, r, code_col)), rules)]
|
||||
|
||||
|
||||
def _short_columns(rows: Grid, data: Sequence[int], code_col: int, skip: int | None) -> list[int]:
|
||||
result: list[int] = []
|
||||
for col in range(code_col):
|
||||
if col == skip:
|
||||
continue
|
||||
values = [_text(cell_at(rows, r, col)) for r in data]
|
||||
filled = [v for v in values if v]
|
||||
if filled and all(len(v) <= _SHORT_NOTE for v in filled):
|
||||
result.append(col)
|
||||
return result
|
||||
|
||||
|
||||
def _text_column(rows: Grid, data: Sequence[int], code_col: int, skip: set[int]) -> int | None:
|
||||
counts: Counter[int] = Counter()
|
||||
for col in range(code_col):
|
||||
if col in skip:
|
||||
continue
|
||||
counts[col] = sum(1 for r in data if any(ch.isalpha() for ch in _text(cell_at(rows, r, col))))
|
||||
best: int | None = None
|
||||
for col in sorted(counts): # ничья — ближайший к коду
|
||||
if counts[col] > 0 and (best is None or counts[col] >= counts[best]):
|
||||
best = col
|
||||
return best
|
||||
|
||||
|
||||
def find_name_column( # noqa: PLR0913, PLR0917 - сигнатура задана планом (Task 8)
|
||||
rows: Grid, header_rows: tuple[int, ...], first: int, last: int, code_col: int, rules: Rules
|
||||
) -> tuple[int | None, int | None]:
|
||||
"""(столбец наименования, столбец «Пояснения»)."""
|
||||
name_col = _caption_column(rows, header_rows, code_col, rules.headers.name_caption, left_of_code=False)
|
||||
notes_col = _caption_column(rows, header_rows, code_col, rules.headers.explanations_caption, left_of_code=True)
|
||||
data = _code_rows(rows, range(first, last + 1), header_rows, code_col, rules)
|
||||
if name_col is None:
|
||||
skip = {notes_col} if notes_col is not None else set(_short_columns(rows, data, code_col, None))
|
||||
name_col = _text_column(rows, data, code_col, skip)
|
||||
if notes_col is None:
|
||||
short = _short_columns(rows, data, code_col, name_col)
|
||||
notes_col = short[-1] if short else None
|
||||
return name_col, notes_col
|
||||
|
||||
|
||||
def name_cell(
|
||||
cells: Sequence[Cell | None], name_col: int | None, code_col: int, explanations_col: int | None
|
||||
) -> Cell | None:
|
||||
"""Ячейка наименования; пустая — ближайшая левее кода ячейка с буквами, кроме «Пояснений»."""
|
||||
if name_col is not None and name_col < len(cells):
|
||||
cell = cells[name_col]
|
||||
if cell is not None and cell.text.strip():
|
||||
return cell
|
||||
for col in range(min(code_col, len(cells)) - 1, -1, -1):
|
||||
cell = cells[col]
|
||||
if col != explanations_col and cell is not None and any(ch.isalpha() for ch in cell.text):
|
||||
return cell
|
||||
return None
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from decimal import Decimal
|
||||
from typing import overload
|
||||
|
||||
from arm_statements.model import (
|
||||
HEADER_FIELDS,
|
||||
@ -138,7 +139,17 @@ def excel( # noqa: PLR0913 - сигнатура задана планом (Task
|
||||
)
|
||||
|
||||
|
||||
def place(cells: Mapping[int, PlaceInput], width: int | None = None) -> list[PlaceInput | None]:
|
||||
@overload
|
||||
def place(cells: Mapping[int, str | Decimal], width: int | None = None) -> list[str | Decimal | None]: ...
|
||||
|
||||
|
||||
@overload
|
||||
def place(cells: Mapping[int, PlaceInput], width: int | None = None) -> list[PlaceInput | None]: ...
|
||||
|
||||
|
||||
def place(
|
||||
cells: Mapping[int, PlaceInput] | Mapping[int, str | Decimal], width: int | None = None
|
||||
) -> list[PlaceInput | None] | list[str | Decimal | None]:
|
||||
"""Строка с ячейками в заданных столбцах, остальные None."""
|
||||
size = width if width is not None else (max(cells) + 1 if cells else 0)
|
||||
return [cells.get(i) for i in range(size)]
|
||||
|
||||
139
arm_statements/tests/test_layout.py
Normal file
139
arm_statements/tests/test_layout.py
Normal file
@ -0,0 +1,139 @@
|
||||
from decimal import Decimal as D
|
||||
|
||||
import pytest
|
||||
|
||||
from arm_statements.layout import (
|
||||
code_column,
|
||||
find_code_column,
|
||||
find_header_rows,
|
||||
find_name_column,
|
||||
first_code_row,
|
||||
is_numbering_row,
|
||||
name_cell,
|
||||
segment_roles,
|
||||
)
|
||||
from arm_statements.loader import CodeSection, Rules
|
||||
from arm_statements.model import Cell, DocumentType, RowRole
|
||||
from arm_statements.tests.builders import excel, ocr, place
|
||||
|
||||
BALANCE_ROWS = [
|
||||
["Наименование показателя", "Код", "На 30.09.2025", "На 31.12.2024"],
|
||||
["1", "2", "3", "4"],
|
||||
["Нематериальные активы", "1110", "10", "9"],
|
||||
["Основные средства", "1150", "20", "19"],
|
||||
["Итого по разделу I", "1100", "30", "28"],
|
||||
]
|
||||
|
||||
|
||||
def test_code_column_by_caption(rules: Rules) -> None:
|
||||
cand = ocr(BALANCE_ROWS)
|
||||
assert (find_code_column(cand.rows, 0, 4, rules), first_code_row(cand.rows, 0, 4, 1, rules)) == (1, 2)
|
||||
|
||||
|
||||
def test_title_block_kody_is_not_code_column(rules: Rules) -> None:
|
||||
title = [
|
||||
place({5: v}) for v in ["Коды", "0710001", "30 09 2025", "12345678", "7700000000", "12267", "16", "384", "", ""]
|
||||
]
|
||||
table = [
|
||||
place({0: n, 2: c, 3: D(a)})
|
||||
for n, c, a in [
|
||||
("Запасы", "1210", "5"),
|
||||
("НДС", "1220", "1"),
|
||||
("Дебиторы", "1230", "4"),
|
||||
("Итого", "1200", "10"),
|
||||
]
|
||||
]
|
||||
cand = excel([("Баланс", [*title, *table])])
|
||||
assert [find_code_column(cand.rows, 0, len(cand.rows) - 1, rules)] == [2]
|
||||
|
||||
|
||||
def test_two_line_document(rules: Rules) -> None:
|
||||
cand = ocr([["Наименование", "Код", "2025"], ["Выручка", "2110", "5"], ["Прибыль", "2400", "1"]])
|
||||
assert find_code_column(cand.rows, 0, 2, rules) == 1
|
||||
|
||||
|
||||
def test_relaxed_search_finds_single_code(rules: Rules) -> None:
|
||||
cand = ocr([["Чистая прибыль", "", "2400", "7", "6"]])
|
||||
assert find_code_column(cand.rows, 0, 0, rules) is None
|
||||
assert [find_code_column(cand.rows, 0, 0, rules, relaxed_section=CodeSection.INCOME)] == [2]
|
||||
|
||||
|
||||
def test_public_code_column(rules: Rules) -> None:
|
||||
assert code_column(ocr(BALANCE_ROWS).rows, rules) == 1
|
||||
single = ocr([["Чистая прибыль", "", "2400", "7", "6"]]).rows
|
||||
assert (code_column(single, rules), code_column(single, rules, relaxed=DocumentType.INCOME)) == (None, 2)
|
||||
assert code_column(single, rules, relaxed=DocumentType.BALANCE) is None and code_column((), rules) is None
|
||||
with pytest.raises(ValueError, match="DISCLOSURE"):
|
||||
code_column(single, rules, relaxed=DocumentType.DISCLOSURE)
|
||||
|
||||
|
||||
def test_numbering_rows(rules: Rules) -> None:
|
||||
def row(*texts: str) -> tuple[Cell | None, ...]:
|
||||
return ocr([list(texts)]).rows[0]
|
||||
|
||||
assert is_numbering_row(row("1", "2", "3", "4", "5"), rules)
|
||||
assert is_numbering_row(row("1", "2", "З", "4"), rules)
|
||||
assert is_numbering_row(row("2", "3", "4", "5"), rules)
|
||||
assert is_numbering_row(row("1", "2", "4", "5"), rules)
|
||||
assert not is_numbering_row(row("1", "3", "5"), rules)
|
||||
assert not is_numbering_row(row("1", "2"), rules)
|
||||
|
||||
|
||||
def test_header_row_roles_and_title(rules: Rules) -> None:
|
||||
rows = [["Бухгалтерский баланс", "", "", ""], *BALANCE_ROWS]
|
||||
cand = ocr(rows)
|
||||
header = find_header_rows(cand.rows, 0, 3, 1, rules)
|
||||
assert header == (1,)
|
||||
roles = segment_roles(cand.rows, 0, 5, header, rules)
|
||||
assert roles == [RowRole.TITLE, RowRole.HEADER, RowRole.NUMBERING, RowRole.DATA, RowRole.DATA, RowRole.DATA]
|
||||
assert RowRole.TITLE not in segment_roles(cand.rows, 3, 5, (), rules)
|
||||
|
||||
|
||||
def test_two_row_header_is_glued(rules: Rules) -> None:
|
||||
cand = ocr(
|
||||
[
|
||||
["", "", "На 30 сентября", "На 31 декабря"],
|
||||
["Наименование", "Код", "2025 г.", "2024 г."],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
]
|
||||
)
|
||||
assert find_header_rows(cand.rows, 0, 2, 1, rules) == (0, 1)
|
||||
single = ocr(
|
||||
[
|
||||
["Раздел", "", "", ""],
|
||||
["Наименование", "Код", "На 30.09.2025", "На 31.12.2024"],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
]
|
||||
)
|
||||
assert find_header_rows(single.rows, 0, 2, 1, rules) == (1,)
|
||||
|
||||
|
||||
def test_name_column_without_caption(rules: Rules) -> None:
|
||||
cand = ocr(
|
||||
[
|
||||
["5.1", "Запасы", "1210", "5"],
|
||||
["5.2", "Дебиторская задолженность", "1230", "4"],
|
||||
["", "Итого", "1200", "9"],
|
||||
]
|
||||
)
|
||||
assert find_name_column(cand.rows, (), 0, 2, 2, rules) == (1, 0)
|
||||
|
||||
|
||||
def test_header_without_code_caption_skips_numbering(rules: Rules) -> None:
|
||||
cand = ocr(
|
||||
[
|
||||
["Титул 2025", "", "", ""],
|
||||
["Показатель", "", "На 30.09.2025", "На 31.12.2024"],
|
||||
["1", "2", "3", "4"],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
]
|
||||
)
|
||||
assert find_header_rows(cand.rows, 0, 3, 1, rules) == (1,)
|
||||
assert find_header_rows(cand.rows, 3, 3, 1, rules) == ()
|
||||
|
||||
|
||||
def test_name_cell_falls_back_to_left_text(rules: Rules) -> None:
|
||||
row = ocr([["5.1", "Запасы", "", "1210"]]).rows[0]
|
||||
cell = name_cell(row, 2, 3, 0)
|
||||
assert cell is not None and cell.text == "Запасы"
|
||||
assert name_cell(ocr([["5.1", "", "", "1210"]]).rows[0], 2, 3, 0) is None
|
||||
Reference in New Issue
Block a user