346 lines
12 KiB
Python
346 lines
12 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
|
||
|
||
|
||
TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = tuple(
|
||
GroupHeader(
|
||
start_col=group.start_index,
|
||
end_col=group.end_index,
|
||
title=group.title,
|
||
)
|
||
for group in COLUMNS.report_groups()
|
||
)
|
||
|
||
|
||
_WIDE_TEXT_KEYWORDS = (
|
||
"Наименование",
|
||
"Адрес",
|
||
"Назначение платежа",
|
||
"Доп. сведения",
|
||
"Дополнительн",
|
||
"Сведения",
|
||
"Иные сведения",
|
||
"ФИО",
|
||
"Место государственной регистрации",
|
||
"Орган, выдавший документ",
|
||
"Сведения миграционной карты",
|
||
)
|
||
_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": "vcenter",
|
||
"text_wrap": True,
|
||
"bg_color": "#E2F0D9",
|
||
}
|
||
)
|
||
self._group_format = self.workbook.add_format(
|
||
{
|
||
"bold": True,
|
||
"align": "center",
|
||
"valign": "vcenter",
|
||
"text_wrap": True,
|
||
"bg_color": "#FFFFFF",
|
||
}
|
||
)
|
||
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,
|
||
),
|
||
)
|
||
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
|
||
if first_col == last_col:
|
||
self.sheet.write(0, first_col, header.title, self._group_format)
|
||
continue
|
||
self.sheet.merge_range(
|
||
0,
|
||
first_col,
|
||
0,
|
||
last_col,
|
||
header.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
|
||
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 _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)
|