fix(normalizer): values по числу колонок, предупреждения редакций и колонок (T8.1–T8.6)

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:37:30 +04:00
parent 37b36047fe
commit 8ebf572a22
2 changed files with 230 additions and 14 deletions

View File

@ -71,9 +71,7 @@ def _resolve_type(label: str, parsed: ParsedTable, rules: OcrRules, build: _Buil
return "OTHER" return "OTHER"
def _columns( def _columns(parsed: ParsedTable, lookalikes: Mapping[str, str], build: _Build) -> list[StatementColumn]:
parsed: ParsedTable, lookalikes: Mapping[str, str], build: _Build
) -> tuple[list[StatementColumn], list[int]]:
"""Колонки значений первой шапки: `label` — текст ячейки шапки, `period_date` — `parse_date`.""" """Колонки значений первой шапки: `label` — текст ячейки шапки, `period_date` — `parse_date`."""
header = parsed.rows[0] header = parsed.rows[0]
columns: list[StatementColumn] = [] columns: list[StatementColumn] = []
@ -86,7 +84,9 @@ def _columns(
columns.append(StatementColumn(column_index=position, label=label, period_date=period)) columns.append(StatementColumn(column_index=position, label=label, period_date=period))
if parsed.synthetic_header: if parsed.synthetic_header:
build.warn("шапка синтезирована: колонки без дат") build.warn("шапка синтезирована: колонки без дат")
return columns, list(header.layout.value_idxs) if not columns:
build.warn("нет колонок значений")
return columns
def _latest_period(columns: list[StatementColumn]) -> date | None: def _latest_period(columns: list[StatementColumn]) -> date | None:
@ -131,23 +131,21 @@ def _value(cell: Cell | None, column_index: int, build: _Build) -> StatementValu
) )
def _line( def _line(row: ParsedRow, column_count: int, edition: Edition | None, rules: OcrRules, build: _Build) -> StatementLine:
row: ParsedRow, value_idxs: list[int], edition: Edition | None, rules: OcrRules, build: _Build
) -> StatementLine:
row_cells = row.cells row_cells = row.cells
code_cell = row_cells[row.layout.code_idx] code_cell = row_cells[row.layout.code_idx]
code = clean_value(code_cell.value) code = clean_value(code_cell.value)
code_clean = code == code_cell.value.strip() code_clean = code == code_cell.value.strip()
label_idx = row.layout.label_idx label_idx = row.layout.label_idx
name_cell = row_cells.get(label_idx) if label_idx is not None else None 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 edition is not None and code in edition.lines:
if known_here and edition is not None:
name = edition.lines[code] name = edition.lines[code]
# spec: уверенность кода и словарного имени — confidence ячейки, если текст чистый и код в справочнике # spec: уверенность кода и словарного имени — confidence ячейки, если текст чистый и код в справочнике
code_confidence = code_cell.confidence if code_clean else min(code_cell.confidence, CAPPED_CONFIDENCE) code_confidence = code_cell.confidence if code_clean else min(code_cell.confidence, CAPPED_CONFIDENCE)
name_confidence = code_confidence name_confidence = code_confidence
else: else:
elsewhere = rules.editions_with(code) # пустая редакция уже даёт одно «редакция <id> не заполнена» — по строке не предупреждаем
elsewhere = rules.editions_with(code) if edition is not None and edition.lines else []
if elsewhere: if elsewhere:
build.warn(f"код {code} из редакции {', '.join(elsewhere)}") build.warn(f"код {code} из редакции {', '.join(elsewhere)}")
name = collapse_whitespace(name_cell.value) if name_cell is not None else "" name = collapse_whitespace(name_cell.value) if name_cell is not None else ""
@ -156,7 +154,13 @@ def _line(
name_confidence = name_cell.confidence if name_cell is not None else 0.0 name_confidence = name_cell.confidence if name_cell is not None else 0.0
if code_confidence < FULL_CONFIDENCE or name_confidence < FULL_CONFIDENCE: if code_confidence < FULL_CONFIDENCE or name_confidence < FULL_CONFIDENCE:
build.fully_recognized = False build.fully_recognized = False
values = [_value(row_cells.get(source_idx), position, build) for position, source_idx in enumerate(value_idxs)] # values — по позиции колонки документа (R-3): лишние индексы раскладки строки отбрасываются, недостающие —
# «ячейки нет»
value_idxs = row.layout.value_idxs
values = [
_value(row_cells.get(value_idxs[position]) if position < len(value_idxs) else None, position, build)
for position in range(column_count)
]
return StatementLine( return StatementLine(
line_code=code, line_code=code,
line_code_confidence=code_confidence, line_code_confidence=code_confidence,
@ -197,8 +201,10 @@ def build_document(
build = _Build(warnings=[]) build = _Build(warnings=[])
label = _label_text(table) label = _label_text(table)
document_type = _resolve_type(label, parsed, rules, build) document_type = _resolve_type(label, parsed, rules, build)
columns, value_idxs = _columns(parsed, rules.lookalikes, build) columns = _columns(parsed, rules.lookalikes, build)
data_rows = [row for row in parsed.rows[1:] if row.kind is not RowKind.HEADER] data_rows = [row for row in parsed.rows[1:] if row.kind is not RowKind.HEADER]
if any(len(row.layout.value_idxs) != len(columns) for row in data_rows):
build.warn("повторная шапка изменила число колонок значений") # одно предупреждение на таблицу
codes = [clean_value(cell_text(row.cells, row.layout.code_idx)) for row in data_rows] codes = [clean_value(cell_text(row.cells, row.layout.code_idx)) for row in data_rows]
edition = _choose_edition(header, columns, codes, rules, build) edition = _choose_edition(header, columns, codes, rules, build)
lines: list[StatementLine] = [] lines: list[StatementLine] = []
@ -209,7 +215,7 @@ def build_document(
build.warn(f"код {code} повторяется: оставлена первая строка") build.warn(f"код {code} повторяется: оставлена первая строка")
continue continue
seen.add(code) seen.add(code)
lines.append(_line(row, list(row.layout.value_idxs) or value_idxs, edition, rules, build)) lines.append(_line(row, len(columns), edition, rules, build))
statement_header = _statement_header(label, header, build) statement_header = _statement_header(label, header, build)
document = StatementDocument( document = StatementDocument(
document_type=document_type, document_type=document_type,

View File

@ -1,12 +1,15 @@
"""Сборка `statementDocument` из таблицы OCR (spec «Нормализация ответа ИС», D-6, D-7).""" """Сборка `statementDocument` из таблицы OCR (spec «Нормализация ответа ИС», D-6, D-7)."""
from dataclasses import replace
from datetime import date from datetime import date
from decimal import Decimal from decimal import Decimal
from pathlib import Path from pathlib import Path
import pytest
from src.normalizer.ocr.document import BuiltDocument, build_document from src.normalizer.ocr.document import BuiltDocument, build_document
from src.normalizer.ocr.header import FileHeader from src.normalizer.ocr.header import FileHeader
from src.normalizer.rules import load_rules from src.normalizer.rules import Edition, OcrRules, load_rules
API_DIR = Path(__file__).resolve().parents[3] API_DIR = Path(__file__).resolve().parents[3]
RULES = load_rules(API_DIR / "rules") RULES = load_rules(API_DIR / "rules")
@ -212,3 +215,210 @@ def test_missing_label_omits_form_name() -> None:
def test_table_rows_not_list_is_none() -> None: 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 assert build_document({"label": {"form_name": "x"}, "rows": "oops"}, 0, "job.json", HEADER_2024, RULES) is None
REPEATED_HEADER = "повторная шапка изменила число колонок значений"
def _head(*dates: str) -> dict:
cells = [(0, "Наименование показателя", 100), (1, "Код", 100)]
cells += [(2 + i, text, 100) for i, text in enumerate(dates)]
return _row(0, *cells)
class TestRepeatedHeader:
"""R-3 (design #21): `columns` — по первой шапке, `values` каждой строки — по позиции этой колонки."""
def test_wider_first_header_pads_missing_values(self) -> None:
table = _table(
"Бухгалтерский баланс",
_head("На 31 декабря 2024 г.", "На 31 декабря 2023 г.", "На 31 декабря 2022 г."),
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100), (3, "2", 100), (4, "3", 100)),
_row(
2,
(0, "Наименование показателя", 100),
(1, "Код", 100),
(2, "На 31 декабря 2024 г.", 100),
(3, "На 31 декабря 2023 г.", 100),
),
_row(3, (0, "НМА", 100), (1, "1110", 100), (2, "4", 100), (3, "5", 100)),
)
built = _build(table)
assert [line.line_code for line in built.document.lines] == ["1150", "1110"]
assert all(len(line.values) == 3 for line in built.document.lines)
last = built.document.lines[1].values[2]
assert (last.amount, last.string_value, last.confidence, last.is_recognized) == (None, None, 0, False)
assert built.document.lines[1].values[1].amount == Decimal("5")
assert built.warnings.count(REPEATED_HEADER) == 1 and built.fully_recognized is False
def test_narrower_first_header_drops_extra_indexes(self) -> None:
table = _table(
"Бухгалтерский баланс",
_head("На 31 декабря 2024 г.", "На 31 декабря 2023 г."),
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100), (3, "2", 100)),
_row(
2,
(0, "Наименование показателя", 100),
(1, "Код", 100),
(2, "На 31 декабря 2024 г.", 100),
(3, "На 31 декабря 2023 г.", 100),
(4, "На 31 декабря 2022 г.", 100),
),
_row(3, (0, "НМА", 100), (1, "1110", 100), (2, "4", 100), (3, "5", 100), (4, "6", 100)),
)
built = _build(table)
assert [c.column_index for c in built.document.columns] == [0, 1]
for line in built.document.lines:
assert [v.column_index for v in line.values] == [0, 1] # нет «сироты» с column_index = 2
assert built.document.lines[1].values[1].amount == Decimal("5")
assert built.warnings.count(REPEATED_HEADER) == 1 and built.fully_recognized is False
def test_same_layout_after_repeated_header_is_silent(self) -> None:
table = _table(
"Бухгалтерский баланс",
*CLEAN_BALANCE["rows"][:2],
_row(
5,
(0, "Наименование показателя", 100),
(1, "Код", 100),
(2, "На 31 декабря 2024 г.", 100),
(3, "На 31 декабря 2023 г.", 100),
),
_row(6, (0, "Итого по разделу I", 100), (1, "1100", 100), (2, "1", 100), (3, "2", 100)),
)
built = _build(table)
assert [line.line_code for line in built.document.lines] == ["1150", "1100"]
assert built.warnings == [] and built.fully_recognized is True
def test_latin_o_in_code_drops_row() -> None:
"""R-4 (design #22): лукалайки применяются только к датам — «11O0» не целое число, строка отбрасывается."""
table = _table(
"Бухгалтерский баланс",
CLEAN_BALANCE["rows"][0],
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100), (3, "2", 100)),
_row(2, (0, "Итого", 100), (1, "11O0", 100), (2, "1", 100), (3, "2", 100)),
)
assert [line.line_code for line in _build(table).document.lines] == ["1150"]
class TestEditionWarnings:
def test_empty_edition_gives_single_warning(self) -> None:
"""R-5 (design #23): пустая fsbu4 — одно «не заполнена», без «код N из редакции 66n» по строкам."""
header = FileHeader("ООО", 100.0, "7700000000", 100.0, "тыс. руб.", None, None, False)
table = _table(
"Бухгалтерский баланс",
_head("На 30 июня 2025 г"),
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100)),
_row(2, (0, "НМА", 100), (1, "1110", 100), (2, "2", 100)),
)
built = _build(table, header)
assert built.edition_id == "fsbu4"
assert built.warnings == ["редакция fsbu4 не заполнена"]
assert not any("из редакции" in w for w in built.warnings)
def test_code_from_other_nonempty_edition_is_reported(self) -> None:
rules = OcrRules(
editions=[
Edition(id="old", valid_until=date(2024, 12, 31), lines={"1150": "Основные средства"}),
Edition(id="new", valid_from=date(2025, 1, 1), lines={"1110": "Нематериальные активы"}),
],
lookalikes=RULES.lookalikes,
labels=RULES.labels,
)
table = _table(
"Бухгалтерский баланс",
CLEAN_BALANCE["rows"][0],
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100), (3, "2", 100)),
_row(2, (0, "НМА", 100), (1, "1110", 100), (2, "3", 100), (3, "4", 100)),
)
built = build_document(table, 0, "job.json", HEADER_2024, rules)
assert built is not None and built.edition_id == "old"
assert built.warnings == ["код 1110 из редакции new"]
assert built.document.lines[1].line_name == "НМА" and built.document.lines[1].line_code_confidence == 99
def test_table_without_value_columns() -> None:
"""R-6 (design #25): нет колонок значений — предупреждение и needs_check; values пустые."""
table = _table(
"Бухгалтерский баланс",
_row(0, (0, "Наименование показателя", 100), (1, "Код", 100)),
_row(1, (0, "ОС", 100), (1, "1150", 100)),
)
built = _build(table)
assert built.document.columns == [] and built.document.lines[0].values == []
assert "нет колонок значений" in built.warnings and built.fully_recognized is False
assert REPEATED_HEADER not in built.warnings
class TestHeaderConditions:
"""D-6: каждое отклонение шапки файла лишает документ статуса «всё распознано»."""
@pytest.mark.parametrize(
"changes",
[
{"inn_confidence": 90.0},
{"inn": None, "inn_confidence": None},
{"organization_confidence": 90.0},
{"organization_name": None, "organization_confidence": None},
{"report_date": None, "report_date_recognized": False},
{"report_date_recognized": False},
],
ids=["inn-low", "inn-none", "org-low", "org-none", "no-date", "date-not-recognized"],
)
def test_header_deviation_is_not_fully_recognized(self, changes: dict) -> None:
assert _build(CLEAN_BALANCE, replace(HEADER_2024, **changes)).fully_recognized is False
def test_baseline_is_fully_recognized(self) -> None:
assert _build(CLEAN_BALANCE, HEADER_2024).fully_recognized is True
def test_column_without_date_warning() -> None:
table = _table(
"Бухгалтерский баланс",
_head("На отчетную дату"),
_row(1, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100)),
)
built = _build(table)
assert "колонка 0 без даты: На отчетную дату" in built.warnings
assert built.document.columns[0].period_date is None and built.fully_recognized is False
def test_edition_not_determined_warning() -> None:
header = FileHeader(None, None, None, None, None, None, None, False)
table = _table(
"Приложение",
_row(0, (0, "Наименование показателя", 100), (1, "Код", 100), (2, "На 1 января", 100)),
_row(1, (0, "a", 100), (1, "5100", 100), (2, "1", 100)),
_row(2, (0, "b", 100), (1, "5200", 100), (2, "2", 100)),
)
built = _build(table, header)
assert built.edition_id is None and "редакция не определена" in built.warnings
assert built.document.lines[0].line_name == "a" and built.fully_recognized is False
class TestKnownCodeConfidence:
"""R-2, ветка «код в выбранной редакции»: confidence ячейки, если текст чистый, иначе не выше 99."""
def test_clean_code_keeps_cell_confidence(self) -> None:
table = _table(
"Бухгалтерский баланс",
CLEAN_BALANCE["rows"][0],
_row(1, (0, "ОС", 80), (1, "1150", 80), (2, "1", 100), (3, "2", 100)),
)
line = _build(table).document.lines[0]
assert line.line_code_confidence == 80 and line.line_name_confidence == 80
assert line.line_name == "Основные средства"
def test_dirty_code_is_capped_at_99(self) -> None:
"""«+1150» проходит `int()`, `clean_value` убирает «+» — текст не равен сырому, уверенность ≤ 99."""
table = _table(
"Бухгалтерский баланс",
CLEAN_BALANCE["rows"][0],
_row(1, (0, "ОС", 100), (1, "+1150", 100), (2, "1", 100), (3, "2", 100)),
)
built = _build(table)
line = built.document.lines[0]
assert line.line_code == "1150" and line.line_name == "Основные средства"
assert line.line_code_confidence == 99 and line.line_name_confidence == 99
assert built.fully_recognized is False