""" 业绩报表导出服务模块。 职责: 将业绩统计数据导出为 CSV 或 Excel(xlsx)格式的文件,供用户下载。 负责文件命名、格式校验、内容写入和格式美化。 依赖: - openpyxl:Excel 文件生成 - csv:标准库 CSV 写入 被引用方: - ReportService.export_performance():调用本服务生成导出文件 """ import csv from datetime import datetime from pathlib import Path from openpyxl import Workbook class ExportService: """业绩报表导出服务。 根据传入的业绩报表数据,生成 CSV 或 Excel 格式的导出文件, 并返回文件元信息(路径、大小、MIME 类型等)供上层服务上传到 OSS。 导出文件存储在本地 /tmp/order-flow-exports 目录下。 """ def __init__(self) -> None: from backend.app.core.config import get_settings settings = get_settings() # 复用 LOCAL_UPLOAD_DIR 的父目录,保持路径风格一致 self.export_root = Path(settings.local_upload_dir).parent / "exports" self.export_root.mkdir(parents=True, exist_ok=True) def build_performance_export(self, report: dict, export_format: str = "csv") -> dict: """根据业绩报表数据生成导出文件。 根据指定的导出格式(csv 或 xlsx),将报表数据写入本地文件, 并返回文件元信息供上层服务上传到 OSS。 参数: report (dict): 业绩报表数据,包含 stat_type(统计类型)和 list(数据列表) export_format (str): 导出格式,支持 "csv" 和 "xlsx",默认 "csv" 返回: dict: 包含 file_name、file_path、file_size、content_type、 object_key、export_format 等字段 被调用方: - ReportService.export_performance() → 业绩报表导出 API """ timestamp = datetime.now().strftime("%Y%m%d%H%M%S") suffix = report["stat_type"] normalized_format = self._normalize_export_format(export_format) file_name = f"performance-{suffix}-{timestamp}.{normalized_format}" file_path = self.export_root / file_name if normalized_format == "xlsx": self._write_xlsx(file_path, report) content_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" else: self._write_csv(file_path, report) content_type = "text/csv" return { "file_name": file_name, "file_path": str(file_path), "file_size": file_path.stat().st_size, "content_type": content_type, "object_key": f"exports/performance/{file_name}", "export_format": normalized_format, } def _normalize_export_format(self, export_format: str) -> str: """校验并标准化导出格式。 将传入的格式字符串转为小写并去除空白,若不在支持的格式列表中 则默认回退为 csv。 参数: export_format (str): 用户指定的导出格式 返回: str: 标准化后的格式字符串("csv" 或 "xlsx") """ normalized = (export_format or "csv").strip().lower() if normalized not in {"csv", "xlsx"}: return "csv" return normalized def _write_csv(self, file_path: Path, report: dict) -> None: """将报表数据写入 CSV 文件。 使用 UTF-8-BOM 编码写入,确保 Excel 打开时中文不乱码。 第一行为表头,后续每行对应一条统计周期数据。 参数: file_path (Path): 目标 CSV 文件路径 report (dict): 报表数据,包含 list 列表 """ with file_path.open("w", encoding="utf-8-sig", newline="") as csvfile: writer = csv.writer(csvfile) writer.writerow(["统计周期", "订单数", "订单金额", "固定提成", "分类明细"]) for item in report["list"]: writer.writerow(self._build_row(item)) writer.writerow(self._build_summary_row(report["list"])) def _write_xlsx(self, file_path: Path, report: dict) -> None: """将报表数据写入 Excel 文件。 使用 openpyxl 创建工作簿,写入表头和数据行,并对表头加粗、 列宽做最小可读性优化,避免客户打开后像原始数据转储。 参数: file_path (Path): 目标 xlsx 文件路径 report (dict): 报表数据,包含 list 列表 """ workbook = Workbook() sheet = workbook.active sheet.title = "业绩统计" sheet.append(["统计周期", "订单数", "订单金额", "固定提成", "分类明细"]) for item in report["list"]: sheet.append(self._build_row(item)) summary = self._build_summary_row(report["list"]) sheet.append(summary) # 给导出表加最小可读性优化,避免客户打开后仍像原始数据转储。 for cell in sheet[1]: cell.font = cell.font.copy(bold=True) # 合计行加粗 last_row = sheet.max_row for cell in sheet[last_row]: cell.font = cell.font.copy(bold=True) sheet.column_dimensions["A"].width = 18 sheet.column_dimensions["B"].width = 12 sheet.column_dimensions["C"].width = 16 sheet.column_dimensions["D"].width = 16 sheet.column_dimensions["E"].width = 48 workbook.save(file_path) def _build_summary_row(self, items: list) -> list: """构建合计行。""" total_count = sum(item["order_count"] for item in items) total_amount = sum(item["order_amount"] for item in items) total_commission = sum(item["commission_amount"] for item in items) return [ "合计", total_count, f"{total_amount:.2f}", f"{total_commission:.2f}", "", ] def _build_row(self, item: dict) -> list: """将单条报表数据项转换为表格行。 将分类明细的金额列表用中文分号连接为可读字符串, 数值字段格式化为保留两位小数的字符串。 参数: item (dict): 单条报表数据,包含 stat_period、order_count、 order_amount、commission_amount、category_amounts 返回: list: 包含 [统计周期, 订单数, 订单金额, 固定提成, 分类明细] 的列表 """ category_text = ";".join( f"{category['category_name']}:{category['amount']:.2f}" for category in item.get("category_amounts", []) ) return [ item["stat_period"], item["order_count"], f"{item['order_amount']:.2f}", f"{item['commission_amount']:.2f}", category_text, ] export_service = ExportService()