功能：新增托盘二维码与用户分配

shierfeng-fastapi-backend/module_business/controller/pallet_controller.py

@@ -0,0 +1,133 @@
+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.entity.vo.master_vo import (
+    DeletePalletModel,
+    PalletCreateModel,
+    PalletEditModel,
+    PalletModel,
+    PalletPageQueryModel,
+)
+from module_business.service.pallet_service import PalletService
+from utils.response_util import ResponseUtil
+
+pallet_controller = APIRouterPro(
+    prefix='/master/pallet', tags=['基础资料-托盘管理'], dependencies=[PreAuthDependency()]
+)
+
+
+@pallet_controller.get(
+    '/list',
+    summary='获取托盘分页列表接口',
+    description='用于获取托盘分页列表',
+    response_model=PageResponseModel[PalletModel],
+    dependencies=[UserInterfaceAuthDependency('master:pallet:list')],
+)
+async def get_master_pallet_list(
+    request: Request,
+    pallet_page_query: Annotated[PalletPageQueryModel, Query()],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+) -> Response:
+    result = await PalletService.list_pallet(query_db, pallet_page_query, is_page=True)
+    return ResponseUtil.success(model_content=result)
+
+
+@pallet_controller.get(
+    '/{pallet_id}/qrcode',
+    summary='补打托盘二维码接口',
+    description='使用托盘原有固定标识重新生成二维码',
+    response_class=Response,
+    responses={200: {'content': {'image/png': {}}}},
+    dependencies=[UserInterfaceAuthDependency('master:pallet:print')],
+)
+async def query_master_pallet_qrcode(
+    request: Request,
+    pallet_id: Annotated[int, Path(gt=0, description='托盘ID')],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+) -> Response:
+    png = await PalletService.get_pallet_qrcode(query_db, pallet_id)
+    return Response(content=png, media_type='image/png')
+
+
+@pallet_controller.get(
+    '/{pallet_id}',
+    summary='获取托盘详情接口',
+    description='用于获取指定托盘的详细信息',
+    response_model=DataResponseModel[PalletModel],
+    dependencies=[UserInterfaceAuthDependency('master:pallet:query')],
+)
+async def query_detail_master_pallet(
+    request: Request,
+    pallet_id: Annotated[int, Path(gt=0, description='托盘ID')],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+) -> Response:
+    result = await PalletService.detail_pallet(query_db, pallet_id)
+    return ResponseUtil.success(data=result)
+
+
+@pallet_controller.post(
+    '',
+    summary='新增托盘接口',
+    description='用于新增托盘及初始库存',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('master:pallet:add')],
+)
+@Log(title='托盘管理', business_type=BusinessType.INSERT)
+async def add_master_pallet(
+    request: Request,
+    add_pallet: PalletCreateModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await PalletService.add_pallet(query_db, add_pallet, current_user.user.user_name)
+    return ResponseUtil.success(msg=result.message)
+
+
+@pallet_controller.put(
+    '',
+    summary='编辑托盘接口',
+    description='仅编辑托盘容量、资产状态和备注',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('master:pallet:edit')],
+)
+@Log(title='托盘管理', business_type=BusinessType.UPDATE)
+async def edit_master_pallet(
+    request: Request,
+    edit_pallet: PalletEditModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await PalletService.edit_pallet(query_db, edit_pallet, current_user.user.user_name)
+    return ResponseUtil.success(msg=result.message)
+
+
+@pallet_controller.delete(
+    '/{pallet_ids}',
+    summary='删除托盘接口',
+    description='用于删除尚未入库且没有流转事件的托盘',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('master:pallet:remove')],
+)
+@Log(title='托盘管理', business_type=BusinessType.DELETE)
+async def delete_master_pallet(
+    request: Request,
+    pallet_ids: Annotated[
+        str,
+        Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的托盘ID'),
+    ],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    delete_pallet = DeletePalletModel(palletIds=[int(pallet_id) for pallet_id in pallet_ids.split(',')])
+    result = await PalletService.remove_pallet(query_db, delete_pallet)
+    return ResponseUtil.success(msg=result.message)

shierfeng-fastapi-backend/module_business/controller/site_controller.py

@@ -12,13 +12,61 @@ 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.entity.vo.master_vo import DeleteSiteModel, SiteModel, SitePageQueryModel
+from module_business.entity.vo.master_vo import (
+    BusinessUserOptionModel,
+    DeleteSiteModel,
+    SiteModel,
+    SitePageQueryModel,
+    UserAssignModel,
+)
 from module_business.service.master_service import MasterService
+from module_business.service.pallet_service import PalletService
 from utils.response_util import ResponseUtil
 
+ASSIGN_LOG = Log
+ASSIGN_CURRENT_USER = CurrentUserDependency
+
 site_controller = APIRouterPro(
     prefix='/master/site', tags=['基础资料-客户现场管理'], dependencies=[PreAuthDependency()]
 )
+site_user_controller = APIRouterPro(
+    prefix='/master/site', tags=['基础资料-客户现场用户分配'], dependencies=[PreAuthDependency()]
+)
+
+
+@site_user_controller.get(
+    '/{site_id}/users',
+    summary='获取客户现场可分配用户接口',
+    description='获取启用且未删除的用户及客户现场分配状态',
+    response_model=DataResponseModel[list[BusinessUserOptionModel]],
+    dependencies=[UserInterfaceAuthDependency('master:site:assign')],
+)
+async def get_master_site_users(
+    request: Request,
+    site_id: Annotated[int, Path(gt=0, description='客户现场ID')],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+) -> Response:
+    users = await PalletService.list_site_users(query_db, site_id)
+    return ResponseUtil.success(data=users)
+
+
+@site_user_controller.put(
+    '/{site_id}/users',
+    summary='分配客户现场用户接口',
+    description='完整替换客户现场的用户关联',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('master:site:assign')],
+)
+@ASSIGN_LOG(title='客户现场用户分配', business_type=BusinessType.GRANT)
+async def assign_master_site_users(
+    request: Request,
+    site_id: Annotated[int, Path(gt=0, description='客户现场ID')],
+    assignment: UserAssignModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, ASSIGN_CURRENT_USER()],
+) -> Response:
+    result = await PalletService.assign_site_users(query_db, site_id, assignment)
+    return ResponseUtil.success(msg=result.message)
 
 
 @site_controller.get(

shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py

@@ -12,13 +12,61 @@ 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.entity.vo.master_vo import DeleteWarehouseModel, WarehouseModel, WarehousePageQueryModel
+from module_business.entity.vo.master_vo import (
+    BusinessUserOptionModel,
+    DeleteWarehouseModel,
+    UserAssignModel,
+    WarehouseModel,
+    WarehousePageQueryModel,
+)
 from module_business.service.master_service import MasterService
+from module_business.service.pallet_service import PalletService
 from utils.response_util import ResponseUtil
 
+ASSIGN_LOG = Log
+ASSIGN_CURRENT_USER = CurrentUserDependency
+
 warehouse_controller = APIRouterPro(
     prefix='/master/warehouse', tags=['基础资料-仓库管理'], dependencies=[PreAuthDependency()]
 )
+warehouse_user_controller = APIRouterPro(
+    prefix='/master/warehouse', tags=['基础资料-仓库用户分配'], dependencies=[PreAuthDependency()]
+)
+
+
+@warehouse_user_controller.get(
+    '/{warehouse_id}/users',
+    summary='获取仓库可分配用户接口',
+    description='获取启用且未删除的用户及仓库分配状态',
+    response_model=DataResponseModel[list[BusinessUserOptionModel]],
+    dependencies=[UserInterfaceAuthDependency('master:warehouse:assign')],
+)
+async def get_master_warehouse_users(
+    request: Request,
+    warehouse_id: Annotated[int, Path(gt=0, description='仓库ID')],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+) -> Response:
+    users = await PalletService.list_warehouse_users(query_db, warehouse_id)
+    return ResponseUtil.success(data=users)
+
+
+@warehouse_user_controller.put(
+    '/{warehouse_id}/users',
+    summary='分配仓库用户接口',
+    description='完整替换仓库的用户关联',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('master:warehouse:assign')],
+)
+@ASSIGN_LOG(title='仓库用户分配', business_type=BusinessType.GRANT)
+async def assign_master_warehouse_users(
+    request: Request,
+    warehouse_id: Annotated[int, Path(gt=0, description='仓库ID')],
+    assignment: UserAssignModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, ASSIGN_CURRENT_USER()],
+) -> Response:
+    result = await PalletService.assign_warehouse_users(query_db, warehouse_id, assignment)
+    return ResponseUtil.success(msg=result.message)
 
 
 @warehouse_controller.get(

shierfeng-fastapi-backend/module_business/dao/master_dao.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Any, Literal
 
 from pydantic import BaseModel
@@ -6,9 +7,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 
 from common.vo import PageModel
+from module_admin.entity.do.user_do import SysUser
 from module_business.entity.do.master_do import (
     BizCustomer,
     BizCustomerSite,
+    BizPallet,
     BizProductionLine,
     BizStrain,
     BizUserSite,
@@ -29,6 +32,8 @@ from module_business.entity.vo.master_vo import (
     CustomerPageQueryModel,
     LineModel,
     LinePageQueryModel,
+    PalletEditModel,
+    PalletPageQueryModel,
     SiteModel,
     SitePageQueryModel,
     StrainModel,
@@ -357,6 +362,128 @@ class MasterDao:
     async def delete_warehouse(cls, db: AsyncSession, warehouse_id: int) -> None:
         await cls._delete(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id)
 
+    @classmethod
+    async def list_pallet(
+        cls,
+        db: AsyncSession,
+        query_object: PalletPageQueryModel,
+        is_page: bool = True,
+    ) -> PageModel | list[dict[str, Any]]:
+        conditions = []
+        if query_object.pallet_code:
+            conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
+        if query_object.asset_status:
+            conditions.append(BizPallet.asset_status == query_object.asset_status)
+        statement = select(BizPallet).where(*conditions).order_by(BizPallet.pallet_id)
+        return await cls._paginate(db, statement, query_object, is_page)
+
+    @classmethod
+    async def get_pallet(
+        cls,
+        db: AsyncSession,
+        pallet_id: int,
+        *,
+        for_update: bool = False,
+    ) -> BizPallet | None:
+        statement = select(BizPallet).where(BizPallet.pallet_id == pallet_id)
+        if for_update:
+            statement = statement.with_for_update()
+        result = await db.execute(statement)
+        return result.scalar_one_or_none()
+
+    @classmethod
+    async def find_pallet_by_code(cls, db: AsyncSession, pallet_code: str) -> BizPallet | None:
+        return await cls._find(db, BizPallet, BizPallet.pallet_code, pallet_code)
+
+    @classmethod
+    async def add_pallet(cls, db: AsyncSession, pallet: BizPallet) -> BizPallet:
+        db.add(pallet)
+        await db.flush()
+        return pallet
+
+    @classmethod
+    async def update_pallet(
+        cls,
+        db: AsyncSession,
+        pallet: PalletEditModel,
+        update_by: str,
+        update_time: datetime,
+    ) -> None:
+        await db.execute(
+            update(BizPallet)
+            .where(BizPallet.pallet_id == pallet.pallet_id)
+            .values(
+                capacity_packages=pallet.capacity_packages,
+                asset_status=pallet.asset_status,
+                remark=pallet.remark,
+                update_by=update_by,
+                update_time=update_time,
+            )
+        )
+
+    @classmethod
+    async def delete_pallet(cls, db: AsyncSession, pallet_id: int) -> None:
+        await cls._delete(db, BizPallet, BizPallet.pallet_id, pallet_id)
+
+    @classmethod
+    async def get_pallet_inventory(
+        cls,
+        db: AsyncSession,
+        pallet_id: int,
+        *,
+        for_update: bool = False,
+    ) -> BizPalletInventory | None:
+        statement = select(BizPalletInventory).where(BizPalletInventory.pallet_id == pallet_id)
+        if for_update:
+            statement = statement.with_for_update()
+        result = await db.execute(statement)
+        return result.scalar_one_or_none()
+
+    @classmethod
+    async def add_pallet_inventory(
+        cls,
+        db: AsyncSession,
+        inventory: BizPalletInventory,
+    ) -> BizPalletInventory:
+        db.add(inventory)
+        await db.flush()
+        return inventory
+
+    @classmethod
+    async def delete_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> None:
+        await cls._delete(db, BizPalletInventory, BizPalletInventory.pallet_id, pallet_id)
+
+    @classmethod
+    async def count_pallet_events(cls, db: AsyncSession, pallet_id: int) -> int:
+        result = await db.execute(
+            select(func.count('*')).select_from(BizPalletEvent).where(BizPalletEvent.pallet_id == pallet_id)
+        )
+        return result.scalar() or 0
+
+    @classmethod
+    async def list_assignable_users(cls, db: AsyncSession) -> list[Any]:
+        result = await db.execute(
+            select(SysUser.user_id, SysUser.user_name, SysUser.nick_name)
+            .where(SysUser.status == '0', SysUser.del_flag == '0')
+            .order_by(SysUser.user_id)
+        )
+        return list(result.all())
+
+    @classmethod
+    async def list_assignable_users_by_ids(cls, db: AsyncSession, user_ids: list[int]) -> list[Any]:
+        if not user_ids:
+            return []
+        result = await db.execute(
+            select(SysUser.user_id, SysUser.user_name, SysUser.nick_name)
+            .where(
+                SysUser.user_id.in_(user_ids),
+                SysUser.status == '0',
+                SysUser.del_flag == '0',
+            )
+            .order_by(SysUser.user_id)
+        )
+        return list(result.all())
+
     @classmethod
     async def count_references(
         cls,

shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py

@@ -389,3 +389,19 @@ class UserAssignModel(BaseModel):
     user_ids: list[PositiveId] = Field(min_length=1, description='用户ID列表')
 
     _validate_unique_ids = field_validator('user_ids')(_ensure_unique_ids)
+
+
+class BusinessUserOptionModel(BaseModel):
+    """仓库或客户现场可分配用户的最小只读模型。"""
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        from_attributes=True,
+        populate_by_name=True,
+        extra='forbid',
+    )
+
+    user_id: PositiveId = Field(description='用户ID')
+    user_name: str = Field(min_length=1, max_length=30, pattern=r'\S', description='用户账号')
+    nick_name: str | None = Field(default=None, max_length=30, description='用户昵称')
+    assigned: bool = Field(description='是否已分配')

shierfeng-fastapi-backend/module_business/service/pallet_service.py

@@ -0,0 +1,282 @@
+import secrets
+from datetime import datetime
+from io import BytesIO
+from typing import Any, Literal
+
+import qrcode
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.vo import CrudResponseModel, PageModel
+from exceptions.exception import ServiceException
+from module_business.dao.access_dao import AccessDao
+from module_business.dao.master_dao import MasterDao
+from module_business.domain.enums import PalletFlowState
+from module_business.entity.do.master_do import BizPallet
+from module_business.entity.do.operation_do import BizPalletInventory
+from module_business.entity.vo.master_vo import (
+    BusinessUserOptionModel,
+    DeletePalletModel,
+    PalletCreateModel,
+    PalletEditModel,
+    PalletModel,
+    PalletPageQueryModel,
+    UserAssignModel,
+)
+
+AssignmentResource = Literal['warehouse', 'site']
+
+_ASSIGNMENT_LABELS: dict[AssignmentResource, str] = {
+    'warehouse': '仓库',
+    'site': '客户现场',
+}
+
+
+class PalletService:
+    """托盘资产及业务用户分配服务层。"""
+
+    @staticmethod
+    def render_qr_png(qr_token: str) -> bytes:
+        """将固定二维码标识渲染为 PNG；补打不会生成新标识。"""
+        buffer = BytesIO()
+        qrcode.make(qr_token).save(buffer, format='PNG')
+        return buffer.getvalue()
+
+    @classmethod
+    async def get_pallet_qrcode(cls, db: AsyncSession, pallet_id: int) -> bytes:
+        pallet = await MasterDao.get_pallet(db, pallet_id)
+        if pallet is None:
+            raise ServiceException(message='托盘不存在')
+        return cls.render_qr_png(pallet.qr_token)
+
+    @staticmethod
+    def _extract_pallet_ids(value: int | DeletePalletModel) -> list[int]:
+        pallet_ids = [value] if isinstance(value, int) else list(value.pallet_ids)
+        if not pallet_ids or any(pallet_id <= 0 for pallet_id in pallet_ids):
+            raise ServiceException(message='托盘ID必须为正整数')
+        if len(pallet_ids) != len(set(pallet_ids)):
+            raise ServiceException(message='托盘ID不能重复')
+        return sorted(pallet_ids)
+
+    @classmethod
+    async def _can_remove(cls, db: AsyncSession, pallet_id: int) -> bool:
+        inventory = await MasterDao.get_pallet_inventory(db, pallet_id)
+        if inventory is None or inventory.flow_state != PalletFlowState.UNSTOCKED.value:
+            return False
+        return await MasterDao.count_pallet_events(db, pallet_id) == 0
+
+    @classmethod
+    async def _to_pallet_model(cls, db: AsyncSession, record: Any) -> PalletModel:
+        model = PalletModel.model_validate(record)
+        return model.model_copy(update={'can_remove': await cls._can_remove(db, model.pallet_id)})
+
+    @classmethod
+    async def _lock_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> tuple[BizPallet, BizPalletInventory]:
+        """按库存、托盘顺序加锁，避免与流转事务形成反序死锁。"""
+        inventory = await MasterDao.get_pallet_inventory(db, pallet_id, for_update=True)
+        if inventory is None:
+            if await MasterDao.get_pallet(db, pallet_id) is None:
+                raise ServiceException(message='托盘不存在')
+            raise ServiceException(message='托盘库存不存在')
+        pallet = await MasterDao.get_pallet(db, pallet_id, for_update=True)
+        if pallet is None:
+            raise ServiceException(message='托盘不存在')
+        return pallet, inventory
+
+    @classmethod
+    async def list_pallet(
+        cls,
+        db: AsyncSession,
+        query_object: PalletPageQueryModel,
+        is_page: bool = True,
+    ) -> PageModel | list[PalletModel]:
+        result = await MasterDao.list_pallet(db, query_object, is_page)
+        if isinstance(result, PageModel):
+            rows = [await cls._to_pallet_model(db, row) for row in result.rows]
+            return result.model_copy(update={'rows': rows})
+        return [await cls._to_pallet_model(db, row) for row in result]
+
+    @classmethod
+    async def detail_pallet(cls, db: AsyncSession, pallet_id: int) -> PalletModel:
+        pallet = await MasterDao.get_pallet(db, pallet_id)
+        if pallet is None:
+            raise ServiceException(message='托盘不存在')
+        return await cls._to_pallet_model(db, pallet)
+
+    @classmethod
+    async def add_pallet(
+        cls,
+        db: AsyncSession,
+        pallet: PalletCreateModel,
+        current_user_name: str,
+    ) -> CrudResponseModel:
+        try:
+            if await MasterDao.find_pallet_by_code(db, pallet.pallet_code) is not None:
+                raise ServiceException(message='托盘编号已存在')
+
+            current_time = datetime.now()
+            pallet_record = BizPallet(
+                pallet_code=pallet.pallet_code,
+                qr_token=secrets.token_urlsafe(32),
+                capacity_packages=pallet.capacity_packages,
+                asset_status='0',
+                remark=pallet.remark,
+                create_by=current_user_name,
+                create_time=current_time,
+                update_by=current_user_name,
+                update_time=current_time,
+            )
+            pallet_record = await MasterDao.add_pallet(db, pallet_record)
+            inventory = BizPalletInventory(
+                pallet_id=pallet_record.pallet_id,
+                flow_state='UNSTOCKED',
+                remaining_packages=0,
+                version=0,
+                updated_at=current_time,
+            )
+            await MasterDao.add_pallet_inventory(db, inventory)
+            await db.commit()
+            return CrudResponseModel(is_success=True, message='新增成功')
+        except IntegrityError:
+            await db.rollback()
+            raise ServiceException(message='新增托盘失败，托盘编号或二维码标识已存在') from None
+        except Exception:
+            await db.rollback()
+            raise
+
+    @classmethod
+    async def edit_pallet(
+        cls,
+        db: AsyncSession,
+        pallet: PalletEditModel,
+        current_user_name: str,
+    ) -> CrudResponseModel:
+        try:
+            _, inventory = await cls._lock_pallet_inventory(db, pallet.pallet_id)
+            if pallet.capacity_packages < inventory.remaining_packages:
+                raise ServiceException(message='托盘容量不能小于当前剩余包数')
+            await MasterDao.update_pallet(
+                db,
+                pallet,
+                update_by=current_user_name,
+                update_time=datetime.now(),
+            )
+            await db.commit()
+            return CrudResponseModel(is_success=True, message='更新成功')
+        except IntegrityError:
+            await db.rollback()
+            raise ServiceException(message='修改托盘失败，托盘状态已变化') from None
+        except Exception:
+            await db.rollback()
+            raise
+
+    @classmethod
+    async def remove_pallet(
+        cls,
+        db: AsyncSession,
+        value: int | DeletePalletModel,
+    ) -> CrudResponseModel:
+        try:
+            pallet_ids = cls._extract_pallet_ids(value)
+            for pallet_id in pallet_ids:
+                _, inventory = await cls._lock_pallet_inventory(db, pallet_id)
+                if inventory.flow_state != PalletFlowState.UNSTOCKED.value:
+                    raise ServiceException(message='托盘未处于未入库状态，不能删除')
+                if await MasterDao.count_pallet_events(db, pallet_id) > 0:
+                    raise ServiceException(message='托盘已有流转事件，不能删除')
+
+            for pallet_id in pallet_ids:
+                await MasterDao.delete_pallet_inventory(db, pallet_id)
+                await MasterDao.delete_pallet(db, pallet_id)
+            await db.commit()
+            return CrudResponseModel(is_success=True, message='删除成功')
+        except IntegrityError:
+            await db.rollback()
+            raise ServiceException(message='删除托盘失败，托盘状态或关联数据已变化') from None
+        except Exception:
+            await db.rollback()
+            raise
+
+    @classmethod
+    async def _list_resource_users(
+        cls,
+        db: AsyncSession,
+        resource: AssignmentResource,
+        resource_id: int,
+    ) -> list[BusinessUserOptionModel]:
+        if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None:
+            raise ServiceException(message=f'{_ASSIGNMENT_LABELS[resource]}不存在')
+
+        users = await MasterDao.list_assignable_users(db)
+        result = []
+        association_method = getattr(AccessDao, f'list_user_{resource}_ids')
+        for user in users:
+            assigned_ids = await association_method(db, user.user_id)
+            result.append(
+                BusinessUserOptionModel(
+                    user_id=user.user_id,
+                    user_name=user.user_name,
+                    nick_name=user.nick_name,
+                    assigned=resource_id in assigned_ids,
+                )
+            )
+        return result
+
+    @classmethod
+    async def list_warehouse_users(
+        cls,
+        db: AsyncSession,
+        warehouse_id: int,
+    ) -> list[BusinessUserOptionModel]:
+        return await cls._list_resource_users(db, 'warehouse', warehouse_id)
+
+    @classmethod
+    async def list_site_users(
+        cls,
+        db: AsyncSession,
+        site_id: int,
+    ) -> list[BusinessUserOptionModel]:
+        return await cls._list_resource_users(db, 'site', site_id)
+
+    @classmethod
+    async def _assign_resource_users(
+        cls,
+        db: AsyncSession,
+        resource: AssignmentResource,
+        resource_id: int,
+        assignment: UserAssignModel,
+    ) -> CrudResponseModel:
+        label = _ASSIGNMENT_LABELS[resource]
+        try:
+            if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None:
+                raise ServiceException(message=f'{label}不存在')
+            users = await MasterDao.list_assignable_users_by_ids(db, assignment.user_ids)
+            if {user.user_id for user in users} != set(assignment.user_ids):
+                raise ServiceException(message='存在不存在、停用或已删除的用户')
+            await getattr(AccessDao, f'replace_{resource}_users')(db, resource_id, assignment.user_ids)
+            await db.commit()
+            return CrudResponseModel(is_success=True, message='分配成功')
+        except IntegrityError:
+            await db.rollback()
+            raise ServiceException(message=f'分配{label}用户失败，关联数据无效') from None
+        except Exception:
+            await db.rollback()
+            raise
+
+    @classmethod
+    async def assign_warehouse_users(
+        cls,
+        db: AsyncSession,
+        warehouse_id: int,
+        assignment: UserAssignModel,
+    ) -> CrudResponseModel:
+        return await cls._assign_resource_users(db, 'warehouse', warehouse_id, assignment)
+
+    @classmethod
+    async def assign_site_users(
+        cls,
+        db: AsyncSession,
+        site_id: int,
+        assignment: UserAssignModel,
+    ) -> CrudResponseModel:
+        return await cls._assign_resource_users(db, 'site', site_id, assignment)

shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py

@@ -0,0 +1,979 @@
+import inspect
+import secrets
+import unittest
+from datetime import datetime
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi import FastAPI
+from fastapi.routing import APIRoute
+from pydantic import ValidationError
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.aspect.interface_auth import CheckUserInterfaceAuth
+from common.aspect.pre_auth import PreAuth
+from common.vo import CrudResponseModel, DataResponseModel, PageModel, PageResponseModel, ResponseBaseModel
+from exceptions.exception import ServiceException
+from module_admin.entity.do.user_do import SysUser
+from module_admin.service.login_service import LoginService
+from module_business.controller import pallet_controller as pallet_controller_module
+from module_business.controller import site_controller as site_controller_module
+from module_business.controller import warehouse_controller as warehouse_controller_module
+from module_business.dao.access_dao import AccessDao
+from module_business.dao.master_dao import MasterDao
+from module_business.entity.do.master_do import BizPallet
+from module_business.entity.do.operation_do import BizPalletInventory
+from module_business.entity.vo.master_vo import (
+    BusinessUserOptionModel,
+    DeletePalletModel,
+    PalletCreateModel,
+    PalletEditModel,
+    PalletModel,
+    PalletPageQueryModel,
+    UserAssignModel,
+)
+from module_business.service import pallet_service as pallet_service_module
+from module_business.service.pallet_service import PalletService
+from utils.page_util import PageUtil
+
+
+def compile_sql(statement: Any) -> str:
+    return ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split())
+
+
+def integrity_error() -> IntegrityError:
+    return IntegrityError('statement', {}, RuntimeError('constraint'))
+
+
+def make_pallet_record(pallet_id: int, capacity_packages: int = 20) -> SimpleNamespace:
+    return SimpleNamespace(
+        pallet_id=pallet_id,
+        pallet_code=f'P-{pallet_id:03d}',
+        qr_token=f'fixed-token-{pallet_id}',
+        capacity_packages=capacity_packages,
+        asset_status='0',
+        remark=None,
+    )
+
+
+class TestPalletServiceContract(unittest.TestCase):
+    def test_service_is_available(self) -> None:
+        self.assertIsNotNone(PalletService)
+
+
+class TestPalletVoContract(unittest.TestCase):
+    def test_business_user_option_only_exposes_safe_fields(self) -> None:
+        source = SimpleNamespace(
+            user_id=7,
+            user_name='operator',
+            nick_name=None,
+            assigned=True,
+            password='secret',
+            email='operator@example.com',
+            phonenumber='13800000000',
+            token='token',
+        )
+
+        option = BusinessUserOptionModel.model_validate(source)
+
+        self.assertEqual(
+            {'userId': 7, 'userName': 'operator', 'nickName': None, 'assigned': True},
+            option.model_dump(by_alias=True),
+        )
+        self.assertTrue(
+            {'password', 'email', 'phonenumber', 'phone', 'token'}.isdisjoint(BusinessUserOptionModel.model_fields)
+        )
+
+    def test_business_user_option_validates_identity_fields(self) -> None:
+        invalid_values = (
+            {'userId': 0, 'userName': 'operator', 'assigned': False},
+            {'userId': 1, 'userName': '', 'assigned': False},
+            {'userId': 1, 'userName': '   ', 'assigned': False},
+        )
+        for value in invalid_values:
+            with self.subTest(value=value), self.assertRaises(ValidationError):
+                BusinessUserOptionModel.model_validate(value)
+
+    def test_pallet_edit_rejects_code_and_qr_token(self) -> None:
+        base = {'palletId': 1, 'capacityPackages': 20, 'assetStatus': '0'}
+        for identity_field in ({'palletCode': 'P-001'}, {'qrToken': 'fixed-token'}):
+            with self.subTest(identity_field=identity_field), self.assertRaises(ValidationError):
+                PalletEditModel.model_validate({**base, **identity_field})
+
+
+class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.db = MagicMock(spec=AsyncSession)
+
+    async def test_list_pallet_filters_and_orders_by_pallet_id(self) -> None:
+        query = PalletPageQueryModel(palletCode='P-00', assetStatus='0')
+        expected = object()
+
+        with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=expected)) as paginate:
+            actual = await MasterDao.list_pallet(self.db, query, is_page=True)
+
+        self.assertIs(expected, actual)
+        statement = paginate.await_args.args[1]
+        sql = compile_sql(statement)
+        self.assertIn('FROM biz_pallet', sql)
+        self.assertIn("biz_pallet.pallet_code LIKE '%P-00%'", sql)
+        self.assertIn("biz_pallet.asset_status = '0'", sql)
+        self.assertIn('ORDER BY biz_pallet.pallet_id', sql)
+
+    async def test_add_pallet_and_inventory_only_flush_current_transaction(self) -> None:
+        pallet = BizPallet(
+            pallet_code='P-001',
+            qr_token='fixed-token',
+            capacity_packages=20,
+            asset_status='0',
+        )
+        inventory = BizPalletInventory(
+            pallet_id=3,
+            flow_state='UNSTOCKED',
+            remaining_packages=0,
+            version=0,
+        )
+
+        actual = await MasterDao.add_pallet(self.db, pallet)
+        await MasterDao.add_pallet_inventory(self.db, inventory)
+
+        self.assertIs(pallet, actual)
+        self.assertEqual([pallet, inventory], [call.args[0] for call in self.db.add.call_args_list])
+        self.assertEqual(2, self.db.flush.await_count)
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_not_awaited()
+
+    async def test_update_pallet_sql_never_contains_identity_fields(self) -> None:
+        edit = PalletEditModel(palletId=3, capacityPackages=18, assetStatus='1', remark='停用')
+        updated_at = datetime(2026, 7, 26, 10, 30)
+
+        await MasterDao.update_pallet(self.db, edit, update_by='admin', update_time=updated_at)
+
+        statement = self.db.execute.await_args.args[0]
+        sql = compile_sql(statement)
+        self.assertIn('UPDATE biz_pallet', sql)
+        self.assertIn('capacity_packages=18', sql)
+        self.assertIn("asset_status='1'", sql)
+        self.assertIn("remark='停用'", sql)
+        self.assertIn("update_by='admin'", sql)
+        self.assertIn('WHERE biz_pallet.pallet_id = 3', sql)
+        self.assertNotIn('pallet_code=', sql)
+        self.assertNotIn('qr_token=', sql)
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_not_awaited()
+
+    async def test_inventory_delete_and_event_count_target_one_pallet(self) -> None:
+        count_result = MagicMock()
+        count_result.scalar.return_value = 2
+        self.db.execute.side_effect = [MagicMock(), count_result]
+
+        await MasterDao.delete_pallet_inventory(self.db, pallet_id=9)
+        count = await MasterDao.count_pallet_events(self.db, pallet_id=9)
+
+        delete_statement = self.db.execute.await_args_list[0].args[0]
+        self.assertTrue(delete_statement.table.compare(BizPalletInventory.__table__))
+        self.assertIn('biz_pallet_inventory.pallet_id = 9', compile_sql(delete_statement))
+        count_statement = self.db.execute.await_args_list[1].args[0]
+        self.assertIn('FROM biz_pallet_event', compile_sql(count_statement))
+        self.assertIn('biz_pallet_event.pallet_id = 9', compile_sql(count_statement))
+        self.assertEqual(2, count)
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_not_awaited()
+
+    async def test_inventory_write_read_uses_row_lock(self) -> None:
+        result = MagicMock()
+        inventory = SimpleNamespace(pallet_id=9)
+        result.scalar_one_or_none.return_value = inventory
+        self.db.execute.return_value = result
+
+        actual = await MasterDao.get_pallet_inventory(self.db, pallet_id=9, for_update=True)
+
+        self.assertIs(inventory, actual)
+        statement = self.db.execute.await_args.args[0]
+        self.assertIn('biz_pallet_inventory.pallet_id = 9', compile_sql(statement))
+        self.assertIn('FOR UPDATE', compile_sql(statement))
+
+    async def test_pallet_write_read_uses_row_lock(self) -> None:
+        result = MagicMock()
+        pallet = make_pallet_record(9)
+        result.scalar_one_or_none.return_value = pallet
+        self.db.execute.return_value = result
+
+        actual = await MasterDao.get_pallet(self.db, pallet_id=9, for_update=True)
+
+        self.assertIs(pallet, actual)
+        statement = self.db.execute.await_args.args[0]
+        self.assertIn('biz_pallet.pallet_id = 9', compile_sql(statement))
+        self.assertIn('FOR UPDATE', compile_sql(statement))
+
+    async def test_assignable_user_queries_filter_disabled_and_deleted_rows(self) -> None:
+        list_result = MagicMock()
+        list_result.all.return_value = ()
+        selected_result = MagicMock()
+        selected_result.all.return_value = ()
+        self.db.execute.side_effect = [list_result, selected_result]
+
+        self.assertEqual([], await MasterDao.list_assignable_users(self.db))
+        self.assertEqual([], await MasterDao.list_assignable_users_by_ids(self.db, [2, 5]))
+
+        for statement in (call.args[0] for call in self.db.execute.await_args_list):
+            sql = compile_sql(statement)
+            self.assertIn('FROM sys_user', sql)
+            self.assertIn("sys_user.status = '0'", sql)
+            self.assertIn("sys_user.del_flag = '0'", sql)
+            self.assertNotIn('password', sql)
+            self.assertEqual(
+                [SysUser.user_id, SysUser.user_name, SysUser.nick_name],
+                list(statement.selected_columns),
+            )
+        self.assertIn('ORDER BY sys_user.user_id', compile_sql(self.db.execute.await_args_list[0].args[0]))
+        selected_sql = compile_sql(self.db.execute.await_args_list[1].args[0])
+        self.assertIn('sys_user.user_id IN (2, 5)', selected_sql)
+
+
+class TestPalletCreationService(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.db = MagicMock(spec=AsyncSession)
+        self.create = PalletCreateModel(palletCode='P-001', capacityPackages=20, remark='测试托盘')
+        self.fixed_time = datetime(2026, 7, 26, 11, 30)
+
+    async def test_add_generates_server_token_and_initial_inventory_in_one_transaction(self) -> None:
+        actions: list[str] = []
+        captured: dict[str, Any] = {}
+
+        async def add_pallet(_db: AsyncSession, pallet: BizPallet) -> BizPallet:
+            actions.append('pallet')
+            captured['pallet'] = pallet
+            pallet.pallet_id = 17
+            return pallet
+
+        async def add_inventory(_db: AsyncSession, inventory: BizPalletInventory) -> BizPalletInventory:
+            actions.append('inventory')
+            captured['inventory'] = inventory
+            return inventory
+
+        async def commit() -> None:
+            actions.append('commit')
+
+        self.db.commit.side_effect = commit
+        with (
+            patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
+            patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=add_pallet)),
+            patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock(side_effect=add_inventory)),
+            patch.object(secrets, 'token_urlsafe', return_value='server-fixed-token') as token_urlsafe,
+            patch.object(pallet_service_module, 'datetime', create=True) as service_datetime,
+        ):
+            service_datetime.now.return_value = self.fixed_time
+            result = await PalletService.add_pallet(self.db, self.create, current_user_name='admin')
+
+        token_urlsafe.assert_called_once_with(32)
+        pallet = captured['pallet']
+        self.assertEqual('P-001', pallet.pallet_code)
+        self.assertEqual('server-fixed-token', pallet.qr_token)
+        self.assertEqual(20, pallet.capacity_packages)
+        self.assertEqual('0', pallet.asset_status)
+        self.assertEqual('admin', pallet.create_by)
+        self.assertEqual(self.fixed_time, pallet.create_time)
+        self.assertEqual('admin', pallet.update_by)
+        self.assertEqual(self.fixed_time, pallet.update_time)
+        inventory = captured['inventory']
+        self.assertEqual(17, inventory.pallet_id)
+        self.assertEqual('UNSTOCKED', inventory.flow_state)
+        self.assertEqual(0, inventory.remaining_packages)
+        self.assertEqual(0, inventory.version)
+        self.assertEqual(['pallet', 'inventory', 'commit'], actions)
+        self.db.commit.assert_awaited_once()
+        self.db.rollback.assert_not_awaited()
+        self.assertIsInstance(result, CrudResponseModel)
+        self.assertTrue(result.is_success)
+
+    async def test_add_rolls_back_when_pallet_flush_fails(self) -> None:
+        failure = RuntimeError('flush failed')
+        add_inventory = AsyncMock()
+        with (
+            patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
+            patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=failure)),
+            patch.object(MasterDao, 'add_pallet_inventory', new=add_inventory),
+            self.assertRaisesRegex(RuntimeError, 'flush failed'),
+        ):
+            await PalletService.add_pallet(self.db, self.create, current_user_name='admin')
+
+        add_inventory.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_add_rolls_back_when_commit_fails(self) -> None:
+        async def assign_id(_db: AsyncSession, pallet: BizPallet) -> BizPallet:
+            pallet.pallet_id = 17
+            return pallet
+
+        self.db.commit.side_effect = RuntimeError('commit failed')
+        with (
+            patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
+            patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=assign_id)),
+            patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock()),
+            self.assertRaisesRegex(RuntimeError, 'commit failed'),
+        ):
+            await PalletService.add_pallet(self.db, self.create, current_user_name='admin')
+
+        self.db.rollback.assert_awaited_once()
+
+    async def test_add_converts_integrity_error_to_stable_service_exception(self) -> None:
+        with (
+            patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
+            patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=integrity_error())),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.add_pallet(self.db, self.create, current_user_name='admin')
+
+        self.assertEqual('新增托盘失败，托盘编号或二维码标识已存在', raised.exception.message)
+        self.db.rollback.assert_awaited_once()
+        self.db.commit.assert_not_awaited()
+
+    async def test_add_rejects_duplicate_code_before_writing(self) -> None:
+        add_pallet = AsyncMock()
+        with (
+            patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=object())),
+            patch.object(MasterDao, 'add_pallet', new=add_pallet),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.add_pallet(self.db, self.create, current_user_name='admin')
+
+        self.assertEqual('托盘编号已存在', raised.exception.message)
+        add_pallet.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+
+class TestPalletQrService(unittest.IsolatedAsyncioTestCase):
+    def test_render_qr_png_returns_real_stable_png(self) -> None:
+        first = PalletService.render_qr_png('fixed-original-token')
+        second = PalletService.render_qr_png('fixed-original-token')
+
+        self.assertEqual(b'\x89PNG\r\n\x1a\n', first[:8])
+        self.assertEqual(first, second)
+
+    async def test_get_qrcode_uses_original_persisted_token(self) -> None:
+        db = MagicMock(spec=AsyncSession)
+        with (
+            patch.object(
+                MasterDao,
+                'get_pallet',
+                new=AsyncMock(return_value=SimpleNamespace(qr_token='persisted-token')),
+            ),
+            patch.object(PalletService, 'render_qr_png', return_value=b'png-bytes') as render,
+            patch.object(secrets, 'token_urlsafe') as token_urlsafe,
+        ):
+            actual = await PalletService.get_pallet_qrcode(db, pallet_id=7)
+
+        self.assertEqual(b'png-bytes', actual)
+        render.assert_called_once_with('persisted-token')
+        token_urlsafe.assert_not_called()
+
+    async def test_get_qrcode_rejects_missing_pallet(self) -> None:
+        db = MagicMock(spec=AsyncSession)
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.get_pallet_qrcode(db, pallet_id=404)
+
+        self.assertEqual('托盘不存在', raised.exception.message)
+
+
+class TestPalletQueryAndEditService(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.db = MagicMock(spec=AsyncSession)
+
+    async def test_list_computes_can_remove_from_inventory_and_events(self) -> None:
+        page = PageModel[Any](
+            rows=[make_pallet_record(1), make_pallet_record(2)],
+            pageNum=1,
+            pageSize=10,
+            total=2,
+            hasNext=False,
+        )
+        query = PalletPageQueryModel()
+        with (
+            patch.object(MasterDao, 'list_pallet', new=AsyncMock(return_value=page)),
+            patch.object(
+                MasterDao,
+                'get_pallet_inventory',
+                new=AsyncMock(
+                    side_effect=[
+                        SimpleNamespace(flow_state='UNSTOCKED'),
+                        SimpleNamespace(flow_state='UNSTOCKED'),
+                    ]
+                ),
+            ),
+            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=[0, 1])),
+        ):
+            actual = await PalletService.list_pallet(self.db, query, is_page=True)
+
+        self.assertIsInstance(actual, PageModel)
+        self.assertTrue(all(isinstance(row, PalletModel) for row in actual.rows))
+        self.assertEqual([True, False], [row.can_remove for row in actual.rows])
+
+    async def test_detail_computes_false_when_inventory_is_not_unstocked(self) -> None:
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
+            patch.object(
+                MasterDao,
+                'get_pallet_inventory',
+                new=AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE')),
+            ),
+            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)),
+        ):
+            actual = await PalletService.detail_pallet(self.db, pallet_id=3)
+
+        self.assertIsInstance(actual, PalletModel)
+        self.assertFalse(actual.can_remove)
+
+    async def test_detail_rejects_missing_pallet(self) -> None:
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.detail_pallet(self.db, pallet_id=404)
+
+        self.assertEqual('托盘不存在', raised.exception.message)
+
+    async def test_edit_updates_only_mutable_fields_with_server_audit(self) -> None:
+        edit = PalletEditModel(palletId=3, capacityPackages=12, assetStatus='1', remark='停用')
+        updated_at = datetime(2026, 7, 26, 12, 30)
+        update_pallet = AsyncMock()
+        lock_actions: list[str] = []
+
+        async def get_pallet(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
+            lock_actions.append('pallet')
+            return make_pallet_record(3)
+
+        async def get_inventory(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
+            lock_actions.append('inventory')
+            return SimpleNamespace(remaining_packages=12)
+
+        get_pallet_mock = AsyncMock(side_effect=get_pallet)
+        get_inventory_mock = AsyncMock(side_effect=get_inventory)
+        with (
+            patch.object(MasterDao, 'get_pallet', new=get_pallet_mock),
+            patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory_mock),
+            patch.object(MasterDao, 'update_pallet', new=update_pallet),
+            patch.object(pallet_service_module, 'datetime') as service_datetime,
+        ):
+            service_datetime.now.return_value = updated_at
+            result = await PalletService.edit_pallet(self.db, edit, current_user_name='admin')
+
+        update_pallet.assert_awaited_once_with(
+            self.db,
+            edit,
+            update_by='admin',
+            update_time=updated_at,
+        )
+        self.assertEqual(['inventory', 'pallet'], lock_actions)
+        get_pallet_mock.assert_awaited_once_with(self.db, 3, for_update=True)
+        get_inventory_mock.assert_awaited_once_with(self.db, 3, for_update=True)
+        self.db.commit.assert_awaited_once()
+        self.db.rollback.assert_not_awaited()
+        self.assertTrue(result.is_success)
+
+    async def test_edit_rejects_capacity_below_remaining_packages(self) -> None:
+        edit = PalletEditModel(palletId=3, capacityPackages=11, assetStatus='0')
+        update_pallet = AsyncMock()
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
+            patch.object(
+                MasterDao,
+                'get_pallet_inventory',
+                new=AsyncMock(return_value=SimpleNamespace(remaining_packages=12)),
+            ),
+            patch.object(MasterDao, 'update_pallet', new=update_pallet),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.edit_pallet(self.db, edit, current_user_name='admin')
+
+        self.assertEqual('托盘容量不能小于当前剩余包数', raised.exception.message)
+        update_pallet.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_edit_rejects_missing_inventory(self) -> None:
+        edit = PalletEditModel(palletId=3, capacityPackages=20, assetStatus='0')
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
+            patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.edit_pallet(self.db, edit, current_user_name='admin')
+
+        self.assertEqual('托盘库存不存在', raised.exception.message)
+        self.db.rollback.assert_awaited_once()
+
+
+class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.db = MagicMock(spec=AsyncSession)
+
+    async def test_batch_remove_preflights_all_then_deletes_inventory_before_pallet(self) -> None:
+        actions: list[str] = []
+
+        async def get_pallet(
+            _db: AsyncSession,
+            pallet_id: int,
+            *,
+            for_update: bool = False,
+        ) -> SimpleNamespace:
+            actions.append(f'get-pallet-{pallet_id}-{for_update}')
+            return make_pallet_record(pallet_id)
+
+        async def get_inventory(
+            _db: AsyncSession,
+            pallet_id: int,
+            *,
+            for_update: bool = False,
+        ) -> SimpleNamespace:
+            actions.append(f'get-inventory-{pallet_id}-{for_update}')
+            return SimpleNamespace(flow_state='UNSTOCKED')
+
+        async def count_events(_db: AsyncSession, pallet_id: int) -> int:
+            actions.append(f'count-events-{pallet_id}')
+            return 0
+
+        async def delete_inventory(_db: AsyncSession, pallet_id: int) -> None:
+            actions.append(f'delete-inventory-{pallet_id}')
+
+        async def delete_pallet(_db: AsyncSession, pallet_id: int) -> None:
+            actions.append(f'delete-pallet-{pallet_id}')
+
+        async def commit() -> None:
+            actions.append('commit')
+
+        self.db.commit.side_effect = commit
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(side_effect=get_pallet)),
+            patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(side_effect=get_inventory)),
+            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=count_events)),
+            patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock(side_effect=delete_inventory)),
+            patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=delete_pallet)),
+        ):
+            result = await PalletService.remove_pallet(self.db, DeletePalletModel(palletIds=[5, 4]))
+
+        self.assertEqual(
+            [
+                'get-inventory-4-True',
+                'get-pallet-4-True',
+                'count-events-4',
+                'get-inventory-5-True',
+                'get-pallet-5-True',
+                'count-events-5',
+                'delete-inventory-4',
+                'delete-pallet-4',
+                'delete-inventory-5',
+                'delete-pallet-5',
+                'commit',
+            ],
+            actions,
+        )
+        self.db.commit.assert_awaited_once()
+        self.db.rollback.assert_not_awaited()
+        self.assertTrue(result.is_success)
+
+    async def test_remove_rejects_invalid_state_before_any_delete(self) -> None:
+        cases = (
+            ('托盘不存在', None, None, 0),
+            ('托盘库存不存在', make_pallet_record(4), None, 0),
+            (
+                '托盘未处于未入库状态，不能删除',
+                make_pallet_record(4),
+                SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE'),
+                0,
+            ),
+            (
+                '托盘已有流转事件，不能删除',
+                make_pallet_record(4),
+                SimpleNamespace(flow_state='UNSTOCKED'),
+                1,
+            ),
+        )
+        for message, pallet, inventory, event_count in cases:
+            with self.subTest(message=message):
+                self.db.reset_mock()
+                delete_inventory = AsyncMock()
+                delete_pallet = AsyncMock()
+                with (
+                    patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)),
+                    patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)),
+                    patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=event_count)),
+                    patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory),
+                    patch.object(MasterDao, 'delete_pallet', new=delete_pallet),
+                    self.assertRaises(ServiceException) as raised,
+                ):
+                    await PalletService.remove_pallet(self.db, 4)
+
+                self.assertEqual(message, raised.exception.message)
+                delete_inventory.assert_not_awaited()
+                delete_pallet.assert_not_awaited()
+                self.db.commit.assert_not_awaited()
+                self.db.rollback.assert_awaited_once()
+
+    async def test_remove_integrity_error_rolls_back_with_stable_failure(self) -> None:
+        with (
+            patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(4))),
+            patch.object(
+                MasterDao,
+                'get_pallet_inventory',
+                new=AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED')),
+            ),
+            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)),
+            patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock()),
+            patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=integrity_error())),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.remove_pallet(self.db, 4)
+
+        self.assertEqual('删除托盘失败，托盘状态或关联数据已变化', raised.exception.message)
+        self.db.rollback.assert_awaited_once()
+        self.db.commit.assert_not_awaited()
+
+
+class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
+    def setUp(self) -> None:
+        self.db = MagicMock(spec=AsyncSession)
+        self.users = [
+            SimpleNamespace(
+                user_id=2,
+                user_name='alice',
+                nick_name='艾丽丝',
+                password='secret-a',
+                email='alice@example.com',
+                phonenumber='13800000000',
+            ),
+            SimpleNamespace(
+                user_id=5,
+                user_name='bob',
+                nick_name=None,
+                password='secret-b',
+                email='bob@example.com',
+                phonenumber='13900000000',
+            ),
+        ]
+
+    async def test_assignment_get_returns_only_safe_fields_with_assigned_marker(self) -> None:
+        cases = (
+            ('warehouse', 7, [7], []),
+            ('site', 9, [9], []),
+        )
+        for resource, resource_id, first_ids, second_ids in cases:
+            with self.subTest(resource=resource):
+                association_method = f'list_user_{resource}_ids'
+                with (
+                    patch.object(
+                        MasterDao,
+                        f'get_{resource}',
+                        new=AsyncMock(return_value=SimpleNamespace()),
+                    ),
+                    patch.object(MasterDao, 'list_assignable_users', new=AsyncMock(return_value=self.users)),
+                    patch.object(
+                        AccessDao,
+                        association_method,
+                        new=AsyncMock(side_effect=[first_ids, second_ids]),
+                    ) as list_ids,
+                ):
+                    actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id)
+
+                self.assertEqual(
+                    [
+                        {'userId': 2, 'userName': 'alice', 'nickName': '艾丽丝', 'assigned': True},
+                        {'userId': 5, 'userName': 'bob', 'nickName': None, 'assigned': False},
+                    ],
+                    [option.model_dump(by_alias=True) for option in actual],
+                )
+                self.assertTrue(all(isinstance(option, BusinessUserOptionModel) for option in actual))
+                self.assertEqual(
+                    [(self.db, 2), (self.db, 5)],
+                    [call.args for call in list_ids.await_args_list],
+                )
+
+    async def test_assignment_get_rejects_missing_resource_before_user_query(self) -> None:
+        cases = (('warehouse', 7, '仓库不存在'), ('site', 9, '客户现场不存在'))
+        for resource, resource_id, message in cases:
+            with self.subTest(resource=resource):
+                list_users = AsyncMock()
+                with (
+                    patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)),
+                    patch.object(MasterDao, 'list_assignable_users', new=list_users),
+                    self.assertRaises(ServiceException) as raised,
+                ):
+                    await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id)
+
+                self.assertEqual(message, raised.exception.message)
+                list_users.assert_not_awaited()
+
+    async def test_assignment_put_validates_users_replaces_and_commits(self) -> None:
+        assignment = UserAssignModel(userIds=[2, 5])
+        for resource, resource_id in (('warehouse', 7), ('site', 9)):
+            with self.subTest(resource=resource):
+                self.db.reset_mock()
+                replace = AsyncMock()
+                with (
+                    patch.object(
+                        MasterDao,
+                        f'get_{resource}',
+                        new=AsyncMock(return_value=SimpleNamespace()),
+                    ),
+                    patch.object(
+                        MasterDao,
+                        'list_assignable_users_by_ids',
+                        new=AsyncMock(return_value=self.users),
+                    ) as list_users,
+                    patch.object(AccessDao, f'replace_{resource}_users', new=replace),
+                ):
+                    result = await getattr(PalletService, f'assign_{resource}_users')(
+                        self.db,
+                        resource_id,
+                        assignment,
+                    )
+
+                list_users.assert_awaited_once_with(self.db, [2, 5])
+                replace.assert_awaited_once_with(self.db, resource_id, [2, 5])
+                self.db.commit.assert_awaited_once()
+                self.db.rollback.assert_not_awaited()
+                self.assertTrue(result.is_success)
+
+    async def test_assignment_put_rejects_missing_disabled_or_deleted_user(self) -> None:
+        replace = AsyncMock()
+        assignment = UserAssignModel(userIds=[2, 5])
+        with (
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())),
+            patch.object(
+                MasterDao,
+                'list_assignable_users_by_ids',
+                new=AsyncMock(return_value=[self.users[0]]),
+            ),
+            patch.object(AccessDao, 'replace_warehouse_users', new=replace),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.assign_warehouse_users(self.db, 7, assignment)
+
+        self.assertEqual('存在不存在、停用或已删除的用户', raised.exception.message)
+        replace.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_replace_failure_rolls_back_and_preserves_original_exception(self) -> None:
+        assignment = UserAssignModel(userIds=[2, 5])
+        with (
+            patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace())),
+            patch.object(
+                MasterDao,
+                'list_assignable_users_by_ids',
+                new=AsyncMock(return_value=self.users),
+            ),
+            patch.object(
+                AccessDao,
+                'replace_site_users',
+                new=AsyncMock(side_effect=RuntimeError('replace failed')),
+            ),
+            self.assertRaisesRegex(RuntimeError, 'replace failed'),
+        ):
+            await PalletService.assign_site_users(self.db, 9, assignment)
+
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_assignment_commit_failure_rolls_back(self) -> None:
+        assignment = UserAssignModel(userIds=[2, 5])
+        self.db.commit.side_effect = RuntimeError('commit failed')
+        with (
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())),
+            patch.object(
+                MasterDao,
+                'list_assignable_users_by_ids',
+                new=AsyncMock(return_value=self.users),
+            ),
+            patch.object(AccessDao, 'replace_warehouse_users', new=AsyncMock()),
+            self.assertRaisesRegex(RuntimeError, 'commit failed'),
+        ):
+            await PalletService.assign_warehouse_users(self.db, 7, assignment)
+
+        self.db.rollback.assert_awaited_once()
+
+    async def test_assignment_integrity_error_has_stable_message_and_rolls_back(self) -> None:
+        assignment = UserAssignModel(userIds=[2, 5])
+        with (
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())),
+            patch.object(
+                MasterDao,
+                'list_assignable_users_by_ids',
+                new=AsyncMock(return_value=self.users),
+            ),
+            patch.object(
+                AccessDao,
+                'replace_warehouse_users',
+                new=AsyncMock(side_effect=integrity_error()),
+            ),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.assign_warehouse_users(self.db, 7, assignment)
+
+        self.assertEqual('分配仓库用户失败，关联数据无效', raised.exception.message)
+        self.db.rollback.assert_awaited_once()
+
+
+class TestPalletAndAssignmentControllers(unittest.IsolatedAsyncioTestCase):
+    @staticmethod
+    def _routers() -> tuple[Any, ...]:
+        return (
+            pallet_controller_module.pallet_controller,
+            warehouse_controller_module.warehouse_user_controller,
+            site_controller_module.site_user_controller,
+        )
+
+    @classmethod
+    def _routes(cls) -> dict[tuple[str, str], APIRoute]:
+        routes = {}
+        for router in cls._routers():
+            for route in router.routes:
+                if isinstance(route, APIRoute):
+                    routes[(route.path, next(iter(route.methods)))] = route
+        return routes
+
+    def test_openapi_has_all_pallet_and_assignment_operations(self) -> None:
+        app = FastAPI()
+        for router in self._routers():
+            app.include_router(router)
+        schema = app.openapi()
+
+        expected_operations = {
+            ('/master/pallet/list', 'get'),
+            ('/master/pallet/{pallet_id}', 'get'),
+            ('/master/pallet', 'post'),
+            ('/master/pallet', 'put'),
+            ('/master/pallet/{pallet_ids}', 'delete'),
+            ('/master/pallet/{pallet_id}/qrcode', 'get'),
+            ('/master/warehouse/{warehouse_id}/users', 'get'),
+            ('/master/warehouse/{warehouse_id}/users', 'put'),
+            ('/master/site/{site_id}/users', 'get'),
+            ('/master/site/{site_id}/users', 'put'),
+        }
+        for path, method in expected_operations:
+            with self.subTest(path=path, method=method):
+                self.assertIn(path, schema['paths'])
+                self.assertIn(method, schema['paths'][path])
+        self.assertIn(
+            'image/png',
+            schema['paths']['/master/pallet/{pallet_id}/qrcode']['get']['responses']['200']['content'],
+        )
+
+    def test_route_response_models_are_exact(self) -> None:
+        routes = self._routes()
+        self.assertEqual(
+            PageResponseModel[PalletModel],
+            routes[('/master/pallet/list', 'GET')].response_model,
+        )
+        self.assertEqual(
+            DataResponseModel[PalletModel],
+            routes[('/master/pallet/{pallet_id}', 'GET')].response_model,
+        )
+        for path, method in (
+            ('/master/pallet', 'POST'),
+            ('/master/pallet', 'PUT'),
+            ('/master/pallet/{pallet_ids}', 'DELETE'),
+            ('/master/warehouse/{warehouse_id}/users', 'PUT'),
+            ('/master/site/{site_id}/users', 'PUT'),
+        ):
+            self.assertEqual(ResponseBaseModel, routes[(path, method)].response_model)
+        user_response = DataResponseModel[list[BusinessUserOptionModel]]
+        self.assertEqual(
+            user_response,
+            routes[('/master/warehouse/{warehouse_id}/users', 'GET')].response_model,
+        )
+        self.assertEqual(
+            user_response,
+            routes[('/master/site/{site_id}/users', 'GET')].response_model,
+        )
+        self.assertIsNone(routes[('/master/pallet/{pallet_id}/qrcode', 'GET')].response_model)
+
+    def test_permissions_are_exact_and_all_routes_require_pre_auth(self) -> None:
+        expected_permissions = {
+            ('/master/pallet/list', 'GET'): 'master:pallet:list',
+            ('/master/pallet/{pallet_id}', 'GET'): 'master:pallet:query',
+            ('/master/pallet', 'POST'): 'master:pallet:add',
+            ('/master/pallet', 'PUT'): 'master:pallet:edit',
+            ('/master/pallet/{pallet_ids}', 'DELETE'): 'master:pallet:remove',
+            ('/master/pallet/{pallet_id}/qrcode', 'GET'): 'master:pallet:print',
+            ('/master/warehouse/{warehouse_id}/users', 'GET'): 'master:warehouse:assign',
+            ('/master/warehouse/{warehouse_id}/users', 'PUT'): 'master:warehouse:assign',
+            ('/master/site/{site_id}/users', 'GET'): 'master:site:assign',
+            ('/master/site/{site_id}/users', 'PUT'): 'master:site:assign',
+        }
+
+        actual_permissions = {}
+        for key, route in self._routes().items():
+            dependencies = [
+                dependency.dependency
+                for dependency in route.dependencies
+                if isinstance(dependency.dependency, CheckUserInterfaceAuth)
+            ]
+            self.assertEqual(1, len(dependencies))
+            self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies))
+            actual_permissions[key] = dependencies[0].perm
+
+        self.assertEqual(expected_permissions, actual_permissions)
+        self.assertEqual(8, len(set(actual_permissions.values())))
+
+    def test_write_routes_have_log_and_current_user(self) -> None:
+        write_routes = (
+            ('/master/pallet', 'POST'),
+            ('/master/pallet', 'PUT'),
+            ('/master/pallet/{pallet_ids}', 'DELETE'),
+            ('/master/warehouse/{warehouse_id}/users', 'PUT'),
+            ('/master/site/{site_id}/users', 'PUT'),
+        )
+        routes = self._routes()
+        for key in write_routes:
+            route = routes[key]
+            self.assertTrue(hasattr(route.endpoint, '__wrapped__'))
+            self.assertTrue(
+                any(
+                    dependency.name == 'current_user' and dependency.call == LoginService.get_current_user
+                    for dependency in route.dependant.dependencies
+                )
+            )
+
+    def test_qrcode_route_is_declared_before_dynamic_detail(self) -> None:
+        route_paths = [route.path for route in pallet_controller_module.pallet_controller.routes]
+        self.assertLess(
+            route_paths.index('/master/pallet/{pallet_id}/qrcode'),
+            route_paths.index('/master/pallet/{pallet_id}'),
+        )
+
+    async def test_qrcode_endpoint_returns_real_image_png_body(self) -> None:
+        png = PalletService.render_qr_png('persisted-token')
+        db = MagicMock(spec=AsyncSession)
+        with patch.object(PalletService, 'get_pallet_qrcode', new=AsyncMock(return_value=png)) as get_qrcode:
+            response = await pallet_controller_module.query_master_pallet_qrcode(
+                request=MagicMock(),
+                pallet_id=7,
+                query_db=db,
+            )
+
+        get_qrcode.assert_awaited_once_with(db, 7)
+        self.assertEqual('image/png', response.media_type)
+        self.assertEqual(b'\x89PNG\r\n\x1a\n', response.body[:8])
+
+    def test_controller_sources_do_not_require_system_user_list_or_cache(self) -> None:
+        sources = '\n'.join(
+            inspect.getsource(module)
+            for module in (
+                pallet_controller_module,
+                warehouse_controller_module,
+                site_controller_module,
+            )
+        )
+        self.assertNotIn('system:user:list', sources)
+        self.assertNotIn('ApiCache', sources)
+
+
+if __name__ == '__main__':
+    unittest.main()