From ae1d01bf079c70c154940f076d01442f6963e429 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Sun, 27 Sep 2026 18:25:58 +0800 Subject: [PATCH] feat(pl): filter recharge applies by mineVehicles plates Co-authored-by: Cursor --- .../services/pl_vehicle_recharge_service.py | 142 ++++++++++++++++-- .../test_pl_vehicle_recharge_mine_vehicles.py | 88 +++++++++++ 2 files changed, 219 insertions(+), 11 deletions(-) diff --git a/backend/app/services/pl_vehicle_recharge_service.py b/backend/app/services/pl_vehicle_recharge_service.py index c41c74bfc..a3e7246a7 100644 --- a/backend/app/services/pl_vehicle_recharge_service.py +++ b/backend/app/services/pl_vehicle_recharge_service.py @@ -22,6 +22,7 @@ from app.schemas.pl.vehicle_recharge import ( VehicleRechargeApplyCreate, VehicleRechargeApplyRead, VehicleRechargeApplyUpdate, + VehicleRechargeApprovalFlowItem, VehicleRechargeApprovalRead, VehicleRechargeDisburseBoardItem, VehicleRechargeRecordCreate, @@ -29,6 +30,7 @@ from app.schemas.pl.vehicle_recharge import ( VehicleRechargeRecordUpdate, ) from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope +from app.services.pl_vehicle_mine_vehicles import license_plates_for_user_personnel _EDITABLE_STATUSES = frozenset({"draft", "rejected"}) _BIZ_TYPE_VEHICLE_RECHARGE = "vehicle_recharge" @@ -115,7 +117,12 @@ def _resolve_applicant_id( def _get_apply_row( - db: Session, tenant_no: str, apply_id: str, user: DbUserContext + db: Session, + tenant_no: str, + apply_id: str, + user: DbUserContext, + *, + mine_vehicles: bool = False, ) -> PlVehicleRechargeApply: row = ( db.query(PlVehicleRechargeApply) @@ -128,7 +135,12 @@ def _get_apply_row( ) if not row: raise HTTPException(status_code=404, detail="申请单不存在") - assert_pl_row_in_scope(db, row, PlVehicleRechargeApply, user, tenant_no) + if mine_vehicles: + plates = license_plates_for_user_personnel(db, tenant_no, user.id) + if row.license_plate not in plates: + raise HTTPException(status_code=403, detail="无权查看该申请") + else: + assert_pl_row_in_scope(db, row, PlVehicleRechargeApply, user, tenant_no) return row @@ -178,10 +190,10 @@ def approval_to_read(row: PlVehicleRechargeApproval) -> VehicleRechargeApprovalR ) -def _list_approvals( +def _list_approval_rows( db: Session, tenant_no: str, apply_id: str -) -> list[VehicleRechargeApprovalRead]: - rows = ( +) -> list[PlVehicleRechargeApproval]: + return ( db.query(PlVehicleRechargeApproval) .filter( PlVehicleRechargeApproval.apply_id == apply_id, @@ -195,7 +207,93 @@ def _list_approvals( ) .all() ) - return [approval_to_read(row) for row in rows] + + +def _list_approvals( + db: Session, tenant_no: str, apply_id: str +) -> list[VehicleRechargeApprovalRead]: + return [approval_to_read(row) for row in _list_approval_rows(db, tenant_no, apply_id)] + + +def _approval_flow_item_from_row( + row: PlVehicleRechargeApproval, + *, + node_order: int, +) -> VehicleRechargeApprovalFlowItem: + return VehicleRechargeApprovalFlowItem( + id=row.id, + applyId=row.apply_id, + nodeName=row.node_name or "节点", + nodeOrder=node_order, + approvalRound=row.approval_round, + approvalStatus=row.approval_status, + approvalOpinion=row.approval_opinion, + approverName=row.approver_name or "", + approvalTime=row.approval_time, + ) + + +def _build_approval_flow( + db: Session, + row: PlVehicleRechargeApply, + approval_rows: list[PlVehicleRechargeApproval], +) -> list[VehicleRechargeApprovalFlowItem]: + """当前轮：流程全部节点；已审带信息，未审为 pending。""" + if not row.wf_id or not (row.approval_round or 0): + return [] + nodes = _workflow_nodes(db, row.wf_id, row.tenant_no) + if not nodes: + return [] + current_round = row.approval_round or 0 + round_approvals = [a for a in approval_rows if (a.approval_round or 0) == current_round] + by_node_id: dict[str, PlVehicleRechargeApproval] = {} + for a in round_approvals: + if not a.node_id: + continue + existing = by_node_id.get(a.node_id) + if existing is None or (a.approval_time or datetime.min) >= ( + existing.approval_time or datetime.min + ): + by_node_id[a.node_id] = a + result: list[VehicleRechargeApprovalFlowItem] = [] + for node in nodes: + approval = by_node_id.get(node.id) + if approval: + result.append( + _approval_flow_item_from_row(approval, node_order=node.node_order or 0) + ) + else: + result.append( + VehicleRechargeApprovalFlowItem( + id="0", + applyId=row.id, + nodeName=node.node_name, + nodeOrder=node.node_order or 0, + approvalRound=current_round, + approvalStatus="pending", + approvalOpinion=None, + approverName=node.approver_name or "", + approvalTime=None, + ) + ) + return result + + +def _build_approval_history( + db: Session, + row: PlVehicleRechargeApply, + approval_rows: list[PlVehicleRechargeApproval], +) -> list[VehicleRechargeApprovalFlowItem]: + """全部历史审批动作（含打回前各轮），按时间正序。""" + node_order_map: dict[str, int] = {} + if row.wf_id: + for node in _workflow_nodes(db, row.wf_id, row.tenant_no): + node_order_map[node.id] = node.node_order or 0 + result: list[VehicleRechargeApprovalFlowItem] = [] + for idx, a in enumerate(approval_rows): + order = node_order_map.get(a.node_id or "", 0) or (idx + 1) + result.append(_approval_flow_item_from_row(a, node_order=order)) + return result def _get_linked_record( @@ -219,6 +317,8 @@ def apply_to_read( include_detail: bool = False, ) -> VehicleRechargeApplyRead: approvals: list[VehicleRechargeApprovalRead] = [] + approval_flow: list[VehicleRechargeApprovalFlowItem] = [] + approval_history: list[VehicleRechargeApprovalFlowItem] = [] record_read: Optional[VehicleRechargeRecordRead] = None current_node_name: Optional[str] = None current_node_action_type: Optional[str] = None @@ -235,7 +335,10 @@ def apply_to_read( current_node_name = node.node_name current_node_action_type = (node.action_type or "approve").strip() or "approve" if include_detail: - approvals = _list_approvals(db, row.tenant_no, row.id) + approval_rows = _list_approval_rows(db, row.tenant_no, row.id) + approvals = [approval_to_read(a) for a in approval_rows] + approval_flow = _build_approval_flow(db, row, approval_rows) + approval_history = _build_approval_history(db, row, approval_rows) record_row = _get_linked_record(db, row.tenant_no, row.id) if record_row: record_read = record_to_read(record_row, db, apply_no=row.apply_no) @@ -256,6 +359,8 @@ def apply_to_read( currentNodeActionType=current_node_action_type, approvalRound=row.approval_round or 0, approvals=approvals, + approvalFlow=approval_flow, + approvalHistory=approval_history, record=record_read, remark=row.remark, createTime=row.create_time, @@ -343,7 +448,7 @@ def _resolve_apply_for_record( if apply.status == "in_approval": raise HTTPException( status_code=400, - detail="审批中的申请请在「充值审批」中完成发放登记", + detail="审批中的申请请在「ETC/加油充值审批」中完成发放登记", ) if apply.status != "pending": raise HTTPException(status_code=400, detail="仅待发放申请可关联") @@ -389,6 +494,7 @@ def list_applies( max_page_size: Optional[int] = None, sort_field: Optional[str] = None, sort_order: Optional[str] = None, + mine_vehicles: bool = False, ) -> tuple[list[VehicleRechargeApplyRead], int]: query = db.query(PlVehicleRechargeApply).filter( PlVehicleRechargeApply.tenant_no == tenant_no, @@ -424,7 +530,13 @@ def list_applies( if ids: query = query.filter(PlVehicleRechargeApply.id.in_(ids)) - query = apply_pl_scope(query, PlVehicleRechargeApply, user, db, tenant_no) + if mine_vehicles: + plates = license_plates_for_user_personnel(db, tenant_no, user.id) + if not plates: + return [], 0 + query = query.filter(PlVehicleRechargeApply.license_plate.in_(plates)) + else: + query = apply_pl_scope(query, PlVehicleRechargeApply, user, db, tenant_no) total = query.count() page = max(1, page) cap = int(max_page_size) if max_page_size else 500 @@ -439,9 +551,16 @@ def list_applies( def get_apply( - db: Session, tenant_no: str, user: DbUserContext, apply_id: str + db: Session, + tenant_no: str, + user: DbUserContext, + apply_id: str, + *, + mine_vehicles: bool = False, ) -> VehicleRechargeApplyRead: - row = _get_apply_row(db, tenant_no, apply_id, user) + row = _get_apply_row( + db, tenant_no, apply_id, user, mine_vehicles=mine_vehicles + ) return apply_to_read(row, db, include_detail=True) @@ -658,6 +777,7 @@ def board_item_to_read( amount=record_row.amount if record_row else None, recordCreateByName=_resolve_user_name(db, record_row.create_by) if record_row else None, recordRemark=record_row.remark if record_row else None, + recordAttachmentUrls=record_row.attachment_urls if record_row else None, ) diff --git a/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py index f0222388a..c75059275 100644 --- a/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py +++ b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py @@ -96,3 +96,91 @@ def test_license_plates_empty_when_no_personnel(): db = _make_session() db.commit() assert license_plates_for_user_personnel(db, "t1", "u-nobody") == set() + + +def _seed_apply(db, *, apply_id, plate, applicant_id, status="draft"): + db.add( + PlVehicleRechargeApply( + id=apply_id, + apply_no=f"SXSQ{apply_id}", + license_plate=plate, + recharge_type="etc", + apply_date=date.today(), + applicant_id=applicant_id, + status=status, + tenant_no="t1", + create_by=applicant_id, + create_dept_id="d1", + create_time=datetime.now(), + update_time=datetime.now(), + del_flag=0, + ) + ) + + +def test_list_applies_mine_vehicles_includes_others_on_my_plate(monkeypatch): + from app.services import pl_vehicle_recharge_service as svc + + db = _make_session() + _seed_person_vehicle(db) + _seed_apply(db, apply_id="a1", plate="鄂A11111", applicant_id="u-office") + _seed_apply(db, apply_id="a2", plate="鄂A99999", applicant_id="u-office") + db.commit() + monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *_a, **_k: query) + monkeypatch.setattr(svc, "assert_pl_row_in_scope", lambda *_a, **_k: None) + monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X") + monkeypatch.setattr( + svc, + "apply_to_read", + lambda row, db, **kw: SimpleNamespace( + id=row.id, licensePlate=row.license_plate, applicantId=row.applicant_id + ), + ) + + items, total = svc.list_applies( + db, "t1", _user(), mine_vehicles=True, page=1, page_size=20 + ) + assert total == 1 + assert items[0].id == "a1" + + +def test_get_apply_mine_vehicles_allows_others_on_my_plate(monkeypatch): + from app.services import pl_vehicle_recharge_service as svc + + db = _make_session() + _seed_person_vehicle(db) + _seed_apply(db, apply_id="a1", plate="鄂A11111", applicant_id="u-office") + db.commit() + monkeypatch.setattr( + svc, + "assert_pl_row_in_scope", + lambda *_a, **_k: (_ for _ in ()).throw( + AssertionError("should not use data scope when mine_vehicles") + ), + ) + monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X") + monkeypatch.setattr( + svc, + "apply_to_read", + lambda row, db, **kw: SimpleNamespace(id=row.id), + ) + + result = svc.get_apply(db, "t1", _user(), "a1", mine_vehicles=True) + assert result.id == "a1" or getattr(result, "id", None) == "a1" + + +def test_get_apply_mine_vehicles_rejects_foreign_plate(monkeypatch): + from fastapi import HTTPException + + from app.services import pl_vehicle_recharge_service as svc + + db = _make_session() + _seed_person_vehicle(db) + _seed_apply(db, apply_id="a2", plate="鄂A99999", applicant_id="u-office") + db.commit() + monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X") + + with pytest.raises(HTTPException) as exc: + svc.get_apply(db, "t1", _user(), "a2", mine_vehicles=True) + assert exc.value.status_code == 403 + assert "无权" in exc.value.detail