From 80622ed33145261307ccb1238ce83605694357a9 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 7 Oct 2026 00:30:05 +0400 Subject: [PATCH] =?UTF-8?q?feat(normalizer):=20=D1=81=D0=B1=D0=BE=D1=80?= =?UTF-8?q?=D0=BA=D0=B0=20statementDocument=20=D0=B8=D0=B7=20=D1=82=D0=B0?= =?UTF-8?q?=D0=B1=D0=BB=D0=B8=D1=86=D1=8B=20OCR=20=E2=80=94=20=D1=82=D0=B8?= =?UTF-8?q?=D0=BF,=20=D0=BA=D0=BE=D0=BB=D0=BE=D0=BD=D0=BA=D0=B8,=20=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=BE=D0=BA=D0=B8,=20=D1=80=D0=B5=D0=B4=D0=B0?= =?UTF-8?q?=D0=BA=D1=86=D0=B8=D0=B8=20(D-6,=20D-7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/src/normalizer/ocr/document.py | 221 +++++++++++++++++++++ api/tests/unit/normalizer/test_document.py | 214 ++++++++++++++++++++ 2 files changed, 435 insertions(+) create mode 100644 api/src/normalizer/ocr/document.py create mode 100644 api/tests/unit/normalizer/test_document.py diff --git a/api/src/normalizer/ocr/document.py b/api/src/normalizer/ocr/document.py new file mode 100644 index 0000000..1a2ab5c --- /dev/null +++ b/api/src/normalizer/ocr/document.py @@ -0,0 +1,221 @@ +"""Сборка `statementDocument` v1.2 из одной таблицы `data_ocr.table_structure[i]` (spec «Нормализация ответа ИС», +D-6, D-7): тип по label или по кодам, колонки из раскладки, строки только с кодом, значения с `is_recognized`, +имена строк из справочника редакции.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import date +from typing import Literal + +from src.contract.result import StatementColumn, StatementDocument, StatementHeader, StatementLine, StatementValue +from src.normalizer.ocr.amounts import is_blank, parse_amount +from src.normalizer.ocr.dates import parse_date +from src.normalizer.ocr.header import FileHeader +from src.normalizer.ocr.layout import Cell, cell_text +from src.normalizer.ocr.rows import ParsedRow, ParsedTable, RowKind, parse_table_rows, table_type_by_codes +from src.normalizer.ocr.text import clean_value, collapse_whitespace, fix_lookalikes, normalize_header +from src.normalizer.rules import Edition, OcrRules + +__all__ = ["BuiltDocument", "build_document"] + +FULL_CONFIDENCE = 100.0 +CAPPED_CONFIDENCE = 99.0 +DocumentType = Literal["BALANCE", "INCOME", "OTHER"] + + +@dataclass(frozen=True) +class BuiltDocument: + document: StatementDocument + warnings: list[str] + fully_recognized: bool + edition_id: str | None + + +@dataclass +class _Build: + """Накопитель предупреждений и признака «всё распознано» на время сборки одного документа.""" + + warnings: list[str] + fully_recognized: bool = True + + def warn(self, text: str) -> None: + self.warnings.append(text) + self.fully_recognized = False + + +def _label_text(table: Mapping[str, object]) -> str: + label = table.get("label") + form_name = label.get("form_name") if isinstance(label, Mapping) else None + return collapse_whitespace(form_name) if isinstance(form_name, str) else "" + + +def _type_by_label(label: str, rules: OcrRules) -> DocumentType | None: + normalized = normalize_header(label) + for needle, kind in rules.labels.items(): + if needle in normalized: + return kind + return None + + +def _resolve_type(label: str, parsed: ParsedTable, rules: OcrRules, build: _Build) -> DocumentType: + by_label = _type_by_label(label, rules) + by_codes = table_type_by_codes(parsed) + if by_label is not None: + if by_codes is not None and by_codes != by_label: + build.warn(f"тип по кодам ({by_codes}) не совпал с label ({label})") + return by_label + if by_codes is not None: + build.warn("тип определён по кодам") + return by_codes + build.warn("тип не определён: OTHER") + return "OTHER" + + +def _columns( + parsed: ParsedTable, lookalikes: Mapping[str, str], build: _Build +) -> tuple[list[StatementColumn], list[int]]: + """Колонки значений первой шапки: `label` — текст ячейки шапки, `period_date` — `parse_date`.""" + header = parsed.rows[0] + columns: list[StatementColumn] = [] + for position, source_idx in enumerate(header.layout.value_idxs): + cell = header.cells.get(source_idx) + label = collapse_whitespace(cell.value) if cell is not None else "" + period = parse_date(fix_lookalikes(label, lookalikes)) if label else None + if period is None and not parsed.synthetic_header: + build.warn(f"колонка {position} без даты: {label}") + columns.append(StatementColumn(column_index=position, label=label, period_date=period)) + if parsed.synthetic_header: + build.warn("шапка синтезирована: колонки без дат") + return columns, list(header.layout.value_idxs) + + +def _latest_period(columns: list[StatementColumn]) -> date | None: + dates = [column.period_date for column in columns if column.period_date is not None] + return max(dates) if dates else None + + +def _choose_edition( + header: FileHeader, columns: list[StatementColumn], codes: list[str], rules: OcrRules, build: _Build +) -> Edition | None: + edition = rules.edition_for(header.report_date) or rules.edition_for(_latest_period(columns)) + if edition is None: + edition = rules.edition_by_codes(codes) + if edition is None: + build.warn("редакция не определена") + return None + build.warn(f"редакция определена по кодам: {edition.id}") + if not edition.lines: + build.warn(f"редакция {edition.id} не заполнена") + return edition + + +def _value(cell: Cell | None, column_index: int, build: _Build) -> StatementValue: + if cell is None: + build.fully_recognized = False + return StatementValue( + column_index=column_index, amount=None, string_value=None, confidence=0, is_recognized=False + ) + parsed = parse_amount(cell.value) + text = collapse_whitespace(cell.value) + # DBML: string_value — «как распознано, если не число» (решение №16): только когда сумма не разобрана + string_value = text if parsed.amount is None and text and not is_blank(text) else None + recognized = cell.confidence == FULL_CONFIDENCE and parsed.verified + if not recognized: + build.fully_recognized = False + return StatementValue( + column_index=column_index, + amount=parsed.amount, + string_value=string_value, + confidence=cell.confidence, + is_recognized=recognized, + ) + + +def _line( + row: ParsedRow, value_idxs: list[int], edition: Edition | None, rules: OcrRules, build: _Build +) -> StatementLine: + row_cells = row.cells + code_cell = row_cells[row.layout.code_idx] + code = clean_value(code_cell.value) + code_clean = code == code_cell.value.strip() + label_idx = row.layout.label_idx + name_cell = row_cells.get(label_idx) if label_idx is not None else None + known_here = edition is not None and code in edition.lines + if known_here and edition is not None: + name = edition.lines[code] + # spec: уверенность кода и словарного имени — confidence ячейки, если текст чистый и код в справочнике + code_confidence = code_cell.confidence if code_clean else min(code_cell.confidence, CAPPED_CONFIDENCE) + name_confidence = code_confidence + else: + elsewhere = rules.editions_with(code) + if elsewhere: + build.warn(f"код {code} из редакции {', '.join(elsewhere)}") + name = collapse_whitespace(name_cell.value) if name_cell is not None else "" + # кода нет в выбранной редакции: «иначе min(confidence, 99)» — всегда, имя из OCR со своим confidence + code_confidence = min(code_cell.confidence, CAPPED_CONFIDENCE) + name_confidence = name_cell.confidence if name_cell is not None else 0.0 + if code_confidence < FULL_CONFIDENCE or name_confidence < FULL_CONFIDENCE: + build.fully_recognized = False + values = [_value(row_cells.get(source_idx), position, build) for position, source_idx in enumerate(value_idxs)] + return StatementLine( + line_code=code, + line_code_confidence=code_confidence, + line_name=name, + line_name_confidence=name_confidence, + values=values, + ) + + +def _statement_header(label: str, header: FileHeader, build: _Build) -> StatementHeader: + if header.inn_confidence != FULL_CONFIDENCE or header.organization_confidence != FULL_CONFIDENCE: + build.fully_recognized = False + if header.report_date is None or not header.report_date_recognized: + build.fully_recognized = False + # form_name в схеме без null: нет label — поле не передаётся (явный None контрактная модель отвергает) + form_name = {"form_name": label} if label else {} + return StatementHeader( + **form_name, + organization_name=header.organization_name, + organization_name_confidence=header.organization_confidence, + inn=header.inn, + inn_confidence=header.inn_confidence, + report_date_text=header.report_date_text, + report_date=header.report_date, + units=header.units, + ) + + +def build_document( + table: Mapping[str, object], index: int, file_name: str, header: FileHeader, rules: OcrRules +) -> BuiltDocument | None: + rows = table.get("rows") + if not isinstance(rows, list): + return None + parsed = parse_table_rows(rows, rules.known_codes()) + if parsed is None or len(parsed.rows) < 2: # noqa: PLR2004 — шапка и хотя бы одна строка данных + return None + build = _Build(warnings=[]) + label = _label_text(table) + document_type = _resolve_type(label, parsed, rules, build) + columns, value_idxs = _columns(parsed, rules.lookalikes, build) + data_rows = [row for row in parsed.rows[1:] if row.kind is not RowKind.HEADER] + codes = [clean_value(cell_text(row.cells, row.layout.code_idx)) for row in data_rows] + edition = _choose_edition(header, columns, codes, rules, build) + lines: list[StatementLine] = [] + seen: set[str] = set() + for row in data_rows: + code = clean_value(cell_text(row.cells, row.layout.code_idx)) + if code in seen: # решение №15: DBML уникален по (meta_id, line_code), как §12 п. 5 у Excel-пути + build.warn(f"код {code} повторяется: оставлена первая строка") + continue + seen.add(code) + lines.append(_line(row, list(row.layout.value_idxs) or value_idxs, edition, rules, build)) + statement_header = _statement_header(label, header, build) + document = StatementDocument( + document_type=document_type, + locator=f"file:{file_name};table:{index}", + header=statement_header, + columns=columns, + lines=lines, + ) + return BuiltDocument(document, build.warnings, build.fully_recognized, edition.id if edition else None) diff --git a/api/tests/unit/normalizer/test_document.py b/api/tests/unit/normalizer/test_document.py new file mode 100644 index 0000000..9793886 --- /dev/null +++ b/api/tests/unit/normalizer/test_document.py @@ -0,0 +1,214 @@ +"""Сборка `statementDocument` из таблицы OCR (spec «Нормализация ответа ИС», D-6, D-7).""" + +from datetime import date +from decimal import Decimal +from pathlib import Path + +from src.normalizer.ocr.document import BuiltDocument, build_document +from src.normalizer.ocr.header import FileHeader +from src.normalizer.rules import load_rules + +API_DIR = Path(__file__).resolve().parents[3] +RULES = load_rules(API_DIR / "rules") +HEADER_2024 = FileHeader( + "ООО Ромашка", 100.0, "7700000000", 100.0, "тыс. руб.", "на 31 декабря 2024 г.", date(2024, 12, 31), True +) + + +def _row(index: int, *cells: tuple[int, str, float]) -> dict: + return {"row_index": index, "cells": [{"column_index": i, "value": v, "confidence": c} for i, v, c in cells]} + + +def _table(label: str, *rows: dict) -> dict: + return {"label": {"form_name": label, "confidence": 100}, "rows": list(rows)} + + +CLEAN_BALANCE = _table( + "Бухгалтерский баланс", + _row( + 0, + (0, "Наименование показателя", 100), + (1, "Код", 100), + (2, "На 31 декабря 2024 г.", 100), + (3, "На 31 декабря 2023 г.", 100), + ), + _row(1, (0, "Основные средства", 100), (1, "1150", 100), (2, "1 234", 100), (3, "(56)", 100)), + _row(2, (0, "Итого по разделу I", 100), (1, "1100", 100), (2, "1 234", 100), (3, "-", 100)), +) + + +def _build(table: dict, header: FileHeader = HEADER_2024, index: int = 0) -> BuiltDocument: + built = build_document(table, index, "job.json", header, RULES) + assert built is not None + return built + + +class TestCleanTable: + def test_document_fields(self) -> None: + built = _build(CLEAN_BALANCE) + doc = built.document + assert doc.document_type == "BALANCE" + assert doc.locator == "file:job.json;table:0" + assert doc.header.form_name == "Бухгалтерский баланс" + assert doc.header.inn == "7700000000" and doc.header.inn_confidence == 100 + assert doc.header.report_date == date(2024, 12, 31) and doc.header.report_date_text == "на 31 декабря 2024 г." + assert doc.header.units == "тыс. руб." and doc.header.okud is None + assert [(c.column_index, c.label, c.period_date) for c in doc.columns] == [ + (0, "На 31 декабря 2024 г.", date(2024, 12, 31)), + (1, "На 31 декабря 2023 г.", date(2023, 12, 31)), + ] + assert doc.checks is None + + def test_lines_and_values(self) -> None: + doc = _build(CLEAN_BALANCE).document + assert [line.line_code for line in doc.lines] == ["1150", "1100"] + first = doc.lines[0] + assert ( + first.line_name == "Основные средства" + and first.line_code_confidence == 100 + and first.line_name_confidence == 100 + ) + values = first.values + assert [v.column_index for v in values] == [0, 1] + assert ( + values[0].amount == Decimal("1234") and values[0].string_value is None and values[0].is_recognized is True + ) + assert values[1].amount == Decimal("-56") and values[1].is_recognized is True + blank = doc.lines[1].values[1] + assert blank.amount is None and blank.string_value is None and blank.is_recognized is True + + def test_fully_recognized_and_edition(self) -> None: + built = _build(CLEAN_BALANCE) + assert built.fully_recognized is True and built.warnings == [] and built.edition_id == "66n" + + +class TestNoise: + def test_unknown_code_uses_ocr_name_and_caps_confidence(self) -> None: + table = _table( + "Бухгалтерский баланс", + CLEAN_BALANCE["rows"][0], + _row(1, (0, "Расшифровка", 100), (1, "11501", 100), (2, "1", 100), (3, "2", 100)), + _row(2, (0, "ОС", 100), (1, "1150", 90), (2, "1", 100), (3, "2", 100)), + ) + built = _build(table) + doc = built.document + # кода нет в справочнике — имя из OCR, уверенность кода не выше 99 (spec: «иначе min(confidence, 99)») + assert doc.lines[0].line_code == "11501" and doc.lines[0].line_name == "Расшифровка" + assert doc.lines[0].line_code_confidence == 99 and doc.lines[0].line_name_confidence == 100 + assert doc.lines[1].line_code == "1150" and doc.lines[1].line_code_confidence == 90 # уже < 100, не поднимается + assert doc.lines[1].line_name == "Основные средства" + assert built.fully_recognized is False + + def test_non_integer_code_row_is_dropped(self) -> None: + """Перенос `_classify_row`: «1150 :» не целое число → строка считается повторным заголовком и пропускается.""" + table = _table( + "Бухгалтерский баланс", + CLEAN_BALANCE["rows"][0], + _row(1, (0, "НМА", 100), (1, "1110", 100), (2, "1", 100), (3, "2", 100)), + _row(2, (0, "ОС", 100), (1, "1150 :", 100), (2, "1", 100), (3, "2", 100)), + ) + assert [line.line_code for line in _build(table).document.lines] == ["1110"] + + def test_duplicate_code_keeps_first(self) -> None: + """Решение №15: DBML уникален по (meta_id, line_code) — повтор кода даёт одну строку и предупреждение.""" + table = _table( + "Бухгалтерский баланс", + CLEAN_BALANCE["rows"][0], + _row(1, (0, "Основные средства", 100), (1, "1150", 100), (2, "1", 100), (3, "2", 100)), + _row(2, (0, "Основные средства", 100), (1, "1150", 100), (2, "7", 100), (3, "8", 100)), + ) + built = _build(table) + assert [line.line_code for line in built.document.lines] == ["1150"] + assert built.document.lines[0].values[0].amount == Decimal("1") + assert "код 1150 повторяется: оставлена первая строка" in built.warnings and built.fully_recognized is False + + def test_missing_cells_keep_column_count(self) -> None: + """Review Focus 2: у строки нет ячеек под колонками — values всё равно по числу колонок.""" + table = _table("Бухгалтерский баланс", CLEAN_BALANCE["rows"][0], _row(1, (0, "ОС", 100), (1, "1150", 100))) + values = _build(table).document.lines[0].values + assert [(v.column_index, v.amount, v.string_value, v.confidence, v.is_recognized) for v in values] == [ + (0, None, None, 0, False), + (1, None, None, 0, False), + ] + + def test_unverified_amount_is_not_recognized(self) -> None: + table = _table( + "Бухгалтерский баланс", + CLEAN_BALANCE["rows"][0], + _row(1, (0, "ОС", 100), (1, "1150", 100), (2, "(7744536}", 100), (3, "x", 100)), + ) + values = _build(table).document.lines[0].values + assert ( + values[0].amount == Decimal("7744536") + and values[0].is_recognized is False + and values[0].string_value is None + ) + assert values[1].amount is None and values[1].is_recognized is False and values[1].string_value == "x" + + def test_synthetic_header_columns_without_dates(self) -> None: + table = _table( + "Бухгалтерский баланс", + _row(0, (1, "1150", 100), (0, "ОС", 100), (2, "1", 100)), + _row(1, (1, "1110", 100), (0, "НМА", 100), (2, "2", 100)), + ) + built = _build(table) + assert [(c.column_index, c.label, c.period_date) for c in built.document.columns] == [(0, "", None)] + assert "шапка синтезирована" in " ".join(built.warnings) and built.fully_recognized is False + + +class TestTypeAndEdition: + def test_type_by_codes_when_label_unknown(self) -> None: + built = _build(_table("Бvxгалтерский баланс", *CLEAN_BALANCE["rows"])) + assert built.document.document_type == "BALANCE" and "тип определён по кодам" in built.warnings + + def test_label_wins_over_codes_with_warning(self) -> None: + built = _build(_table("Отчет о финансовых результатах", *CLEAN_BALANCE["rows"])) + assert built.document.document_type == "INCOME" + assert any(w.startswith("тип по кодам (BALANCE) не совпал с label") for w in built.warnings) + + def test_other_when_nothing_known(self) -> None: + table = _table( + "Приложение", + CLEAN_BALANCE["rows"][0], + _row(1, (0, "a", 100), (1, "5100", 100), (2, "1", 100), (3, "2", 100)), + _row(2, (0, "b", 100), (1, "5200", 100), (2, "1", 100), (3, "2", 100)), + ) + built = _build(table) + assert built.document.document_type == "OTHER" and "тип не определён: OTHER" in built.warnings + + def test_edition_2025_is_empty(self) -> None: + header = FileHeader("ООО", 100.0, "7700000000", 100.0, "тыс. руб.", None, None, False) + table = _table( + "Бухгалтерский баланс", + _row(0, (0, "Наименование показателя", 100), (1, "Код", 100), (2, "На 30 июня 2025 г.", 100)), + _row(1, (0, "Основные средства", 100), (1, "1150", 100), (2, "1", 100)), + ) + built = _build(table, header) + assert built.edition_id == "fsbu4" and "редакция fsbu4 не заполнена" in built.warnings + assert built.document.lines[0].line_name == "Основные средства" # из OCR + assert built.document.lines[0].line_name_confidence == 100 # confidence ячейки наименования + assert built.fully_recognized is False + + def test_edition_by_codes_when_no_dates(self) -> None: + header = FileHeader(None, None, None, None, None, None, None, False) + table = _table("Бухгалтерский баланс", _row(0, (1, "1150", 100), (0, "ОС", 100), (2, "1", 100))) + built = _build(table, header) + assert built.edition_id == "66n" and "редакция определена по кодам: 66n" in built.warnings + + def test_unparsed_table_is_none(self) -> None: + assert ( + build_document(_table("x", _row(0, (0, "a", 100), (1, "b", 100))), 0, "job.json", HEADER_2024, RULES) + is None + ) + + +def test_missing_label_omits_form_name() -> None: + """Контракт: `form_name` без `null` — при отсутствии label поле не задаётся, сборка не падает.""" + table = {"rows": CLEAN_BALANCE["rows"]} + built = _build(table) + assert built.document.header.form_name is None + assert "form_name" not in built.document.to_wire()["header"] + + +def test_table_rows_not_list_is_none() -> None: + assert build_document({"label": {"form_name": "x"}, "rows": "oops"}, 0, "job.json", HEADER_2024, RULES) is None