From c858a85d852eece3c28164ae6de536f7acabcfb4 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 20:24:47 +0400 Subject: [PATCH] =?UTF-8?q?feat(statements):=20=D0=B0=D0=B4=D0=B0=D0=BF?= =?UTF-8?q?=D1=82=D0=B5=D1=80=20=D1=88=D0=B0=D0=B3=D0=B0=201=20=E2=80=94?= =?UTF-8?q?=20=D0=BA=D0=BD=D0=B8=D0=B3=D0=B0=20=E2=86=92=20FormCandidate,?= =?UTF-8?q?=20=D1=81=D0=BA=D0=BB=D0=B5=D0=B9=D0=BA=D0=B0=20=D0=BB=D0=B8?= =?UTF-8?q?=D1=81=D1=82=D0=BE=D0=B2,=20segment=5Fbreaks=20(#54)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Тест test_npz_candidates красный: НПЗ — BLOCKED (Р-31), см. отчёт задачи 5. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- api/src/parser/detect.py | 10 +- api/src/parser/statements/adapter.py | 221 ++++++++++++++++++ .../fixtures/test_statements_adapter_real.py | 42 ++++ .../unit/parser/statements/test_adapter.py | 153 ++++++++++++ 4 files changed, 424 insertions(+), 2 deletions(-) create mode 100644 api/src/parser/statements/adapter.py create mode 100644 api/tests/fixtures/test_statements_adapter_real.py create mode 100644 api/tests/unit/parser/statements/test_adapter.py diff --git a/api/src/parser/detect.py b/api/src/parser/detect.py index 46b5bda..249aa0a 100644 --- a/api/src/parser/detect.py +++ b/api/src/parser/detect.py @@ -31,6 +31,7 @@ __all__ = [ "first_data_row", "inn_from_cell", "is_header_repeat", + "is_osv_sheet", ] CollapsedGroup = Literal["opening", "closing"] @@ -154,11 +155,16 @@ _STATEMENT_KINDS = { } +def is_osv_sheet(sheet: Sheet, rules: Rules) -> bool: + """Признак ведомости (`sheet_markers`) в первых `SHEET_MARKER_ROWS` строках листа.""" + head = [text for row in range(1, min(sheet.n_rows, SHEET_MARKER_ROWS) + 1) for _, text in _texts(sheet, row)] + return any(_search(rules.headers.sheet_markers, text) for text in head) + + def classify_sheet(sheet: Sheet, rules: Rules) -> SheetKind: """ОСВ — если в первых строках есть признак ведомости; иначе форма ББ/ОФР, её продолжение или прочее — по `statements.detect` (признаки формы ищутся по всему листу до таблицы).""" - head = [text for row in range(1, min(sheet.n_rows, SHEET_MARKER_ROWS) + 1) for _, text in _texts(sheet, row)] - if any(_search(rules.headers.sheet_markers, text) for text in head): + if is_osv_sheet(sheet, rules): return SheetKind.OSV return _STATEMENT_KINDS[classify_statement_sheet(sheet, rules)] diff --git a/api/src/parser/statements/adapter.py b/api/src/parser/statements/adapter.py new file mode 100644 index 0000000..cbf59cf --- /dev/null +++ b/api/src/parser/statements/adapter.py @@ -0,0 +1,221 @@ +"""Адаптер книги -> `FormCandidate` (план 4, задача 5; спека «Шаг 1 — многостраничная форма», §11.2 п. 2–4). + +`plan_sheets` раскладывает листы по видам (ОСВ / форма / продолжение / прочее), `build_candidates` собирает из листов +форм кандидатов для пакета: блок формы -> кандидат, лист-продолжение присоединяется к открытой форме (`segment_breaks` +— границы листов в общей сетке) либо становится отдельным кандидатом с предупреждением. +""" + +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field + +from arm_statements import Cell as StatementCell +from arm_statements import ( + DocumentType, + Evidence, + FormCandidate, + HeaderCells, + Source, + StatementWarning, + WarningCode, + continuation_type, + make_warning, + sheet_joins, +) +from src.parser.deadline import Deadline +from src.parser.detect import SheetKind +from src.parser.grid import Sheet +from src.parser.statements.cells import grid_rows +from src.parser.statements.forms import ( + StatementRules, + classify_statement_sheet, + code_rows, + codes_of, + continuation_marker, + find_blocks, + is_weak_continuation, + sheet_amount_columns, + sheet_code_column, +) +from src.parser.statements.header_cells import header_cells + +__all__ = ["SheetPlan", "build_candidates", "plan_sheets"] + +MAX_FORMAT_LENGTH = 30 +_OPEN_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME}) + +_Row = tuple[StatementCell | None, ...] + + +@dataclass(frozen=True) +class SheetPlan: + osv: tuple[Sheet, ...] + statements: tuple[tuple[Sheet, SheetKind], ...] + other: tuple[Sheet, ...] + + +def plan_sheets(sheets: Sequence[Sheet], is_osv: Callable[[Sheet], bool], rules: StatementRules) -> SheetPlan: + """Листы по порядку книги: ОСВ; форма; продолжение (в том числе слабое — лист из 1–2 кодов сразу после листа + формы или продолжения); прочее. Лист ОСВ или «прочее» прерывает «сразу после».""" + osv: list[Sheet] = [] + statements: list[tuple[Sheet, SheetKind]] = [] + other: list[Sheet] = [] + last_type: DocumentType | None = None # тип последнего блока последнего листа формы; None — формы «сразу до» нет + for sheet in sheets: + if is_osv(sheet): + osv.append(sheet) + last_type = None + continue + kind = classify_statement_sheet(sheet, rules) + if kind == "statement": + statements.append((sheet, SheetKind.STATEMENT)) + blocks = find_blocks(sheet, rules) + last_type = blocks[-1].type_hint if blocks else None + elif kind == "continuation": + statements.append((sheet, SheetKind.STATEMENT_CONTINUATION)) + elif last_type is not None and is_weak_continuation(sheet, rules, last_type): + statements.append((sheet, SheetKind.STATEMENT_CONTINUATION)) + else: + other.append(sheet) + last_type = None + return SheetPlan(tuple(osv), tuple(statements), tuple(other)) + + +@dataclass +class _Draft: + """Кандидат в сборке.""" + + type_hint: DocumentType | None + sheet_names: list[str] + evidence: Evidence + header: HeaderCells + rows: list[_Row] + form_amount_columns: int = 0 + segment_breaks: list[int] = field(default_factory=list) + formats: set[str] = field(default_factory=set) + warnings: list[StatementWarning] = field(default_factory=list) + + @property + def locator(self) -> str: + return "sheet:" + ",".join(self.sheet_names) + + +def _shorten(fmt: str) -> str: + return fmt if len(fmt) <= MAX_FORMAT_LENGTH else fmt[: MAX_FORMAT_LENGTH - 1] + "…" + + +def _form_block_drafts(sheet: Sheet, rules: StatementRules) -> list[_Draft]: + drafts: list[_Draft] = [] + for block in find_blocks(sheet, rules): + formats: set[str] = set() + col = sheet_code_column(sheet, rules, block.first_row, block.last_row) + rows_with_codes = [] if col is None else code_rows(sheet, rules, col, block.first_row, block.last_row) + header_last = max(block.first_row, rows_with_codes[0] - 1) if rows_with_codes else block.last_row + header, evidence = header_cells(sheet, rules, block, header_last, formats) + amounts = 0 if col is None else sheet_amount_columns(sheet, rules, col, block.first_row, block.last_row) + drafts.append( + _Draft( + type_hint=block.type_hint, + sheet_names=[sheet.name], + evidence=evidence, + header=header, + rows=grid_rows(sheet, block.first_row, block.last_row, sheet.n_cols, formats), + form_amount_columns=amounts, + formats=formats, + ) + ) + return drafts + + +def _separate(sheet: Sheet, type_hint: DocumentType | None) -> _Draft: + formats: set[str] = set() + return _Draft( + type_hint=type_hint, + sheet_names=[sheet.name], + evidence=Evidence(), + header=HeaderCells(), + rows=grid_rows(sheet, 1, sheet.n_rows, sheet.n_cols, formats), + formats=formats, + ) + + +def _sheet_joins(sheet: Sheet, form: _Draft, rules: StatementRules) -> bool: + """Лист без маркера присоединяется по кодам и числу колонок сумм (как сегмент шага 2 пакета).""" + form_type = form.type_hint + assert form_type is not None # noqa: S101 - открытая форма всегда с типом + col = sheet_code_column(sheet, rules, 1, sheet.n_rows) + if col is None: + col = sheet_code_column(sheet, rules, 1, sheet.n_rows, relaxed=form_type) + if col is None: + return False + codes = codes_of(sheet, rules, col, 1, sheet.n_rows) + amounts = sheet_amount_columns(sheet, rules, col, 1, sheet.n_rows) + return sheet_joins(form_type, codes, amounts, form.form_amount_columns, rules) + + +def _continuation(sheet: Sheet, drafts: list[_Draft], rules: StatementRules) -> None: + open_form = next((draft for draft in reversed(drafts) if draft.type_hint in _OPEN_TYPES), None) + marker = continuation_marker(sheet, rules) + marker_type = None if marker is None else continuation_type(marker.text, rules) + if open_form is None: + drafts.append(_separate(sheet, marker_type)) + return + locator = f"sheet:{sheet.name}" + if marker_type is not None and marker_type != open_form.type_hint: + separate = _separate(sheet, marker_type) + message = ( + f"лист «{sheet.name}»: тип формы в маркере продолжения ({marker_type}) не совпадает с типом " + f"открытой формы ({open_form.type_hint})" + ) + separate.warnings.append(make_warning(WarningCode.CONTINUATION_TYPE_CONFLICT, message, rules, locator=locator)) + drafts.append(separate) + return + if marker is None and not _sheet_joins(sheet, open_form, rules): + separate = _separate(sheet, None) + message = ( + f"лист «{sheet.name}» не присоединён к форме «{open_form.sheet_names[0]}»: коды или число колонок " + "сумм не подходят" + ) + separate.warnings.append(make_warning(WarningCode.SHEET_NOT_JOINED, message, rules, locator=locator)) + drafts.append(separate) + return + open_form.segment_breaks.append(len(open_form.rows)) + open_form.rows.extend(grid_rows(sheet, 1, sheet.n_rows, sheet.n_cols, open_form.formats)) + open_form.sheet_names.append(sheet.name) + + +def _finish(draft: _Draft, rules: StatementRules, *, xlsb: bool) -> FormCandidate: + width = max((len(row) for row in draft.rows), default=0) + rows = tuple(row + (None,) * (width - len(row)) for row in draft.rows) + warnings = list(draft.warnings) + locator = draft.locator + if xlsb: + message = "формат чисел xlsb не читается: значения как в ячейках" + warnings.append(make_warning(WarningCode.XLSB_NO_FORMATS, message, rules, locator=locator)) + if draft.formats: + shown = ", ".join(f"«{_shorten(fmt)}»" for fmt in sorted(draft.formats)) + message = f"формат чисел не разобран, значения как в ячейках: {shown}" + warnings.append(make_warning(WarningCode.NUMBER_FORMAT_UNPARSED, message, rules, locator=locator)) + return FormCandidate( + source=Source.EXCEL, + locator=locator, + type_hint=draft.type_hint, + evidence=draft.evidence, + header=draft.header, + rows=rows, + segment_breaks=tuple(draft.segment_breaks), + warnings=tuple(warnings), + ) + + +def build_candidates( + sheets: Sequence[tuple[Sheet, SheetKind]], rules: StatementRules, deadline: Deadline, *, xlsb: bool +) -> list[FormCandidate]: + """Кандидаты форм из листов `STATEMENT` / `STATEMENT_CONTINUATION` в порядке книги.""" + drafts: list[_Draft] = [] + for sheet, kind in sheets: + deadline.check(f"лист «{sheet.name}»: форма") + if kind is SheetKind.STATEMENT: + drafts.extend(_form_block_drafts(sheet, rules)) + elif kind is SheetKind.STATEMENT_CONTINUATION: + _continuation(sheet, drafts, rules) + return [_finish(draft, rules, xlsb=xlsb) for draft in drafts] diff --git a/api/tests/fixtures/test_statements_adapter_real.py b/api/tests/fixtures/test_statements_adapter_real.py new file mode 100644 index 0000000..822533e --- /dev/null +++ b/api/tests/fixtures/test_statements_adapter_real.py @@ -0,0 +1,42 @@ +"""Адаптер книги -> `FormCandidate` на реальных выгрузках (план 4, задача 5).""" + +from pathlib import Path + +import pytest + +from arm_statements import DocumentType, default_rules +from src.parser.deadline import Deadline +from src.parser.detect import is_osv_sheet +from src.parser.readers import read_book +from src.parser.rules import load_rules +from src.parser.statements.adapter import build_candidates, plan_sheets + +pytestmark = pytest.mark.fixtures +STATEMENTS = "statements-xlsx" +NPZ = "npz/2025-Q3/Бухгалтерская отчетность за Январь - Сентябрь 2025 г. (НОВОРОССИЙСКИЙ ПРОКАТНЫЙ ЗАВОД).xlsx" +UNAGRANDE_BALANCE = "unagrande-group/2025-Q3/ББ. Лист 1 3кв.25.xls" + + +def _candidates(path: Path): # noqa: ANN202 + sheets, _ = read_book(path.read_bytes(), path.suffix.lstrip(".").upper()) # type: ignore[arg-type] + osv_rules = load_rules() + rules = default_rules() + plan = plan_sheets(sheets, lambda sheet: is_osv_sheet(sheet, osv_rules), rules) + return build_candidates(plan.statements, rules, Deadline(600.0), xlsb=False) + + +def test_npz_candidates(fixtures_dir: Path) -> None: + candidates = _candidates(fixtures_dir / STATEMENTS / NPZ) + assert [(c.type_hint, c.locator.count(",") + 1, len(c.segment_breaks)) for c in candidates] == [ + (DocumentType.BALANCE, 5, 4), + (DocumentType.INCOME, 3, 2), + ] + codes = {w.code for c in candidates for w in c.warnings} + assert not codes & {"SHEET_NOT_JOINED", "CONTINUATION_TYPE_CONFLICT"} + + +def test_unagrande_title_block_header(fixtures_dir: Path) -> None: + candidates = _candidates(fixtures_dir / STATEMENTS / UNAGRANDE_BALANCE) + header = candidates[0].header + assert header.okopf is not None and header.okopf.text == "12267" + assert header.inn is not None and "7728757141" in header.inn.text diff --git a/api/tests/unit/parser/statements/test_adapter.py b/api/tests/unit/parser/statements/test_adapter.py new file mode 100644 index 0000000..9c8d208 --- /dev/null +++ b/api/tests/unit/parser/statements/test_adapter.py @@ -0,0 +1,153 @@ +"""Адаптер книги -> `FormCandidate`: план листов, склейка листов-продолжений, предупреждения (план 4, задача 5).""" + +import pytest +from gridbuilder import at, sheet_from_rows, with_formats + +from arm_statements import DocumentType, Severity, Source, default_rules +from src.parser.deadline import Deadline +from src.parser.detect import SheetKind, is_osv_sheet +from src.parser.grid import Sheet +from src.parser.rules import load_rules +from src.parser.statements.adapter import build_candidates, plan_sheets +from src.parser.statements.forms import StatementRules + +NO_DEADLINE = Deadline(10_000.0) +STATEMENT = SheetKind.STATEMENT +CONTINUATION = SheetKind.STATEMENT_CONTINUATION +BALANCE_CODES = (1110, 1150, 1170, 1100) +INCOME_CODES = (2110, 2120, 2400) + + +@pytest.fixture(scope="module") +def rules() -> StatementRules: + return default_rules() + + +def _form(title: str, codes: tuple[int, ...], name: str, *, amounts: int = 3, marker: str | None = None) -> Sheet: + head = [[marker]] if marker else [[title]] + caption = at({1: "Показатель", 3: "Код", **{4 + i: f"На 3{i + 1}.12.2024" for i in range(amounts)}}) + body = [at({1: "Статья", 3: code, **{4 + i: 10 + i for i in range(amounts)}}) for code in codes] + return sheet_from_rows([*head, caption, *body], name=name) + + +def test_balance_with_marked_continuation_and_income(rules: StatementRules) -> None: + balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс") + marked = _form("", (1210, 1230, 1600), "Баланс (2)", marker="Форма 0710001 с.2") + ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР") + candidates = build_candidates( + [(balance, STATEMENT), (marked, CONTINUATION), (ofr, STATEMENT)], rules, NO_DEADLINE, xlsb=False + ) + assert [(c.type_hint, c.locator) for c in candidates] == [ + (DocumentType.BALANCE, "sheet:Баланс,Баланс (2)"), + (DocumentType.INCOME, "sheet:ОФР"), + ] + assert candidates[0].segment_breaks == (balance.n_rows,) + assert len({len(row) for row in candidates[0].rows}) == 1 + assert candidates[0].source is Source.EXCEL + assert len(candidates[0].rows) == balance.n_rows + marked.n_rows + + +def test_refs_of_cells(rules: StatementRules) -> None: + rows = [at({1: "Бухгалтерский баланс"})] + [[] for _ in range(24)] + rows.append(at({3: "Код", 4: "На 31.12.2024"})) + rows.append(at({1: "Статья", 3: 1110, 4: 1})) + rows += [at({1: "Статья", 3: code, 4: 1}) for code in (1150, 1170, 1100)] + assert rows[26] == at({1: "Статья", 3: 1110, 4: 1}) + for name, ref in (("Баланс", "Баланс!C27"), ("Бухгалтерский баланс. Лист 1", "'Бухгалтерский баланс. Лист 1'!C27")): + [candidate] = build_candidates([(sheet_from_rows(rows, name=name), STATEMENT)], rules, NO_DEADLINE, xlsb=False) + cells = [cell for row in candidate.rows for cell in row if cell is not None and cell.text == "1110"] + assert [cell.ref for cell in cells] == [ref] + + +def test_marker_type_conflict_is_separate_candidate(rules: StatementRules) -> None: + balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс") + other = _form("", INCOME_CODES, "Лист 2", marker="Форма 0710002 с.2") + first, second = build_candidates([(balance, STATEMENT), (other, CONTINUATION)], rules, NO_DEADLINE, xlsb=False) + assert first.warnings == () + assert second.type_hint is DocumentType.INCOME + assert [(w.code, w.severity) for w in second.warnings] == [("CONTINUATION_TYPE_CONFLICT", Severity.ERROR)] + assert second.segment_breaks == () + assert len(second.rows) == other.n_rows + + +def test_single_line_continuation_joins(rules: StatementRules) -> None: + ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР") + tail = sheet_from_rows([at({1: "Чистая прибыль", 3: 2400, 4: 1, 5: 2, 6: 3})], name="ОФР (2)") + [candidate] = build_candidates([(ofr, STATEMENT), (tail, CONTINUATION)], rules, NO_DEADLINE, xlsb=False) + assert candidate.segment_breaks == (ofr.n_rows,) + assert candidate.warnings == () + + +def test_wider_sheet_with_three_codes_is_not_joined(rules: StatementRules) -> None: + balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс", amounts=3) + wide = sheet_from_rows( + [at({1: "Статья", 3: code, 4: 1, 5: 2, 6: 3, 7: 4}) for code in (1210, 1230, 1250)], name="Лист 2" + ) + first, second = build_candidates([(balance, STATEMENT), (wide, CONTINUATION)], rules, NO_DEADLINE, xlsb=False) + assert first.segment_breaks == () + assert second.type_hint is None + assert [(w.code, w.severity) for w in second.warnings] == [("SHEET_NOT_JOINED", Severity.ERROR)] + + +def test_lonely_continuation_has_no_warning(rules: StatementRules) -> None: + marked = _form("", (1210, 1230, 1600), "Баланс (2)", marker="Форма 0710001 с.2") + [candidate] = build_candidates([(marked, CONTINUATION)], rules, NO_DEADLINE, xlsb=False) + assert candidate.type_hint is DocumentType.BALANCE + assert candidate.warnings == () + assert candidate.header.inn is None + assert len(candidate.rows) == marked.n_rows + + +def test_two_forms_on_one_sheet_share_locator(rules: StatementRules) -> None: + balance = [at({1: "Бухгалтерский баланс"}), at({3: "Код", 4: "На 31.12.2024"})] + balance += [at({1: "Статья", 3: code, 4: 1}) for code in BALANCE_CODES] + income = [at({1: "Отчет о финансовых результатах"}), at({3: "Код", 4: "За 2024"})] + income += [at({1: "Статья", 3: code, 4: 1}) for code in INCOME_CODES] + sheet = sheet_from_rows([*balance, *income], name="Формы") + candidates = build_candidates([(sheet, STATEMENT)], rules, NO_DEADLINE, xlsb=False) + assert [(c.type_hint, c.locator) for c in candidates] == [ + (DocumentType.BALANCE, "sheet:Формы"), + (DocumentType.INCOME, "sheet:Формы"), + ] + assert len(candidates[0].rows) == len(balance) + assert len(candidates[1].rows) == len(income) + + +def test_xlsb_and_unparsed_formats_warnings(rules: StatementRules) -> None: + balance = _form("Бухгалтерский баланс", BALANCE_CODES, "Баланс", amounts=1) + tail = _form("", (1210, 1230, 1250), "Баланс (2)", amounts=1, marker="Форма 0710001 с.2") + formats = {(row, 4): "0.0%" for row in range(3, balance.n_rows + 1)} + balance = with_formats(balance, formats) + tail = with_formats(tail, {(row, 4): "0.0%" for row in range(3, tail.n_rows + 1)}) + ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР", amounts=1) + candidates = build_candidates( + [(balance, STATEMENT), (tail, CONTINUATION), (ofr, STATEMENT)], rules, NO_DEADLINE, xlsb=True + ) + codes = [[w.code for w in c.warnings] for c in candidates] + assert codes == [["XLSB_NO_FORMATS", "NUMBER_FORMAT_UNPARSED"], ["XLSB_NO_FORMATS"]] + assert "«0.0%»" in candidates[0].warnings[1].message + assert candidates[0].warnings[1].message.count("0.0%") == 1 + + +def test_plan_sheets_weak_continuation_only_after_form(rules: StatementRules) -> None: + ofr = _form("Отчет о финансовых результатах", INCOME_CODES, "ОФР") + weak = sheet_from_rows([["Чистая прибыль", "2400", 10, 8]], name="2400") + notes = sheet_from_rows([["Примечания"], ["Текст пояснения к отчетности"]], name="Примечания") + osv = sheet_from_rows([["Оборотно-сальдовая ведомость"]], name="ОСВ") + + def is_osv(sheet: Sheet) -> bool: + return sheet.name == "ОСВ" + + plan = plan_sheets([ofr, weak], is_osv, rules) + assert [(s.name, k) for s, k in plan.statements] == [("ОФР", STATEMENT), ("2400", CONTINUATION)] + plan = plan_sheets([osv, weak], is_osv, rules) + assert (plan.osv, plan.statements, plan.other) == ((osv,), (), (weak,)) + plan = plan_sheets([ofr, notes, weak], is_osv, rules) + assert [s.name for s in plan.other] == ["Примечания", "2400"] + + +def test_is_osv_sheet_by_marker() -> None: + rules = load_rules() + osv = sheet_from_rows([["Оборотно-сальдовая ведомость по счету 60"]]) + assert is_osv_sheet(osv, rules) + assert not is_osv_sheet(sheet_from_rows([["Бухгалтерский баланс"]]), rules)