5be03db68d
Решения владельца 27.10.2026 (docs #172): субсчёт с 3 цифрами в сегменте под счётом шапки, код контрагента + название + договоры (стратегия code_blocks), суммы субсчёта из «ИТОГ по …», «ИТОГ» — итог листа, «Поставщики» — вид субконто в таблице, метка «За период» со значением в соседней ячейке, диапазон «01.01.25..30.06.25», год из двух цифр. Синтетика s45 (период, организация) и s46 (блоки), спека osv-parsing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
428 lines
22 KiB
Python
428 lines
22 KiB
Python
"""Словари `rules/*.yaml` и загрузчик (design D-7): синонимы заголовков, группировок, периодов и типов данных.
|
||
|
||
Словари — единственный носитель синонимов: код парсера только вызывает `Pattern.search` и не знает ни одного слова
|
||
из ведомости. Новый синоним — строка в YAML.
|
||
|
||
Правила нормализации (одни и те же для проверяемого текста и для самих паттернов в YAML):
|
||
текст приводится `normalize_text` (нижний регистр, «ё» -> «е», дефисы/тире/неразрывные пробелы/переносы -> пробел,
|
||
пробелы схлопнуты), а паттерн компилируется как есть с `re.IGNORECASE`. Поэтому паттерн, содержащий «ё», дефис или
|
||
неразрывный пробел, никогда не сработал бы; загрузчик отвергает такие паттерны с указанием файла и ключа.
|
||
"""
|
||
|
||
import bisect
|
||
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",
|
||
"MonthRange",
|
||
"OsvCheckRules",
|
||
"OsvGroupRules",
|
||
"OsvHeaderRules",
|
||
"Pattern",
|
||
"PeriodRule",
|
||
"PeriodRules",
|
||
"Rules",
|
||
"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"
|
||
|
||
GroupType = Literal["ACCOUNT", "DEPARTMENT", "COUNTERPARTY", "CONTRACT", "OTHER"]
|
||
SumGroup = Literal["opening", "turnover", "closing"]
|
||
HeaderMarker = Literal["account", "period", "data_type", "currency", "org_params", "subkonto", "chart_of_accounts"]
|
||
RowDataType = Literal["CURRENCY", "QUANTITY"]
|
||
Quarter = Literal["Q1", "Q2", "Q3", "Q4"]
|
||
|
||
_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]")
|
||
|
||
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]
|
||
# Итог субсчёта «ИТОГ по 60.111» (группа account — код субсчёта); пусто — правило не действует
|
||
subaccount_sums: 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"}),
|
||
("subaccount_sums", {"account"}),
|
||
):
|
||
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
|
||
|
||
|
||
def _check_account_codes(field_name: str, codes: Iterable[str]) -> None:
|
||
bad = [code for code in codes if not _ACCOUNT_CODE.fullmatch(code)]
|
||
if bad:
|
||
raise ValueError(f"{field_name}: ожидаются коды счетов вида «91» или «76.09», {bad}")
|
||
|
||
|
||
def _is_under_account(account_code: str | None, roots: Iterable[str]) -> bool:
|
||
"""Код счёта из файла («91», «91.01») совпадает с одним из `roots` либо является его субсчётом (не «910»)."""
|
||
if not account_code:
|
||
return False
|
||
code = account_code.strip()
|
||
return any(code == root or code.startswith(f"{root}.") for root in roots)
|
||
|
||
|
||
class OsvGroupRules(_Model):
|
||
mapping: list[tuple[Pattern, GroupType]]
|
||
default: GroupType = "OTHER"
|
||
# Расчётные счета: аналитика первого уровня — контрагенты, даже если вид субконто в шапке не назван (#59, R-6)
|
||
counterparty_accounts: list[str] = []
|
||
|
||
@model_validator(mode="after")
|
||
def _check_counterparty_accounts(self) -> "OsvGroupRules":
|
||
_check_account_codes("counterparty_accounts", self.counterparty_accounts)
|
||
return self
|
||
|
||
def is_counterparty_account(self, account_code: str | None) -> bool:
|
||
"""Счёт из `counterparty_accounts` или его субсчёт («76» -> 76, 76.2, 76.Н.1; не 760)."""
|
||
return _is_under_account(account_code, self.counterparty_accounts)
|
||
|
||
def classify(self, text: str) -> GroupType:
|
||
"""Тип группировки по названию из шапки; не совпавшее ни с одним паттерном — `default`."""
|
||
found = first_match(self.mapping, text)
|
||
return self.default if found is None else found
|
||
|
||
def known(self, text: str) -> GroupType | None:
|
||
"""Тип группировки по явной записи `mapping`; не совпавшее ни с одной записью — `None` (не `default`)."""
|
||
return first_match(self.mapping, text)
|
||
|
||
|
||
class OsvCheckRules(_Model):
|
||
accounts_without_analytic_balance: list[str]
|
||
|
||
@model_validator(mode="after")
|
||
def _check_codes(self) -> "OsvCheckRules":
|
||
_check_account_codes("accounts_without_analytic_balance", self.accounts_without_analytic_balance)
|
||
return self
|
||
|
||
def has_no_analytic_balance(self, account_code: str | None) -> bool:
|
||
"""Код счёта из файла («91», «91.01») относится к счёту из списка: совпадает с ним либо его субсчёт."""
|
||
return _is_under_account(account_code, self.accounts_without_analytic_balance)
|
||
|
||
|
||
class PeriodRule(_Model):
|
||
pattern: Pattern
|
||
quarter: Quarter
|
||
is_cumulative: bool
|
||
|
||
|
||
_MONTHS_IN_YEAR = 12
|
||
_CENTURY = 2000 # двузначный год даты «…25» — 2025
|
||
_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
|
||
# Двузначный год даты (группа yy), если года из четырёх цифр в тексте нет
|
||
short_year_date: 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 year_of(self, text: str) -> int | None:
|
||
"""Год периода: четыре цифры (`year`), иначе двузначный год даты конца диапазона «01.01.25..30.06.25» (20yy)."""
|
||
found = self.year.search(text)
|
||
if found is not None:
|
||
return int(found.group(0))
|
||
short = self.short_year_date.search(text)
|
||
return None if short is None else _CENTURY + int(short.group("yy"))
|
||
|
||
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]]
|
||
indicator_column: list[tuple[Pattern, DataType]]
|
||
indicator_column_untyped: list[Pattern]
|
||
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
|
||
checks: OsvCheckRules
|
||
periods: PeriodRules
|
||
data_types: DataTypeRules
|
||
|
||
|
||
def load_rules(path: Path = DEFAULT_RULES_DIR) -> Rules:
|
||
"""Читает пять словарей ОСВ каталога и валидирует их; любая проблема — `RuntimeError` с файлом и ключом.
|
||
Словари форм ББ/ОФР — в пакете `arm_statements` (`default_rules()`), здесь их нет.
|
||
"""
|
||
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),
|
||
)
|
||
|
||
|
||
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
|