From 833544def33c8c334583aee878214144d45bf833 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Fri, 21 Aug 2026 18:34:35 +0800 Subject: [PATCH] fix(bx): include export_common helper and advance export endpoints Previous reimbursement export commit depended on these files but they were left unstaged. Co-authored-by: Cursor --- backend/app/api/bx/advance.py | 193 ++++++++++++++++++++++++++++ backend/app/api/bx/export_common.py | 128 ++++++++++++++++++ 2 files changed, 321 insertions(+) create mode 100644 backend/app/api/bx/export_common.py diff --git a/backend/app/api/bx/advance.py b/backend/app/api/bx/advance.py index 76c84722c..704b4b70d 100644 --- a/backend/app/api/bx/advance.py +++ b/backend/app/api/bx/advance.py @@ -4,6 +4,14 @@ from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy import or_ from sqlalchemy.orm import Session +from app.api.bx.export_common import ( + EXPORT_MAX_ROWS, + ExportColumn, + ExportValidationError, + BxApplyListExportBody, + dump_export_row, + respond_xlsx, +) from app.core.bx_auth import BxUserContext, get_bx_current_user from app.core.database import get_db from app.core.tenant_context import get_tenant_no @@ -38,6 +46,57 @@ def _fill_offset_amounts(db: Session, tenant_no: str, rows, out_rows: List[Advan out_row.offset_amount = offset_map.get(out_row.id, 0.0) +ADVANCE_EXPORT_COLUMNS = [ + ExportColumn(title="申请单号", field="applyCode"), + ExportColumn(title="申请时间", field="applyTime"), + ExportColumn(title="申请人", field="applicantName"), + ExportColumn(title="申请原因", field="reason"), + ExportColumn(title="申请金额（元）", field="amount"), + ExportColumn(title="实际发放金额（元）", field="actualAmount"), + ExportColumn(title="冲抵总金额（元）", field="offsetAmount"), + ExportColumn(title="状态", field="status"), + ExportColumn(title="当前节点", field="currentNode"), +] + + +def _advance_list_filters(body: BxApplyListExportBody) -> list: + filters = [] + if body.status: + filters.append(BxAdvanceApply.status == body.status) + if body.keyword: + filters.append( + or_( + BxAdvanceApply.apply_code.contains(body.keyword), + BxAdvanceApply.applicant_name.contains(body.keyword), + ) + ) + if body.applicant_id: + filters.append(BxAdvanceApply.applicant_id == str(body.applicant_id).strip()) + if body.ids: + filters.append(BxAdvanceApply.id.in_([str(i) for i in body.ids])) + append_create_time_filters( + filters, + BxAdvanceApply, + apply_date=body.apply_date, + apply_date_from=body.apply_date_from, + apply_date_to=body.apply_date_to, + ) + return filters + + +def _export_advance_rows( + db: Session, + tenant_no: str, + rows, + *, + with_offset: bool = True, +) -> List[dict]: + out_rows = [AdvanceApplyOut.model_validate(r) for r in rows] + if with_offset: + _fill_offset_amounts(db, tenant_no, rows, out_rows) + return [dump_export_row(r) for r in out_rows] + + @router.get("/", response_model=ApiResponse[PagedData[AdvanceApplyOut]]) def list_advances( page: int = 1, @@ -207,6 +266,140 @@ def approved_by_me( return ApiResponse(data=PagedData(rows=out_rows, total=total, page=page, page_size=pageSize)) +@router.post("/export") +def export_advances( + body: BxApplyListExportBody, + db: Session = Depends(get_db), + user: BxUserContext = Depends(get_bx_current_user), +): + try: + tenant_no = _tenant(db, user) + filters = _advance_list_filters(body) + rows, _ = AdvanceService.get_paged( + db, tenant_no, page=1, page_size=EXPORT_MAX_ROWS, filters=filters, user=user + ) + export_rows = _export_advance_rows(db, tenant_no, rows) + return respond_xlsx( + db, + tenant_no=tenant_no, + sheet_title="备用金申请", + columns=ADVANCE_EXPORT_COLUMNS, + rows=export_rows, + filename_stem="备用金申请", + requested_columns=body.columns, + ) + except ExportValidationError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + +@router.post("/my/export") +def export_my_advances( + body: BxApplyListExportBody, + db: Session = Depends(get_db), + user: BxUserContext = Depends(get_bx_current_user), +): + try: + tenant_no = _tenant(db, user) + filters = [ + or_( + BxAdvanceApply.applicant_id == user.id, + BxAdvanceApply.create_by == user.id, + ) + ] + filters.extend(_advance_list_filters(body)) + rows, _ = AdvanceService.get_paged( + db, tenant_no, page=1, page_size=EXPORT_MAX_ROWS, filters=filters, user=user + ) + export_rows = _export_advance_rows(db, tenant_no, rows) + return respond_xlsx( + db, + tenant_no=tenant_no, + sheet_title="我的备用金", + columns=ADVANCE_EXPORT_COLUMNS, + rows=export_rows, + filename_stem="我的备用金申请", + requested_columns=body.columns, + ) + except ExportValidationError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + +@router.post("/pending-approval/export") +def export_pending_approval_advances( + body: BxApplyListExportBody, + db: Session = Depends(get_db), + user: BxUserContext = Depends(get_bx_current_user), +): + try: + tenant_no = _tenant(db, user) + applicant_id = str(body.applicant_id).strip() if body.applicant_id else None + rows, _ = AdvanceService.pending_approval( + db, + user, + tenant_no, + page=1, + page_size=EXPORT_MAX_ROWS, + keyword=body.keyword or "", + applicant_id=applicant_id, + apply_date=body.apply_date, + apply_date_from=body.apply_date_from, + apply_date_to=body.apply_date_to, + ) + if body.ids: + id_set = {str(i) for i in body.ids} + rows = [r for r in rows if str(r.id) in id_set] + export_rows = _export_advance_rows(db, tenant_no, rows, with_offset=False) + return respond_xlsx( + db, + tenant_no=tenant_no, + sheet_title="备用金待审批", + columns=ADVANCE_EXPORT_COLUMNS, + rows=export_rows, + filename_stem="备用金待审批", + requested_columns=body.columns, + ) + except ExportValidationError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + +@router.post("/approved-by-me/export") +def export_approved_by_me_advances( + body: BxApplyListExportBody, + db: Session = Depends(get_db), + user: BxUserContext = Depends(get_bx_current_user), +): + try: + tenant_no = _tenant(db, user) + applicant_id = str(body.applicant_id).strip() if body.applicant_id else None + rows, _ = AdvanceService.approved_by_me( + db, + user, + tenant_no, + page=1, + page_size=EXPORT_MAX_ROWS, + keyword=body.keyword or "", + applicant_id=applicant_id, + apply_date=body.apply_date, + apply_date_from=body.apply_date_from, + apply_date_to=body.apply_date_to, + ) + if body.ids: + id_set = {str(i) for i in body.ids} + rows = [r for r in rows if str(r.id) in id_set] + export_rows = _export_advance_rows(db, tenant_no, rows, with_offset=False) + return respond_xlsx( + db, + tenant_no=tenant_no, + sheet_title="备用金已审批", + columns=ADVANCE_EXPORT_COLUMNS, + rows=export_rows, + filename_stem="备用金已审批", + requested_columns=body.columns, + ) + except ExportValidationError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + @router.get("/{advance_id}", response_model=ApiResponse[AdvanceApplyOut]) def get_advance( advance_id: str, diff --git a/backend/app/api/bx/export_common.py b/backend/app/api/bx/export_common.py new file mode 100644 index 000000000..5330e580f --- /dev/null +++ b/backend/app/api/bx/export_common.py @@ -0,0 +1,128 @@ +"""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", +]