"""Словари `rules/*.yaml` и загрузчик (design D-7): синонимы заголовков, группировок, периодов и типов данных. Словари — единственный носитель синонимов: код парсера только вызывает `Pattern.search` и не знает ни одного слова из ведомости. Новый синоним — строка в YAML. Правила нормализации (одни и те же для проверяемого текста и для самих паттернов в YAML): текст приводится `normalize_text` (нижний регистр, «ё» -> «е», дефисы/тире/неразрывные пробелы/переносы -> пробел, пробелы схлопнуты), а паттерн компилируется как есть с `re.IGNORECASE`. Поэтому паттерн, содержащий «ё», дефис или неразрывный пробел, никогда не сработал бы; загрузчик отвергает такие паттерны с указанием файла и ключа. """ import re from collections.abc import Iterable from dataclasses import dataclass 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", "OsvGroupRules", "OsvHeaderRules", "Pattern", "PeriodRule", "PeriodRules", "Rules", "first_match", "load_rules", "normalize_text", ] 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" 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"] _ISO_CODE_LENGTH = 3 # Разделители, сводимые к одному пробелу: любые пробельные (включая 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]") Value = TypeVar("Value") def normalize_text(s: str) -> str: """Нижний регистр, «ё» -> «е», пробельные символы и дефисы/тире -> один пробел, невидимые переносы удалены.""" lowered = _INVISIBLE.sub("", s).lower().replace("ё", "е") return _SEPARATORS.sub(" ", lowered).strip() 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 __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] statement_markers: list[Pattern] groups: dict[SumGroup, list[Pattern]] debit: list[Pattern] credit: list[Pattern] service_columns: dict[str, list[Pattern]] total_row: 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'") 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 PeriodRule(_Model): pattern: Pattern quarter: Quarter is_cumulative: bool class PeriodRules(_Model): rules: list[PeriodRule] year: Pattern 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]] 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 @dataclass(frozen=True) class Rules: headers: OsvHeaderRules groups: OsvGroupRules periods: PeriodRules data_types: DataTypeRules def load_rules(path: Path = DEFAULT_RULES_DIR) -> Rules: """Читает четыре словаря каталога и валидирует их; любая проблема — `RuntimeError` с файлом и ключом.""" return Rules( headers=_load(path, HEADERS_FILE, OsvHeaderRules), groups=_load(path, GROUPS_FILE, OsvGroupRules), periods=_load(path, PERIODS_FILE, PeriodRules), data_types=_load(path, DATA_TYPES_FILE, DataTypeRules), ) 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