arm-core/api/arm_statements/recompute.py

"""Точка входа пересчёта формы по сохранённым данным (спека «Точка входа пакета для пересчёта…»).

Ядро собирает `StoredForm` из своих строк БД; сервисы функцией не пользуются. Пересчёт не меняет `line_status`,
суммы и коды — только выводит признаки (Р-34: подтверждение кода — `code_fixed`, уверенность кода не меняется).
"""

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import date
from decimal import Decimal

from arm_statements.checks import checks_from
from arm_statements.evaluation import EvalLine, EvalValue
from arm_statements.loader import Rules, default_rules
from arm_statements.model import (
    DocStatus,
    DocumentType,
    Edition,
    FormVariant,
    LineStatus,
    PeriodKind,
    Severity,
    StatementChecks,
)
from arm_statements.review import evaluate
from arm_statements.status import StatusInput, StatusLine, StatusValue, document_status
from arm_statements.structure import edition_by_date


@dataclass(frozen=True, slots=True)
class StoredColumn:
    key: str
    column_index: int | None
    period_date: date | None
    period_kind: PeriodKind


@dataclass(frozen=True, slots=True)
class StoredValue:
    column_key: str
    amount: Decimal | None
    text: str | None  # source_text; у значения правки None
    confidence: float | None
    is_recognized: bool
    value_fixed: bool  # is_modified или is_confirmed


@dataclass(frozen=True, slots=True)
class StoredLine:
    key: str
    row_index: int | None  # None — строка правки
    line_status: LineStatus
    line_code: str | None
    line_code_confidence: float | None
    line_name: str | None
    code_fixed: bool  # line_code_modified или is_confirmed строки
    parent_key: str | None
    values: tuple[StoredValue, ...]


@dataclass(frozen=True, slots=True)
class StoredWarning:
    code: str
    severity: Severity


@dataclass(frozen=True, slots=True)
class StoredForm:
    document_type: DocumentType
    effective_date: date | None
    has_organization_or_inn: bool
    period_known: bool
    needs_review_fields: tuple[str, ...]
    accepted: bool
    warnings: tuple[StoredWarning, ...]
    columns: tuple[StoredColumn, ...]
    lines: tuple[StoredLine, ...]


@dataclass(frozen=True, slots=True)
class RecomputeResult:
    form_variant: FormVariant | None
    edition: Edition | None
    checks: StatementChecks
    line_needs_review: Mapping[str, bool]
    value_needs_review: Mapping[tuple[str, str], bool]
    detail_parents: Mapping[str, str | None]
    status: DocStatus


def _sort_key(line: StoredLine) -> tuple[bool, int]:
    return (line.row_index is None, line.row_index or 0)


def _eval_line(line: StoredLine) -> EvalLine:
    return EvalLine(
        key=line.key,
        order=line.row_index,
        line_status=line.line_status,
        line_code=line.line_code,
        code_confidence=line.line_code_confidence,
        code_fixed=line.code_fixed,
        values=tuple(
            EvalValue(v.column_key, v.amount, v.text, v.confidence, v.is_recognized, fixed=v.value_fixed)
            for v in line.values
        ),
    )


def _detail_parents(lines: Sequence[StoredLine]) -> dict[str, str | None]:
    """Ближайшая выше по `row_index` строка `ok` с `row_index`; у `detail` без `row_index` и без такой — `None`."""
    parents: dict[str, str | None] = {}
    last_ok: str | None = None
    for line in lines:  # уже отсортированы: строки правки в конце
        if line.row_index is None:
            if line.line_status is LineStatus.DETAIL:
                parents[line.key] = None
            continue
        if line.line_status is LineStatus.DETAIL:
            parents[line.key] = last_ok
        elif line.line_status is LineStatus.OK:
            last_ok = line.key
    return parents


def recompute(form: StoredForm, rules: Rules | None = None) -> RecomputeResult:
    """Шаги 6–8 по сохранённым данным: вариант, редакция, §7, `needs_review`, родители `detail`, статус."""
    rules = rules if rules is not None else default_rules()
    if form.document_type is DocumentType.DISCLOSURE:
        ordered = sorted(form.lines, key=_sort_key)
        return RecomputeResult(
            form_variant=None,
            edition=edition_by_date(form.effective_date),
            checks=StatementChecks((), ()),
            line_needs_review={line.key: False for line in ordered},
            value_needs_review={(line.key, v.column_key): False for line in ordered for v in line.values},
            detail_parents=_detail_parents(ordered),
            status=DocStatus.DONE,
        )
    lines = sorted(form.lines, key=_sort_key)
    evaluation = evaluate(
        [_eval_line(line) for line in lines],
        [column.key for column in form.columns],
        form.document_type,
        form.effective_date,
        rules,
        accepted=form.accepted,
    )
    checks = checks_from(evaluation.outcomes, {column.key: column.column_index for column in form.columns})
    status = document_status(
        StatusInput(
            document_type=form.document_type,
            layout_found=any(line.row_index is not None for line in lines),
            has_organization_or_inn=form.has_organization_or_inn,
            period_known=form.period_known,
            needs_review_fields=form.needs_review_fields,
            lines=tuple(
                StatusLine(line.line_status, evaluation.line_needs_review.get(line.key, False)) for line in lines
            ),
            values=tuple(
                StatusValue(line.line_status, evaluation.value_needs_review.get((line.key, v.column_key), False))
                for line in lines
                for v in line.values
            ),
            hard_failures=evaluation.hard_failures,
            accepted=form.accepted,
            warning_severities=tuple(w.severity for w in form.warnings),
        )
    )
    return RecomputeResult(
        form_variant=evaluation.variant,
        edition=evaluation.edition,
        checks=checks,
        line_needs_review=evaluation.line_needs_review,
        value_needs_review=evaluation.value_needs_review,
        detail_parents=_detail_parents(lines),
        status=status,
    )