default
9cd9dd9d76
- 新增 DeviceBeaconBinding 模型，信标-设备多对多绑定 CRUD - 蓝牙打卡模式批量配置/恢复正常模式 API - 反向同步: 查询设备 BTMACSET 配置更新数据库绑定 (独立 session 解决事务隔离) - 设备列表快捷操作弹窗修复 (fire-and-forget IIFE 替代阻塞轮询) - 保存按钮防抖: 围栏/信标绑定保存点击后 disabled + 转圈防重复提交 - 审计日志中间件 + 系统配置/备份/固件 API - 设备分组管理 + 告警规则配置 - 5个数据导出 API (CSV UTF-8 BOM) - 位置热力图 + 告警条件删除 + 位置清理 via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run>
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"
|