Andrey Epifancev 56c0f01d86 chore(arm_statements): копия пакета arm-statements@6bce80b (#44)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 19:38:33 +04:00

517 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Загрузчик словарей `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()