c1288d08fb
- C-1 (№30): сумма с более чем двумя знаками после запятой не принимается (amount=null, string_value, is_recognized=false); confidence ячеек и шапки приводится к [0, 100], NaN и бесконечность дают 0 - F-A (№28): ошибочный файл набора при успешном понижает статус до needs_check - F-B: «ИНН не распознан» только когда ИНН не извлечён, иначе «ИНН распознан неуверенно» - №29: предупреждения «строк с нечисловым кодом пропущено: N» и «строк без кода с суммами пропущено: N» - I-3 (№31): тексты дат длиннее 200 символов не разбираются и не проходят fix_lookalikes - T1.3 (№33): загрузчик yaml отклоняет повторяющиеся ключи; тавтологичный тест дублей заменён настоящим - гигиена: докстринг amounts.py описывает фактическое поведение (все точки удаляются), мёртвые noqa в ocr/__init__.py Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
221 lines
12 KiB
Python
221 lines
12 KiB
Python
"""`OcrNormalizer` (spec «Статус результата нормализации», «Результат по контракту», D-6)."""
|
||
|
||
import copy
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
import src.normalizer.ocr as ocr_module
|
||
from src.contract import ErrorCode, ResultStatus
|
||
from src.gateway.hotfolder import AnswerFile
|
||
from src.gateway.merge import merge_answers
|
||
from src.normalizer import NO_TABLES_MESSAGE, OcrNormalizer
|
||
from src.normalizer.rules import load_rules
|
||
from tests.worker_helpers import SUCCESS_ANSWER
|
||
|
||
API_DIR = Path(__file__).resolve().parents[3]
|
||
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
|
||
JOB = "3f1c2b7a-0000-4000-8000-000000000001"
|
||
RULES = load_rules(API_DIR / "rules")
|
||
|
||
|
||
def _merged(*files: tuple[str, Any]) -> dict[str, Any]:
|
||
answers = [AnswerFile(name, json.dumps(body, ensure_ascii=False).encode()) for name, body in files]
|
||
return dict(merge_answers(JOB, answers))
|
||
|
||
|
||
def _real_set(*names: str) -> dict[str, Any]:
|
||
if not REAL_FIXTURES.is_dir():
|
||
pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
|
||
files = []
|
||
for name in names:
|
||
suffix = name.removesuffix(".json").partition("_")[2].partition("_")[2]
|
||
target = f"{JOB}_{suffix}.json" if suffix else f"{JOB}.json"
|
||
files.append(AnswerFile(target, (REAL_FIXTURES / name).read_bytes()))
|
||
return dict(merge_answers(JOB, files))
|
||
|
||
|
||
CLEAN: dict[str, Any] = {
|
||
"file_uid": JOB,
|
||
"error": {"code": "", "message": ""},
|
||
"data_ocr": {
|
||
"label": {"form_name": "Бухгалтерская отчетность", "confidence": 100},
|
||
"organization": {"form_name": "ООО Ромашка", "confidence": 100},
|
||
"inn": {"form_name": "7700000000", "confidence": 100},
|
||
"units": {"form_name": "тыс. руб.", "confidence": 100},
|
||
"date": {"form_name": "на 31 декабря 2024 г.", "confidence": 100},
|
||
"table_structure": [
|
||
{
|
||
"label": {"form_name": "Бухгалтерский баланс", "confidence": 100},
|
||
"rows": [
|
||
{
|
||
"row_index": 0,
|
||
"cells": [
|
||
{"column_index": 0, "value": "Наименование показателя", "confidence": 100},
|
||
{"column_index": 1, "value": "Код", "confidence": 100},
|
||
{"column_index": 2, "value": "На 31 декабря 2024 г.", "confidence": 100},
|
||
],
|
||
},
|
||
{
|
||
"row_index": 1,
|
||
"cells": [
|
||
{"column_index": 0, "value": "Основные средства", "confidence": 100},
|
||
{"column_index": 1, "value": "1150", "confidence": 100},
|
||
{"column_index": 2, "value": "1 234", "confidence": 100},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
class TestStatus:
|
||
def test_done_when_everything_recognized(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", CLEAN)))
|
||
assert outcome.status is ResultStatus.DONE and outcome.warnings == []
|
||
assert len(outcome.documents) == 1 and outcome.documents[0].document_type == "BALANCE"
|
||
assert outcome.metrics is not None and outcome.metrics.rows == 1
|
||
|
||
def test_needs_check_with_table_warning(self) -> None:
|
||
noisy = copy.deepcopy(CLEAN)
|
||
noisy["data_ocr"]["table_structure"][0]["rows"][1]["cells"][2]["confidence"] = 90
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", noisy)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
|
||
def test_no_tables_is_recognition_failed(self) -> None:
|
||
empty = copy.deepcopy(CLEAN)
|
||
empty["data_ocr"]["table_structure"] = []
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", empty)))
|
||
assert outcome.status is ResultStatus.ERROR
|
||
assert outcome.error is not None and outcome.error.code is ErrorCode.RECOGNITION_FAILED
|
||
assert outcome.error.message == NO_TABLES_MESSAGE
|
||
|
||
def test_skipped_table_is_warned(self) -> None:
|
||
bad = copy.deepcopy(CLEAN)
|
||
empty_table = {"label": {"form_name": "x"}, "rows": [{"row_index": 0, "cells": []}]}
|
||
bad["data_ocr"]["table_structure"].append(empty_table)
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", bad)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert f"{JOB}.json: table:1: нет шапки и столбца кода" in outcome.warnings
|
||
|
||
def test_exception_in_one_table_becomes_warning(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
real = ocr_module.build_document
|
||
|
||
def boom(table: Any, index: int, *args: Any, **kwargs: Any) -> Any:
|
||
if index == 1:
|
||
raise RuntimeError("сломалось")
|
||
return real(table, index, *args, **kwargs)
|
||
|
||
monkeypatch.setattr(ocr_module, "build_document", boom)
|
||
two = copy.deepcopy(CLEAN)
|
||
two["data_ocr"]["table_structure"].append(copy.deepcopy(two["data_ocr"]["table_structure"][0]))
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", two)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK and len(outcome.documents) == 1
|
||
assert f"{JOB}.json: table:1: BACKEND_PARSING_ERROR: RuntimeError: сломалось" in outcome.warnings
|
||
|
||
def test_file_level_warnings(self) -> None:
|
||
noisy = copy.deepcopy(CLEAN)
|
||
noisy["data_ocr"]["inn"] = {"form_name": "+", "confidence": 0}
|
||
noisy["data_ocr"]["date"] = {"form_name": "Общество", "confidence": 100}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", noisy)))
|
||
assert f"{JOB}.json: ИНН не распознан" in outcome.warnings
|
||
assert not any("ИНН распознан неуверенно" in w for w in outcome.warnings)
|
||
assert f"{JOB}.json: дата отчёта не разобрана" in outcome.warnings
|
||
|
||
def test_inn_found_but_unsure_is_worded_differently(self) -> None:
|
||
"""F-B: «не распознан» — только когда ИНН не извлечён; извлечённый с confidence < 100 — «неуверенно»."""
|
||
unsure = copy.deepcopy(CLEAN)
|
||
unsure["data_ocr"]["inn"] = {"form_name": "7700000000", "confidence": 75}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", unsure)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert f"{JOB}.json: ИНН распознан неуверенно" in outcome.warnings
|
||
assert not any("ИНН не распознан" in w for w in outcome.warnings)
|
||
|
||
def test_inn_with_noise_is_unsure(self) -> None:
|
||
"""ИНН извлечён из мусора (confidence ограничивается 99) — тоже «неуверенно», а не «не распознан»."""
|
||
noisy = copy.deepcopy(CLEAN)
|
||
noisy["data_ocr"]["inn"] = {"form_name": "77 0000 0000 |", "confidence": 100}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", noisy)))
|
||
assert outcome.warnings == [f"{JOB}.json: ИНН распознан неуверенно"]
|
||
|
||
def test_inn_missing_field_is_not_recognized(self) -> None:
|
||
missing = copy.deepcopy(CLEAN)
|
||
del missing["data_ocr"]["inn"]
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", missing)))
|
||
assert outcome.warnings == [f"{JOB}.json: ИНН не распознан"]
|
||
|
||
|
||
class TestErrorFilesStillHandled:
|
||
def test_all_files_failed_keeps_first_code(self) -> None:
|
||
failed = {"file_uid": JOB, "error": {"code": "INVALID_DOCUMENT", "message": "нет"}, "data_ocr": {}}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", failed)))
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_DOCUMENT
|
||
assert outcome.warnings == [f"{JOB}.json: INVALID_DOCUMENT: нет"]
|
||
|
||
def test_mixed_files(self) -> None:
|
||
failed = {"file_uid": JOB, "error": {"code": "INVALID_DOCUMENT", "message": "нет"}, "data_ocr": {}}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", CLEAN), (f"{JOB}_ofr.json", failed)))
|
||
# №28: потерянный ошибочный файл не остаётся под `done`, даже если успешный файл распознан полностью
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert outcome.error is None and len(outcome.documents) == 1
|
||
assert outcome.warnings == [f"{JOB}_ofr.json: INVALID_DOCUMENT: нет"]
|
||
assert OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", CLEAN))).status is ResultStatus.DONE
|
||
|
||
def test_mixed_files_error_first_in_order(self) -> None:
|
||
"""Порядок файлов не важен: ошибочный первым, успешный вторым — тоже needs_check с документом."""
|
||
failed = {"file_uid": JOB, "error": {"code": "INVALID_DOCUMENT", "message": "нет"}, "data_ocr": {}}
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}_a.json", failed), (f"{JOB}_b.json", CLEAN)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK and len(outcome.documents) == 1
|
||
|
||
def test_not_merged_structure(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize({"job_id": JOB})
|
||
assert outcome.status is ResultStatus.ERROR and outcome.error is not None
|
||
assert outcome.error.code is ErrorCode.INVALID_FORMAT
|
||
|
||
def test_success_answer_helper_from_worker_tests(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize(_merged((f"{JOB}.json", SUCCESS_ANSWER)))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK # confidence < 100 у шапки
|
||
|
||
|
||
@pytest.mark.fixtures
|
||
class TestRealFixtures:
|
||
def test_aep6(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize(_real_set("aep_6.json"))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
types = [d.document_type for d in outcome.documents]
|
||
assert types == ["BALANCE", "INCOME"]
|
||
balance, income = outcome.documents
|
||
assert [(c.column_index, c.label, str(c.period_date)) for c in balance.columns] == [
|
||
(0, "На 30 июня 2025 г", "2025-06-30"),
|
||
(1, "На 31 декабря 2024 г.", "2024-12-31"),
|
||
(2, "На 31 декабря 2023 г.", "2023-12-31"),
|
||
]
|
||
codes = [line.line_code for line in balance.lines]
|
||
assert "1105" in codes and "1110" in codes
|
||
assert all(line.line_code for line in balance.lines)
|
||
assert [str(c.period_date) for c in income.columns] == ["2025-06-30", "2024-06-30"]
|
||
revenue = next(line for line in income.lines if line.line_code == "2110")
|
||
assert str(revenue.values[0].amount) == "13070"
|
||
assert balance.header.units == "тыс. руб." and balance.header.inn is None and balance.header.report_date is None
|
||
|
||
def test_aep3(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize(_real_set("aep_3.json"))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
balance = next(d for d in outcome.documents if d.document_type == "BALANCE")
|
||
assert all(c.period_date is None for c in balance.columns)
|
||
assert balance.header.inn == "3121200351"
|
||
assert balance.header.inn_confidence is not None and balance.header.inn_confidence < 100
|
||
assert str(balance.header.report_date) == "2024-12-31"
|
||
assert any(line.line_code == "1120" for line in balance.lines)
|
||
assert any("шапка синтезирована" in w for w in outcome.warnings)
|
||
|
||
def test_aep3_with_failed_ofr(self) -> None:
|
||
outcome = OcrNormalizer(RULES).normalize(_real_set("aep_3.json", "aep_3_ofr.json"))
|
||
assert outcome.status is ResultStatus.NEEDS_CHECK
|
||
assert any(w.startswith(f"{JOB}_ofr.json: INVALID_DOCUMENT:") for w in outcome.warnings)
|