"""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) ids: Optional[List[str]] = None columns: Optional[List[str]] = None class BxApplyListExportBody(BxExportBodyBase): 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") 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", ]