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 <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
dbba755fd8
commit
833544def3
2 files changed
+321
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
Reference in new issue
Block a user