156 lines
5.1 KiB
Python
156 lines
5.1 KiB
Python
"""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",
|
|
]
|