ad86a55716
Копия api/arm_statements: 3390f92..25b2bb8 (1 коммитов), скрипт docs/scripts/sync-statements.sh. Коммиты arm-statements: - 72e2e72 fix(#8): шаг 2 — нумерация со сбоем OCR под заголовком, повтор шапки со склеенной нумерацией (AURORA-1517) statements-check arm-core: 0 — копия синхронна с arm-statements@origin/main (25b2bb8) Co-Authored-By: sync-statements.sh <noreply@aepif.ru>
295 lines
13 KiB
Python
295 lines
13 KiB
Python
"""Шаг 2: сегменты сетки, повторные шапки, границы листов и сопоставление колонок с первым сегментом."""
|
||
|
||
from collections.abc import Sequence
|
||
from dataclasses import dataclass, replace
|
||
from typing import NamedTuple
|
||
|
||
from arm_statements.codes import code_parts, is_code_cell, majority_section, parse_code
|
||
from arm_statements.columns import (
|
||
ColumnDraft,
|
||
DocumentDates,
|
||
document_dates,
|
||
finalize_columns,
|
||
merge_outside_table,
|
||
outside_table_warning,
|
||
segment_columns,
|
||
trailing_outside_refs,
|
||
)
|
||
from arm_statements.issues import WarningCode, make_warning
|
||
from arm_statements.layout import (
|
||
Grid,
|
||
caption_text,
|
||
cell_at,
|
||
find_code_column,
|
||
find_header_rows,
|
||
find_name_column,
|
||
first_code_row,
|
||
is_numbering_row,
|
||
segment_roles,
|
||
)
|
||
from arm_statements.loader import CodeSection, Rules
|
||
from arm_statements.model import (
|
||
FormCandidate,
|
||
PeriodKind,
|
||
RowRole,
|
||
StatementColumn,
|
||
StatementHeader,
|
||
StatementWarning,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class SegmentLayout:
|
||
"""Раскладка сегмента; `value_cols` — (столбец сетки сегмента, `column_index` колонки документа)."""
|
||
|
||
first_row: int
|
||
last_row: int
|
||
header_rows: tuple[int, ...]
|
||
code_col: int | None
|
||
name_col: int | None
|
||
explanations_col: int | None
|
||
value_cols: tuple[tuple[int, int], ...]
|
||
extra_cols: tuple[int, ...]
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Layout:
|
||
"""Раскладка кандидата; `roles` — по всем строкам сетки."""
|
||
|
||
code_found: bool
|
||
column_section: CodeSection | None
|
||
columns: tuple[StatementColumn, ...]
|
||
segments: tuple[SegmentLayout, ...]
|
||
roles: tuple[RowRole, ...]
|
||
dates: DocumentDates
|
||
warnings: tuple[StatementWarning, ...]
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _Analysed:
|
||
"""Сегмент до сопоставления колонок."""
|
||
|
||
first: int
|
||
last: int
|
||
header_rows: tuple[int, ...]
|
||
code_col: int | None
|
||
name_col: int | None
|
||
explanations_col: int | None
|
||
roles: tuple[RowRole, ...]
|
||
drafts: tuple[ColumnDraft, ...]
|
||
draft_warnings: tuple[StatementWarning, ...]
|
||
has_content: bool
|
||
|
||
|
||
class _Bounds(NamedTuple):
|
||
last: int
|
||
reference: frozenset[str]
|
||
|
||
|
||
def _text(rows: Grid, row: int, col: int) -> str:
|
||
cell = cell_at(rows, row, col)
|
||
return "" if cell is None else cell.text.strip()
|
||
|
||
|
||
def _chunks(rows: Grid, breaks: Sequence[int]) -> list[tuple[int, int]]:
|
||
"""Куски сетки по границам листов; границы вне сетки и повторы отбрасываются."""
|
||
cuts = sorted({b for b in breaks if 0 < b < len(rows)})
|
||
starts = [0, *cuts]
|
||
ends = [*(c - 1 for c in cuts), len(rows) - 1]
|
||
return list(zip(starts, ends, strict=True))
|
||
|
||
|
||
def _caption_cells(rows: Grid, header_rows: Sequence[int]) -> frozenset[str]:
|
||
cells = (caption_text(_text(rows, r, c)) for r in header_rows for c in range(len(rows[r])))
|
||
return frozenset(c for c in cells if c)
|
||
|
||
|
||
def _is_repeat_header(rows: Grid, row: int, code_col: int | None, reference: frozenset[str], rules: Rules) -> bool:
|
||
"""Строка-шапка: ячейка подписи кода или не менее двух ячеек, равных ячейкам шапки первого сегмента."""
|
||
if is_numbering_row(rows[row], rules):
|
||
return False
|
||
if code_col is not None and is_code_cell(_text(rows, row, code_col), rules):
|
||
return False
|
||
normalized = [caption_text(c.text) for c in rows[row] if c is not None and c.text.strip()]
|
||
if any(rules.headers.code_caption.search(text) for text in normalized):
|
||
return True
|
||
return sum(1 for text in normalized if text and text in reference) >= _MIN_SAME_CELLS
|
||
|
||
|
||
_MIN_SAME_CELLS = 2
|
||
|
||
|
||
def _segment_bounds(rows: Grid, start: int, chunk_last: int, reference: frozenset[str], rules: Rules) -> _Bounds:
|
||
"""Последняя строка сегмента: до следующей шапки, стоящей после первой строки с кодом."""
|
||
code_col = find_code_column(rows, start, chunk_last, rules)
|
||
first_code = None if code_col is None else first_code_row(rows, start, chunk_last, code_col, rules)
|
||
if code_col is None or first_code is None:
|
||
return _Bounds(chunk_last, reference)
|
||
ref = reference or _caption_cells(rows, find_header_rows(rows, start, first_code, code_col, rules))
|
||
for row in range(first_code + 1, chunk_last + 1):
|
||
if _is_repeat_header(rows, row, code_col, ref, rules):
|
||
return _Bounds(row - 1, ref)
|
||
return _Bounds(chunk_last, ref)
|
||
|
||
|
||
def _segment_codes(rows: Grid, first: int, last: int, code_col: int, rules: Rules) -> list[str]:
|
||
codes: list[str] = []
|
||
for row in range(first, last + 1):
|
||
text = _text(rows, row, code_col)
|
||
if is_code_cell(text, rules):
|
||
codes.extend(c for part in code_parts(text) if (c := parse_code(part, rules).code) is not None)
|
||
return codes
|
||
|
||
|
||
def _analyse( # noqa: PLR0913, PLR0917 - внутренняя сборка сегмента
|
||
candidate: FormCandidate, first: int, last: int, section: CodeSection | None, rules: Rules
|
||
) -> _Analysed:
|
||
rows = candidate.rows
|
||
code_col = find_code_column(rows, first, last, rules)
|
||
if code_col is None and section is not None:
|
||
code_col = find_code_column(rows, first, last, rules, relaxed_section=section)
|
||
has_content = any(c is not None and c.text.strip() for r in range(first, last + 1) for c in rows[r])
|
||
first_code = None if code_col is None else first_code_row(rows, first, last, code_col, rules)
|
||
if code_col is None or first_code is None:
|
||
return _Analysed(first, last, (), None, None, None, (RowRole.DATA,) * (last - first + 1), (), (), has_content)
|
||
header_rows = find_header_rows(rows, first, first_code, code_col, rules)
|
||
roles = tuple(segment_roles(rows, first, last, header_rows, rules))
|
||
data_rows = [r for r, role in zip(range(first, last + 1), roles, strict=True) if role is RowRole.DATA]
|
||
name_col, notes_col = find_name_column(rows, header_rows, first, last, code_col, rules)
|
||
drafts, warnings = segment_columns(rows, header_rows, data_rows, code_col, candidate.source, rules)
|
||
return _Analysed(
|
||
first, last, header_rows, code_col, name_col, notes_col, roles, tuple(drafts), tuple(warnings), has_content
|
||
)
|
||
|
||
|
||
def _split_and_analyse(candidate: FormCandidate, rules: Rules) -> tuple[list[_Analysed], CodeSection | None]:
|
||
rows = candidate.rows
|
||
result: list[_Analysed] = []
|
||
reference: frozenset[str] = frozenset()
|
||
section: CodeSection | None = None
|
||
for chunk_first, chunk_last in _chunks(rows, candidate.segment_breaks):
|
||
start = chunk_first
|
||
while start <= chunk_last:
|
||
bounds = _segment_bounds(rows, start, chunk_last, reference, rules)
|
||
reference = reference or bounds.reference
|
||
seg = _analyse(candidate, start, bounds.last, section, rules)
|
||
if section is None and seg.code_col is not None:
|
||
section = majority_section(_segment_codes(rows, seg.first, seg.last, seg.code_col, rules), rules)
|
||
result.append(seg)
|
||
start = bounds.last + 1
|
||
return result, section
|
||
|
||
|
||
def _map_by_date(
|
||
drafts: Sequence[ColumnDraft], columns: Sequence[StatementColumn]
|
||
) -> tuple[tuple[int, int], ...] | None:
|
||
"""Сопоставление по `period_date`: у каждой колонки дата есть и совпадает ровно с одной колонкой документа."""
|
||
if not drafts or any(d.date is None for d in drafts):
|
||
return None
|
||
pairs: list[tuple[int, int]] = []
|
||
for draft in drafts:
|
||
found = [c.column_index for c in columns if c.period_date == draft.date]
|
||
if len(found) != 1:
|
||
return None
|
||
pairs.append((draft.grid_col, found[0]))
|
||
if len({target for _, target in pairs}) != len(pairs):
|
||
return None
|
||
return tuple(pairs)
|
||
|
||
|
||
def _map_segment(
|
||
seg: _Analysed, columns: tuple[StatementColumn, ...], rules: Rules
|
||
) -> tuple[SegmentLayout, StatementWarning | None]:
|
||
"""Колонки сегмента → колонки документа; по порядку и разное число — `SEGMENT_COLUMNS_MISMATCH`."""
|
||
drafts = seg.drafts
|
||
mapped = _map_by_date(drafts, columns) if seg.header_rows else None
|
||
warning: StatementWarning | None = None
|
||
extra: tuple[int, ...] = ()
|
||
if mapped is None:
|
||
size = min(len(drafts), len(columns))
|
||
mapped = tuple((d.grid_col, c.column_index) for d, c in zip(drafts[:size], columns[:size], strict=False))
|
||
extra = tuple(d.grid_col for d in drafts[size:])
|
||
if len(drafts) != len(columns):
|
||
tail = f"; лишние столбцы сегмента: {', '.join(str(c) for c in extra)}" if extra else ""
|
||
message = (
|
||
f"в сегменте со строки {seg.first} колонок значений {len(drafts)}, "
|
||
f"в первом сегменте {len(columns)}{tail}"
|
||
)
|
||
warning = make_warning(WarningCode.SEGMENT_COLUMNS_MISMATCH, message, rules, locator=None)
|
||
layout = SegmentLayout(
|
||
seg.first, seg.last, seg.header_rows, seg.code_col, seg.name_col, seg.explanations_col, mapped, extra
|
||
)
|
||
return layout, warning
|
||
|
||
|
||
def _trim_outside(seg: _Analysed, width: int, candidate: FormCandidate, rules: Rules) -> _Analysed:
|
||
"""Сегмент без подписанных колонок: колонки сверх `width` документа — числа правее таблицы, не колонки."""
|
||
if seg.code_col is None or len(seg.drafts) <= width or any(d.kind is not PeriodKind.NONE for d in seg.drafts):
|
||
return seg
|
||
tail = seg.drafts[width:]
|
||
if any(d.label for d in tail): # столбец с текстом подписи — не формула аналитика, остаётся расхождением
|
||
return seg
|
||
data = [r for r, role in zip(range(seg.first, seg.last + 1), seg.roles, strict=True) if role is RowRole.DATA]
|
||
refs = trailing_outside_refs(
|
||
candidate.rows, data, [d.grid_col for d in tail], seg.code_col, candidate.source, rules
|
||
)
|
||
return replace(
|
||
seg, drafts=seg.drafts[:width], draft_warnings=(*seg.draft_warnings, outside_table_warning(refs, rules))
|
||
)
|
||
|
||
|
||
def find_layout(candidate: FormCandidate, header: StatementHeader, rules: Rules) -> Layout:
|
||
"""Раскладка кандидата: сегменты, колонки документа, роли всех строк и предупреждения шага 2."""
|
||
segments, section = _split_and_analyse(candidate, rules)
|
||
coded = [s for s in segments if s.code_col is not None]
|
||
defining = next((s for s in coded if s.header_rows), coded[0] if coded else None)
|
||
warnings: list[StatementWarning] = []
|
||
columns: tuple[StatementColumn, ...] = ()
|
||
dates = document_dates([], header)
|
||
if defining is not None:
|
||
dates = document_dates(defining.drafts, header)
|
||
columns, finalize_warnings = finalize_columns(defining.drafts, dates, rules)
|
||
# числа правее таблицы — по всем сегментам (формула аналитика может стоять на любой странице формы);
|
||
# сегмент без подписанных колонок (лист-продолжение) сверх колонок документа тоже «правее таблицы»
|
||
if any(d.kind is not PeriodKind.NONE for d in defining.drafts):
|
||
segments = [
|
||
seg if seg is defining else _trim_outside(seg, len(columns), candidate, rules) for seg in segments
|
||
]
|
||
warnings.extend(
|
||
merge_outside_table(
|
||
[
|
||
w
|
||
for seg in segments
|
||
for w in seg.draft_warnings
|
||
if seg is defining or w.code == WarningCode.NUMBERS_OUTSIDE_TABLE
|
||
],
|
||
rules,
|
||
)
|
||
)
|
||
warnings.extend(finalize_warnings)
|
||
if not any(s.header_rows for s in coded):
|
||
warnings.append(
|
||
make_warning(WarningCode.HEADER_ROW_NOT_FOUND, "строка заголовка таблицы не найдена", rules)
|
||
)
|
||
laid: list[SegmentLayout] = []
|
||
for seg in segments:
|
||
if seg.code_col is None:
|
||
laid.append(SegmentLayout(seg.first, seg.last, (), None, None, None, (), ()))
|
||
if coded and seg.has_content:
|
||
message = f"в сегменте со строки {seg.first} не найден столбец кодов"
|
||
warnings.append(make_warning(WarningCode.SEGMENT_WITHOUT_CODES, message, rules))
|
||
elif seg is defining:
|
||
own = tuple((c.column_index, c.column_index) for c in columns)
|
||
laid.append(
|
||
SegmentLayout(
|
||
seg.first, seg.last, seg.header_rows, seg.code_col, seg.name_col, seg.explanations_col, own, ()
|
||
)
|
||
)
|
||
else:
|
||
segment_layout, mismatch = _map_segment(seg, columns, rules)
|
||
laid.append(segment_layout)
|
||
if mismatch is not None:
|
||
warnings.append(mismatch)
|
||
roles = tuple(role for seg in segments for role in seg.roles)
|
||
return Layout(bool(coded), section, columns, tuple(laid), roles, dates, tuple(warnings))
|