feat: шаг 2 — колонки периодов, относительные даты бланка 66н, неуверенные и мусорные подписи
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
f4533f45e1
commit
995a778fa4
162
arm_statements/columns.py
Normal file
162
arm_statements/columns.py
Normal file
@ -0,0 +1,162 @@
|
||||
"""Шаг 2: колонки значений, подписи и даты колонок."""
|
||||
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from arm_statements.amounts import parse_amount
|
||||
from arm_statements.codes import is_code_cell
|
||||
from arm_statements.dates import expected_dates, parse_period_label, resolve_relative
|
||||
from arm_statements.issues import WarningCode, make_warning
|
||||
from arm_statements.layout import Grid, cell_at
|
||||
from arm_statements.loader import Rules
|
||||
from arm_statements.model import Cell, PeriodKind, Source, StatementColumn, StatementHeader, StatementWarning
|
||||
|
||||
_FULL_CONFIDENCE = 100.0
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ColumnDraft:
|
||||
"""Колонка сегмента до сопоставления; `date` — только у абсолютной подписи."""
|
||||
|
||||
grid_col: int
|
||||
label: str
|
||||
label_confidence: float | None
|
||||
kind: PeriodKind
|
||||
date: date | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class DocumentDates:
|
||||
effective: date | None
|
||||
confident_effective: date | None
|
||||
latest_absolute: date | None
|
||||
confident_latest_absolute: date | None
|
||||
|
||||
|
||||
def _text(cell: Cell | None) -> str:
|
||||
return "" if cell is None else cell.text.strip()
|
||||
|
||||
|
||||
def _code_rows(rows: Grid, data_rows: Sequence[int], code_col: int, rules: Rules) -> list[int]:
|
||||
return [r for r in data_rows if is_code_cell(_text(cell_at(rows, r, code_col)), rules)]
|
||||
|
||||
|
||||
def _is_number(cell: Cell | None, source: Source, rules: Rules) -> bool:
|
||||
if cell is None:
|
||||
return False
|
||||
if isinstance(cell.value, Decimal):
|
||||
return True
|
||||
return parse_amount(cell.text, source, rules).amount is not None
|
||||
|
||||
|
||||
def has_numbers( # noqa: PLR0913, PLR0917 - сигнатура задана планом (Task 9)
|
||||
rows: Grid, data_rows: Sequence[int], col: int, code_col: int, source: Source, rules: Rules
|
||||
) -> bool:
|
||||
"""В строках с кодом есть ячейка столбца с числом (значение-число или разобранная сумма)."""
|
||||
return any(_is_number(cell_at(rows, r, col), source, rules) for r in _code_rows(rows, data_rows, code_col, rules))
|
||||
|
||||
|
||||
def _width(rows: Grid, row_indexes: Sequence[int]) -> int:
|
||||
return max((len(rows[r]) for r in row_indexes if 0 <= r < len(rows)), default=0)
|
||||
|
||||
|
||||
def amount_columns(rows: Grid, code_col: int, rules: Rules, *, source: Source = Source.EXCEL) -> tuple[int, ...]:
|
||||
"""Столбцы правее кодов, где в строках с кодом есть числа, по возрастанию."""
|
||||
every = range(len(rows))
|
||||
return tuple(
|
||||
col
|
||||
for col in range(code_col + 1, _width(rows, every))
|
||||
if has_numbers(rows, every, col, code_col, source, rules)
|
||||
)
|
||||
|
||||
|
||||
def _label(rows: Grid, header_rows: tuple[int, ...], col: int) -> Cell | None:
|
||||
"""Подпись колонки: строки заголовка через пробел, уверенность — минимум; одна ячейка берётся как есть."""
|
||||
cells = [c for r in header_rows if (c := cell_at(rows, r, col)) is not None and c.text.strip()]
|
||||
if not cells:
|
||||
return None
|
||||
if len(cells) == 1:
|
||||
return cells[0]
|
||||
text = " ".join(c.text.strip() for c in cells)
|
||||
return Cell(text, None, min(c.confidence for c in cells), cells[0].ref)
|
||||
|
||||
|
||||
def segment_columns( # noqa: PLR0913, PLR0917 - сигнатура задана планом (Task 9)
|
||||
rows: Grid,
|
||||
header_rows: tuple[int, ...],
|
||||
data_rows: Sequence[int],
|
||||
code_col: int,
|
||||
source: Source,
|
||||
rules: Rules,
|
||||
) -> tuple[list[ColumnDraft], list[StatementWarning]]:
|
||||
"""Колонки значений сегмента и предупреждения о пропущенных/недатированных столбцах."""
|
||||
drafts: list[ColumnDraft] = []
|
||||
warnings: list[StatementWarning] = []
|
||||
for col in range(code_col + 1, _width(rows, [*header_rows, *data_rows])):
|
||||
numbers = has_numbers(rows, data_rows, col, code_col, source, rules)
|
||||
if not header_rows:
|
||||
if numbers:
|
||||
drafts.append(ColumnDraft(col, "", None, PeriodKind.NONE, None))
|
||||
continue
|
||||
cell = _label(rows, header_rows, col)
|
||||
parsed = parse_period_label(cell, rules)
|
||||
if cell is not None and parsed.excluded:
|
||||
message = f"столбец {col} («{cell.text}») пропущен: не период"
|
||||
warnings.append(make_warning(WarningCode.COLUMN_SKIPPED, message, rules, column_index=col))
|
||||
elif cell is not None and parsed.kind is not PeriodKind.NONE:
|
||||
drafts.append(ColumnDraft(col, cell.text, cell.confidence, parsed.kind, parsed.date))
|
||||
elif numbers:
|
||||
label, conf = (cell.text, cell.confidence) if cell is not None else ("", None)
|
||||
message = f"у столбца {col} с числами нет разобранной подписи периода"
|
||||
warnings.append(make_warning(WarningCode.COLUMN_WITHOUT_DATE, message, rules, column_index=col))
|
||||
drafts.append(ColumnDraft(col, label, conf, PeriodKind.NONE, None))
|
||||
return drafts, warnings
|
||||
|
||||
|
||||
def _latest(drafts: Sequence[ColumnDraft], *, confident: bool) -> date | None:
|
||||
days = [
|
||||
d.date
|
||||
for d in drafts
|
||||
if d.kind is PeriodKind.ABSOLUTE
|
||||
and d.date is not None
|
||||
and (not confident or d.label_confidence == _FULL_CONFIDENCE)
|
||||
]
|
||||
return max(days, default=None)
|
||||
|
||||
|
||||
def document_dates(drafts: Sequence[ColumnDraft], header: StatementHeader) -> DocumentDates:
|
||||
"""Эффективная дата и её «уверенный» двойник: неуверенное само себя не подтверждает."""
|
||||
latest = _latest(drafts, confident=False)
|
||||
confident_latest = _latest(drafts, confident=True)
|
||||
report = header.report_date
|
||||
sure_report = report.value if report.confidence == _FULL_CONFIDENCE else None
|
||||
return DocumentDates(
|
||||
effective=report.value or latest,
|
||||
confident_effective=sure_report or confident_latest,
|
||||
latest_absolute=latest,
|
||||
confident_latest_absolute=confident_latest,
|
||||
)
|
||||
|
||||
|
||||
def finalize_columns(
|
||||
drafts: Sequence[ColumnDraft], dates: DocumentDates, rules: Rules
|
||||
) -> tuple[tuple[StatementColumn, ...], list[StatementWarning]]:
|
||||
"""Колонки документа: даты относительных подписей и проверка неуверенных абсолютных."""
|
||||
expected = None if dates.confident_effective is None else expected_dates(dates.confident_effective)
|
||||
columns: list[StatementColumn] = []
|
||||
warnings: list[StatementWarning] = []
|
||||
for draft in drafts:
|
||||
if draft.kind is PeriodKind.ABSOLUTE:
|
||||
period = draft.date
|
||||
unsure = draft.label_confidence is not None and draft.label_confidence < _FULL_CONFIDENCE
|
||||
if unsure and (expected is None or period not in expected):
|
||||
message = f"дата подписи столбца {draft.grid_col} («{draft.label}») не подтверждена"
|
||||
warnings.append(
|
||||
make_warning(WarningCode.COLUMN_DATE_UNCERTAIN, message, rules, column_index=draft.grid_col)
|
||||
)
|
||||
else:
|
||||
period = resolve_relative(draft.kind, dates.effective)
|
||||
columns.append(StatementColumn(draft.grid_col, draft.label, draft.label_confidence, period, draft.kind))
|
||||
return tuple(columns), warnings
|
||||
146
arm_statements/tests/test_columns.py
Normal file
146
arm_statements/tests/test_columns.py
Normal file
@ -0,0 +1,146 @@
|
||||
from collections.abc import Sequence
|
||||
from datetime import date
|
||||
from decimal import Decimal as D
|
||||
from typing import Any
|
||||
|
||||
from arm_statements.columns import amount_columns, document_dates, finalize_columns, segment_columns
|
||||
from arm_statements.header import parse_header
|
||||
from arm_statements.loader import Rules
|
||||
from arm_statements.model import HeaderCells, PeriodKind, Severity, Source, StatementColumn, StatementWarning
|
||||
from arm_statements.tests.builders import excel, hdr, ocr
|
||||
|
||||
OCR, EXCEL = Source.OCR, Source.EXCEL
|
||||
|
||||
|
||||
def run( # noqa: PLR0913, PLR0917 - вспомогательная функция тестов, сигнатура задана планом (Task 9)
|
||||
rows: Sequence[Sequence[Any]],
|
||||
source: Source,
|
||||
rules: Rules,
|
||||
header: HeaderCells | None = None,
|
||||
header_rows: tuple[int, ...] = (0,),
|
||||
data_rows: tuple[int, ...] = (1, 2),
|
||||
) -> tuple[tuple[StatementColumn, ...], list[StatementWarning]]:
|
||||
cand = ocr(rows) if source is OCR else excel([("Лист1", rows)])
|
||||
drafts, warnings = segment_columns(cand.rows, header_rows, data_rows, 1 if source is OCR else 2, source, rules)
|
||||
head = parse_header(header or hdr(), rules)
|
||||
columns, more = finalize_columns(drafts, document_dates(drafts, head), rules)
|
||||
return columns, [*warnings, *more]
|
||||
|
||||
|
||||
def test_blank_66n_relative_labels(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", "На отчетную дату", "На 31 декабря предыдущего года",
|
||||
"На 31 декабря года, предшествующего предыдущему"],
|
||||
["Запасы", "1210", "5", "4", "3"],
|
||||
["Итого", "1200", "5", "4", "3"],
|
||||
] # fmt: skip
|
||||
columns, warnings = run(rows, OCR, rules, header=hdr(report_date="30.09.2025"))
|
||||
assert [c.period_date for c in columns] == [date(2025, 9, 30), date(2024, 12, 31), date(2023, 12, 31)]
|
||||
assert [c.period_kind for c in columns] == [
|
||||
PeriodKind.CURRENT,
|
||||
PeriodKind.PREV_YEAR_END,
|
||||
PeriodKind.PREV_PREV_YEAR_END,
|
||||
]
|
||||
assert warnings == []
|
||||
|
||||
|
||||
def test_excel_skipped_change_column(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "", "Код", "На 30.09.2025", "На 31.12.2024", "Изменение, %"],
|
||||
["Запасы", "", "1210", D("5"), D("4"), D("25")],
|
||||
["Итого", "", "1200", D("5"), D("4"), D("25")],
|
||||
]
|
||||
columns, warnings = run(rows, EXCEL, rules)
|
||||
assert [c.column_index for c in columns] == [3, 4]
|
||||
assert [(w.code, w.severity, w.column_index) for w in warnings] == [("COLUMN_SKIPPED", Severity.INFO, 5)]
|
||||
assert {c.label_confidence for c in columns} == {100.0}
|
||||
|
||||
|
||||
def test_junk_label_over_numbers_is_kept(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", "На 30.09.2025", ("| | г", 17.0)],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
["Итого", "1200", "5", "4"],
|
||||
]
|
||||
columns, warnings = run(rows, OCR, rules, header=hdr(report_date="30.09.2025"))
|
||||
junk = columns[1]
|
||||
assert (junk.label, junk.period_date, junk.period_kind) == ("| | г", None, PeriodKind.NONE)
|
||||
assert [(w.code, w.severity, w.column_index) for w in warnings] == [("COLUMN_WITHOUT_DATE", Severity.ERROR, 3)]
|
||||
|
||||
|
||||
def test_empty_label_over_numbers_is_kept(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", "На 30.09.2025", ""],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
["Итого", "1200", "5", "4"],
|
||||
]
|
||||
columns, warnings = run(rows, OCR, rules)
|
||||
assert [(c.column_index, c.label, c.label_confidence, c.period_kind) for c in columns] == [
|
||||
(2, "На 30.09.2025", 100.0, PeriodKind.ABSOLUTE),
|
||||
(3, "", None, PeriodKind.NONE),
|
||||
]
|
||||
assert [(w.code, w.column_index) for w in warnings] == [("COLUMN_WITHOUT_DATE", 3)]
|
||||
|
||||
|
||||
def test_two_row_header_is_joined_with_min_confidence(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["", "", ("На 31 декабря", 90.0)],
|
||||
["Наименование", "Код", ("2024 г.", 80.0)],
|
||||
["Запасы", "1210", "5"],
|
||||
]
|
||||
cand = ocr(rows)
|
||||
drafts, _ = segment_columns(cand.rows, (0, 1), (2,), 1, OCR, rules)
|
||||
assert [(d.grid_col, d.label, d.label_confidence, d.date) for d in drafts] == [
|
||||
(2, "На 31 декабря 2024 г.", 80.0, date(2024, 12, 31))
|
||||
]
|
||||
|
||||
|
||||
def test_sparse_ocr_columns(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", "На 30.09.2025", None, "На 31.12.2024", "На 31.12.2023"],
|
||||
["Запасы", "1210", "5", None, "4", "3"],
|
||||
["Итого", "1200", "5", None, "4", "3"],
|
||||
]
|
||||
columns, _ = run(rows, OCR, rules)
|
||||
assert [c.column_index for c in columns] == [2, 4, 5]
|
||||
|
||||
|
||||
def test_uncertain_label_does_not_confirm_itself(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", ("На 30.09.2О25", 60.0), ""],
|
||||
["Запасы", "1210", "5", "4"],
|
||||
["Итого", "1200", "5", "4"],
|
||||
]
|
||||
cand = ocr(rows)
|
||||
drafts, _ = segment_columns(cand.rows, (0,), (1, 2), 1, OCR, rules)
|
||||
dates = document_dates(drafts, parse_header(hdr(), rules))
|
||||
assert (dates.effective, dates.confident_effective) == (date(2025, 9, 30), None)
|
||||
_, warnings = finalize_columns(drafts, dates, rules)
|
||||
assert [(w.code, w.column_index) for w in warnings] == [("COLUMN_DATE_UNCERTAIN", 2)]
|
||||
|
||||
|
||||
def test_uncertain_label_confirmed_by_report_date(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "Код", "На 30.09.2025", ("На 31.12.2024", 60.0), ("На 31.03.2025", 60.0)],
|
||||
["Запасы", "1210", "5", "4", "3"],
|
||||
["Итого", "1200", "5", "4", "3"],
|
||||
]
|
||||
_, warnings = run(rows, OCR, rules, header=hdr(report_date="30.09.2025"))
|
||||
assert [(w.code, w.column_index) for w in warnings] == [("COLUMN_DATE_UNCERTAIN", 4)]
|
||||
|
||||
|
||||
def test_segment_without_header(rules: Rules) -> None:
|
||||
cand = ocr([["Запасы", "1210", "5", "", "4"], ["Итого", "1200", "5", "", "4"]])
|
||||
drafts, warnings = segment_columns(cand.rows, (), (0, 1), 1, OCR, rules)
|
||||
assert [(d.grid_col, d.label, d.kind) for d in drafts] == [(2, "", PeriodKind.NONE), (4, "", PeriodKind.NONE)]
|
||||
assert warnings == []
|
||||
|
||||
|
||||
def test_public_amount_columns(rules: Rules) -> None:
|
||||
rows: list[list[Any]] = [
|
||||
["Наименование", "", "Код", "На 30.09.2025", "На 31.12.2024", "Изменение"],
|
||||
["Запасы", "", "1210", D("5"), D("4"), "рост"],
|
||||
["Итого", "", "1200", D("5"), "", "рост"],
|
||||
]
|
||||
assert amount_columns(excel([("Лист1", rows)]).rows, 2, rules) == (3, 4)
|
||||
assert amount_columns(ocr([["Запасы", "1210", "5", "", "4"]]).rows, 1, rules, source=OCR) == (2, 4)
|
||||
Loading…
x
Reference in New Issue
Block a user