2a7ceca334
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
189 lines
9.6 KiB
Python
189 lines
9.6 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 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)
|