"""`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 f"{JOB}.json: дата отчёта не разобрана" in outcome.warnings 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))) assert outcome.status is ResultStatus.DONE # ошибочный файл — предупреждение, не needs_check assert outcome.warnings == [f"{JOB}_ofr.json: INVALID_DOCUMENT: нет"] 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)