diff --git a/arm_statements/columns.py b/arm_statements/columns.py new file mode 100644 index 0000000..75ae3ff --- /dev/null +++ b/arm_statements/columns.py @@ -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 diff --git a/arm_statements/tests/test_columns.py b/arm_statements/tests/test_columns.py new file mode 100644 index 0000000..2a12edc --- /dev/null +++ b/arm_statements/tests/test_columns.py @@ -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)