arm-statements/arm_statements/checks.py

"""Шаг 7: самопроверки §7 и вариант формы."""

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

from arm_statements.codes import document_type_of, section_of
from arm_statements.evaluation import EvalLine, is_check_line
from arm_statements.loader import RelationRule, Rules
from arm_statements.model import DocumentType, FailedRelation, FormVariant, StatementChecks

_ZERO = Decimal("0")
_HUNDRED = Decimal("100")
_PARENT_LEN = 4
_SUBLINE_LEN = 5
_GROUP_LEN = 3  # первые цифры, общие у строки и её подстрок (1230 → 12301…12314)


def form_variant(document_type: DocumentType, ok_codes: Iterable[str]) -> FormVariant | None:
    """Вариант формы по структуре кодов (спека «Шаг 6»); ОКУД признаком не служит."""
    codes = set(ok_codes)
    if document_type is DocumentType.BALANCE:
        simplified = "1600" in codes and not {"1100", "1200"} & codes
    elif document_type is DocumentType.INCOME:
        simplified = bool({"2400", "2110"} & codes) and not {"2100", "2200", "2300"} & codes
    else:
        return None
    return FormVariant.SIMPLIFIED if simplified else FormVariant.FULL


def subline_parents(codes: Sequence[str]) -> list[str | None]:
    """Родитель каждой позиции: у пятизначного кода — ближайший четырёхзначный выше с теми же тремя цифрами,
    иначе первые четыре цифры; у прочих — `None`."""
    result: list[str | None] = []
    last_parent: str | None = None
    for code in codes:
        if len(code) == _PARENT_LEN:
            last_parent = code
            result.append(None)
        elif len(code) == _SUBLINE_LEN:
            if last_parent is not None and last_parent[:_GROUP_LEN] == code[:_GROUP_LEN]:
                result.append(last_parent)
            else:
                result.append(code[:_PARENT_LEN])
        else:
            result.append(None)
    return result


@dataclass(frozen=True, slots=True)
class RelationOutcome:
    column: str
    rule: RelationRule
    codes: tuple[str, ...]
    computed: bool
    expected: Decimal | None  # None — цель пуста
    actual: Decimal | None  # None — пусты все члены
    within_tolerance: bool
    within_rounding: bool

    @property
    def hard_failed(self) -> bool:
        return self.rule.strict and self.computed and not self.within_tolerance and not self.within_rounding


def _column_sums(lines: Sequence[EvalLine], column: str) -> tuple[dict[str, Decimal], set[str]]:
    """Код → сумма `amount` по строкам `is_check_line` в колонке; второй результат — коды с непустой суммой."""
    sums: dict[str, Decimal] = {}
    for item in lines:
        if not is_check_line(item) or item.line_code is None:
            continue
        for value in item.values:
            if value.column == column and value.amount is not None:
                sums[item.line_code] = sums.get(item.line_code, _ZERO) + value.amount
    return sums, set(sums)


def _rhs(sums: Mapping[str, Decimal], rule: RelationRule, *, absolute: bool) -> Decimal:
    total = _ZERO
    for sign, code in rule.terms:
        value = sums.get(code, _ZERO)
        if absolute and code in rule.abs_terms:
            value = abs(value)
        total += sign * value
    return total


def _actual(sums: Mapping[str, Decimal], rule: RelationRule, expected: Decimal, tolerance: Decimal) -> Decimal:
    """Правая часть: сошедшийся вариант (как записано, затем по модулю), иначе ближайший к цели."""
    plain = _rhs(sums, rule, absolute=False)
    if abs(expected - plain) <= tolerance or not rule.abs_terms:
        return plain
    with_abs = _rhs(sums, rule, absolute=True)
    if abs(expected - with_abs) <= tolerance or abs(expected - with_abs) < abs(expected - plain):
        return with_abs
    return plain


def _outcome(
    column: str, rule: RelationRule, sums: Mapping[str, Decimal], filled: set[str], rules: Rules
) -> RelationOutcome:
    codes = (rule.target, *(code for _, code in rule.terms))
    target_filled = rule.target in filled
    members_filled = any(code in filled for _, code in rule.terms)
    if not target_filled and not members_filled:
        return RelationOutcome(column, rule, codes, False, None, None, True, True)
    expected = sums[rule.target] if target_filled else None
    if expected is None or not members_filled:
        actual = _rhs(sums, rule, absolute=False) if members_filled else None
        return RelationOutcome(column, rule, codes, True, expected, actual, False, False)
    actual = _actual(sums, rule, expected, rules.checks.tolerance)
    diff = abs(expected - actual)
    scale = max(abs(expected), *(abs(sums.get(code, _ZERO)) for _, code in rule.terms))
    within_rounding = diff <= scale * rules.checks.rounding_tolerance_percent / _HUNDRED
    return RelationOutcome(column, rule, codes, True, expected, actual, diff <= rules.checks.tolerance, within_rounding)


def _subline_rules(
    lines: Sequence[EvalLine], document_type: DocumentType, variant: FormVariant, rules: Rules
) -> list[RelationRule]:
    """«родитель = Σ подстрок» для родителей из отрезка формы (мягкие), в порядке строк документа."""
    codes = [item.line_code for item in lines if is_check_line(item) and item.line_code is not None]
    children: dict[str, list[str]] = {}
    for code, parent in zip(codes, subline_parents(codes), strict=True):
        if parent is None or code in children.get(parent, []):
            continue
        section = section_of(parent, rules)
        if section is None or document_type_of(section) is not document_type:
            continue
        children.setdefault(parent, []).append(code)
    return [
        RelationRule(
            form=document_type,
            variant=variant,
            relation=f"{parent} = {' + '.join(kids)}",
            target=parent,
            terms=tuple((1, kid) for kid in kids),
            strict=False,
            abs_terms=frozenset(),
        )
        for parent, kids in children.items()
    ]


def evaluate_relations(
    lines: Sequence[EvalLine],
    columns: Sequence[str],
    document_type: DocumentType,
    variant: FormVariant | None,
    rules: Rules,
) -> list[RelationOutcome]:
    """Все соотношения набора формы и подстрок по каждой колонке. `lines` — в порядке документа."""
    effective = variant or FormVariant.FULL
    static = list(rules.checks.relations_for(document_type, effective))
    generated = _subline_rules(lines, document_type, effective, rules) if rules.checks.subline_check else []
    outcomes: list[RelationOutcome] = []
    for column in columns:
        sums, filled = _column_sums(lines, column)
        outcomes.extend(_outcome(column, rule, sums, filled, rules) for rule in static)
        # подстроки — только если хоть одна подстрока в колонке непуста (AURORA-1527)
        outcomes.extend(
            _outcome(column, rule, sums, filled, rules)
            for rule in generated
            if any(code in filled for _, code in rule.terms)
        )
    return outcomes


def checks_from(outcomes: Sequence[RelationOutcome], column_index_of: Mapping[str, int | None]) -> StatementChecks:
    hard: list[FailedRelation] = []
    soft: list[FailedRelation] = []
    for item in outcomes:
        if not item.computed or item.within_tolerance:
            continue
        failure = FailedRelation(
            column_index=column_index_of.get(item.column),
            column_key=item.column,
            relation=item.rule.relation,
            codes=item.codes,
            expected=item.expected,
            actual=item.actual,
        )
        (hard if item.hard_failed else soft).append(failure)
    return StatementChecks(relations_failed=tuple(hard), soft_relations_failed=tuple(soft))