"""Загрузчик словарей `rules/*.yaml` в типизированный `Rules` с проверкой при загрузке.""" import re from collections.abc import Mapping from dataclasses import dataclass from decimal import Decimal from enum import StrEnum from functools import cache from pathlib import Path from typing import Literal, TypeVar import yaml from arm_statements.issues import WarningCode from arm_statements.model import DocumentType, Edition, FormVariant, PeriodKind, Severity LABEL_KEYS: tuple[str, ...] = ( "organization", "inn", "okpo", "okopf", "okfs", "okopf_okfs", "okei", "units", "address", "date", ) _CHECKS = "statement_checks.yaml" _PERIODS = "statement_periods.yaml" _HEADERS = "statement_headers.yaml" _UNITS = "statement_units.yaml" _RANGES = "statement_code_ranges.yaml" _ORDERS = "statement_code_orders.yaml" _WARNINGS = "statement_warnings.yaml" _LOOKALIKES = "ocr_lookalikes.yaml" _RELATION = re.compile(r"^(\d{4}) = (\d{4})((?: [+-] \d{4})*)$") _RELATION_TERM = re.compile(r"([+-]) (\d{4})") _PERIOD_GROUPS = frozenset({"d", "m", "mw", "y", "q", "n"}) _ABSOLUTE_KINDS = frozenset({"range_end", "day_month_year", "quarter", "half", "months", "month_range", "year"}) _MONTH_MAX = 12 _PERCENT_MAX = 100 _E = TypeVar("_E", bound=StrEnum) class RulesError(ValueError): """Словарь не прошёл проверку (в сообщении — файл и ключ).""" class CodeSection(StrEnum): BALANCE = "BALANCE" INCOME = "INCOME" CAPITAL = "CAPITAL" CASH_FLOW = "CASH_FLOW" NOTES = "NOTES" TARGET_USE = "TARGET_USE" @dataclass(frozen=True, slots=True) class CodeRange: section: CodeSection low: int high: int @dataclass(frozen=True, slots=True) class CodeRanges: ranges: tuple[CodeRange, ...] min_share: Decimal min_codes: int min_codes_continuation: int @dataclass(frozen=True, slots=True) class RelationRule: form: DocumentType variant: FormVariant relation: str target: str terms: tuple[tuple[int, str], ...] strict: bool abs_terms: frozenset[str] @dataclass(frozen=True, slots=True) class CheckRules: tolerance: Decimal rounding_tolerance_percent: Decimal subline_check: bool relations: tuple[RelationRule, ...] def relations_for(self, form: DocumentType, variant: FormVariant) -> tuple[RelationRule, ...]: return tuple(r for r in self.relations if r.form == form and r.variant == variant) def abs_codes(self, form: DocumentType, variant: FormVariant) -> frozenset[str]: found: set[str] = set() for rule in self.relations_for(form, variant): found |= rule.abs_terms return frozenset(found) @dataclass(frozen=True, slots=True) class PeriodPattern: kind: str pattern: re.Pattern[str] @dataclass(frozen=True, slots=True) class PeriodRules: relative: tuple[tuple[PeriodKind, re.Pattern[str]], ...] absolute: tuple[PeriodPattern, ...] months: tuple[tuple[str, int], ...] min_year: int max_year: int max_text_length: int @dataclass(frozen=True, slots=True) class HeaderRules: form_titles: Mapping[DocumentType, tuple[re.Pattern[str], ...]] okud: Mapping[DocumentType, tuple[re.Pattern[str], ...]] okud_code: re.Pattern[str] continuation: tuple[re.Pattern[str], ...] code_caption: re.Pattern[str] name_caption: re.Pattern[str] explanations_caption: re.Pattern[str] column_exclusions: tuple[re.Pattern[str], ...] labels: Mapping[str, tuple[re.Pattern[str], ...]] auditor: tuple[re.Pattern[str], ...] legal_form: tuple[re.Pattern[str], ...] @dataclass(frozen=True, slots=True) class UnitRules: spellings: tuple[tuple[re.Pattern[str], str], ...] okei: Mapping[str, str] thousands: str fractional_share_percent: int @dataclass(frozen=True, slots=True) class CodeOrders: editions: tuple[Edition, ...] sequences: Mapping[tuple[Edition, DocumentType, FormVariant], tuple[str, ...]] def sequence(self, edition: Edition, form: DocumentType, variant: FormVariant) -> tuple[str, ...]: return self.sequences.get((edition, form, variant), ()) @dataclass(frozen=True, slots=True) class WarningRule: code: str severity: Severity scope: Literal["document", "file"] @dataclass(frozen=True, slots=True) class Rules: checks: CheckRules periods: PeriodRules headers: HeaderRules units: UnitRules code_ranges: CodeRanges code_orders: CodeOrders lookalikes: Mapping[str, str] warnings: Mapping[str, WarningRule] # --- проверяющие помощники: Any из YAML дальше них не уходит --- def _fail(file: str, key: str, message: str) -> RulesError: return RulesError(f"{file}: {key}: {message}") def _read(directory: Path, name: str) -> dict[str, object]: path = directory / name try: data = yaml.safe_load(path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError) as exc: raise _fail(name, "<файл>", f"не удалось прочитать: {exc}") from exc if not isinstance(data, dict): raise _fail(name, "<корень>", "ожидается словарь") return {str(k): v for k, v in data.items()} def _map(file: str, key: str, node: object) -> dict[str, object]: if not isinstance(node, dict): raise _fail(file, key, "ожидается словарь") return {str(k): v for k, v in node.items()} def _seq(file: str, key: str, node: object) -> list[object]: if not isinstance(node, list): raise _fail(file, key, "ожидается список") return list(node) def _get(file: str, data: Mapping[str, object], key: str) -> object: if key not in data: raise _fail(file, key, "ключ отсутствует") return data[key] def _str(file: str, key: str, node: object) -> str: if not isinstance(node, str): raise _fail(file, key, "ожидается строка") return node def _int(file: str, key: str, node: object) -> int: if isinstance(node, bool) or not isinstance(node, int): raise _fail(file, key, "ожидается целое число") return node def _bool(file: str, key: str, node: object) -> bool: if not isinstance(node, bool): raise _fail(file, key, "ожидается true/false") return node def _number(file: str, key: str, node: object) -> Decimal: if isinstance(node, bool) or not isinstance(node, int | float): raise _fail(file, key, "ожидается число") return Decimal(str(node)) def _rx(file: str, key: str, node: object) -> re.Pattern[str]: text = _str(file, key, node) try: return re.compile(text, re.IGNORECASE) except re.error as exc: raise _fail(file, key, f"регулярка не компилируется: {exc}") from exc def _rx_list(file: str, key: str, node: object) -> tuple[re.Pattern[str], ...]: return tuple(_rx(file, f"{key}[{i}]", item) for i, item in enumerate(_seq(file, key, node))) def _enum(file: str, key: str, enum_type: type[_E], node: object) -> _E: try: return enum_type(_str(file, key, node)) except ValueError as exc: raise _fail(file, key, f"недопустимое значение {node!r}") from exc def _doc_type_patterns(file: str, key: str, node: object) -> dict[DocumentType, tuple[re.Pattern[str], ...]]: return { _enum(file, f"{key}.{name}", DocumentType, name): _rx_list(file, f"{key}.{name}", value) for name, value in _map(file, key, node).items() } # --- разделы --- def _parse_relation(file: str, key: str, text: str) -> tuple[str, tuple[tuple[int, str], ...]]: match = _RELATION.match(text) if match is None: raise _fail(file, key, f"соотношение {text!r} не вида «<код> = <код> [+|-] <код> …»") terms = [(1, match.group(2))] terms += [(1 if sign == "+" else -1, code) for sign, code in _RELATION_TERM.findall(match.group(3))] return match.group(1), tuple(terms) def _load_checks(directory: Path) -> CheckRules: file = _CHECKS data = _read(directory, file) relations: list[RelationRule] = [] for i, item in enumerate(_seq(file, "relations", _get(file, data, "relations"))): key = f"relations[{i}]" entry = _map(file, key, item) form = _enum(file, f"{key}.form", DocumentType, _get(file, entry, "form")) variant = _enum(file, f"{key}.variant", FormVariant, _get(file, entry, "variant")) text = _str(file, f"{key}.relation", _get(file, entry, "relation")) target, terms = _parse_relation(file, f"{key}.relation", text) raw_abs = _seq(file, f"{key}.abs_terms", entry.get("abs_terms", [])) abs_terms = frozenset(_str(file, f"{key}.abs_terms", code) for code in raw_abs) if not abs_terms <= {code for _, code in terms}: raise _fail(file, f"{key}.abs_terms", "коды должны быть членами правой части соотношения") relations.append( RelationRule( form=form, variant=variant, relation=text, target=target, terms=terms, strict=_bool(file, f"{key}.strict", _get(file, entry, "strict")), abs_terms=abs_terms, ) ) return CheckRules( tolerance=_number(file, "tolerance", _get(file, data, "tolerance")), rounding_tolerance_percent=_number( file, "rounding_tolerance_percent", _get(file, data, "rounding_tolerance_percent") ), subline_check=_bool(file, "subline_check", _get(file, data, "subline_check")), relations=tuple(relations), ) def _load_periods(directory: Path) -> PeriodRules: file = _PERIODS data = _read(directory, file) relative: list[tuple[PeriodKind, re.Pattern[str]]] = [] for i, item in enumerate(_seq(file, "relative", _get(file, data, "relative"))): key = f"relative[{i}]" entry = _map(file, key, item) kind = _enum(file, f"{key}.kind", PeriodKind, _get(file, entry, "kind")) relative.append((kind, _rx(file, f"{key}.pattern", _get(file, entry, "pattern")))) absolute: list[PeriodPattern] = [] for i, item in enumerate(_seq(file, "absolute", _get(file, data, "absolute"))): key = f"absolute[{i}]" entry = _map(file, key, item) abs_kind = _str(file, f"{key}.kind", _get(file, entry, "kind")) if abs_kind not in _ABSOLUTE_KINDS: raise _fail(file, f"{key}.kind", f"недопустимый вид {abs_kind!r}") pattern = _rx(file, f"{key}.pattern", _get(file, entry, "pattern")) if not set(pattern.groupindex) <= _PERIOD_GROUPS: raise _fail(file, f"{key}.pattern", f"допустимые имена групп: {sorted(_PERIOD_GROUPS)}") absolute.append(PeriodPattern(abs_kind, pattern)) months: list[tuple[str, int]] = [] for stem, number in _map(file, "months", _get(file, data, "months")).items(): month = _int(file, f"months.{stem}", number) if not 1 <= month <= _MONTH_MAX: raise _fail(file, f"months.{stem}", "номер месяца вне 1..12") months.append((stem, month)) years = _map(file, "years", _get(file, data, "years")) min_year = _int(file, "years.min", _get(file, years, "min")) max_year = _int(file, "years.max", _get(file, years, "max")) if min_year > max_year: raise _fail(file, "years", "min больше max") return PeriodRules( relative=tuple(relative), absolute=tuple(absolute), months=tuple(months), min_year=min_year, max_year=max_year, max_text_length=_int(file, "max_text_length", _get(file, data, "max_text_length")), ) def _load_headers(directory: Path) -> HeaderRules: file = _HEADERS data = _read(directory, file) captions = _map(file, "captions", _get(file, data, "captions")) labels_raw = _map(file, "labels", _get(file, data, "labels")) if tuple(labels_raw) != LABEL_KEYS: raise _fail(file, "labels", f"ключи должны быть ровно {LABEL_KEYS}, получены {tuple(labels_raw)}") return HeaderRules( form_titles=_doc_type_patterns(file, "form_titles", _get(file, data, "form_titles")), okud=_doc_type_patterns(file, "okud", _get(file, data, "okud")), okud_code=_rx(file, "okud_code", _get(file, data, "okud_code")), continuation=_rx_list(file, "continuation", _get(file, data, "continuation")), code_caption=_rx(file, "captions.code", _get(file, captions, "code")), name_caption=_rx(file, "captions.name", _get(file, captions, "name")), explanations_caption=_rx(file, "captions.explanations", _get(file, captions, "explanations")), column_exclusions=_rx_list(file, "captions.exclusions", _get(file, captions, "exclusions")), labels={name: _rx_list(file, f"labels.{name}", value) for name, value in labels_raw.items()}, auditor=_rx_list(file, "auditor", _get(file, data, "auditor")), legal_form=_rx_list(file, "legal_form", _get(file, data, "legal_form")), ) def _load_units(directory: Path) -> UnitRules: file = _UNITS data = _read(directory, file) spellings: list[tuple[re.Pattern[str], str]] = [] canonical_all: set[str] = set() for i, item in enumerate(_seq(file, "units", _get(file, data, "units"))): key = f"units[{i}]" entry = _map(file, key, item) canonical = _str(file, f"{key}.canonical", _get(file, entry, "canonical")) canonical_all.add(canonical) spellings.extend((rx, canonical) for rx in _rx_list(file, f"{key}.patterns", _get(file, entry, "patterns"))) okei = { code: _str(file, f"okei.{code}", value) for code, value in _map(file, "okei", _get(file, data, "okei")).items() } for code, canonical in okei.items(): if canonical not in canonical_all: raise _fail(file, f"okei.{code}", f"единица {canonical!r} не среди canonical") thousands = _str(file, "thousands", _get(file, data, "thousands")) if thousands not in canonical_all: raise _fail(file, "thousands", f"единица {thousands!r} не среди canonical") percent = _int(file, "fractional_share_percent", _get(file, data, "fractional_share_percent")) if not 1 <= percent <= _PERCENT_MAX: raise _fail(file, "fractional_share_percent", "ожидается 1..100") return UnitRules(tuple(spellings), okei, thousands, percent) def _load_ranges(directory: Path) -> CodeRanges: file = _RANGES data = _read(directory, file) ranges: list[CodeRange] = [] for name, bounds in _map(file, "sections", _get(file, data, "sections")).items(): section = _enum(file, f"sections.{name}", CodeSection, name) pair = _seq(file, f"sections.{name}", bounds) if len(pair) != len(("low", "high")): raise _fail(file, f"sections.{name}", "ожидается [low, high]") low, high = (_int(file, f"sections.{name}", v) for v in pair) if low > high: raise _fail(file, f"sections.{name}", "low больше high") ranges.append(CodeRange(section, low, high)) min_codes = _int(file, "min_codes", _get(file, data, "min_codes")) continuation = _int(file, "min_codes_continuation", _get(file, data, "min_codes_continuation")) if not 1 <= continuation <= min_codes: raise _fail(file, "min_codes_continuation", "должно быть 1 <= min_codes_continuation <= min_codes") return CodeRanges( ranges=tuple(ranges), min_share=_number(file, "min_share", _get(file, data, "min_share")), min_codes=min_codes, min_codes_continuation=continuation, ) def _load_orders(directory: Path) -> CodeOrders: file = _ORDERS data = _read(directory, file) if _get(file, data, "version") != 1: raise _fail(file, "version", "поддерживается только версия 1") repeats = _map(file, "repeats", _get(file, data, "repeats")) editions_raw = _map(file, "editions", _get(file, data, "editions")) for ed_name, forms_node in repeats.items(): if ed_name not in editions_raw: raise _fail(file, f"repeats.{ed_name}", "редакции нет в editions") known_forms = _map(file, f"editions.{ed_name}", editions_raw[ed_name]) for form_name in _map(file, f"repeats.{ed_name}", forms_node): if form_name not in known_forms: raise _fail(file, f"repeats.{ed_name}.{form_name}", "формы нет в editions этой редакции") editions: list[Edition] = [] sequences: dict[tuple[Edition, DocumentType, FormVariant], tuple[str, ...]] = {} for ed_name, forms_node in editions_raw.items(): edition = _enum(file, f"editions.{ed_name}", Edition, ed_name) editions.append(edition) forms = _map(file, f"editions.{ed_name}", forms_node) edition_repeats = _map(file, f"repeats.{ed_name}", repeats.get(ed_name, {})) for form_name, variants_node in forms.items(): form = _enum(file, f"editions.{ed_name}.{form_name}", DocumentType, form_name) repeat_codes = [ _str(file, f"repeats.{ed_name}.{form_name}", c) for c in _seq(file, f"repeats.{ed_name}.{form_name}", edition_repeats.get(form_name, [])) ] variants = _map(file, f"editions.{ed_name}.{form_name}", variants_node) for var_name, codes_node in variants.items(): key = f"editions.{ed_name}.{form_name}.{var_name}" variant = _enum(file, key, FormVariant, var_name) codes = tuple(_str(file, key, c) for c in _seq(file, key, codes_node)) for code in codes: if not re.fullmatch(r"\d{4}", code): raise _fail(file, key, f"код {code!r} не из четырёх цифр") _check_repeat_counts(file, key, variant, codes, repeat_codes) sequences[(edition, form, variant)] = codes return CodeOrders(tuple(editions), sequences) def _check_repeat_counts(file: str, key: str, variant: FormVariant, codes: tuple[str, ...], repeats: list[str]) -> None: expected = {code: 1 + repeats.count(code) for code in set(codes) | set(repeats)} for code, limit in sorted(expected.items()): actual = codes.count(code) if variant is FormVariant.FULL and actual != limit: raise _fail(file, key, f"код {code} встречается {actual} раз, ожидается {limit} (1 + repeats)") if variant is FormVariant.SIMPLIFIED and actual > limit: raise _fail(file, key, f"код {code} встречается {actual} раз, допустимо не более {limit}") def _load_warnings(directory: Path) -> dict[str, WarningRule]: file = _WARNINGS data = _read(directory, file) warnings: dict[str, WarningRule] = {} for code, node in _map(file, "warnings", _get(file, data, "warnings")).items(): entry = _map(file, f"warnings.{code}", node) severity = _enum(file, f"warnings.{code}.severity", Severity, _get(file, entry, "severity")) scope = _str(file, f"warnings.{code}.scope", _get(file, entry, "scope")) if scope == "document": warnings[code] = WarningRule(code, severity, "document") elif scope == "file": warnings[code] = WarningRule(code, severity, "file") else: raise _fail(file, f"warnings.{code}.scope", f"недопустимая область {scope!r}") if set(warnings) != {c.value for c in WarningCode}: diff = sorted(set(warnings) ^ {c.value for c in WarningCode}) raise _fail(file, "warnings", f"коды словаря и WarningCode расходятся: {diff}") return warnings def _load_lookalikes(directory: Path) -> dict[str, str]: file = _LOOKALIKES data = _read(directory, file) return { char: _str(file, f"lookalikes.{char}", value) for char, value in _map(file, "lookalikes", _get(file, data, "lookalikes")).items() } def load_rules(directory: Path | None = None) -> Rules: """Прочитать и проверить словари каталога (по умолчанию — `rules/` пакета).""" base = directory if directory is not None else Path(__file__).parent / "rules" return Rules( checks=_load_checks(base), periods=_load_periods(base), headers=_load_headers(base), units=_load_units(base), code_ranges=_load_ranges(base), code_orders=_load_orders(base), lookalikes=_load_lookalikes(base), warnings=_load_warnings(base), ) @cache def default_rules() -> Rules: return load_rules()