shanxia2.0/backend/app/api/bx/export_common.py

"""bx 列表 Excel 导出公共辅助。"""
from __future__ import annotations

from datetime import date, datetime
from typing import Any, List, Mapping, Optional, Sequence

from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session

from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)

BX_STATUS_LABELS = {
    "draft": "待提交",
    "pending": "审批中",
    "approved": "审批通过",
    "rejected": "审批驳回",
    "issued": "已发放",
    "paid": "已发放",
    "offset": "已冲抵",
    "processed": "已处理",
}

ISSUE_STATUS_LABELS = {
    "issued": "已发放",
    "pending": "待发放",
}


class BxExportBodyBase(BaseModel):
    model_config = ConfigDict(populate_by_name=True, extra="allow")

    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None


class BxApplyListExportBody(BxExportBodyBase):
    model_config = ConfigDict(populate_by_name=True, extra="allow")

    status: Optional[str] = None
    keyword: str = ""
    applicant_id: Optional[str] = Field(default=None, alias="applicantId")
    apply_date: Optional[str] = Field(default=None, alias="applyDate")
    apply_date_from: Optional[str] = Field(default=None, alias="applyDateFrom")
    apply_date_to: Optional[str] = Field(default=None, alias="applyDateTo")
    expense_id: Optional[str] = Field(default=None, alias="expenseId")
    expense_name: Optional[str] = Field(default=None, alias="expenseName")
    apply_code: Optional[str] = Field(default=None, alias="applyCode")
    title: Optional[str] = None
    current_node: Optional[str] = Field(default=None, alias="currentNode")
    status_in: Optional[str] = Field(default=None, alias="statusIn")
    amount_op: Optional[str] = Field(default=None, alias="amountOp")
    amount: Optional[str] = None
    amount_min: Optional[str] = Field(default=None, alias="amountMin")
    amount_max: Optional[str] = Field(default=None, alias="amountMax")
    reason: Optional[str] = None
    applicant_name: Optional[str] = Field(default=None, alias="applicantName")
    actual_amount_op: Optional[str] = Field(default=None, alias="actualAmountOp")
    actual_amount: Optional[str] = Field(default=None, alias="actualAmount")
    actual_amount_min: Optional[str] = Field(default=None, alias="actualAmountMin")
    actual_amount_max: Optional[str] = Field(default=None, alias="actualAmountMax")
    offset_amount_op: Optional[str] = Field(default=None, alias="offsetAmountOp")
    offset_amount: Optional[str] = Field(default=None, alias="offsetAmount")
    offset_amount_min: Optional[str] = Field(default=None, alias="offsetAmountMin")
    offset_amount_max: Optional[str] = Field(default=None, alias="offsetAmountMax")
    issue_time_from: Optional[str] = Field(default=None, alias="issueTimeFrom")
    issue_time_to: Optional[str] = Field(default=None, alias="issueTimeTo")
    issue_user_id: Optional[str] = Field(default=None, alias="issueUserId")
    issue_user_name: Optional[str] = Field(default=None, alias="issueUserName")
    issue_by: Optional[str] = Field(default=None, alias="issueBy")
    sort_field: Optional[str] = Field(default=None, alias="sortField")
    sort_order: Optional[str] = Field(default=None, alias="sortOrder")


def status_label(status: Any) -> str:
    if status is None:
        return ""
    if status in (0, "0"):
        return "禁用"
    if status in (1, "1"):
        return "启用"
    key = str(status)
    return BX_STATUS_LABELS.get(key, key)


def issue_status_label(status: Any) -> str:
    if status == "issued":
        return "已发放"
    return "待发放"


def dump_export_row(
    obj: Any,
    *,
    status_fields: Sequence[str] = ("status",),
    extra: Optional[Mapping[str, Any]] = None,
) -> dict:
    if hasattr(obj, "model_dump"):
        data = obj.model_dump(by_alias=True)
    elif isinstance(obj, Mapping):
        data = dict(obj)
    else:
        data = dict(obj)
    for key, value in list(data.items()):
        if isinstance(value, datetime):
            data[key] = value.strftime("%Y-%m-%d %H:%M:%S")
        elif isinstance(value, date):
            data[key] = value.isoformat()
    for field in status_fields:
        if field in data:
            data[field] = status_label(data.get(field))
    if extra:
        data.update(extra)
    return data


def respond_xlsx(
    db: Session,
    *,
    tenant_no: str,
    sheet_title: str,
    columns: Sequence[ExportColumn],
    rows: Sequence[Mapping[str, Any]],
    filename_stem: str,
    requested_columns: Optional[Sequence[str]] = None,
):
    cols = filter_export_columns(columns, requested_columns)
    date_str = datetime.now().strftime("%Y%m%d")
    return export_rows_to_xlsx_response(
        db,
        tenant_no=tenant_no,
        sheet_title=sheet_title,
        columns=cols,
        rows=rows,
        filename_stem=f"{filename_stem}_{date_str}",
    )


__all__ = [
    "EXPORT_MAX_ROWS",
    "ExportColumn",
    "ExportValidationError",
    "BxExportBodyBase",
    "BxApplyListExportBody",
    "status_label",
    "issue_status_label",
    "dump_export_row",
    "respond_xlsx",
    "ISSUE_STATUS_LABELS",
]