diff --git a/arm_statements/structure.py b/arm_statements/structure.py new file mode 100644 index 0000000..c24ccb6 --- /dev/null +++ b/arm_statements/structure.py @@ -0,0 +1,177 @@ +"""Шаг 8 — подтверждение неуверенного кода структурой бланка и выбор редакции (спека «Шаг 8»).""" + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import date + +from arm_statements.checks import form_variant, subline_parents +from arm_statements.evaluation import EvalLine, is_check_line +from arm_statements.loader import Rules +from arm_statements.model import DocumentType, Edition, FormVariant + +FSBU4_FROM = date(2025, 1, 1) # эффективная дата после 2024-12-31 → fsbu4 + +_CODE_LEN = 4 +_SUBLINE_LEN = 5 +_CONFIDENT = 100 + + +def edition_by_date(effective: date | None) -> Edition | None: + if effective is None: + return None + return Edition.FSBU4 if effective >= FSBU4_FROM else Edition.RAS_66N + + +@dataclass(frozen=True, slots=True) +class StructureResult: + edition: Edition | None + unconfirmed: frozenset[str] # ключи строк с неуверенным кодом, не подтверждённым структурой + + +# Элемент последовательности: (код, уверенный). +_Item = tuple[str, bool] + + +def _optimum(items: Sequence[_Item], order: Sequence[str], scale: int, skip: int | None = None) -> int: + """Взвешенная LCS `items` × `order` по равенству кода; вес позиции `scale * уверенный + 1`. + + `skip` — индекс позиции `items`, которой запрещено входить. Сложность O(|items|·|order|).""" + if not items or not order: + return 0 + previous = [0] * (len(order) + 1) + for index, (code, confident) in enumerate(items): + weight = None if index == skip else scale * int(confident) + 1 + current = [0] * (len(order) + 1) + for j, order_code in enumerate(order, start=1): + best = max(previous[j], current[j - 1]) + if weight is not None and code == order_code: + best = max(best, previous[j - 1] + weight) + current[j] = best + previous = current + return previous[-1] + + +def _editions(rules: Rules) -> tuple[Edition, ...]: + return rules.code_orders.editions or tuple(Edition) + + +def _pick(optima: dict[Edition, int], effective: date | None) -> tuple[Edition | None, tuple[Edition, ...]]: + """Редакция с большим оптимумом; равенство — по дате. Второе значение — редакции, чьи оптимумы лучшие.""" + top = max(optima.values()) + tied = tuple(edition for edition, value in optima.items() if value == top) + if len(tied) == 1: + return tied[0], tied + by_date = edition_by_date(effective) + return (by_date if by_date in tied else None), tied + + +def _filtered(items: Sequence[_Item], order: Sequence[str]) -> tuple[list[_Item], list[int]]: + allowed = set(order) + kept = [(item, index) for index, item in enumerate(items) if item[0] in allowed] + return [item for item, _ in kept], [index for _, index in kept] + + +def _select( + document_type: DocumentType, + items: Sequence[_Item], + variant: FormVariant, + effective: date | None, + rules: Rules, +) -> tuple[Edition | None, tuple[Edition, ...], dict[Edition, int]]: + scale = len(items) + 1 + optima = { + edition: _optimum( + _filtered(items, order := rules.code_orders.sequence(edition, document_type, variant))[0], + order, + scale, + ) + for edition in _editions(rules) + } + edition, tied = _pick(optima, effective) + return edition, tied, optima + + +def choose_edition( + document_type: DocumentType, + codes: Sequence[tuple[str, bool]], + effective: date | None, + rules: Rules, + *, + variant: FormVariant | None = None, +) -> Edition | None: + """Шаги 2–3 и 6 без подтверждения кодов; `codes` — пары `(код, уверенный)` строк `ok` по порядку.""" + if document_type is DocumentType.DISCLOSURE: + return edition_by_date(effective) + if variant is None: + variant = form_variant(document_type, [code for code, _ in codes]) or FormVariant.FULL + items = [(code, confident) for code, confident in codes if len(code) == _CODE_LEN] + return _select(document_type, items, variant, effective, rules)[0] + + +def _is_uncertain(line: EvalLine) -> bool: + return (line.code_confidence is None or line.code_confidence < _CONFIDENT) and not line.code_fixed + + +def _confirm_sublines(checked: Sequence[EvalLine], codes: Sequence[str], confirmed_keys: set[str]) -> set[str]: + """Пятизначные неуверенные коды: подтверждены родителем, стоящим раньше (§11.9); возвращает неподтверждённые.""" + unconfirmed: set[str] = set() + parents = subline_parents(codes) + for position, line in enumerate(checked): + code = line.line_code or "" + if len(code) == _CODE_LEN or not _is_uncertain(line): + continue + parent = parents[position] + found = None + if len(code) == _SUBLINE_LEN and parent is not None: + found = next((e for e in reversed(checked[:position]) if e.line_code == parent), None) + if found is not None and (not _is_uncertain(found) or found.key in confirmed_keys): + confirmed_keys.add(line.key) + else: + unconfirmed.add(line.key) + return unconfirmed + + +def confirm_codes( + lines: Sequence[EvalLine], + document_type: DocumentType, + variant: FormVariant | None, + effective: date | None, + rules: Rules, +) -> StructureResult: + if document_type is DocumentType.DISCLOSURE: + return StructureResult(edition_by_date(effective), frozenset()) + checked = sorted( + (line for line in lines if is_check_line(line) and line.order is not None), + key=lambda line: line.order or 0, + ) + codes = [line.line_code or "" for line in checked] + if variant is None: + variant = form_variant(document_type, codes) or FormVariant.FULL + four = [line for line in checked if len(line.line_code or "") == _CODE_LEN] + items: list[_Item] = [(line.line_code or "", not _is_uncertain(line)) for line in four] + edition, tied, _ = _select(document_type, items, variant, effective, rules) + + # Четырёхзначные неуверенные коды: входят во все лучшие подпоследовательности выбранной редакции + # (при равенстве без даты — во все лучшие обеих редакций, P1-4). + unconfirmed: set[str] = set() + confirmed_keys: set[str] = set() + scale = len(items) + 1 + for index, line in enumerate(four): + if not _is_uncertain(line): + confirmed_keys.add(line.key) + continue + confirmed = True + for candidate in tied if edition is None else (edition,): + order = rules.code_orders.sequence(candidate, document_type, variant) + if items[index][0] not in order: + confirmed = False + break + kept, indices = _filtered(items, order) + total = _optimum(kept, order, scale) + if _optimum(kept, order, scale, skip=indices.index(index)) >= total: + confirmed = False + break + (confirmed_keys if confirmed else unconfirmed).add(line.key) + + unconfirmed |= _confirm_sublines(checked, codes, confirmed_keys) + return StructureResult(edition, frozenset(unconfirmed)) diff --git a/arm_statements/tests/test_structure.py b/arm_statements/tests/test_structure.py new file mode 100644 index 0000000..3fe0f26 --- /dev/null +++ b/arm_statements/tests/test_structure.py @@ -0,0 +1,106 @@ +from datetime import date + +from arm_statements.evaluation import EvalLine +from arm_statements.loader import Rules +from arm_statements.model import DocumentType, Edition, FormVariant, LineStatus +from arm_statements.structure import choose_edition, confirm_codes + +BAL = DocumentType.BALANCE +Q3_2025 = date(2025, 9, 30) +FULL_CONF = 100.0 + + +def seq(*items: tuple[str, float]) -> list[EvalLine]: + return [EvalLine(f"r{i}", i, LineStatus.OK, code, conf, False, ()) for i, (code, conf) in enumerate(items)] + + +def unconfirmed(lines: list[EvalLine], rules: Rules, effective: date | None = Q3_2025) -> set[str]: + return set(confirm_codes(lines, BAL, FormVariant.FULL, effective, rules).unconfirmed) + + +def test_uncertain_code_in_place(rules: Rules) -> None: + assert unconfirmed(seq(("1220", 100.0), ("1230", 50.0), ("1240", 100.0)), rules) == set() + + +def test_swapped_uncertain_neighbours(rules: Rules) -> None: + lines = seq(("1210", 100.0), ("1230", 50.0), ("1220", 50.0), ("1240", 100.0)) + assert unconfirmed(lines, rules) == {"r1", "r2"} + + +def test_conflicting_anchors(rules: Rules) -> None: + lines = seq(("1210", 100.0), ("1240", 100.0), ("1230", 100.0), ("1250", 50.0)) + assert unconfirmed(lines, rules) == set() + + +def test_legal_repeat_1220_fsbu4(rules: Rules) -> None: + lines = seq(("1215", 100.0), ("1220", 100.0), ("1220", 50.0), ("1230", 100.0)) + assert unconfirmed(lines, rules) == set() + + +def test_repeat_outside_orders(rules: Rules) -> None: + lines = seq(("1220", 100.0), ("1230", 100.0), ("1230", 50.0), ("1240", 100.0)) + assert unconfirmed(lines, rules) == {"r2"} + + +def test_66n_blank_in_2025(rules: Rules) -> None: + lines = seq(("1110", 100.0), ("1120", 60.0), ("1130", 100.0)) + result = confirm_codes(lines, BAL, FormVariant.FULL, Q3_2025, rules) + assert (result.edition, result.unconfirmed) == (Edition.RAS_66N, frozenset()) + + +def test_tie_resolved_by_date(rules: Rules) -> None: + lines = seq(("1210", 100.0), ("1230", 100.0)) + assert confirm_codes(lines, BAL, FormVariant.FULL, Q3_2025, rules).edition is Edition.FSBU4 + assert confirm_codes(lines, BAL, FormVariant.FULL, date(2024, 12, 31), rules).edition is Edition.RAS_66N + + +def test_choose_edition_public(rules: Rules) -> None: + blank_66n = [("1110", True), ("1120", False), ("1130", True)] + assert choose_edition(BAL, blank_66n, Q3_2025, rules) is Edition.RAS_66N + tie = [("1210", True), ("1230", True)] + assert choose_edition(BAL, tie, Q3_2025, rules) is Edition.FSBU4 + assert choose_edition(BAL, tie, date(2024, 12, 31), rules) is Edition.RAS_66N + assert choose_edition(BAL, tie, None, rules) is None + assert choose_edition(BAL, [], Q3_2025, rules) is Edition.FSBU4 + assert choose_edition(DocumentType.DISCLOSURE, [("5101", True)], date(2024, 12, 31), rules) is Edition.RAS_66N + + +def test_choose_edition_equals_confirm_codes(rules: Rules) -> None: + lines = seq(("1110", 100.0), ("1120", 60.0), ("1130", 100.0)) + pairs = [(ln.line_code or "", (ln.code_confidence or 0.0) >= FULL_CONF) for ln in lines] + expected = confirm_codes(lines, BAL, FormVariant.FULL, Q3_2025, rules).edition + assert choose_edition(BAL, pairs, Q3_2025, rules, variant=FormVariant.FULL) is expected + + +def test_confident_nonstandard_code_not_flagged(rules: Rules) -> None: + assert unconfirmed(seq(("1230", 100.0), ("1231", 100.0), ("11101", 100.0)), rules) == set() + + +def test_subline_follows_confirmed_parent(rules: Rules) -> None: + lines = seq(("1220", 100.0), ("1230", 50.0), ("12301", 60.0), ("1240", 100.0)) + assert unconfirmed(lines, rules) == set() + orphan = seq(("1220", 100.0), ("12301", 60.0), ("1240", 100.0)) + assert unconfirmed(orphan, rules) == {"r1"} + + +def test_tie_without_date_union_of_editions(rules: Rules) -> None: + lines = seq(("1210", 100.0), ("1230", 50.0)) + result = confirm_codes(lines, BAL, FormVariant.FULL, None, rules) + assert result.edition is None + assert result.unconfirmed == frozenset() + + +def test_disclosure_has_no_unconfirmed(rules: Rules) -> None: + lines = seq(("5101", 50.0)) + result = confirm_codes(lines, DocumentType.DISCLOSURE, None, Q3_2025, rules) + assert (result.edition, result.unconfirmed) == (Edition.FSBU4, frozenset()) + + +def test_uncertain_code_outside_order_unconfirmed(rules: Rules) -> None: + assert unconfirmed(seq(("1230", 100.0), ("9999", 50.0), ("1240", 100.0)), rules) == {"r1"} + + +def test_performance_realistic_size(rules: Rules) -> None: + order = rules.code_orders.sequence(Edition.FSBU4, BAL, FormVariant.FULL) + lines = seq(*[(code, 50.0) for code in order] * 1) + assert confirm_codes(lines, BAL, FormVariant.FULL, Q3_2025, rules).unconfirmed == frozenset()