81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
import csv
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from openpyxl import Workbook
|
||
|
||
|
||
class ExportService:
|
||
def __init__(self) -> None:
|
||
self.export_root = Path("D:/tmp/order-flow-exports")
|
||
self.export_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
def build_performance_export(self, report: dict, export_format: str = "csv") -> dict:
|
||
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:
|
||
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:
|
||
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))
|
||
|
||
def _write_xlsx(self, file_path: Path, report: dict) -> None:
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "业绩统计"
|
||
sheet.append(["统计周期", "订单数", "订单金额", "固定提成", "分类明细"])
|
||
for item in report["list"]:
|
||
sheet.append(self._build_row(item))
|
||
# 给导出表加最小可读性优化，避免客户打开后仍像原始数据转储。
|
||
for cell in sheet[1]:
|
||
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_row(self, item: dict) -> 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()
|