SFM/app/pipeline/report.py
2026-07-06 13:27:10 +03:00

74 lines
2.4 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.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,
)
cells: list[WriteOnlyCell] = []
for column_name in self.columns:
cell = WriteOnlyCell(self.sheet, value=data.get(column_name, ""))
if "Дата" in column_name:
cell.number_format = "DD.MM.YYYY"
if "Сумм" in column_name or "Сумма" in column_name:
cell.number_format = "#,##0.00"
cells.append(cell)
self.sheet.append(cells)
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 write_report(rows: Iterable[ReportRow], destination: Path) -> None:
writer = StreamingReportWriter()
for row in rows:
writer.append_row(row)
writer.save(destination)