RSHB/SFM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
Raykov-MS 2026-08-01 10:00:29 +03:00
parent d1cb2dda71
commit d6b3e73cb7
7 changed files with 3307 additions and 529 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
app/pipeline/column_constants.py Normal file
View File

@ -0,0 +1,120 @@
from __future__ import annotations
from .column_registry import COLUMNS
def _index(legacy_position: int) -> int:
return COLUMNS.index_of(f"legacy_{legacy_position:03d}")
# Служебные поля и операция.
COL_FILE_NAME = _index(1)
COL_RECORD_ID = _index(7)
COL_FTR_SIGN = _index(10)
COL_SUSPENSION_BASIS = _index(11)
COL_OPERATION_SIGN = _index(14)
COL_ESP_SIGN = _index(15)
COL_OPERATION_CODE = _index(19)
COL_EXTRA_CODES = _index(20)
COL_UNUSUAL_CODES = _index(21)
COL_CURRENCY = _index(22)
COL_AMOUNT_CURRENCY = _index(23)
COL_AMOUNT_RUB = _index(24)
COL_SALE_CURRENCY = _index(25)
COL_SALE_AMOUNT = _index(26)
COL_CURRENCY_SIGN = _index(27)
COL_SUSPICIOUS_ID = _index(28)
COL_METAL_CODE = _index(29)
COL_METAL_NAME = _index(30)
COL_ITEM_TYPE = _index(31)
# Перевод.
COL_TRANSFER_TYPE = _index(35)
COL_TERRITORY_CODE = _index(36)
COL_OPERATOR_TYPE = _index(37)
COL_PAYER_ACCOUNT = _index(38)
COL_PAYER_ESP = _index(39)
COL_PAYER_BANK_NAME = _index(40)
COL_PAYER_BANK_BIK = _index(41)
COL_RECIPIENT_BANK_NAME = _index(42)
COL_RECIPIENT_BANK_BIK = _index(43)
COL_PAYER_BANK_ACCOUNT = _index(44)
COL_RECIPIENT_BANK_ACCOUNT = _index(45)
COL_RECIPIENT_ACCOUNT = _index(46)
COL_RECIPIENT_ESP = _index(47)
COL_TRANSFER_STATUS = _index(62)
# Карты и место операции.
COL_TERRITORY_CODE_CARD = _index(96)
COL_CARD_HOLDER_INFO = _index(103)
COL_EMPLOYEE_SIGN = _index(104)
COL_FOREIGN_BANK_NAME = _index(105)
# Участник.
COL_PARTICIPANT_TYPE = _index(109)
COL_RESIDENT_SIGN = _index(110)
COL_CLIENT_SIGN = _index(111)
COL_IDENT_FL = _index(119)
COL_INN = _index(120)
COL_KPP = _index(122)
COL_BIRTH_REG_DATE = _index(123)
COL_BRANCH_SIGN = _index(124)
COL_INSURANCE_NUMBER = _index(125)
COL_COUNTRY_CODE = _index(126)
COL_ADDRESS_ONE_LINE = _index(141)
COL_ADDRESS_STRUCTURED = _index(142)
COL_REG_PLACE_ONE_LINE = _index(152)
COL_REG_PLACE_STRUCTURED = _index(153)
# ЕИО и ценные бумаги.
COL_EIO_INN = _index(163)
COL_EIO_REG_PLACE_ONE_LINE = _index(178)
COL_CP_ISSUER_INN = _index(202)
# Блоки схемы.
BLOCK_EIO = COLUMNS.block("eio")
BLOCK_CP = COLUMNS.block("cp")
# Точные поля наличных, которые исторически очищаются для безналичных переводов.
CASHLESS_CLEARED_COLUMNS = tuple(
_index(position) for position in (50, 61, 65, 76, 84, 95, 99, 101)
)
OKATO_COUNTRY_PAIRS: tuple[tuple[int, int], ...] = tuple(
(_index(country), _index(okato))
for country, okato in (
(52, 53),
(67, 68),
(86, 87),
(144, 145),
(170, 171),
)
)
# Текущая перестановка адресного блока при выгрузке в целевой шаблон.
PARTICIPANT_ADDRESS_OUTPUT_SOURCES: dict[int, int] = {
_index(141): _index(142),
**{_index(target): _index(target + 1) for target in range(142, 151)},
_index(151): _index(141),
}
# Бизнес-константы.
ACCOUNT_PLACEHOLDER = "00000000000000000000"
CURRENCY_OPERATION_CODES = frozenset(str(code) for code in range(6101, 6127))
CASHLESS_TRANSFER_TYPES = frozenset(
{"1", "2", "3", "4", "7", "10", "11", "12", "13", "14"}
)
VALID_FTR_SIGNS = frozenset({"0", "1", "2", "3", "4", "5"})
VALID_SUSPENSION_BASES = frozenset({"1", "2", "3"})
VALID_ESP_SIGNS = frozenset({"1", "2", "3", "4"})
PRECIOUS_ITEM_CODES = frozenset({"5020", "5021", "5022", "5023"})
SUSPICIOUS_CODE = "6001"
TERRITORY_CODE = "5016"
PRECIOUS_METAL_CODE = "C99"
EMPLOYEE_ABSENT_CODE = "0"
RUSSIA_COUNTRY_CODE = "643"
PARTICIPANT_TYPE_UL = "1"
PARTICIPANT_TYPE_FL = "2"
PARTICIPANT_TYPE_IP = "3"
PARTICIPANT_TYPE_FLCHP = "4"

