RSHB/arm-scan-gateway
9
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(normalizer): сборка statementDocument из таблицы OCR — тип, колонки, строки, редакции (D-6, D-7)

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-07 00:30:05 +04:00
parent 6882493d98
commit 80622ed331
2 changed files with 435 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

221
api/src/normalizer/ocr/document.py Normal file
View File

@ -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)

214
api/tests/unit/normalizer/test_document.py Normal file
View File

@ -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