"""报表 Excel 模板构建(纯函数,可单测)。 版式(FR-032「导出模板围绕财务复核和领导签字设计」): 公司信息页眉 → 报表标题 → 筛选条件 → 导出时间/人/模板版本 → 表头 → 数据行 → 合计行 → 财务复核栏 / 领导签字栏 / 日期栏。 精度红线:金额写 ``Decimal``(openpyxl 原生支持,不经 float)+ ``number_format='#,##0.00'``;合计在 Python 侧求和后两位收口,**不写 Excel 公式**(避免打开时重算漂移与公式注入面)。文本列以 ``= + - @`` 开头时前置单引号,防公式注入。 """ from dataclasses import dataclass from datetime import datetime from decimal import ROUND_HALF_UP, Decimal from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from app.core.report_meta import CUSTOMER_TYPE_LABELS, PRODUCT_CATEGORY_LABELS _HEADER_FILL = PatternFill("solid", fgColor="DDE3F0") _THIN = Side(style="thin", color="B0B7C3") _BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN) _TITLE_FONT = Font(size=14, bold=True) _COMPANY_FONT = Font(size=16, bold=True) _HEADER_FONT = Font(bold=True) _AMOUNT_FORMAT = "#,##0.00" _DATE_FORMAT = "YYYY-MM-DD" _FORMULA_PREFIXES = ("=", "+", "-", "@", "\t") BUSINESS_STATUS_LABELS = {1: "进行中", 2: "已完成", 3: "已终止"} @dataclass(frozen=True, slots=True) class _Column: """列定义:key 取行 dict 字段,kind 决定单元格类型与格式。""" key: str title: str kind: str # text / amount / int / date / customer / category / status width: float _COLUMNS: dict[str, tuple[_Column, ...]] = { "personal-detail": ( _Column("business_no", "合同编号", "text", 20), _Column("ship_date", "发货日", "date", 12), _Column("receipt_date", "回款日", "date", 12), _Column("payment_days", "回款天数", "int", 10), _Column("product_category", "产品类别", "category", 14), _Column("customer_type", "客户类型", "customer", 14), _Column("role_name", "岗位", "text", 12), _Column("rate_text", "计提比例", "text", 14), _Column("weight_r", "支付权重R", "int", 10), _Column("performance_amount", "绩效金额", "amount", 14), _Column("payable_amount", "应付绩效", "amount", 14), ), "department-summary": ( _Column("manager_name", "经理团队", "text", 16), _Column("member_name", "成员", "text", 14), _Column("role_code", "岗位编码", "text", 12), _Column("performance_sum", "绩效金额", "amount", 14), _Column("payable_sum", "应付绩效", "amount", 14), _Column("adjust_net", "调整净额", "amount", 14), _Column("net_total", "净合计", "amount", 14), ), "business-stats": ( _Column("business_no", "业务编号", "text", 20), _Column("business_date", "业务日期", "date", 12), _Column("customer_name", "客户", "text", 22), _Column("owner_personnel_name", "归属人", "text", 12), _Column("status", "状态", "status", 10), _Column("total_sale_amount", "销售额", "amount", 14), _Column("total_ship_amount", "发货额", "amount", 14), _Column("total_received_amount", "回款额", "amount", 14), _Column("performance_amount", "绩效金额", "amount", 14), _Column("payable_amount", "应付绩效", "amount", 14), ), "payment-stats": ( _Column("label", "回款天数区间", "text", 20), _Column("receipt_amount", "回款金额", "amount", 16), _Column("receipt_count", "回款笔数", "int", 10), _Column("performance_amount", "绩效金额", "amount", 14), _Column("payable_amount", "应付绩效", "amount", 14), _Column("ledger_count", "台账行数", "int", 10), ), } def build_workbook( *, code: str, title: str, filters: str, rows: list[dict], summary: dict | None, buckets: dict | None, template_version: str, operator_name: str, company_name: str, ) -> bytes: """构建并返回 xlsx 字节流(业务数据只读,不触碰数据库)。""" workbook = Workbook() sheet = workbook.active sheet.title = _sheet_name(code) columns = _COLUMNS[code] _write_header(sheet, columns, company_name, title, filters, template_version, operator_name) if rows: for row in rows: _write_row(sheet, columns, row) else: sheet.append(["本筛选条件无数据"]) _merge_row(sheet, len(columns)) _write_total(sheet, columns, rows) _write_signature(sheet, len(columns)) _autofit(sheet, columns) # 回款统计附维度对比 sheet(客户类型/产品类别),与区间表共用页眉格式 if code == "payment-stats" and buckets: for dim_key, dim_title in ( ("by_customer_type", "客户类型对比"), ("by_product_category", "产品类别对比"), ): extra = workbook.create_sheet(dim_title) _write_dimension_sheet( extra, dim_title, company_name, buckets.get(dim_key) or [], template_version, operator_name, ) buffer = BytesIO() workbook.save(buffer) return buffer.getvalue() def _sheet_name(code: str) -> str: return { "personal-detail": "个人绩效明细", "department-summary": "部门绩效汇总", "business-stats": "业务统计", "payment-stats": "回款统计", }[code] def _write_header( sheet, columns: tuple[_Column, ...], company_name: str, title: str, filters: str, template_version: str, operator_name: str, ) -> None: """页眉四行 + 表头行;返回后 sheet.max_row 即数据行起点。""" sheet.append([company_name]) _merge_row(sheet, len(columns)) sheet.cell(row=1, column=1).font = _COMPANY_FONT sheet.cell(row=1, column=1).alignment = Alignment(horizontal="left", vertical="center") sheet.append([title]) _merge_row(sheet, len(columns)) sheet.cell(row=2, column=1).font = _TITLE_FONT sheet.cell(row=2, column=1).alignment = Alignment(horizontal="center", vertical="center") sheet.append([f"筛选条件:{filters}"]) _merge_row(sheet, len(columns)) sheet.append( [ f"导出时间:{datetime.now():%Y-%m-%d %H:%M:%S} | 导出人:{operator_name} | " f"模板版本:{template_version}" ] ) _merge_row(sheet, len(columns)) sheet.append([]) sheet.append([column.title for column in columns]) header_row = sheet.max_row for cell in sheet[header_row]: cell.font = _HEADER_FONT cell.fill = _HEADER_FILL cell.border = _BORDER cell.alignment = Alignment(horizontal="center", vertical="center") sheet.freeze_panes = sheet.cell(row=header_row + 1, column=1) def _merge_row(sheet, column_count: int) -> None: """合并当前行(公司名/标题/筛选行跨全表宽)。""" row = sheet.max_row if column_count > 1: sheet.merge_cells(start_row=row, start_column=1, end_row=row, end_column=column_count) def _write_row(sheet, columns: tuple[_Column, ...], row: dict) -> None: """数据行:按列类型写单元格(金额 Decimal、日期 date、文本防注入)。""" values = [] for column in columns: raw = row.get(column.key) values.append(_cell_value(column, raw)) sheet.append(values) for index, column in enumerate(columns, start=1): cell = sheet.cell(row=sheet.max_row, column=index) cell.border = _BORDER if column.kind == "amount": cell.number_format = _AMOUNT_FORMAT elif column.kind == "date": cell.number_format = _DATE_FORMAT def _cell_value(column: _Column, raw): """单值转换:标签列转中文,文本列防公式注入。""" if raw is None: return None if column.kind == "amount": return raw if isinstance(raw, Decimal) else Decimal(str(raw)) if column.kind == "date": return raw if column.kind == "int": return int(raw) if column.kind == "customer": return CUSTOMER_TYPE_LABELS.get(str(raw), str(raw)) if column.kind == "category": return PRODUCT_CATEGORY_LABELS.get(str(raw), str(raw)) if column.kind == "status": return BUSINESS_STATUS_LABELS.get(int(raw), str(raw)) return _safe_text(raw) def _safe_text(value) -> str: """文本防公式注入:以 = + - @ 或制表符开头时前置单引号。""" text = str(value) return "'" + text if text.startswith(_FORMULA_PREFIXES) else text def _write_total(sheet, columns: tuple[_Column, ...], rows: list[dict]) -> None: """合计行:金额列按行求和后两位收口(不写 Excel 公式)。 「合计」标签只写在第一个非金额列;其余非金额列留空——否则每遇到 一个文本列都会重复写一次标签,合计行会满屏「合计」。 """ if not rows: return totals = [] label_written = False for column in columns: if column.kind != "amount": totals.append("合计" if not label_written else None) label_written = True continue total = sum( ( row.get(column.key) for row in rows if isinstance(row.get(column.key), (Decimal, int)) ), Decimal("0"), ) totals.append(total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) sheet.append(totals) for index, column in enumerate(columns, start=1): cell = sheet.cell(row=sheet.max_row, column=index) cell.font = _HEADER_FONT cell.border = _BORDER if column.kind == "amount": cell.number_format = _AMOUNT_FORMAT def _write_signature(sheet, column_count: int) -> None: """财务复核栏 / 领导签字栏 / 日期栏(FR-032 模板要素)。""" sheet.append([]) sheet.append(["财务复核:________________", "", "日期:______年__月__日"]) sheet.append(["领导签字:________________", "", "日期:______年__月__日"]) for row_index in (sheet.max_row - 1, sheet.max_row): sheet.cell(row=row_index, column=1).alignment = Alignment(horizontal="left") def _autofit(sheet, columns: tuple[_Column, ...]) -> None: """列宽按定义表设置(CJK 按 2 字符宽,不用默认 8.43)。""" for index, column in enumerate(columns, start=1): sheet.column_dimensions[get_column_letter(index)].width = column.width def _write_dimension_sheet( sheet, title: str, company_name: str, rows: list[dict], template_version: str, operator_name: str, ) -> None: """维度对比 sheet(客户类型/产品类别)。""" columns = ( _Column("label", "维度", "text", 16), _Column("receipt_amount", "回款金额", "amount", 16), _Column("performance_amount", "绩效金额", "amount", 14), _Column("payable_amount", "应付绩效", "amount", 14), _Column("ledger_count", "台账行数", "int", 10), ) _write_header(sheet, columns, company_name, title, "全部", template_version, operator_name) for row in rows: _write_row(sheet, columns, row) _write_total(sheet, columns, rows) _write_signature(sheet, len(columns)) _autofit(sheet, columns)