365
app/pipeline/column_registry.py Normal file
View File

@ -0,0 +1,365 @@
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_authorization": ("Места приема/выдачи наличных и авторизация ЭСП"),
"same_operator_place": "Внесение/выдача у одного оператора (место операции)",
"cash_form_basis": "Наличная форма и основание операции",
"participant_base": "Участник операции (базовые сведения)",
"participant_identity": ("Участник операции (идентификация, документы, адреса)"),
"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)

827
app/pipeline/mapping.py
View File

File diff suppressed because it is too large Load Diff

2252
app/pipeline/mapping_table.json
View File

File diff suppressed because it is too large Load Diff

121
app/pipeline/report.py
View File

@ -2,15 +2,22 @@ from __future__ import annotations
import re
import shutil
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Iterable
import xlsxwriter
from .column_constants import (
COL_KPP,
COL_PARTICIPANT_TYPE,
PARTICIPANT_ADDRESS_OUTPUT_SOURCES,
PARTICIPANT_TYPE_UL,
)
from .column_registry import COLUMNS
from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row_by_index
@ -30,72 +37,21 @@ class GroupHeader:
title: str
# Верхняя группирующая шапка согласно согласованным логическим категориям.
TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = (
GroupHeader(start_col=1, end_col=1, title="Служебные поля"),
GroupHeader(start_col=2, end_col=6, title="Реквизиты сообщения/КО/филиала"),
TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = tuple(
GroupHeader(
start_col=7,
end_col=13,
title="Идентификация записи и контрольные признаки",
),
GroupHeader(start_col=14, end_col=36, title="Параметры операции"),
GroupHeader(
start_col=37,
end_col=48,
title="Перевод ДС и реквизиты расчетов",
),
GroupHeader(
start_col=49,
end_col=66,
title="Места приема/выдачи наличных и авторизация ЭСП",
),
GroupHeader(
start_col=67,
end_col=75,
title="Внесение/выдача у одного оператора (место операции)",
),
GroupHeader(
start_col=76,
end_col=86,
title="Наличная форма и основание операции",
),
GroupHeader(
start_col=87,
end_col=126,
title="Участник операции (базовые сведения)",
),
GroupHeader(
start_col=127,
end_col=154,
title="Участник операции (идентификация, документы, адреса)",
),
GroupHeader(
start_col=155,
end_col=193,
title="ЕИО/Бенефициар (идентификация, документы, адреса)",
),
GroupHeader(
start_col=194,
end_col=204,
title="Сведения о ценных бумагах (ЦП)",
),
GroupHeader(
start_col=205,
end_col=205,
title=(
"Колонки вне согласованной группировки (если количество полей расширится)"
),
),
start_col=group.start_index,
end_col=group.end_index,
title=group.title,
)
for group in COLUMNS.report_groups()
)
class StreamingReportWriter:
def __init__(self) -> None:
self.columns = list(FIXED_REPORT_COLUMNS)
temp_file = NamedTemporaryFile(suffix=".xlsx", delete=False)
self._temp_report_path = Path(temp_file.name)
temp_file.close()
with NamedTemporaryFile(suffix=".xlsx", delete=False) as temp_file:
self._temp_report_path = Path(temp_file.name)
self.workbook = xlsxwriter.Workbook(
str(self._temp_report_path),
{"constant_memory": True},
@ -138,9 +94,10 @@ class StreamingReportWriter:
operation_fields=row.operation_fields,
participant_fields=row.participant_fields,
)
for index, column_name in enumerate(self.columns):
raw_value = self._resolve_output_value(index + 1, data_by_index)
default_format = self._column_formats[index]
for output_index, column_name in enumerate(self.columns, start=1):
zero_based_index = output_index - 1
raw_value = self._resolve_output_value(output_index, data_by_index)
default_format = self._column_formats[zero_based_index]
value, cell_format = self._normalize_cell_value(
column_name=column_name,
raw_value=raw_value,
@ -148,23 +105,25 @@ class StreamingReportWriter:
)
if isinstance(value, datetime) and cell_format is self._date_format:
self.sheet.write_datetime(
self._next_data_row, index, value, cell_format
self._next_data_row, zero_based_index, value, cell_format
)
elif self._is_text_identifier_column(column_name):
self.sheet.write_string(
self._next_data_row,
index,
zero_based_index,
self._normalize_identifier_string(
column_name,
value,
column_index=index + 1,
column_index=output_index,
row_by_index=data_by_index,
),
)
elif cell_format is not None:
self.sheet.write(self._next_data_row, index, value, cell_format)
self.sheet.write(
self._next_data_row, zero_based_index, value, cell_format
)
else:
self.sheet.write(self._next_data_row, index, value)
self.sheet.write(self._next_data_row, zero_based_index, value)
self._next_data_row += 1
def _resolve_output_value(
@ -172,15 +131,10 @@ class StreamingReportWriter:
column_index: int,
row_by_index: dict[int, str],
) -> object:
# В целевом шаблоне адресный блок участника ожидается со структурным
# адресом в колонке 141, поэтому на этапе выгрузки делаем сдвиг.
if column_index == 141:
return row_by_index.get(142, "")
if 142 <= column_index <= 150:
return row_by_index.get(column_index + 1, "")
if column_index == 151:
return row_by_index.get(141, "")
return row_by_index.get(column_index, "")
source_index = PARTICIPANT_ADDRESS_OUTPUT_SOURCES.get(
column_index, column_index
)
return row_by_index.get(source_index, "")
def save(self, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
@ -261,9 +215,9 @@ class StreamingReportWriter:
if "БИК" in column_name:
return normalized.zfill(9)
if "КПП" in column_name:
participant_type = str(row_by_index.get(109, "")).strip()
participant_type = str(row_by_index.get(COL_PARTICIPANT_TYPE, "")).strip()
normalized_type = participant_type.lstrip("0") or "0"
if column_index == 122 and normalized_type != "1":
if column_index == COL_KPP and normalized_type != PARTICIPANT_TYPE_UL:
return normalized
return normalized.zfill(9)
if "ИНН" in column_name:
@ -300,7 +254,7 @@ class StreamingReportWriter:
date_part = normalized.split()[0]
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d.%m.%Y", "%Y/%m/%d"):
try:
return datetime.strptime(date_part, fmt)
return datetime.strptime(date_part, fmt) # noqa: DTZ007
except ValueError:
continue
return None
@ -309,15 +263,16 @@ class StreamingReportWriter:
try:
self._temp_report_path.unlink(missing_ok=True)
except Exception: # noqa: BLE001
pass
return
def __del__(self) -> None:
if not getattr(self, "_is_closed", True):
try:
self.workbook.close()
except Exception: # noqa: BLE001
pass
self._is_closed = True
self._is_closed = True
else:
self._is_closed = True
self._cleanup_temp_file()

135
tests/unit/test_column_registry.py Normal file
View File

@ -0,0 +1,135 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from app.pipeline.column_registry import (
COLUMNS,
MAPPING_DATA_FILE,
AmbiguousColumnNameError,
ColumnRegistry,
ColumnSchemaError,
)
def test_registry_keeps_stable_file_name_column_id() -> None:
assert COLUMNS.index_of("legacy_001") == 1
assert COLUMNS.meta(1).name == "Имя XML файла"
def test_registry_covers_all_report_columns_without_gaps() -> None:
indexes = tuple(column.index for column in COLUMNS.ordered_columns())
assert indexes == tuple(range(1, 205))
assert all(column.block for column in COLUMNS.ordered_columns())
assert all(column.report_group for column in COLUMNS.ordered_columns())
def test_registry_assigns_card_columns_missing_from_original_plan() -> None:
assert COLUMNS.block("foreign_card") == tuple(range(97, 107))
def test_registry_explicitly_preserves_structured_lookup_compatibility() -> None:
assert COLUMNS.meta(180).allow_short_lookup is True
assert COLUMNS.meta(153).allow_short_lookup is False
def test_registry_rejects_ambiguous_display_name_lookup() -> None:
assert COLUMNS.indices_named("ОКАТО") == (4, 171)
with pytest.raises(AmbiguousColumnNameError):
COLUMNS.index_named("ОКАТО")
def test_registry_report_groups_are_contiguous_and_cover_schema() -> None:
covered_indexes: list[int] = []
for group in COLUMNS.report_groups():
assert group.indexes == tuple(range(group.start_index, group.end_index + 1))
covered_indexes.extend(group.indexes)
assert covered_indexes == list(range(1, 205))
@pytest.mark.parametrize(
("field", "invalid_value"),
(
("index", True),
("column_id", None),
("column_name", ["Имя"]),
("source_scope", {"scope": "any"}),
("structured_order", False),
),
)
def test_registry_rejects_invalid_scalar_types(
tmp_path: Path, field: str, invalid_value: object
) -> None:
data = _mapping_data()
data[0][field] = invalid_value
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def test_registry_rejects_unknown_report_group(tmp_path: Path) -> None:
data = _mapping_data()
data[0]["report_group"] = "unknown_group"
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def test_registry_rejects_unknown_business_block(tmp_path: Path) -> None:
data = _mapping_data()
data[0]["block"] = "unknown_block"
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def test_registry_rejects_duplicate_column_ids_and_index_gaps(
tmp_path: Path,
) -> None:
data = _mapping_data()
data[1]["column_id"] = data[0]["column_id"]
data[1]["index"] = 3
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def test_registry_rejects_structured_group_without_aggregate(
tmp_path: Path,
) -> None:
data = _mapping_data()
aggregate = next(item for item in data if item["index"] == 50)
aggregate["structured_role"] = "component"
aggregate["structured_order"] = 10
aggregate["structured_value"] = False
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def test_registry_rejects_duplicate_structured_component_order(
tmp_path: Path,
) -> None:
data = _mapping_data()
first = next(item for item in data if item["index"] == 51)
second = next(item for item in data if item["index"] == 52)
second["structured_order"] = first["structured_order"]
with pytest.raises(ColumnSchemaError):
ColumnRegistry.load(_write_mapping(tmp_path, data))
def _mapping_data() -> list[dict[str, object]]:
return json.loads(MAPPING_DATA_FILE.read_text(encoding="utf-8"))
def _write_mapping(tmp_path: Path, data: list[dict[str, object]]) -> Path:
path = tmp_path / "mapping_table.json"
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
return path

16
tests/unit/test_report.py
View File

@ -1,11 +1,12 @@
from __future__ import annotations
from datetime import datetime
from datetime import date, datetime
from pathlib import Path
from unittest.mock import patch
from openpyxl import load_workbook
from app.pipeline.column_registry import COLUMNS
from app.pipeline.mapping import FIXED_REPORT_COLUMNS
from app.pipeline.report import TOP_GROUP_HEADERS, ReportRow, write_report
@ -64,6 +65,17 @@ def test_write_report_contains_top_group_header_row(tmp_path: Path) -> None:
assert merged_ranges == expected_ranges
def test_top_group_headers_follow_registry_report_groups() -> None:
actual_ranges = tuple(
(header.start_col, header.end_col) for header in TOP_GROUP_HEADERS
)
registry_ranges = tuple(
(group.start_index, group.end_index) for group in COLUMNS.report_groups()
)
assert actual_ranges == registry_ranges
def test_write_report_converts_date_string_to_excel_date(tmp_path: Path) -> None:
destination = tmp_path / "report.xlsx"
rows = [
@ -83,7 +95,7 @@ def test_write_report_converts_date_string_to_excel_date(tmp_path: Path) -> None
date_col_idx = FIXED_REPORT_COLUMNS.index("Дата сообщения") + 1
cell = sheet.cell(row=3, column=date_col_idx)
assert isinstance(cell.value, datetime)
assert cell.value.date() == datetime(2005, 8, 23).date()
assert cell.value.date() == date(2005, 8, 23)
assert cell.number_format == "DD.MM.YYYY"