from __future__ import annotations import re import shutil from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from tempfile import NamedTemporaryFile from typing import Iterable import xlsxwriter from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row @dataclass class ReportRow: file_name: str record_id: str operation_index: int operation_fields: dict[str, str] participant_fields: dict[str, str] @dataclass(frozen=True) class GroupHeader: start_col: int end_col: int title: str # Верхняя группирующая шапка согласно согласованным логическим категориям. TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = ( GroupHeader(start_col=1, end_col=1, title="Служебные поля"), GroupHeader(start_col=2, end_col=6, title="Реквизиты сообщения/КО/филиала"), 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=( "Колонки вне согласованной группировки (если количество полей расширится)" ), ), ) 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() 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"}) self._column_formats = [self._resolve_format(column) for column in self.columns] self._build_group_header_row() self._build_header_row() # Фиксируем обе строки шапки. 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 = build_fixed_row( 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, ) for index, column_name in enumerate(self.columns): raw_value = data.get(column_name, "") default_format = self._column_formats[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, index, value, cell_format ) elif self._is_text_identifier_column(column_name): self.sheet.write_string( self._next_data_row, index, self._normalize_identifier_string(value), ) elif cell_format is not None: self.sheet.write(self._next_data_row, index, value, cell_format) else: self.sheet.write(self._next_data_row, index, value) self._next_data_row += 1 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 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, value: object) -> str: if value is None: return "" text = str(value).strip() if re.fullmatch(r"\d+\.0+", text): return text.split(".", maxsplit=1)[0] return text 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) except ValueError: continue return None def _cleanup_temp_file(self) -> None: try: self._temp_report_path.unlink(missing_ok=True) except Exception: # noqa: BLE001 pass 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._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)