beeb190601
- Draw a grid on the two header rows only, data cells stay borderless. - Align column names to the top so the start of a long name stays visible. - Narrow the 14 code columns to the length of their values; the rule matches by column name, so identical indicators in every block get equal widths. - Halve both payment system name columns (32 -> 16). - Give "Характеристика операции" the same width as "Назначение платежа". - Drop the group title above "Статус перевода": it duplicated the column name. The group cell stays in place so the header grid is not broken. - Pass the cell format to write_string. Identifier columns (БИК, ИНН, ОГРН, КПП, СВИФТ, account numbers) were written without one, so text wrapping and vertical alignment never applied to them - this is why БИК had no line break. The width padding and the header rotation are single constants: the customer has not confirmed yet whether "N знаков" means the value length (implemented) or an amount to subtract, and whether narrow headers should be rotated.
412 lines
16 KiB
Python
412 lines
16 KiB
Python
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
|
||
|
||
import xlsxwriter
|
||
|
||
from .column_constants import COL_KPP, COL_PARTICIPANT_TYPE, PARTICIPANT_TYPE_UL
|
||
from .column_registry import COLUMNS
|
||
from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row_by_index
|
||
|
||
|
||
@dataclass
|
||
class ReportRow:
|
||
file_name: str
|
||
record_id: str
|
||
operation_index: int
|
||
operation_fields: dict[str, str]
|
||
participant_fields: dict[str, str]
|
||
is_continuation: bool = False
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GroupHeader:
|
||
start_col: int
|
||
end_col: int
|
||
title: str
|
||
group_id: str = ""
|
||
|
||
|
||
TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = tuple(
|
||
GroupHeader(
|
||
start_col=group.start_index,
|
||
end_col=group.end_index,
|
||
title=group.title,
|
||
group_id=group.group_id,
|
||
)
|
||
for group in COLUMNS.report_groups()
|
||
)
|
||
|
||
# Группы верхнего уровня, у которых заказчик просил убрать наименование: над
|
||
# колонкой «Статус перевода» заголовок группы дублировал название самой
|
||
# колонки. Ячейка группы остаётся на месте (с рамкой), но без текста.
|
||
SUPPRESSED_GROUP_IDS = frozenset({"transfer_status"})
|
||
|
||
|
||
_WIDE_TEXT_KEYWORDS = (
|
||
"Наименование",
|
||
"Адрес",
|
||
"Назначение платежа",
|
||
"Характеристика операции",
|
||
"Доп. сведения",
|
||
"Дополнительн",
|
||
"Сведения",
|
||
"Иные сведения",
|
||
"ФИО",
|
||
"Место государственной регистрации",
|
||
"Орган, выдавший документ",
|
||
"Сведения миграционной карты",
|
||
)
|
||
# Требования заказчика от 24.09.2026: у перечисленных показателей колонка
|
||
# сужается до ширины содержимого. Ключ — подстрока имени колонки (сравнение без
|
||
# учёта регистра), значение — длина значения в знаках: числа из списка заказчика
|
||
# совпадают с фактическими данными (ОКАТО «08», код страны «643», признак
|
||
# валютной операции «VO» и т.д.). Правило применяется ко ВСЕМ одноимённым
|
||
# колонкам во всех блоках отчёта, чтобы одинаковые показатели были одинаковой
|
||
# ширины. Порядок значим: побеждает первое совпадение, поэтому более узкие
|
||
# формулировки идут выше — «Признак валютной операции» это 2 знака, а не 3 как
|
||
# у самой валюты.
|
||
_CHARACTER_WIDTH_RULES: tuple[tuple[str, int], ...] = (
|
||
("признак валютной операции", 2),
|
||
("признак операции, связанной", 1),
|
||
("совершения оп. с испол. эсп", 1),
|
||
("тип информации", 1),
|
||
("номер филиала", 2),
|
||
("окато", 2),
|
||
("код страны", 3),
|
||
("валют", 3),
|
||
("вид перевода", 1),
|
||
("код территории", 3),
|
||
("тип оператора", 1),
|
||
("статус перевода", 1),
|
||
("признак дул", 1),
|
||
("код вида документа", 2),
|
||
)
|
||
# Ширины, заданные не «в знаках», а напрямую: обе колонки наименования
|
||
# платёжной системы сужены вдвое от прежних 32.
|
||
_FIXED_WIDTH_RULES: tuple[tuple[str, float], ...] = (
|
||
("наименование плат. системы", 16),
|
||
)
|
||
# Excel отводит часть ширины колонки под отступы и кнопку автофильтра, поэтому
|
||
# к длине значения добавляем небольшой запас: иначе значение в колонке «на один
|
||
# знак» не видно вообще.
|
||
_WIDTH_PADDING = 2
|
||
_MIN_COLUMN_WIDTH = 3
|
||
# Поворот названий колонок в шапке, в градусах: 0 — перенос по словам, 90 —
|
||
# вертикальный текст. Заказчик просил перенос с выравниванием по верхнему краю,
|
||
# но в узких колонках название почти не читается, поэтому переключение на
|
||
# вертикальную шапку оставлено одной константой.
|
||
HEADER_TEXT_ROTATION = 0
|
||
_NARROW_KEYWORDS = (
|
||
"Код",
|
||
"БИК",
|
||
"ИНН",
|
||
"ОГРН",
|
||
"КПП",
|
||
"Индекс",
|
||
"Дом",
|
||
"Корпус",
|
||
"Офис",
|
||
"Квартира",
|
||
"Серия",
|
||
"Тип",
|
||
"Признак",
|
||
"ОКАТО",
|
||
"Валюта",
|
||
"СВИФТ",
|
||
)
|
||
|
||
|
||
class StreamingReportWriter:
|
||
def __init__(self) -> None:
|
||
self.columns = list(FIXED_REPORT_COLUMNS)
|
||
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},
|
||
)
|
||
self._is_closed = False
|
||
self.sheet = self.workbook.add_worksheet("Отчет")
|
||
self._header_format = self.workbook.add_format(
|
||
{
|
||
"bold": True,
|
||
"align": "center",
|
||
"valign": "top",
|
||
"text_wrap": True,
|
||
"bg_color": "#E2F0D9",
|
||
"border": 1,
|
||
"rotation": HEADER_TEXT_ROTATION,
|
||
}
|
||
)
|
||
self._group_format = self.workbook.add_format(
|
||
{
|
||
"bold": True,
|
||
"align": "center",
|
||
"valign": "vcenter",
|
||
"text_wrap": True,
|
||
"bg_color": "#FFFFFF",
|
||
"border": 1,
|
||
}
|
||
)
|
||
self._date_format = self.workbook.add_format({"num_format": "DD.MM.YYYY"})
|
||
self._sum_format = self.workbook.add_format({"num_format": "#,##0.00"})
|
||
# Дефолтный формат для обычных текстовых ячеек — раньше такие ячейки
|
||
# писались вообще без формата (cell_format is None), поэтому перенос
|
||
# текста применялся только к шапке, а не к данным.
|
||
self._text_format = self.workbook.add_format(
|
||
{"text_wrap": True, "valign": "top"}
|
||
)
|
||
self._column_formats = [self._resolve_format(column) for column in self.columns]
|
||
self._build_group_header_row()
|
||
# В режиме constant_memory строка "уходит" на диск сразу после того,
|
||
# как начата следующая — поэтому set_row для строки 0 нужно вызывать
|
||
# ДО построения строки 1, иначе вызов молча ничего не сделает.
|
||
self.sheet.set_row(0, 22)
|
||
self._build_header_row()
|
||
self.sheet.set_row(1, 46)
|
||
for column_index, column_name in enumerate(self.columns):
|
||
column_format = self._column_formats[column_index]
|
||
self.sheet.set_column(
|
||
column_index,
|
||
column_index,
|
||
self._resolve_column_width(column_name, column_format),
|
||
)
|
||
# Шапка из двух строк уже настроена выше (до перехода на следующую
|
||
# строку — см. комментарий про constant_memory).
|
||
# Фиксируем обе строки шапки.
|
||
self.sheet.freeze_panes(2, 0)
|
||
self.sheet.autofilter(1, 0, 1, len(self.columns) - 1)
|
||
self._next_data_row = 2
|
||
|
||
def append_row(self, row: ReportRow) -> None:
|
||
data_by_index = build_fixed_row_by_index(
|
||
file_name=row.file_name,
|
||
record_id=row.record_id,
|
||
operation_index=row.operation_index,
|
||
operation_fields=row.operation_fields,
|
||
participant_fields=row.participant_fields,
|
||
is_continuation=row.is_continuation,
|
||
)
|
||
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,
|
||
default_format=default_format,
|
||
)
|
||
if isinstance(value, datetime) and cell_format is self._date_format:
|
||
self.sheet.write_datetime(
|
||
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,
|
||
zero_based_index,
|
||
self._normalize_identifier_string(
|
||
column_name,
|
||
value,
|
||
column_index=output_index,
|
||
row_by_index=data_by_index,
|
||
),
|
||
default_format,
|
||
)
|
||
elif cell_format is not None:
|
||
self.sheet.write(
|
||
self._next_data_row, zero_based_index, value, cell_format
|
||
)
|
||
else:
|
||
self.sheet.write(self._next_data_row, zero_based_index, value)
|
||
self._next_data_row += 1
|
||
|
||
def _resolve_output_value(
|
||
self,
|
||
column_index: int,
|
||
row_by_index: dict[int, str],
|
||
) -> object:
|
||
return row_by_index.get(column_index, "")
|
||
|
||
def save(self, destination: Path) -> None:
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
if not self._is_closed:
|
||
self.workbook.close()
|
||
self._is_closed = True
|
||
shutil.copyfile(self._temp_report_path, destination)
|
||
finally:
|
||
self._cleanup_temp_file()
|
||
|
||
def _build_group_header_row(self) -> None:
|
||
for header in TOP_GROUP_HEADERS:
|
||
if header.start_col > len(self.columns):
|
||
continue
|
||
first_col = header.start_col - 1
|
||
last_col = min(header.end_col, len(self.columns)) - 1
|
||
title = "" if header.group_id in SUPPRESSED_GROUP_IDS else header.title
|
||
if first_col == last_col:
|
||
self.sheet.write(0, first_col, title, self._group_format)
|
||
continue
|
||
self.sheet.merge_range(
|
||
0,
|
||
first_col,
|
||
0,
|
||
last_col,
|
||
title,
|
||
self._group_format,
|
||
)
|
||
|
||
def _build_header_row(self) -> None:
|
||
for index, name in enumerate(self.columns):
|
||
self.sheet.write(1, index, name, self._header_format)
|
||
|
||
def _resolve_format(self, column_name: str):
|
||
if "Дата" in column_name:
|
||
return self._date_format
|
||
if "Сумм" in column_name or "Сумма" in column_name:
|
||
return self._sum_format
|
||
return self._text_format
|
||
|
||
def _resolve_column_width(self, column_name: str, column_format) -> float:
|
||
if column_format is self._date_format:
|
||
return 12
|
||
if column_format is self._sum_format:
|
||
return 14
|
||
explicit_width = self._resolve_explicit_width(column_name)
|
||
if explicit_width is not None:
|
||
return explicit_width
|
||
if "Номер счета" in column_name:
|
||
return 20
|
||
if "Номер платежной карты" in column_name or "Номер банкомата" in column_name:
|
||
return 16
|
||
if any(keyword in column_name for keyword in _WIDE_TEXT_KEYWORDS):
|
||
return 32
|
||
if any(keyword in column_name for keyword in _NARROW_KEYWORDS):
|
||
return 11
|
||
return 14
|
||
|
||
def _resolve_explicit_width(self, column_name: str) -> float | None:
|
||
normalized = column_name.casefold()
|
||
for keyword, width in _FIXED_WIDTH_RULES:
|
||
if keyword in normalized:
|
||
return width
|
||
for keyword, characters in _CHARACTER_WIDTH_RULES:
|
||
if keyword in normalized:
|
||
return max(characters + _WIDTH_PADDING, _MIN_COLUMN_WIDTH)
|
||
return None
|
||
|
||
def _normalize_cell_value(
|
||
self,
|
||
*,
|
||
column_name: str,
|
||
raw_value: object,
|
||
default_format,
|
||
) -> tuple[object, object | None]:
|
||
if default_format is self._date_format:
|
||
parsed_date = self._parse_date_value(raw_value)
|
||
if parsed_date is not None:
|
||
return parsed_date, self._date_format
|
||
if raw_value in (None, ""):
|
||
return "", None
|
||
return raw_value, None
|
||
return raw_value, default_format
|
||
|
||
def _is_text_identifier_column(self, column_name: str) -> bool:
|
||
keywords = ("ИНН", "ОГРН", "КПП", "БИК", "СВИФТ", "Номер счета")
|
||
return any(keyword in column_name for keyword in keywords)
|
||
|
||
def _normalize_identifier_string(
|
||
self,
|
||
column_name: str,
|
||
value: object,
|
||
*,
|
||
column_index: int,
|
||
row_by_index: dict[int, str],
|
||
) -> str:
|
||
if value is None:
|
||
return ""
|
||
text = str(value).strip()
|
||
normalized = self._to_plain_integer_text(text)
|
||
if normalized is None:
|
||
return text
|
||
|
||
if "Номер счета" in column_name:
|
||
return normalized.zfill(20)
|
||
if "БИК" in column_name:
|
||
return normalized.zfill(9)
|
||
if "КПП" in column_name:
|
||
participant_type = str(row_by_index.get(COL_PARTICIPANT_TYPE, "")).strip()
|
||
normalized_type = participant_type.lstrip("0") or "0"
|
||
if column_index == COL_KPP and normalized_type != PARTICIPANT_TYPE_UL:
|
||
return normalized
|
||
return normalized.zfill(9)
|
||
if "ИНН" in column_name:
|
||
if len(normalized) in {9, 11, 4}:
|
||
return normalized.zfill(len(normalized) + 1)
|
||
return normalized
|
||
return normalized
|
||
|
||
def _to_plain_integer_text(self, text: str) -> str | None:
|
||
if re.fullmatch(r"\d+", text):
|
||
return text
|
||
if re.fullmatch(r"\d+\.0+", text):
|
||
return text.split(".", maxsplit=1)[0]
|
||
if re.fullmatch(r"\d+(?:\.\d+)?[eE][+-]?\d+", text):
|
||
try:
|
||
numeric = Decimal(text)
|
||
except InvalidOperation:
|
||
return None
|
||
if numeric != numeric.to_integral_value():
|
||
return None
|
||
return format(numeric, "f").split(".", maxsplit=1)[0]
|
||
return None
|
||
|
||
def _parse_date_value(self, value: object) -> datetime | None:
|
||
if isinstance(value, datetime):
|
||
return value
|
||
if isinstance(value, date):
|
||
return datetime.combine(value, datetime.min.time())
|
||
if not isinstance(value, str):
|
||
return None
|
||
normalized = value.strip()
|
||
if not normalized:
|
||
return None
|
||
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) # noqa: DTZ007
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
def _cleanup_temp_file(self) -> None:
|
||
try:
|
||
self._temp_report_path.unlink(missing_ok=True)
|
||
except Exception: # noqa: BLE001
|
||
return
|
||
|
||
def __del__(self) -> None:
|
||
if not getattr(self, "_is_closed", True):
|
||
try:
|
||
self.workbook.close()
|
||
except Exception: # noqa: BLE001
|
||
self._is_closed = True
|
||
else:
|
||
self._is_closed = True
|
||
self._cleanup_temp_file()
|
||
|
||
|
||
def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
|
||
writer = StreamingReportWriter()
|
||
for row in rows:
|
||
writer.append_row(row)
|
||
writer.save(destination)
|