arm-statements/arm_statements/contract.py

"""Перевод выхода пакета в форму контракта 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"]),
    )