"""Логические проверки форм ББ/ОФР §7 в ядре (spec statement-checks; design D-4, D-5). Перенос `tabular-parser/api/src/parser/statements/checks.py` по семантике: соотношение по каждой колонке, только если в форме есть строка его цели (пустая сумма `None` = 0: оба производителя отдают значение на каждую колонку каждой строки); отсутствующий код и пустая сумма в правой части — 0; правая часть считается как записано и с `abs_terms` по модулю, выполнено, если хоть один вариант в пределах `tolerance`; жёсткое нарушение не больше `rounding_tolerance_percent` от наибольшего по модулю числа соотношения — мягкое; «сумма пятизначных подстрок = четырёхзначный родитель» — мягкая. Результат не хранится (DBML). """ import functools import re from collections.abc import Mapping, Sequence from dataclasses import dataclass from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Literal, cast import yaml __all__ = [ "CROSS_FORM_RELATION", "CheckRules", "ChecksReport", "FailedCheck", "RelationRule", "cross_form_check", "load_check_rules", "run_checks", ] Form = Literal["BALANCE", "INCOME"] RULES_PATH = Path(__file__).resolve().parents[2] / "rules" / "statement_checks.yaml" CROSS_FORM_RELATION = "1370 − 1370(начало года) = 2400" # диапазоны четырёхзначных кодов форм для генерации подстрок (§12 п. 2; в tabular-parser — # statement_headers.yaml) CODE_RANGES: dict[str, tuple[int, int]] = {"BALANCE": (1100, 1799), "INCOME": (2100, 2999)} _ZERO = Decimal("0") _HUNDRED = Decimal("100") _PARENT_LEN = 4 _GROUP_LEN = 3 _RELATION = re.compile(r"^\s*(?P\d{4})\s*=\s*(?P\d{4,5}(\s*[+\-]\s*\d{4,5})*)\s*$") _TERM = re.compile(r"([+\-])?\s*(\d{4,5})") @dataclass(frozen=True) class RelationRule: form: Form relation: str target: str terms: tuple[tuple[int, str], ...] strict: bool abs_terms: frozenset[str] = frozenset() @dataclass(frozen=True) class CheckRules: tolerance: Decimal rounding_tolerance_percent: Decimal subline_check: bool relations: tuple[RelationRule, ...] def relations_for(self, form: str) -> tuple[RelationRule, ...]: return tuple(rule for rule in self.relations if rule.form == form) @dataclass(frozen=True) class FailedCheck: column_index: int relation: str expected: Decimal actual: Decimal @dataclass(frozen=True) class ChecksReport: failed: tuple[FailedCheck, ...] soft_failed: tuple[FailedCheck, ...] def _parse_relation(form: Form, relation: str, strict: bool, abs_terms: Sequence[str]) -> RelationRule: found = _RELATION.match(relation) if found is None: raise ValueError(f"соотношение «{relation}»: ожидается «<код> = <код> [+|-] <код> …»") terms = tuple((-1 if sign == "-" else 1, code) for sign, code in _TERM.findall(found.group("rhs"))) unknown = set(abs_terms) - {code for _, code in terms} if unknown: raise ValueError(f"abs_terms {sorted(unknown)} нет в правой части «{relation}»") return RelationRule(form, relation, found.group("target"), terms, strict, frozenset(abs_terms)) @functools.lru_cache(maxsize=4) def load_check_rules(path: Path | None = None) -> CheckRules: """Словарь §7 из yaml; любая ошибка — `ValueError` с именем файла.""" source = path or RULES_PATH try: data = yaml.safe_load(source.read_text(encoding="utf-8")) relations = tuple( _parse_relation(item["form"], item["relation"], bool(item["strict"]), item.get("abs_terms", ())) for item in data["relations"] ) for rule in relations: if rule.form not in CODE_RANGES: raise ValueError(f"форма {rule.form!r}: ожидается BALANCE или INCOME") return CheckRules( tolerance=Decimal(str(data["tolerance"])), rounding_tolerance_percent=Decimal(str(data.get("rounding_tolerance_percent", 0))), subline_check=bool(data.get("subline_check", True)), relations=relations, ) except (OSError, KeyError, TypeError, ValueError, InvalidOperation, yaml.YAMLError) as exc: raise ValueError(f"словарь проверок {source.name}: {exc}") from exc def _rhs(amounts: Mapping[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: Mapping[str, Decimal], rule: RelationRule, tolerance: Decimal) -> tuple[bool, Decimal, Decimal]: 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(codes: Sequence[str], form: str) -> list[RelationRule]: """«1110 = 11101 + …» для четырёхзначных кодов формы с пятизначными подстроками. Порядок строк в БД не хранится (DBML порядка не знает): «по порядку» — это порядок кодов, в каком их отдаёт `statement_amounts` (M-9 2b: подстрока относится к ближайшей четырёхзначной строке выше с теми же тремя цифрами).""" low, high = CODE_RANGES[form] 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 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) rules: list[RelationRule] = [] for parent, kids in children.items(): if parent.isdigit() and low <= int(parent) <= high: relation = f"{parent} = {' + '.join(kids)}" rules.append( RelationRule( cast(Form, form), relation, parent, tuple((1, kid) for kid in kids), False, ) ) return rules def run_checks( form: str, lines: Sequence[tuple[str, Mapping[int, Decimal | None]]], column_indexes: Sequence[int], rules: CheckRules, ) -> ChecksReport: """Проверки §7 формы `form` по колонкам `column_indexes`; `lines` — `(код, {колонка: сумма})` в порядке кодов (порядок строк формы в БД не хранится; от него зависит только привязка пятизначных подстрок).""" explicit = list(rules.relations_for(form)) sublines = _subline_rules([code for code, _ in lines], form) if rules.subline_check else [] hard: list[FailedCheck] = [] soft: list[FailedCheck] = [] for column_index in column_indexes: # присутствие цели = наличие строки формы; пустая сумма (None) = 0, как `_amounts` в tabular-parser: оба # производителя отдают значение на каждую колонку каждой кодированной строки (пустая ячейка — amount=None) amounts = {code: (values.get(column_index) or _ZERO) for code, values in lines} for rule in [*explicit, *sublines]: if rule not in sublines and rule.target not in amounts: continue holds, expected, actual = _holds(amounts, rule, rules.tolerance) if holds: continue failed = FailedCheck(column_index, rule.relation, expected, actual) scale = max([abs(expected), *(abs(amounts.get(code, _ZERO)) for _, code in rule.terms)]) rounding = scale * rules.rounding_tolerance_percent / _HUNDRED if rule.strict and abs(expected - actual) > rounding: hard.append(failed) else: soft.append(failed) return ChecksReport(failed=tuple(hard), soft_failed=tuple(soft)) def cross_form_check( balance_1370: tuple[Decimal | None, Decimal | None], income_2400: Decimal | None, tolerance: Decimal, ) -> FailedCheck | None: """Межформенная §7 (мягкая, D-5): `1370(отчётная дата) − 1370(начало года) = 2400(период)`; неполные данные — None.""" current, year_start = balance_1370 if current is None or year_start is None or income_2400 is None: return None delta = current - year_start if abs(delta - income_2400) <= tolerance: return None return FailedCheck(-1, CROSS_FORM_RELATION, delta, income_2400)