SFM/app/pipeline/report.py
2026-07-07 11:05:28 +03:00

84 lines
2.8 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
from openpyxl.styles import Alignment, Font, PatternFill
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]
class StreamingReportWriter:
def __init__(self) -> None:
self.columns = list(FIXED_REPORT_COLUMNS)
self._column_formats = [self._resolve_format(column) for column in self.columns]
self.workbook = Workbook(write_only=True)
self.sheet = self.workbook.create_sheet(title="Отчет")
self.sheet.freeze_panes = "A2"
self.sheet.append(self._build_header_row())
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,
)
values: list[object] = []
for index, column_name in enumerate(self.columns):
raw_value = data.get(column_name, "")
number_format = self._column_formats[index]
if number_format is None:
values.append(raw_value)
continue
cell = WriteOnlyCell(self.sheet, value=raw_value)
cell.number_format = number_format
values.append(cell)
self.sheet.append(values)
def save(self, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
self.workbook.save(destination)
def _build_header_row(self) -> list[WriteOnlyCell]:
fill = PatternFill(start_color="E2F0D9", end_color="E2F0D9", fill_type="solid")
result: list[WriteOnlyCell] = []
for name in self.columns:
cell = WriteOnlyCell(self.sheet, value=name)
cell.font = Font(bold=True)
cell.alignment = Alignment(
horizontal="center",
vertical="center",
wrap_text=True,
)
cell.fill = fill
result.append(cell)
return result
def _resolve_format(self, column_name: str) -> str | None:
if "Дата" in column_name:
return "DD.MM.YYYY"
if "Сумм" in column_name or "Сумма" in column_name:
return "#,##0.00"
return None
def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
writer = StreamingReportWriter()
for row in rows:
writer.append_row(row)
writer.save(destination)