certflow.handlers.csv_export 源代码

"""CSV 导出工具

提供打印日志等数据的 CSV 导出能力，使用 UTF-8 with BOM 编码，
保证 Excel 直接打开中文不乱码。流式逐行写入，适合大数据量。
"""

from __future__ import annotations

import csv
from collections.abc import Callable, Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from certflow.models.print_log import PrintLog


# 打印日志导出列定义：(中文表头, 取值函数)
PRINT_LOG_COLUMNS: list[tuple[str, Callable[[PrintLog], Any]]] = [
    ("ID", lambda log: log.id),
    ("出厂编号SN", lambda log: log.SN or ""),
    ("KKS编码", lambda log: log.KKS or ""),
    ("产品名称", lambda log: log.ChanPinMingCheng or ""),
    ("产品型号", lambda log: log.ChanPinXingHao or ""),
    ("打印机", lambda log: log.PrinterName or ""),
    ("状态", lambda log: log.status or ""),
    ("打印时间", lambda log: str(log.hgz_PrintTime or "")),
    ("内容摘要", lambda log: log.content_summary or ""),
]




def export_print_logs_csv(
    logs: Sequence[PrintLog],
    file_path: str,
    columns: Sequence[tuple[str, Callable[[PrintLog], Any]]] | None = None,
) -> int:
    """将打印日志导出为 CSV（UTF-8 BOM）

    逐行写入，避免一次性加载全部内容到内存。

    Args:
        logs: PrintLog 序列
        file_path: 输出文件路径
        columns: 列定义 [(中文表头, 取值函数)]；为 None 时使用默认 PRINT_LOG_COLUMNS

    Returns:
        int: 导出的数据行数（不含表头）
    """
    cols = columns or PRINT_LOG_COLUMNS
    path = Path(file_path)
    path.parent.mkdir(parents=True, exist_ok=True)

    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        writer.writerow([header for header, _ in cols])
        for log in logs:
            writer.writerow([getter(log) for _, getter in cols])

    return len(logs)