PayrollManagementSystem/backend/app/services/report_export.py

"""报表 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)