7c098dbbf4
Копия api/arm_statements: 3390f92..25b2bb8 (1 коммитов), скрипт docs/scripts/sync-statements.sh. Коммиты arm-statements: - 72e2e72 fix(#8): шаг 2 — нумерация со сбоем OCR под заголовком, повтор шапки со склеенной нумерацией (AURORA-1517) statements-check tabular-parser: 0 — копия синхронна с arm-statements@origin/main (25b2bb8) Co-Authored-By: sync-statements.sh <noreply@aepif.ru>
335 lines
15 KiB
Python
335 lines
15 KiB
Python
"""Шаг 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]+")
|
||
_NUMBERING_TAIL = re.compile(r"\s+[0-9]{1,2}$")
|
||
_FULL_CONFIDENCE = 100.0
|
||
|
||
|
||
def _text(cell: Cell | None) -> str:
|
||
return "" if cell is None else cell.text.strip()
|
||
|
||
|
||
def caption_text(text: str) -> str:
|
||
"""Подпись шапки для сравнения: `normalize_caption` без номера столбца, склеенного снизу («Коды\\n2» → «коды»)."""
|
||
return _NUMBERING_TAIL.sub("", normalize_caption(text))
|
||
|
||
|
||
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, *, noisy: bool = False) -> bool:
|
||
"""≥ 3 непустых ячеек-целых, слева направо +1, начало 1 или 2, не больше одного пропуска.
|
||
|
||
`noisy` — строка сразу под шапкой (AURORA-1517): ячейка со сбоем OCR даёт своё единственное целое («• ■ 1» → 1,
|
||
«3гола» → 3), ячейка без цифр («/», «V») — пропуск, два целых в ячейке — не нумерация.
|
||
"""
|
||
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 not None:
|
||
numbers.append(int(fixed))
|
||
continue
|
||
if not noisy:
|
||
return False
|
||
found = _NUMBER.findall(text)
|
||
if len(found) > 1:
|
||
return False
|
||
numbers.extend(int(number) for number in found)
|
||
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 _lower_candidate(rows: Grid, row: int, first_code: int, rules: Rules) -> int | None:
|
||
below = row + 1
|
||
while below < first_code and is_numbering_row(rows[below], rules):
|
||
below += 1
|
||
return below if below < first_code 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(caption_text(_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)
|
||
lower = _lower_candidate(rows, header, first_code, rules)
|
||
if lower is not None and _glued_with_upper(rows, header, lower, code_col, rules):
|
||
return (header, lower)
|
||
return (header,)
|
||
|
||
|
||
def segment_roles(rows: Grid, first: int, last: int, header_rows: tuple[int, ...], rules: Rules) -> list[RowRole]:
|
||
"""Роли строк `first..last`: выше заголовка — TITLE (только при найденном заголовке); строка сразу под заголовком —
|
||
нумерация и при сбое OCR (`is_numbering_row(noisy=True)`)."""
|
||
roles: list[RowRole] = []
|
||
top = min(header_rows) if header_rows else None
|
||
under = max(header_rows) + 1 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, noisy=row == under):
|
||
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(caption_text(_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
|