""" CSV Export Utilities - CSV 数据导出工具 Shared helpers for streaming CSV responses. """ import csv import io from collections.abc import AsyncIterator, Sequence from datetime import datetime from typing import Any def _format_value(value: Any) -> str: """Format a single value for CSV output.""" if value is None: return "" if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, bool): return "是" if value else "否" if isinstance(value, float): return f"{value:.6f}" if abs(value) < 1000 else f"{value:.2f}" if isinstance(value, (dict, list)): import json return json.dumps(value, ensure_ascii=False) return str(value) def build_csv_content( headers: list[str], rows: Sequence[Any], field_extractors: list[Any], ) -> str: """Build complete CSV string with BOM for Excel compatibility. Args: headers: Column header names (Chinese). rows: ORM model instances or row tuples. field_extractors: List of callables or attribute name strings. """ buf = io.StringIO() buf.write("\ufeff") # UTF-8 BOM for Excel writer = csv.writer(buf) writer.writerow(headers) for row in rows: values = [] for ext in field_extractors: if callable(ext): values.append(_format_value(ext(row))) else: values.append(_format_value(getattr(row, ext, ""))) writer.writerow(values) return buf.getvalue() def csv_filename(prefix: str) -> str: """Generate a timestamped CSV filename.""" ts = datetime.now().strftime("%Y%m%d_%H%M%S") return f"{prefix}_{ts}.csv"