"""Проверка соотношений §7 по колонкам (design D-7, спека D-7). Правила из `statement_checks.yaml`; отсутствующий код = 0. Правая часть вычисляется как записано и с `abs_terms` по модулю (1С минус vs 66н скобки). Допуск `tolerance` для всех правил. Жёсткие нарушения → `relations_failed`, мягкие → `soft_relations_failed`. Правило проверяется, если целевой код в данных (подстроки всегда). OQ-138: жёсткое нарушение не больше `rounding_tolerance_percent` от наибольшего |числа| соотношения — округление, уходит в мягкие. AURORA-1527: «родитель = Σ подстрок» по колонке — только если хотя бы одна подстрока в ней непуста (amount не null). M-9: пятизначная подстрока относится к ближайшей четырёхзначной строке выше с теми же первыми тремя цифрами. """ from collections.abc import Sequence from decimal import Decimal from src.contract.result import FailedRelation, StatementChecks, StatementLine from src.parser.rules import FormType, RelationRule, Rules __all__ = ["run_statement_checks"] _ZERO = Decimal("0") _HUNDRED = Decimal("100") _PARENT_LEN = 4 _GROUP_LEN = 3 # первые цифры кода, общие у строки и её подстрок (1230 → 12301…12314) def _amounts(lines: Sequence[StatementLine], column_count: int) -> list[dict[str, Decimal]]: """Код → сумма по каждой колонке (null → 0).""" columns: list[dict[str, Decimal]] = [{} for _ in range(column_count)] for item in lines: for value in item.values: if 0 <= value.column_index < column_count: columns[value.column_index][item.line_code] = value.amount if value.amount is not None else _ZERO return columns def _filled(lines: Sequence[StatementLine], column_count: int) -> list[set[str]]: """Коды строк, у которых в колонке есть значение (`amount` не null; ноль — значение).""" columns: list[set[str]] = [set() for _ in range(column_count)] for item in lines: for value in item.values: if 0 <= value.column_index < column_count and value.amount is not None: columns[value.column_index].add(item.line_code) return columns def _rhs(amounts: dict[str, Decimal], rule: RelationRule, *, absolute: bool) -> Decimal: total = _ZERO for sign, code in rule.terms: value = amounts.get(code, _ZERO) if absolute and code in rule.abs_terms: value = abs(value) total += sign * value return total def _holds(amounts: dict[str, Decimal], rule: RelationRule, tolerance: Decimal) -> tuple[bool, Decimal, Decimal]: """Проверка соотношения в пределах допуска. Вычисляет правую часть два раза: как записано и с abs() для членов в abs_terms. Правило выполняется, если хотя бы один вариант в пределах tolerance. При нарушении возвращается вариант, ближайший к левой части (при равенстве — «как записано»): аналитик видит реальную величину расхождения. """ expected = amounts.get(rule.target, _ZERO) plain = _rhs(amounts, rule, absolute=False) if abs(expected - plain) <= tolerance: return True, expected, plain if rule.abs_terms: with_abs = _rhs(amounts, rule, absolute=True) if abs(expected - with_abs) <= tolerance: return True, expected, with_abs if abs(expected - with_abs) < abs(expected - plain): return False, expected, with_abs return False, expected, plain def _subline_rules(lines: Sequence[StatementLine], form: FormType, rules: Rules) -> list[RelationRule]: """«1110 = 11101 + 11102 + …» для каждого четырёхзначного кода формы (в диапазоне кодов этой формы), у которого есть пятизначные подстроки.""" low, high = rules.statement_headers.code_ranges[form] codes = [item.line_code for item in lines] parents = {code for code in codes if len(code) == _PARENT_LEN} children: dict[str, list[str]] = {} last_parent: str | None = None for code in codes: # порядок строк документа if len(code) == _PARENT_LEN: last_parent = code continue if len(code) != _PARENT_LEN + 1: continue # M-9: 12310 под 1230 — по порядку строк; без подходящей строки выше — по префиксу (11101 → 1110) if last_parent is not None and last_parent[:_GROUP_LEN] == code[:_GROUP_LEN]: parent = last_parent elif code[:_PARENT_LEN] in parents: parent = code[:_PARENT_LEN] else: continue children.setdefault(parent, []).append(code) return [ RelationRule(form=form, relation=f"{parent} = {' + '.join(kids)}", strict=False) for parent, kids in children.items() if low <= int(parent) <= high ] def run_statement_checks( lines: Sequence[StatementLine], column_count: int, form: FormType, rules: Rules ) -> tuple[StatementChecks, list[str]]: config = rules.statement_checks relations = list(config.relations_for(form)) subline_rules = _subline_rules(lines, form, rules) if config.subline_check else [] relations.extend(subline_rules) hard: list[FailedRelation] = [] soft: list[FailedRelation] = [] warnings: list[str] = [] filled = _filled(lines, column_count) for column_index, amounts in enumerate(_amounts(lines, column_count)): for rule in relations: # Явные правила из конфига пропускаются, если цель отсутствует в данных колонки. # Сгенерированные подстроки всегда проверяются (если они сгенерированы, то цель есть). is_subline = rule in subline_rules if not is_subline and rule.target not in amounts: continue # AURORA-1527: детализации нет (все подстроки родителя в колонке пусты) — соотношение не проверяется if is_subline and not any(code in filled[column_index] for _, code in rule.terms): continue holds, expected, actual = _holds(amounts, rule, config.tolerance) if holds: continue failed = FailedRelation(column_index=column_index, relation=rule.relation, expected=expected, actual=actual) # база — наибольшее по модулю число соотношения: итог может быть малой разностью больших слагаемых scale = max([abs(expected), *(abs(amounts.get(code, _ZERO)) for _, code in rule.terms)]) rounding = scale * config.rounding_tolerance_percent / _HUNDRED if rule.strict and abs(expected - actual) <= rounding: soft.append(failed) warnings.append( f"соотношение «{rule.relation}» сошлось в пределах округления (колонка {column_index}): " f"{expected} ≈ {actual}" ) continue if rule.strict: hard.append(failed) warning_msg = f"соотношение «{rule.relation}» нарушено (колонка {column_index}): {expected} ≠ {actual}" warnings.append(warning_msg) else: soft.append(failed) warning_msg = ( f"мягкое соотношение «{rule.relation}» нарушено (колонка {column_index}): {expected} ≠ {actual}" ) warnings.append(warning_msg) # списки задаются явно: `to_wire` сериализует только заданные поля, пустой список = «проверено, сошлось» return StatementChecks(relations_failed=hard, soft_relations_failed=soft), warnings