diff --git a/arm_statements/checks.py b/arm_statements/checks.py new file mode 100644 index 0000000..16a3cf2 --- /dev/null +++ b/arm_statements/checks.py @@ -0,0 +1,184 @@ +"""Шаг 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)) diff --git a/arm_statements/evaluation.py b/arm_statements/evaluation.py new file mode 100644 index 0000000..c75e4f7 --- /dev/null +++ b/arm_statements/evaluation.py @@ -0,0 +1,36 @@ +"""Общий вид строк шагов 6–8 у сервиса и ядра.""" + +from dataclasses import dataclass +from decimal import Decimal + +from arm_statements.model import LineStatus + + +@dataclass(frozen=True, slots=True) +class EvalValue: + column: str # ключ колонки: str(column_index) у сервиса, ключ хоста у ядра + amount: Decimal | None + text: str | None + confidence: float | None + is_recognized: bool + fixed: bool = False # правлено или подтверждено аналитиком + + +@dataclass(frozen=True, slots=True) +class EvalLine: + key: str # ref у сервиса, ключ хоста у ядра + order: int | None # row_index; у ядра None — строка правки + line_status: LineStatus + line_code: str | None + code_confidence: float | None + code_fixed: bool + values: tuple[EvalValue, ...] + + +def is_check_line(line: EvalLine) -> bool: + """Строка участвует в §7: `ok` с кодом; строка правки (`order is None`) — только с суммой (P1-8).""" + if line.line_status is not LineStatus.OK or line.line_code is None: + return False + if line.order is None: + return any(value.amount is not None for value in line.values) + return True diff --git a/arm_statements/tests/test_checks.py b/arm_statements/tests/test_checks.py new file mode 100644 index 0000000..d7ab03c --- /dev/null +++ b/arm_statements/tests/test_checks.py @@ -0,0 +1,107 @@ +from decimal import Decimal as D + +from arm_statements.checks import checks_from, evaluate_relations, form_variant, subline_parents +from arm_statements.evaluation import EvalLine, EvalValue, is_check_line +from arm_statements.loader import Rules +from arm_statements.model import DocumentType, FormVariant, LineStatus, StatementChecks + +BAL, INC = DocumentType.BALANCE, DocumentType.INCOME + + +def line(code: str | None, *amounts: str | None, status: LineStatus = LineStatus.OK, order: int = 0) -> EvalLine: + values = tuple(EvalValue(str(i), None if a is None else D(a), a, 100.0, True) for i, a in enumerate(amounts)) + return EvalLine(f"k{order}{code}", order, status, code, 100.0, False, values) + + +def failed(lines: list[EvalLine], doc: DocumentType, variant: FormVariant | None, rules: Rules) -> StatementChecks: + outcomes = evaluate_relations(lines, ["0"], doc, variant, rules) + return checks_from(outcomes, {"0": 0}) + + +def test_form_variant() -> None: + assert form_variant(BAL, ["1150", "1210", "1600", "1310", "1700"]) is FormVariant.SIMPLIFIED + assert form_variant(BAL, ["1100", "1600"]) is FormVariant.FULL + assert form_variant(INC, ["2110", "2400"]) is FormVariant.SIMPLIFIED + assert form_variant(INC, ["2110", "2100", "2400"]) is FormVariant.FULL + assert form_variant(DocumentType.DISCLOSURE, ["5101"]) is None + + +def test_empty_target_is_violation(rules: Rules) -> None: + lines = [ + line("1110", "30"), + line("1100", "30"), + line("1210", "70"), + line("1200", "70"), + line(None, "100", status=LineStatus.DETAIL), + ] + checks = failed(lines, BAL, FormVariant.FULL, rules) + assert [f.relation for f in checks.relations_failed] == ["1600 = 1100 + 1200"] + assert (checks.relations_failed[0].expected, checks.relations_failed[0].actual) == (None, D("100")) + + +def test_target_without_members_is_violation(rules: Rules) -> None: + lines = [line("1110", "30"), line("1100", "30"), line("1210", "70"), line("1200", "70"), line("1600", "100")] + checks = failed(lines, BAL, FormVariant.FULL, rules) + assert [f.relation for f in checks.relations_failed] == ["1600 = 1700"] + assert checks.relations_failed[0].codes == ("1600", "1700") + assert (checks.relations_failed[0].expected, checks.relations_failed[0].actual) == (D("100"), None) + + +def test_zero_target_with_empty_members_is_violation(rules: Rules) -> None: + lines = [line("1600", "0"), line("1110", "0"), line("1100", "0")] + hard = failed(lines, BAL, FormVariant.FULL, rules).relations_failed + assert [(f.relation, f.expected, f.actual) for f in hard] == [("1600 = 1700", D("0"), None)] + + +def test_duplicate_code_is_summed(rules: Rules) -> None: + lines = [line("2300", "849"), line("2410", "100"), line("2410", "649"), line("2400", "100")] + outcomes = evaluate_relations(lines, ["0"], INC, FormVariant.FULL, rules) + target = next(o for o in outcomes if o.rule.relation == "2400 = 2300 - 2410 + 2420") + assert target.within_tolerance + + +def test_simplified_income_soft(rules: Rules) -> None: + lines = [line("2110", "1000"), line("2120", "-600"), line("2410", "50"), line("2400", "500")] + checks = failed(lines, INC, FormVariant.SIMPLIFIED, rules) + assert checks.relations_failed == () + assert [f.relation for f in checks.soft_relations_failed] == ["2400 = 2110 - 2120 - 2330 + 2340 - 2350 - 2410"] + + +def test_rounding_goes_soft(rules: Rules) -> None: + lines = [ + line("1600", "17000000009.02"), + line("1700", "17000000000"), + line("1100", "17000000009.02"), + line("1300", "17000000000"), + ] + checks = failed(lines, BAL, FormVariant.FULL, rules) + assert "1600 = 1700" in [f.relation for f in checks.soft_relations_failed] + assert "1600 = 1700" not in [f.relation for f in checks.relations_failed] + + +def test_abs_terms_hold_with_minus(rules: Rules) -> None: + lines = [line("2110", "100"), line("2120", "-60"), line("2100", "40")] + hard = failed(lines, INC, FormVariant.FULL, rules).relations_failed + assert "2100 = 2110 - 2120" not in [f.relation for f in hard] + + +def test_sublines(rules: Rules) -> None: + assert subline_parents(["1230", "12301", "12302", "1240", "11101"]) == [None, "1230", "1230", None, "1110"] + # расхождение 2 (>1): в плане было 10 против 4 + 5 = 9, это в пределах допуска ±1 — спека «Шаг 7» (T13.1) + lines = [line("1230", "10"), line("12301", "4"), line("12302", "4")] + soft = failed(lines, BAL, FormVariant.FULL, rules).soft_relations_failed + assert [f.relation for f in soft] == ["1230 = 12301 + 12302"] + within = [line("1230", "10"), line("12301", "4"), line("12302", "5")] + assert failed(within, BAL, FormVariant.FULL, rules).soft_relations_failed == () + empty = [line("1230", "10"), line("12301", None), line("12302", None)] + assert failed(empty, BAL, FormVariant.FULL, rules).soft_relations_failed == () + + +def test_is_check_line() -> None: + assert is_check_line(line("1600", "1")) + assert not is_check_line(line("1600", "1", status=LineStatus.DETAIL)) + assert not is_check_line(line(None, "1")) + correction = EvalLine("c", None, LineStatus.OK, "1600", 100.0, False, line("1600", None).values) + assert not is_check_line(correction) + with_amount = EvalLine("c", None, LineStatus.OK, "1600", 100.0, False, line("1600", "5").values) + assert is_check_line(with_amount)