60 lines
1.7 KiB
Python
60 lines
1.7 KiB
Python
|
"""
|
||
|
|
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"
|