389 lines
17 KiB
Python
389 lines
17 KiB
Python
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)
|