"""Шаг 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))