"""Словари `rules/*.yaml` и загрузчик (design D-7): синонимы заголовков, группировок, периодов и типов данных. Словари — единственный носитель синонимов: код парсера только вызывает `Pattern.search` и не знает ни одного слова из ведомости. Новый синоним — строка в YAML. Правила нормализации (одни и те же для проверяемого текста и для самих паттернов в YAML): текст приводится `normalize_text` (нижний регистр, «ё» -> «е», дефисы/тире/неразрывные пробелы/переносы -> пробел, пробелы схлопнуты), а паттерн компилируется как есть с `re.IGNORECASE`. Поэтому паттерн, содержащий «ё», дефис или неразрывный пробел, никогда не сработал бы; загрузчик отвергает такие паттерны с указанием файла и ключа. """ import bisect import calendar import datetime import re from collections.abc import Iterable from dataclasses import dataclass from decimal import Decimal from pathlib import Path from typing import Any, Literal, TypeVar, get_args import yaml from pydantic import BaseModel, ConfigDict, ValidationError, model_validator from pydantic_core import core_schema from src.contract.common import DataType __all__ = [ "DEFAULT_RULES_DIR", "DataTypeRules", "MonthRange", "OsvCheckRules", "OsvGroupRules", "OsvHeaderRules", "Pattern", "PeriodRule", "PeriodRules", "RelationRule", "Rules", "StatementCheckRules", "StatementHeaderRules", "StatementPeriodRules", "first_match", "load_rules", "normalize_text", "source_offset", ] DEFAULT_RULES_DIR = Path(__file__).resolve().parents[2] / "rules" HEADERS_FILE = "osv_headers.yaml" GROUPS_FILE = "osv_groups.yaml" PERIODS_FILE = "periods.yaml" DATA_TYPES_FILE = "data_types.yaml" CHECKS_FILE = "osv_checks.yaml" STATEMENT_HEADERS_FILE = "statement_headers.yaml" STATEMENT_PERIODS_FILE = "statement_periods.yaml" STATEMENT_CHECKS_FILE = "statement_checks.yaml" GroupType = Literal["ACCOUNT", "DEPARTMENT", "COUNTERPARTY", "CONTRACT", "OTHER"] SumGroup = Literal["opening", "turnover", "closing"] HeaderMarker = Literal["account", "period", "data_type", "currency", "org_params", "subkonto"] RowDataType = Literal["CURRENCY", "QUANTITY"] Quarter = Literal["Q1", "Q2", "Q3", "Q4"] FormType = Literal["BALANCE", "INCOME", "DISCLOSURE"] StatementLabel = Literal[ "organization", "inn", "okpo", "okopf", "okfs", "okopf_okfs", "okei", "units", "address", "date", ] _ISO_CODE_LENGTH = 3 # Код счёта в словаре проверок: цифры, при необходимости с субсчетами через точку _ACCOUNT_CODE = re.compile(r"\d+(\.\d+)*") # Разделители, сводимые к одному пробелу: любые пробельные (включая NBSP и переносы строк), дефис-минус, дефисы и тире # U+2010–U+2015, знак минус U+2212 _SEPARATORS = re.compile(r"[\s\-\u2010-\u2015\u2212]+") # Невидимые символы переноса: мягкий перенос, пробел нулевой ширины, BOM — удаляются, слово не разрывается _INVISIBLE = re.compile("[\u00ad\u200b\ufeff]") # Что нормализация убирает из текста: таких символов в паттерне быть не может _UNMATCHABLE_IN_PATTERN = re.compile(r"[ёЁ\s\-\u2010-\u2015\u2212\u00ad\u200b\ufeff]") # Регулярные выражения для парсинга соотношений форм _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})") Value = TypeVar("Value") def normalize_text(s: str) -> str: """Нижний регистр, «ё» -> «е», пробельные символы и дефисы/тире -> один пробел, невидимые переносы удалены.""" lowered = _INVISIBLE.sub("", s).lower().replace("ё", "е") return _SEPARATORS.sub(" ", lowered).strip() def source_offset(text: str, normalized_offset: int) -> int: """Смещение в исходном тексте, с которого начинается хвост нормализованного текста с позиции `normalized_offset`. Позиции `Match` относятся к `normalize_text(text)` (нижний регистр, схлопнутые пробелы), а в результат нужен исходный текст («3 квартал 2024 г.», не «3 квартал 2024 г.» в нижнем регистре). Нормализация префикса — префикс нормализации целого, поэтому длина нормализованного префикса монотонна по длине исходного. """ # Пробелы сразу после совпадения (`\s+` в метке) нормализация префикса отбрасывает — не считаем их в позиции target = len(normalize_text(text)[:normalized_offset].rstrip()) return bisect.bisect_left(range(len(text) + 1), target, key=lambda size: len(normalize_text(text[:size]))) class Pattern: """Регулярное выражение словаря: `search` применяется к `normalize_text(text)`, регистр не учитывается.""" __slots__ = ("_regex", "source") def __init__(self, source: str, regex: re.Pattern[str]) -> None: self.source = source self._regex = regex @classmethod def compile(cls, source: str) -> "Pattern": """Компиляция записи словаря. `ValueError` — пробел/«ё»/дефис в самом паттерне или неверная регулярка.""" if _UNMATCHABLE_IN_PATTERN.search(source.replace(" ", "")): raise ValueError( f"паттерн {source!r} не нормализован: «ё», дефисы, тире и переносы строк в паттерне не допускаются " "(проверяемый текст нормализуется: «е» вместо «ё», пробел вместо дефиса); пишите `\\s`, `\\w`" ) try: regex = re.compile(source, re.IGNORECASE) except re.error as exc: raise ValueError(f"паттерн {source!r}: некорректное регулярное выражение ({exc})") from exc return cls(source, regex) def search(self, text: str) -> re.Match[str] | None: """Первое совпадение в нормализованном тексте; позиции `Match` относятся к нормализованной строке.""" return self._regex.search(normalize_text(text)) def search_normalized(self, normalized: str) -> re.Match[str] | None: """Как `search`, но текст уже приведён `normalize_text`: ячейка нормализуется один раз на все паттерны.""" return self._regex.search(normalized) @property def group_names(self) -> frozenset[str]: """Имена именованных групп паттерна: словарь требует их у записей месяцев и диапазона.""" return frozenset(self._regex.groupindex) def __repr__(self) -> str: return f"Pattern({self.source!r})" @classmethod def __get_pydantic_core_schema__(cls, source_type: Any, handler: Any) -> core_schema.CoreSchema: return core_schema.no_info_after_validator_function( cls.compile, core_schema.str_schema(), serialization=core_schema.plain_serializer_function_ser_schema(lambda pattern: pattern.source), ) def first_match(entries: Iterable[tuple[Pattern, Value]], text: str) -> Value | None: """Значение первой записи словаря, паттерн которой совпал с текстом (порядок записей = приоритет).""" for pattern, value in entries: if pattern.search(text): return value return None class _Model(BaseModel): # extra=forbid: опечатка в ключе словаря — ошибка загрузки, а не молча игнорируемый синоним model_config = ConfigDict(frozen=True, extra="forbid") class OsvHeaderRules(_Model): sheet_markers: list[Pattern] groups: dict[SumGroup, list[Pattern]] debit: list[Pattern] credit: list[Pattern] service_columns: dict[str, list[Pattern]] total_row: list[Pattern] total_row_exact: list[Pattern] total_row_secondary: list[Pattern] column_captions: list[Pattern] counterparty_inn_suffix: list[Pattern] inn_only_row: list[Pattern] inn_kpp_cell: list[Pattern] header_markers: dict[HeaderMarker, list[Pattern]] @model_validator(mode="after") def _check_required_keys(self) -> "OsvHeaderRules": for field_name, expected in ( ("groups", set(get_args(SumGroup))), ("header_markers", set(get_args(HeaderMarker))), ): missing = expected - set(getattr(self, field_name)) if missing: raise ValueError(f"{field_name}: нет обязательных ключей {sorted(missing)}") if "inn" not in self.service_columns: raise ValueError("service_columns: нет обязательного ключа 'inn'") for field_name, groups in ( ("counterparty_inn_suffix", {"tail", "inn"}), ("inn_only_row", {"inn"}), ("inn_kpp_cell", {"inn"}), ): for pattern in getattr(self, field_name): missing_groups = groups - pattern.group_names if missing_groups: raise ValueError(f"{field_name}: в паттерне {pattern.source!r} нет групп {sorted(missing_groups)}") return self class OsvGroupRules(_Model): mapping: list[tuple[Pattern, GroupType]] default: GroupType = "OTHER" def classify(self, text: str) -> GroupType: """Тип группировки по названию из шапки; не совпавшее ни с одним паттерном — `default`.""" found = first_match(self.mapping, text) return self.default if found is None else found class OsvCheckRules(_Model): accounts_without_analytic_balance: list[str] @model_validator(mode="after") def _check_codes(self) -> "OsvCheckRules": bad = [code for code in self.accounts_without_analytic_balance if not _ACCOUNT_CODE.fullmatch(code)] if bad: raise ValueError(f"accounts_without_analytic_balance: ожидаются коды счетов вида «91» или «76.09», {bad}") return self def has_no_analytic_balance(self, account_code: str | None) -> bool: """Код счёта из файла («91», «91.01») относится к счёту из списка: совпадает с ним либо его субсчёт.""" if not account_code: return False code = account_code.strip() return any(code == root or code.startswith(f"{root}.") for root in self.accounts_without_analytic_balance) class PeriodRule(_Model): pattern: Pattern quarter: Quarter is_cumulative: bool _MONTHS_IN_YEAR = 12 _RANGE_GROUPS = frozenset({"m1", "y1", "m2", "y2"}) _MONTH_GROUPS = frozenset(f"m{number:02d}" for number in range(1, _MONTHS_IN_YEAR + 1)) @dataclass(frozen=True) class MonthRange: """Диапазон месяцев периода: «Октябрь 2024 г. – Сентябрь 2025 г.» -> (10, 2024, 9, 2025).""" start_month: int start_year: int end_month: int end_year: int @property def months(self) -> int: """Число месяцев диапазона включительно: октябрь 2024 – сентябрь 2025 = 12.""" return (self.end_year - self.start_year) * _MONTHS_IN_YEAR + self.end_month - self.start_month + 1 class PeriodRules(_Model): rules: list[PeriodRule] year: Pattern end: Pattern # Названия месяцев: одна запись с именованными группами m01..m12; диапазон месяцев с годами — группы m1, y1, m2, y2 months: Pattern range: Pattern standalone: list[Pattern] # Текст, в котором один месяц не задаёт границы периода: дата на число, квартал, полугодие, «N месяцев» not_single_month: Pattern @model_validator(mode="after") def _check_month_patterns(self) -> "PeriodRules": if self.months.group_names != _MONTH_GROUPS: raise ValueError("months: ожидаются именованные группы m01…m12 (по одной на месяц)") missing = _RANGE_GROUPS - self.range.group_names if missing: raise ValueError(f"range: нет обязательных именованных групп {sorted(missing)}") return self def month_number(self, text: str) -> int | None: """Номер месяца по названию в любом падеже («октября» -> 10); не месяц — `None`.""" found = self.months.search(text) return None if found is None or found.lastgroup is None else int(found.lastgroup.removeprefix("m")) def month_range(self, text: str) -> MonthRange | None: """Диапазон месяцев с годом конца: начало без года — в том же году, а если месяц начала позже месяца конца (октябрь – сентябрь) — в предыдущем. Диапазон, который начинается с января и не называет год начала («январь-март 2025 г.»), словарь периодов разбирает своими правилами (`match`), поэтому здесь это `None`. """ found = self.range.search(text) if found is None: return None start, end = self.month_number(found.group("m1")), self.month_number(found.group("m2")) if start is None or end is None: return None end_year = int(found.group("y2")) year_text = found.group("y1") if year_text is None and start == 1: return None start_year = int(year_text) if year_text is not None else end_year - (1 if start > end else 0) return MonthRange(start, start_year, end, end_year) def bounds_span(self, text: str) -> MonthRange | None: """Месяцы, которые период называет прямо: диапазон (в т. ч. от января без года начала — «Январь - Май 2025 г.») или один месяц («Февраль 2024 г.»); иначе `None`. Нужен только для границ периода: квартал и признак нарастающего итога по-прежнему берутся из `match`. """ span = self.month_range(text) if span is not None: return span found = self.range.search(text) if found is not None: start, end = self.month_number(found.group("m1")), self.month_number(found.group("m2")) if start == 1 and end is not None and found.group("y1") is None: end_year = int(found.group("y2")) return MonthRange(1, end_year, end, end_year) return None if self.not_single_month.search(text): return None month, year = self.month_number(text), self.year.search(text) if month is None or year is None: return None return MonthRange(month, int(year.group(0)), month, int(year.group(0))) def match(self, text: str) -> PeriodRule | None: """Первое правило, паттерн которого совпал с текстом периода; порядок в YAML — от специфичных к общим.""" for rule in self.rules: if rule.pattern.search(text): return rule return None class DataTypeRules(_Model): header: list[tuple[Pattern, DataType]] standalone_header: list[tuple[Pattern, DataType]] row: list[tuple[Pattern, RowDataType]] currency_codes: list[str] currency_aliases: dict[str, list[Pattern]] @model_validator(mode="after") def _check_currencies(self) -> "DataTypeRules": bad = [code for code in self.currency_codes if len(code) != _ISO_CODE_LENGTH or not code.isupper()] if bad: raise ValueError(f"currency_codes: ожидаются буквенные коды ISO 4217 в верхнем регистре, получено {bad}") unknown = sorted(set(self.currency_aliases) - set(self.currency_codes)) if unknown: raise ValueError(f"currency_aliases: коды {unknown} не перечислены в currency_codes") return self def currency_code(self, text: str) -> str | None: """Код ISO по тексту: сам код («usd», «RUB») или синоним из `currency_aliases` («руб.»); иначе `None`.""" bare = normalize_text(text).upper() if bare in self.currency_codes: return bare for code, patterns in self.currency_aliases.items(): if any(pattern.search(text) for pattern in patterns): return code return None class _LineCodeRules(_Model): ranges: dict[FormType, tuple[int, int]] min_share: float min_codes: int min_codes_continuation: int @model_validator(mode="after") def _check(self) -> "_LineCodeRules": if set(self.ranges) != set(get_args(FormType)): raise ValueError(f"line_codes.ranges: нужны все формы {get_args(FormType)}") if not 0 < self.min_share <= 1 or self.min_codes < 1: raise ValueError("line_codes: min_share в (0, 1], min_codes ≥ 1") if not 1 <= self.min_codes_continuation <= self.min_codes: raise ValueError("line_codes: min_codes_continuation в [1, min_codes]") return self class StatementHeaderRules(_Model): form_titles: dict[FormType, list[Pattern]] okud: dict[FormType, list[Pattern]] okud_code: Pattern units_thousands: Pattern continuation: list[Pattern] code_caption: list[Pattern] explanations_caption: list[Pattern] column_numbering: Pattern labels: dict[StatementLabel, list[Pattern]] auditor: list[Pattern] legal_form: list[Pattern] line_codes: _LineCodeRules fractional_share_percent: int @model_validator(mode="after") def _check_required_keys(self) -> "StatementHeaderRules": for field_name, expected in ( ("form_titles", set(get_args(FormType))), ("okud", set(get_args(FormType))), ("labels", set(get_args(StatementLabel))), ): missing = expected - set(getattr(self, field_name)) if missing: raise ValueError(f"{field_name}: нет обязательных ключей {sorted(missing)}") if not 1 <= self.fractional_share_percent <= 100: # noqa: PLR2004 raise ValueError( f"fractional_share_percent: должна быть в диапазоне [1, 100], получено {self.fractional_share_percent}" ) return self def form_type(self, text: str) -> FormType | None: entries = ((p, k) for k, patterns in self.form_titles.items() for p in patterns) return first_match(entries, text) def okud_type(self, text: str) -> FormType | None: entries = ((p, k) for k, patterns in self.okud.items() for p in patterns) return first_match(entries, text) def is_thousands(self, text: str) -> bool: return self.units_thousands.search(text) is not None def is_continuation(self, text: str) -> bool: return any(pattern.search(text) for pattern in self.continuation) def is_code_caption(self, text: str) -> bool: return any(pattern.search(text) for pattern in self.code_caption) def is_explanations_caption(self, text: str) -> bool: return any(pattern.search(text) for pattern in self.explanations_caption) def label(self, key: StatementLabel, text: str) -> re.Match[str] | None: for pattern in self.labels[key]: found = pattern.search(text) if found is not None: return found return None def is_auditor_label(self, text: str) -> bool: return any(pattern.search(text) for pattern in self.auditor) def has_legal_form(self, text: str) -> bool: return any(pattern.search(text) for pattern in self.legal_form) @property def code_ranges(self) -> dict[FormType, tuple[int, int]]: return dict(self.line_codes.ranges) @property def code_min_share(self) -> float: return self.line_codes.min_share @property def min_codes(self) -> int: return self.line_codes.min_codes @property def min_codes_continuation(self) -> int: return self.line_codes.min_codes_continuation class StatementPeriodRules(_Model): months: Pattern on_date: Pattern numeric_date: Pattern for_months: Pattern for_range: Pattern for_year: Pattern @model_validator(mode="after") def _check_groups(self) -> "StatementPeriodRules": if self.months.group_names != _MONTH_GROUPS: raise ValueError("months: ожидаются именованные группы m01…m12") for name, pattern, groups in ( ("on_date", self.on_date, {"d", "m", "y"}), ("numeric_date", self.numeric_date, {"d", "m", "y"}), ("for_months", self.for_months, {"n", "y"}), ("for_range", self.for_range, {"m", "y"}), ("for_year", self.for_year, {"y"}), ): missing = groups - pattern.group_names if missing: raise ValueError(f"{name}: нет именованных групп {sorted(missing)}") return self def _month(self, text: str) -> int | None: if text.isdigit(): number = int(text) return number if 1 <= number <= _MONTHS_IN_YEAR else None found = self.months.search(text) return None if found is None or found.lastgroup is None else int(found.lastgroup.removeprefix("m")) def date_triple(self, day: str, month: str, year: str) -> datetime.date | None: """Дата из трёх ячеек «число | месяц | год» (строки или числа как текст); невалидная — None.""" month_number = self._month(month.strip()) if month.strip() else None if not day.strip().isdigit() or month_number is None or not year.strip().isdigit(): return None try: return datetime.date(int(year), month_number, int(day)) except ValueError: return None def period_end(self, text: str) -> datetime.date | None: """Дата колонки/подзаголовка: «на <дата>» — она сама; «за N месяцев/за январь–март/за год» — конец периода.""" found = self.on_date.search(text) or self.numeric_date.search(text) if found is not None: return self.date_triple(found.group("d"), found.group("m"), found.group("y")) found = self.for_months.search(text) if found is not None: return _month_end(int(found.group("y")), int(found.group("n"))) found = self.for_range.search(text) if found is not None: month = self._month(found.group("m")) return None if month is None else _month_end(int(found.group("y")), month) found = self.for_year.search(text) if found is not None: return datetime.date(int(found.group("y")), _MONTHS_IN_YEAR, 31) return None def _month_end(year: int, month: int) -> datetime.date | None: if not 1 <= month <= _MONTHS_IN_YEAR: return None last = calendar.monthrange(year, month)[1] return datetime.date(year, month, last) class RelationRule(_Model): form: FormType relation: str strict: bool = True abs_terms: tuple[str, ...] = () @property def target(self) -> str: return _parse_relation(self.relation)[0] @property def terms(self) -> tuple[tuple[int, str], ...]: return _parse_relation(self.relation)[1] @model_validator(mode="after") def _check(self) -> "RelationRule": target, terms = _parse_relation(self.relation) codes = {code for _, code in terms} unknown = set(self.abs_terms) - codes if unknown: raise ValueError(f"abs_terms {sorted(unknown)} нет в правой части «{self.relation}»") if target in codes: raise ValueError(f"«{self.relation}»: код слева встречается справа") return self def _parse_relation(relation: str) -> tuple[str, tuple[tuple[int, str], ...]]: found = _RELATION.match(relation) if found is None: msg = ( f"соотношение «{relation}»: ожидается «<код> = <код> [+|-] <код> …» из четырёхзначных или пятизначных кодов" ) raise ValueError(msg) terms = tuple((-1 if sign == "-" else 1, code) for sign, code in _TERM.findall(found.group("rhs"))) return found.group("target"), terms class StatementCheckRules(_Model): tolerance: Decimal #: OQ-138: доля (в процентах) от наибольшего |числа| соотношения, в пределах которой жёсткое нарушение — округление rounding_tolerance_percent: Decimal = Decimal("0") subline_check: bool = True relations: list[RelationRule] def relations_for(self, form: str) -> list[RelationRule]: return [rule for rule in self.relations if rule.form == form] @model_validator(mode="after") def _check_rounding_tolerance(self) -> "StatementCheckRules": if not Decimal("0") <= self.rounding_tolerance_percent <= Decimal("100"): raise ValueError("rounding_tolerance_percent: ожидается доля в процентах от 0 до 100") return self @model_validator(mode="after") def _check_explicit_rules_have_four_digit_codes(self) -> "StatementCheckRules": """Явные правила из yaml содержат только четырёхзначные коды; пятизначные — только в подстроках.""" parent_len = 4 # четырёхзначные коды для явных правил for rule in self.relations: target, terms = _parse_relation(rule.relation) # Цель всегда четырёхзначная (проверяется в _RELATION) if len(target) != parent_len: raise ValueError(f"явное правило «{rule.relation}»: цель должна быть четырёхзначной") # Члены правой части должны быть четырёхзначными for _, code in terms: if len(code) != parent_len: raise ValueError( f"явное правило «{rule.relation}»: код «{code}» должен быть четырёхзначным; " f"пятизначные коды допустимы только в сгенерированных подстроках" ) return self @dataclass(frozen=True) class Rules: headers: OsvHeaderRules groups: OsvGroupRules checks: OsvCheckRules periods: PeriodRules data_types: DataTypeRules statement_headers: StatementHeaderRules statement_periods: StatementPeriodRules statement_checks: StatementCheckRules def load_rules(path: Path = DEFAULT_RULES_DIR) -> Rules: """Читает восемь словарей каталога и валидирует их; любая проблема — `RuntimeError` с файлом и ключом.""" return Rules( headers=_load(path, HEADERS_FILE, OsvHeaderRules), groups=_load(path, GROUPS_FILE, OsvGroupRules), checks=_load(path, CHECKS_FILE, OsvCheckRules), periods=_load(path, PERIODS_FILE, PeriodRules), data_types=_load(path, DATA_TYPES_FILE, DataTypeRules), statement_headers=_load(path, STATEMENT_HEADERS_FILE, StatementHeaderRules), statement_periods=_load(path, STATEMENT_PERIODS_FILE, StatementPeriodRules), statement_checks=_load(path, STATEMENT_CHECKS_FILE, StatementCheckRules), ) ModelT = TypeVar("ModelT", bound=BaseModel) def _load(directory: Path, name: str, model: type[ModelT]) -> ModelT: file = directory / name try: raw = yaml.safe_load(file.read_text(encoding="utf-8")) except OSError as exc: raise RuntimeError(f"словарь {name}: файл не прочитан ({file}): {exc.strerror or exc}") from exc except yaml.YAMLError as exc: reason = " ".join(str(exc).split()) raise RuntimeError(f"словарь {name}: некорректный YAML ({file}): {reason}") from exc if not isinstance(raw, dict): raise RuntimeError(f"словарь {name}: ожидается словарь верхнего уровня ({file})") try: return model.model_validate(raw) except ValidationError as exc: problems = [ f"словарь {name}: {'.'.join(str(part) for part in error['loc']) or '<корень>'}: {error['msg']}" for error in exc.errors() ] raise RuntimeError("\n".join(problems)) from exc