From 3ee74c6907542e3a3a369d107d82430338b69756 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 14:06:55 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=BF=D0=B5=D1=80=D0=B5=D0=B2=D0=BE?= =?UTF-8?q?=D0=B4=20=D0=B4=D0=BE=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=B0?= =?UTF-8?q?=20=D0=B8=20=D0=BF=D1=80=D0=B5=D0=B4=D1=83=D0=BF=D1=80=D0=B5?= =?UTF-8?q?=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D0=B9=20=D0=B2=20=D1=84=D0=BE?= =?UTF-8?q?=D1=80=D0=BC=D1=83=20=D0=BA=D0=BE=D0=BD=D1=82=D1=80=D0=B0=D0=BA?= =?UTF-8?q?=D1=82=D0=B0=20v2.0,=20=D0=B2=D1=85=D0=BE=D0=B4=20=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D1=82=D1=83=D1=81=D0=B0=20=D0=B8=D0=B7=20=D0=B4=D0=BE?= =?UTF-8?q?=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=B0=20=D0=BA=D0=BE=D0=BD?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D0=BA=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- arm_statements/__init__.py | 7 +- arm_statements/contract.py | 151 ++++++++++++++++++++++++++ arm_statements/tests/test_contract.py | 99 +++++++++++++++++ 3 files changed, 255 insertions(+), 2 deletions(-) create mode 100644 arm_statements/contract.py create mode 100644 arm_statements/tests/test_contract.py diff --git a/arm_statements/__init__.py b/arm_statements/__init__.py index d8027ce..d75b97f 100644 --- a/arm_statements/__init__.py +++ b/arm_statements/__init__.py @@ -1,12 +1,12 @@ """Общий код нормализации форм ББ/ОФР. -Публичный API — только имена `__all__` (раздел «Публичный API пакета» плана 1). Имена задачи 19 -(`to_contract`, `warning_to_contract`, `status_input_from_contract`) добавляются ими. +Публичный API — только имена `__all__` (раздел «Публичный API пакета» плана 1). Контрактные имена — `contract.py`. """ from arm_statements.amounts import parse_amount from arm_statements.codes import parse_code, sheet_joins from arm_statements.columns import amount_columns +from arm_statements.contract import status_input_from_contract, to_contract, warning_to_contract from arm_statements.dates import parse_date, resolve_relative from arm_statements.header import continuation_type, is_continuation, parse_header, type_hint from arm_statements.issues import WarningCode, make_warning @@ -130,6 +130,9 @@ __all__ = [ "sheet_joins", "statements_version", "status_input", + "status_input_from_contract", + "to_contract", "to_jsonable", "type_hint", + "warning_to_contract", ] diff --git a/arm_statements/contract.py b/arm_statements/contract.py new file mode 100644 index 0000000..10e7d80 --- /dev/null +++ b/arm_statements/contract.py @@ -0,0 +1,151 @@ +"""Перевод выхода пакета в форму контракта v2.0 и вход статуса из документа контракта. + +Один перевод на оба сервиса (P1-26): хосты свой `contract_map` не пишут. Схему пакет не читает — +форма держится тестами и разовой сверкой со `statementDocument` схемы v2.0. +""" + +from collections.abc import Mapping +from decimal import Decimal +from typing import Any + +from arm_statements.model import ( + HEADER_FIELDS, + DocumentType, + FailedRelation, + LineStatus, + PeriodKind, + Severity, + StatementDocument, + StatementHeader, + StatementWarning, +) +from arm_statements.status import StatusInput, StatusLine, StatusValue + + +def _money(value: Decimal | None) -> str | None: + return None if value is None else format(value, "f") + + +def warning_to_contract(warning: StatementWarning, *, file_level: bool = False) -> dict[str, Any]: + """`warning` схемы; при `file_level` — `fileWarning` (с `locator`). Ключи без значения опускаются.""" + data: dict[str, Any] = { + "code": warning.code, + "severity": str(warning.severity), + "message": warning.message, + } + if warning.line_refs: + data["line_refs"] = list(warning.line_refs) + if warning.column_index is not None: + data["column_index"] = warning.column_index + if warning.field is not None: + data["field"] = warning.field + if file_level and warning.locator is not None: + data["locator"] = warning.locator + return data + + +def _header(header: StatementHeader) -> dict[str, Any]: + data: dict[str, Any] = {"form_name": header.form_name} + for name in HEADER_FIELDS: + field = getattr(header, name) + value = field.value + data[name] = value.isoformat() if name == "report_date" and value is not None else value + data[f"{name}_text"] = field.text + data[f"{name}_confidence"] = field.confidence + data["needs_review_fields"] = list(header.needs_review_fields) + return data + + +def _relation(relation: FailedRelation) -> dict[str, Any]: + if relation.column_index is None: + raise ValueError(f"у соотношения {relation.relation!r} нет column_index: у сервиса номер колонки есть всегда") + return { + "column_index": relation.column_index, + "relation": relation.relation, + "codes": list(relation.codes), + "expected": _money(relation.expected), + "actual": _money(relation.actual), + } + + +def to_contract(document: StatementDocument) -> dict[str, Any]: + """`statementDocument` схемы v2.0 как json-совместимый `dict`.""" + data: dict[str, Any] = { + "document_type": str(document.document_type), + "form_variant": None if document.form_variant is None else str(document.form_variant), + "locator": document.locator, + "status": str(document.status), + "warnings": [warning_to_contract(w) for w in document.warnings], + "header": _header(document.header), + "columns": [ + { + "column_index": c.column_index, + "label": c.label, + "label_confidence": c.label_confidence, + "period_date": None if c.period_date is None else c.period_date.isoformat(), + "period_kind": str(c.period_kind), + } + for c in document.columns + ], + "lines": [ + { + "row_index": ln.row_index, + "ref": ln.ref, + "line_status": str(ln.line_status), + "line_code": ln.line_code, + "line_code_text": ln.line_code_text, + "line_code_confidence": ln.line_code_confidence, + "line_name": ln.line_name, + "line_name_confidence": ln.line_name_confidence, + "parent_ref": ln.parent_ref, + "needs_review": ln.needs_review, + "values": [ + { + "column_index": v.column_index, + "amount": _money(v.amount), + "string_value": v.string_value, + "text": v.text, + "confidence": v.confidence, + "is_recognized": v.is_recognized, + "needs_review": v.needs_review, + } + for v in ln.values + ], + } + for ln in document.lines + ], + } + if document.checks is not None: + data["checks"] = { + "relations_failed": [_relation(r) for r in document.checks.relations_failed], + "soft_relations_failed": [_relation(r) for r in document.checks.soft_relations_failed], + } + return data + + +def status_input_from_contract(document: Mapping[str, Any]) -> StatusInput: + """Вход функции статуса из `statementDocument` контракта v2.0 (то же правило, что `status_input`). + + Схему не проверяет: битый документ — `KeyError` / `ValueError`. Отсутствующий необязательный ключ и `null` равны. + """ + header = document["header"] + lines = document["lines"] + columns = document["columns"] + period_known = header.get("report_date") is not None or any( + c["period_kind"] == PeriodKind.ABSOLUTE and c.get("period_date") for c in columns + ) + checks = document.get("checks") + return StatusInput( + document_type=DocumentType(document["document_type"]), + layout_found=len(lines) > 0, + has_organization_or_inn=bool(header.get("organization_name") or header.get("inn")), + period_known=period_known, + needs_review_fields=tuple(header.get("needs_review_fields") or ()), + lines=tuple(StatusLine(LineStatus(ln["line_status"]), bool(ln["needs_review"])) for ln in lines), + values=tuple( + StatusValue(LineStatus(ln["line_status"]), bool(v["needs_review"])) for ln in lines for v in ln["values"] + ), + hard_failures=0 if checks is None else len(checks["relations_failed"]), + accepted=False, + warning_severities=tuple(Severity(w["severity"]) for w in document["warnings"]), + ) diff --git a/arm_statements/tests/test_contract.py b/arm_statements/tests/test_contract.py new file mode 100644 index 0000000..fe63ba8 --- /dev/null +++ b/arm_statements/tests/test_contract.py @@ -0,0 +1,99 @@ +import json +from dataclasses import replace +from decimal import Decimal as D + +import pytest + +from arm_statements import ( + DocStatus, + DocumentType, + document_status, + make_warning, + normalize_candidate, + status_input, + status_input_from_contract, + to_contract, + warning_to_contract, +) +from arm_statements.loader import Rules +from arm_statements.model import FailedRelation, StatementChecks, StatementDocument +from arm_statements.tests.builders import hdr, ocr + +HEAD = ["Наименование показателя", "Код", "На 30.09.2025"] +FULL_HEADER = hdr( + organization_name="ООО «Пример»", inn=("7700000000", 80.0), report_date="30.09.2025", units="тыс. руб." +) +HEADER_KEYS = {"form_name", "needs_review_fields"} | { + f"{name}{suffix}" + for name in ("organization_name", "inn", "report_date", "units", "okud", "okpo", "okopf", "okfs", "okei", "address") + for suffix in ("", "_text", "_confidence") +} + + +def balance(rules: Rules, *rows: list[str]) -> StatementDocument: + cand = ocr([HEAD, *rows], header=FULL_HEADER, type_hint=DocumentType.BALANCE, title="Бухгалтерский баланс") + doc = normalize_candidate(cand, rules).document + assert doc is not None + return doc + + +def test_document_in_contract_form(rules: Rules) -> None: + doc = balance(rules, ["Запасы", "1210", "60"], ["Итого II", "1200", "60"], ["НИ", "", "7"]) + data = json.loads(json.dumps(to_contract(doc), ensure_ascii=False)) + assert set(data["header"]) == HEADER_KEYS + assert (data["header"]["form_name"], data["header"]["inn"], data["header"]["inn_confidence"]) == ( + "Бухгалтерский баланс", + "7700000000", + 80.0, + ) + assert data["header"]["report_date"] == "2025-09-30" and data["header"]["okei"] is None + assert data["lines"][0]["values"][0]["amount"] == "60.00" + assert {ln["line_status"] for ln in data["lines"]} >= {"ok"} + assert "column_key" not in json.dumps(data) and "row:" not in json.dumps(data["header"]) + assert "locator" not in json.dumps(data["warnings"]) + assert set(data["checks"]) == {"relations_failed", "soft_relations_failed"} + + +def test_empty_target_is_null_in_contract(rules: Rules) -> None: + doc = balance(rules, ["Запасы", "1210", "60"], ["Итого II", "1200", "60"], ["Итого I", "1100", "40"]) + failed = to_contract(doc)["checks"]["relations_failed"] + target = next(f for f in failed if f["relation"] == "1600 = 1100 + 1200") + assert (target["expected"], target["actual"], target["column_index"]) == (None, "100.00", 2) + + +def test_no_lines_no_checks_key(rules: Rules) -> None: + doc = normalize_candidate( + ocr([["Итого", "15 000"]], type_hint=DocumentType.BALANCE, header=FULL_HEADER), rules + ).document + assert doc is not None and "checks" not in to_contract(doc) and to_contract(doc)["lines"] == [] + + +def test_relation_without_column_index_is_error(rules: Rules) -> None: + doc = balance(rules, ["Запасы", "1210", "60"], ["Итого II", "1200", "60"]) + bad = FailedRelation(None, "c1", "1600 = 1700", ("1600", "1700"), D("1"), None) + with pytest.raises(ValueError, match="column_index"): + to_contract(replace(doc, checks=StatementChecks((bad,), ()))) + + +def test_warning_keys(rules: Rules) -> None: + w = make_warning("ROW_SPLIT", "разлеплено", rules, line_refs=("row:3#1", "row:3#2"), locator="file:a.json") + assert warning_to_contract(w) == { + "code": "ROW_SPLIT", + "severity": "info", + "message": "разлеплено", + "line_refs": ["row:3#1", "row:3#2"], + } + lost = make_warning("TABLE_LOST", "таблица", rules, locator="file:a.json;table:1") + assert warning_to_contract(lost, file_level=True) == { + "code": "TABLE_LOST", + "severity": "error", + "message": "таблица", + "locator": "file:a.json;table:1", + } + + +def test_status_input_roundtrip(rules: Rules) -> None: + doc = balance(rules, ["Запасы", "1210", "60"], ["Итого II", "1200", "60"], ["НИ", "", "7"]) + data = status_input_from_contract(json.loads(json.dumps(to_contract(doc), ensure_ascii=False))) + assert data == status_input(doc) + assert document_status(data) is doc.status is DocStatus.NEEDS_CHECK