Files
webfortilog/app/services/exporter.py
2026-04-24 14:40:32 +02:00

70 lines
2.0 KiB
Python

import csv
import io
from dataclasses import dataclass
from app.constants import VENDOR_FIELDS
@dataclass(slots=True)
class ExportResult:
content: str
columns: list[str]
output_format: str
def preview(self, record_limit: int) -> str:
"""Build a small preview string for the result page."""
if self.output_format == "text":
marker = f"--- record {record_limit + 1} ---"
if marker in self.content:
return self.content.split(marker, 1)[0].rstrip()
return self.content
lines = self.content.splitlines()
if len(lines) <= record_limit + 1:
return self.content
return "\n".join(lines[: record_limit + 1])
def build_export(
records: list[dict[str, str]],
union_keys: list[str],
mode: str,
output_format: str,
) -> ExportResult:
columns = VENDOR_FIELDS if mode == "vendor" else union_keys
if output_format == "text":
return ExportResult(
content=_render_text(records, columns),
columns=columns,
output_format=output_format,
)
return ExportResult(
content=_render_csv(records, columns),
columns=columns,
output_format=output_format,
)
def _render_text(records: list[dict[str, str]], columns: list[str]) -> str:
max_key_length = max((len(column) for column in columns), default=0)
chunks: list[str] = []
for index, record in enumerate(records, start=1):
chunks.append(f"--- record {index} ---")
for column in columns:
value = record.get(column, "")
chunks.append(f" {column.ljust(max_key_length)} = {value}")
return "\n".join(chunks)
def _render_csv(records: list[dict[str, str]], columns: list[str]) -> str:
buffer = io.StringIO()
writer = csv.DictWriter(buffer, fieldnames=columns, extrasaction="ignore")
writer.writeheader()
for record in records:
writer.writerow({column: record.get(column, "") for column in columns})
return buffer.getvalue()