zlwuandCursor
833544def3
Previous reimbursement export commit depended on these files but they were left unstaged. Co-authored-by: Cursor <cursoragent@cursor.com>
129 lines
3.3 KiB
Python
129 lines
3.3 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)
|
|
|
|
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")
|
|
|
|
|
|
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",
|
|
]
|