from __future__ import annotations import json from collections import defaultdict from dataclasses import dataclass from pathlib import Path EXPECTED_COLUMN_COUNT = 204 MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json") REPORT_GROUP_TITLES: dict[str, str] = { "service": "Служебные поля", "message_bank": "Реквизиты сообщения/КО/филиала", "record_control": "Идентификация записи и контрольные признаки", "operation_parameters": "Параметры операции", "transfer_settlement": ( "Сведения о переводах денежных средств, в том числе электронных денежных средств" ), "cash_receipt": "Сведения о месте приема наличных денежных средств", "transfer_status": "Статус перевода денежных средств", "cash_payment": "Сведения о месте выдачи наличных денежных средств", "esp_authorization": "Сведения о месте авторизации ЭСП", "own_account_cash": ( "Сведения о внесении наличных денежных средств на свой банковский счет " "или о получении наличных денежных средств со своего банковского счета " "у одного оператора по переводу денежных средств" ), "foreign_card_operation": ( "Сведения об операции с использованием платежной карты иностранного банка" ), "operation_basis": "Основание совершения операции", "participant_base": "Сведения об участниках операции", "participant_fio": "ФИО участника операции", "participant_identifiers": "Идентификационные сведения участника операции", "participant_identity_document": ("Сведения о документе, удостоверяющем личность"), "participant_stay_document": ( "Сведения о документе, подтверждающем право на пребывание " "(проживание) в Российской Федерации" ), "participant_address": "Адрес участника операции", "participant_state_registration": ( "Место государственной регистрации участника операции" ), "participant_extra": "Дополнительная информация об участнике операции", "eio": "ЕИО/Бенефициар (идентификация, документы, адреса)", "cp": "Сведения о ценных бумагах (ЦП)", } BUSINESS_BLOCKS = frozenset( { "file_metadata", "message_header", "operation_base", "transfer", "cash_receipt", "transfer_status", "cash_payment", "network", "cash_operation", "foreign_card", "operation_basis", "participant_base", "participant_name", "participant_identity", "participant_documents", "participant_address", "participant_extra", "eio", "cp", } ) class ColumnSchemaError(ValueError): """Ошибка целостности схемы колонок отчёта.""" class AmbiguousColumnNameError(LookupError): """Имя колонки соответствует нескольким позициям отчёта.""" @dataclass(frozen=True) class ColumnMeta: index: int column_id: str name: str block: str report_group: str xml_tag: str xml_path: str source_scope: str allow_short_lookup: bool allow_direct_mapping: bool structured_value: bool structured_group: str structured_role: str structured_order: int join_with_space: bool @dataclass(frozen=True) class ReportGroup: group_id: str title: str indexes: tuple[int, ...] @property def start_index(self) -> int: return self.indexes[0] @property def end_index(self) -> int: return self.indexes[-1] class ColumnRegistry: def __init__(self, columns: list[ColumnMeta]) -> None: ordered = tuple(sorted(columns, key=lambda column: column.index)) self._validate_columns(ordered) self._ordered = ordered self._by_index = {column.index: column for column in ordered} self._by_id = {column.column_id: column for column in ordered} by_name: dict[str, list[int]] = defaultdict(list) by_block: dict[str, list[int]] = defaultdict(list) for column in ordered: by_name[column.name].append(column.index) by_block[column.block].append(column.index) self._by_name = {name: tuple(indexes) for name, indexes in by_name.items()} self._by_block = {block: tuple(indexes) for block, indexes in by_block.items()} self._report_groups = self._build_report_groups(ordered) def index_of(self, column_id: str) -> int: """Возвращает текущую позицию стабильной колонки.""" try: return self._by_id[column_id].index except KeyError as exc: raise KeyError(f"Неизвестный column_id: {column_id}") from exc def meta(self, index: int) -> ColumnMeta: """Возвращает метаданные колонки по текущей позиции.""" return self._by_index[index] def by_id(self, column_id: str) -> ColumnMeta: """Возвращает метаданные по стабильному идентификатору.""" try: return self._by_id[column_id] except KeyError as exc: raise KeyError(f"Неизвестный column_id: {column_id}") from exc def block(self, block_name: str) -> tuple[int, ...]: """Возвращает позиции всех колонок бизнес-блока.""" return self._by_block.get(block_name, ()) def indices_named(self, name: str) -> tuple[int, ...]: """Возвращает все позиции с указанным отображаемым именем.""" return self._by_name.get(name, ()) def index_named(self, name: str, *, block: str | None = None) -> int: """Ищет имя без молчаливого выбора среди дубликатов.""" indexes = self.indices_named(name) if block is not None: indexes = tuple( index for index in indexes if self.meta(index).block == block ) if not indexes: raise KeyError(f"Колонка не найдена: {name}") if len(indexes) > 1: raise AmbiguousColumnNameError( f"Имя колонки неоднозначно: {name!r}, позиции {indexes}" ) return indexes[0] def ordered_columns(self) -> tuple[ColumnMeta, ...]: """Возвращает всю схему в порядке вывода.""" return self._ordered def all_columns(self) -> tuple[str, ...]: """Возвращает отображаемые имена в порядке вывода.""" return tuple(column.name for column in self._ordered) def report_groups(self) -> tuple[ReportGroup, ...]: """Возвращает непрерывные группы верхней шапки.""" return self._report_groups @classmethod def load(cls, json_path: Path) -> ColumnRegistry: raw = json.loads(json_path.read_text(encoding="utf-8")) if not isinstance(raw, list): raise ColumnSchemaError("Корень mapping_table.json должен быть списком") columns: list[ColumnMeta] = [] for position, item in enumerate(raw, start=1): if not isinstance(item, dict): raise ColumnSchemaError( f"Запись {position} mapping_table.json должна быть объектом" ) try: columns.append( ColumnMeta( index=cls._required_int(item, "index"), column_id=cls._required_str(item, "column_id"), name=cls._required_str(item, "column_name"), block=cls._required_str(item, "block"), report_group=cls._required_str(item, "report_group"), xml_tag=cls._required_str(item, "xml_tag"), xml_path=cls._required_str(item, "xml_path"), source_scope=cls._required_str(item, "source_scope"), allow_short_lookup=cls._required_bool( item, "allow_short_lookup" ), allow_direct_mapping=cls._required_bool( item, "allow_direct_mapping" ), structured_value=cls._required_bool(item, "structured_value"), structured_group=cls._required_str(item, "structured_group"), structured_role=cls._required_str(item, "structured_role"), structured_order=cls._required_int(item, "structured_order"), join_with_space=cls._required_bool(item, "join_with_space"), ) ) except (KeyError, TypeError, ValueError) as exc: raise ColumnSchemaError( f"Некорректная запись схемы в позиции {position}: {exc}" ) from exc return cls(columns) @staticmethod def _required_bool(item: dict[object, object], key: str) -> bool: value = item[key] if not isinstance(value, bool): raise TypeError(f"{key} должен быть bool") return value @staticmethod def _required_str(item: dict[object, object], key: str) -> str: value = item[key] if not isinstance(value, str): raise TypeError(f"{key} должен быть строкой") return value.strip() @staticmethod def _required_int(item: dict[object, object], key: str) -> int: value = item[key] if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{key} должен быть целым числом") return value @staticmethod def _validate_columns(columns: tuple[ColumnMeta, ...]) -> None: if len(columns) != EXPECTED_COLUMN_COUNT: raise ColumnSchemaError( f"Ожидалось {EXPECTED_COLUMN_COUNT} колонок, получено {len(columns)}" ) indexes = tuple(column.index for column in columns) expected_indexes = tuple(range(1, EXPECTED_COLUMN_COUNT + 1)) if indexes != expected_indexes: raise ColumnSchemaError( "Индексы колонок должны непрерывно идти от 1 до 204" ) column_ids = [column.column_id for column in columns] if len(column_ids) != len(set(column_ids)): raise ColumnSchemaError("column_id должны быть уникальными") valid_sources = {"any", "operation", "participant"} valid_roles = {"", "aggregate", "one_line", "component"} for column in columns: if not column.column_id or not column.name: raise ColumnSchemaError( f"Колонка {column.index}: пустой column_id или column_name" ) if not column.block or not column.report_group: raise ColumnSchemaError( f"Колонка {column.index}: block и report_group обязательны" ) if column.block not in BUSINESS_BLOCKS: raise ColumnSchemaError(f"Колонка {column.index}: неизвестный block") if column.report_group not in REPORT_GROUP_TITLES: raise ColumnSchemaError( f"Колонка {column.index}: неизвестный report_group" ) if column.source_scope not in valid_sources: raise ColumnSchemaError( f"Колонка {column.index}: неизвестный source_scope" ) if column.structured_role not in valid_roles: raise ColumnSchemaError( f"Колонка {column.index}: неизвестная structured_role" ) if bool(column.structured_group) != bool(column.structured_role): raise ColumnSchemaError( f"Колонка {column.index}: группа и роль структуры задаются вместе" ) ColumnRegistry._validate_structured_groups(columns) @staticmethod def _validate_structured_groups(columns: tuple[ColumnMeta, ...]) -> None: grouped: dict[str, list[ColumnMeta]] = defaultdict(list) for column in columns: if column.structured_group: grouped[column.structured_group].append(column) elif column.structured_order != 0 or column.join_with_space: raise ColumnSchemaError( f"Колонка {column.index}: настройки структуры заданы без группы" ) for group_id, members in grouped.items(): aggregates = [ member for member in members if member.structured_role == "aggregate" ] components = [ member for member in members if member.structured_role == "component" ] if len(aggregates) != 1: raise ColumnSchemaError( f"Группа {group_id!r} должна иметь один aggregate" ) aggregate = aggregates[0] if aggregate.column_id != group_id: raise ColumnSchemaError( f"Группа {group_id!r} должна ссылаться на column_id aggregate" ) if not aggregate.structured_value or aggregate.structured_order != 0: raise ColumnSchemaError( f"Aggregate группы {group_id!r} настроен некорректно" ) if not components: raise ColumnSchemaError( f"Группа {group_id!r} должна содержать components" ) for role in ("one_line", "component"): role_members = [ member for member in members if member.structured_role == role ] orders = [member.structured_order for member in role_members] if any(order < 1 for order in orders) or len(orders) != len( set(orders) ): raise ColumnSchemaError( f"Группа {group_id!r}: порядок роли {role} некорректен" ) if any(member.structured_value for member in role_members): raise ColumnSchemaError( f"Группа {group_id!r}: только aggregate может быть structured_value" ) if any(member.join_with_space for member in role_members): raise ColumnSchemaError( f"Группа {group_id!r}: join_with_space допустим только aggregate" ) @staticmethod def _build_report_groups( columns: tuple[ColumnMeta, ...], ) -> tuple[ReportGroup, ...]: result: list[ReportGroup] = [] seen_groups: set[str] = set() current_id = "" current_indexes: list[int] = [] for column in columns: if column.report_group == current_id: current_indexes.append(column.index) continue if current_indexes: result.append( ReportGroup( current_id, REPORT_GROUP_TITLES[current_id], tuple(current_indexes), ) ) seen_groups.add(current_id) if column.report_group in seen_groups: raise ColumnSchemaError( f"report_group {column.report_group!r} разбит на несколько диапазонов" ) current_id = column.report_group current_indexes = [column.index] if current_indexes: result.append( ReportGroup( current_id, REPORT_GROUP_TITLES[current_id], tuple(current_indexes), ) ) return tuple(result) COLUMNS = ColumnRegistry.load(MAPPING_DATA_FILE)