from typing import Annotated from fastapi import Path, Query, Request, Response from sqlalchemy.ext.asyncio import AsyncSession from common.annotation.log_annotation import Log from common.aspect.db_seesion import DBSessionDependency from common.aspect.interface_auth import UserInterfaceAuthDependency from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency from common.enums import BusinessType from common.router import APIRouterPro from common.vo import DataResponseModel, PageResponseModel, ResponseBaseModel from module_admin.entity.vo.user_vo import CurrentUserModel from module_business.dependency.test_batch_dependency import TestBatchScopeDependency from module_business.domain.test_batch_scope import TestBatchScope from module_business.entity.vo.operation_vo import ( BusinessOptionModel, ConfirmCommandModel, PalletScanResponseModel, ShipmentCreateModel, ShipmentModel, ShipmentPageQueryModel, ShipmentScanQueryModel, ShipmentUpdateModel, ) from module_business.service.shipment_service import ShipmentService from utils.response_util import ResponseUtil shipment_controller = APIRouterPro( prefix='/business/shipment', tags=['业务管理-整托发货'], dependencies=[PreAuthDependency()], ) @shipment_controller.get( '/list', summary='获取整托发货单分页列表接口', description='按当前用户仓库与现场双数据范围获取整托发货单', response_model=PageResponseModel[ShipmentModel], dependencies=[UserInterfaceAuthDependency('business:shipment:list')], ) async def get_business_shipment_list( request: Request, shipment_page_query: Annotated[ShipmentPageQueryModel, Query()], query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()], ) -> Response: result = await ShipmentService.get_shipment_list_services( query_db, shipment_page_query, current_user, test_scope, is_page=True, ) return ResponseUtil.success(model_content=result) @shipment_controller.get( '/warehouse-options', summary='获取发货来源仓库选项接口', description='仅返回当前用户有权访问的启用仓库', response_model=DataResponseModel[list[BusinessOptionModel]], dependencies=[UserInterfaceAuthDependency('business:shipment:query')], ) async def get_business_shipment_warehouse_options( request: Request, query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()], ) -> Response: result = await ShipmentService.get_warehouse_options_services(query_db, current_user, test_scope) return ResponseUtil.success(data=result) @shipment_controller.get( '/site-options', summary='获取发货目标现场选项接口', description='仅返回当前用户有权访问的启用客户现场', response_model=DataResponseModel[list[BusinessOptionModel]], dependencies=[UserInterfaceAuthDependency('business:shipment:query')], ) async def get_business_shipment_site_options( request: Request, query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()], ) -> Response: result = await ShipmentService.get_site_options_services(query_db, current_user, test_scope) return ResponseUtil.success(data=result) @shipment_controller.get( '/scan/{qr_token}', summary='扫描整托发货托盘接口', description='查询托盘是否可加入指定仓库至现场的整托发货单', response_model=DataResponseModel[PalletScanResponseModel], dependencies=[UserInterfaceAuthDependency('business:shipment:query')], ) async def scan_business_shipment_pallet( request: Request, qr_token: Annotated[ str, Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'), ], shipment_scan_query: Annotated[ShipmentScanQueryModel, Query()], query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()], ) -> Response: result = await ShipmentService.scan_shipment_pallet_services( query_db, qr_token, shipment_scan_query, current_user, test_scope, ) return ResponseUtil.success(data=result) @shipment_controller.get( '/{shipment_id}', summary='获取整托发货单详情接口', description='按仓库与现场双数据范围获取指定整托发货单', response_model=DataResponseModel[ShipmentModel], dependencies=[UserInterfaceAuthDependency('business:shipment:query')], ) async def query_detail_business_shipment( request: Request, shipment_id: Annotated[int, Path(gt=0, description='发货单ID')], query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()], ) -> Response: result = await ShipmentService.shipment_detail_services(query_db, shipment_id, current_user, test_scope) return ResponseUtil.success(data=result) @shipment_controller.post( '', summary='新增整托发货草稿接口', description='新增多托盘整托发货草稿', response_model=ResponseBaseModel, dependencies=[UserInterfaceAuthDependency('business:shipment:add')], ) @Log(title='整托发货', business_type=BusinessType.INSERT) async def add_business_shipment( request: Request, add_shipment: ShipmentCreateModel, query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)], ) -> Response: result = await ShipmentService.add_shipment_services(query_db, add_shipment, current_user, test_scope) return ResponseUtil.success(msg=result.message) @shipment_controller.put( '', summary='编辑整托发货草稿接口', description='编辑尚未确认的多托盘整托发货草稿', response_model=ResponseBaseModel, dependencies=[UserInterfaceAuthDependency('business:shipment:edit')], ) @Log(title='整托发货', business_type=BusinessType.UPDATE) async def edit_business_shipment( request: Request, edit_shipment: ShipmentUpdateModel, query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)], ) -> Response: result = await ShipmentService.edit_shipment_services(query_db, edit_shipment, current_user, test_scope) return ResponseUtil.success(msg=result.message) @shipment_controller.delete( '/{shipment_ids}', summary='不支持单条物理删除，请通过测试数据批次清理', description='不支持单条物理删除，请通过测试数据批次清理', response_model=ResponseBaseModel, dependencies=[UserInterfaceAuthDependency('business:shipment:list')], ) @Log(title='整托发货', business_type=BusinessType.DELETE) async def delete_business_shipment( request: Request, shipment_ids: Annotated[ str, Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的发货单ID'), ], query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)], ) -> Response: result = await ShipmentService.delete_shipment_services( query_db, [int(shipment_id) for shipment_id in shipment_ids.split(',')], current_user, test_scope, ) return ResponseUtil.success(msg=result.message) @shipment_controller.put( '/{shipment_id}/confirm', summary='确认整托发货单接口', description='原子确认整张发货单并将全部托盘移至目标现场', response_model=ResponseBaseModel, dependencies=[UserInterfaceAuthDependency('business:shipment:confirm')], ) @Log(title='整托发货', business_type=BusinessType.UPDATE) async def confirm_business_shipment( request: Request, shipment_id: Annotated[int, Path(gt=0, description='发货单ID')], confirm_command: ConfirmCommandModel, query_db: Annotated[AsyncSession, DBSessionDependency()], current_user: Annotated[CurrentUserModel, CurrentUserDependency()], test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)], ) -> Response: result = await ShipmentService.confirm_shipment_services( query_db, shipment_id, confirm_command, current_user, test_scope, ) return ResponseUtil.success(msg=result.message)