zkxw/docs/superpowers/plans/2026-05-22-storage-slot-occupancy.md

库位占用记录功能实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为库位管理增加占用记录功能，支持多物品类型（细胞产品 + 其他）的统一出入库追踪和历史记录。

Architecture: 新增 biz_storage_slot_occupancy 表作为多态关联占用记录，通过 item_type + item_id 关联不同来源的物品。在 biz_storage_facility_slot 上增加缓存字段（occupied_count 等）提升查询性能。细胞产品 CRUD 入库/出库时自动同步 occupancy 记录。

Tech Stack: FastAPI, SQLAlchemy, Pydantic v2, MySQL, Vue 3, Element Plus, TypeScript

Design Spec: docs/superpowers/specs/2026-05-22-storage-slot-occupancy-design.md

---

File Structure

操作	文件	职责
Create	docs/sql/2026-05-22-storage-slot-occupancy.sql	DDL 迁移脚本
Modify	backend/app/models/models.py	新增 BizStorageSlotOccupancy 模型 + BizStorageFacilitySlot 增加缓存字段
Modify	backend/app/models/__init__.py	新增导出
Create	backend/app/schemas/storage_occupancy.py	Pydantic schemas
Create	backend/app/crud/storage_occupancy.py	CRUD 层（含缓存刷新）
Modify	backend/app/api/operations.py	新增 4 个 API 路由
Modify	backend/app/crud/cell_product.py	细胞产品入库/出库同步 occupancy
Modify	web-admin/src/api/admin.ts	新增 4 个前端 API 函数
Modify	web-admin/src/views/operations/storage-facility/index.vue	占用明细弹窗 + 手动出入库

---

Task 1: SQL 迁移脚本

Files:

• Create: docs/sql/2026-05-22-storage-slot-occupancy.sql

• [ ] Step 1: 创建 SQL 迁移文件

-- ============================================================
-- 库位占用记录功能迁移
-- 日期: 2026-05-22
-- ============================================================

-- 1. 新增：库位占用记录表
CREATE TABLE biz_storage_slot_occupancy (
    id               VARCHAR(64)  NOT NULL              COMMENT '主键ID',
    slot_id          VARCHAR(64)  NOT NULL              COMMENT '储物格ID (FK → biz_storage_facility_slot)',
    item_type        VARCHAR(32)  NOT NULL              COMMENT '物品类型: cell_product=细胞产品, other=其他',
    item_id          VARCHAR(64)  NOT NULL              COMMENT '物品ID (指向对应业务表主键)',
    item_code        VARCHAR(64)  NULL                  COMMENT '物品编码 (冗余展示)',
    item_name        VARCHAR(200) NULL                  COMMENT '物品名称 (冗余展示)',
    item_source      VARCHAR(32)  NULL                  COMMENT '来源: production=生产入库, manual=手动入库, transfer=转移',
    stored_at        DATETIME     NULL                  COMMENT '入库时间',
    removed_at       DATETIME     NULL                  COMMENT '出库时间 (NULL=当前在库)',
    occupancy_status VARCHAR(16)  NOT NULL DEFAULT 'stored' COMMENT 'stored=在库, removed=已出库',
    tenant_no        VARCHAR(64)  NULL                  COMMENT '租户编号',
    order_num        INT          DEFAULT 0             COMMENT '排序',
    remark           VARCHAR(500) NULL                  COMMENT '备注',
    create_by        VARCHAR(64)  NULL                  COMMENT '创建者',
    create_dept_id   VARCHAR(64)  NULL                  COMMENT '创建部门',
    create_time      DATETIME     DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    update_by        VARCHAR(64)  NULL                  COMMENT '更新者',
    update_time      DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    del_flag         CHAR(1)      DEFAULT '0'           COMMENT '删除标志(0存在 1删除)',
    PRIMARY KEY (id),
    INDEX idx_sso_slot (slot_id, occupancy_status, del_flag),
    INDEX idx_sso_item (item_type, item_id, del_flag),
    INDEX idx_sso_tenant (tenant_no, del_flag)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='库位占用记录表';

-- 2. 增强：biz_storage_facility_slot 增加缓存字段
ALTER TABLE biz_storage_facility_slot
    ADD COLUMN occupied_count INT DEFAULT 0 COMMENT '当前占用数量',
    ADD COLUMN last_item_type VARCHAR(32) NULL COMMENT '最近存储物品类型',
    ADD COLUMN last_stored_at DATETIME NULL COMMENT '最近入库时间';

---

Task 2: 后端 Model — BizStorageSlotOccupancy + BizStorageFacilitySlot 增强

Files:

• Modify: backend/app/models/models.py (在 BizStorageFacilitySlot 类后插入新模型，在 BizStorageFacilitySlot 类中增加字段)

• Modify: backend/app/models/__init__.py

• [ ] Step 1: 在 BizStorageFacilitySlot 模型中增加缓存字段

在 backend/app/models/models.py 中，BizStorageFacilitySlot 类的 del_flag 字段后面，增加三个字段：

    occupied_count = Column(Integer, nullable=True, default=0, comment="当前占用数量")
    last_item_type = Column(String(32), nullable=True, comment="最近存储物品类型")
    last_stored_at = Column(DateTime, nullable=True, comment="最近入库时间")

即修改后的 BizStorageFacilitySlot 类完整如下：

class BizStorageFacilitySlot(Base):
    """存储设施小格 biz_storage_facility_slot"""

    __tablename__ = "biz_storage_facility_slot"

    id = Column(String(64), primary_key=True, comment="主键ID")
    facility_node_id = Column(String(64), nullable=False, comment="设施树叶子节点ID")
    slot_code = Column(String(64), nullable=True, comment="小格编码/坐标")
    slot_status = Column(String(16), nullable=True, comment="占用状态等扩展")
    occupied_count = Column(Integer, nullable=True, default=0, comment="当前占用数量")
    last_item_type = Column(String(32), nullable=True, comment="最近存储物品类型")
    last_stored_at = Column(DateTime, nullable=True, comment="最近入库时间")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")

• Step 2: 在 BizStorageFacilitySlot 类之后插入 BizStorageSlotOccupancy 模型

在 BizStorageFacilitySlot 类结束（del_flag 行之后）和 BizExhibitionAppointment 类之间，插入新模型：

class BizStorageSlotOccupancy(Base):
    """库位占用记录 biz_storage_slot_occupancy"""

    __tablename__ = "biz_storage_slot_occupancy"

    id = Column(String(64), primary_key=True, comment="主键ID")
    slot_id = Column(String(64), nullable=False, comment="储物格ID")
    item_type = Column(String(32), nullable=False, comment="物品类型: cell_product/other")
    item_id = Column(String(64), nullable=False, comment="物品ID")
    item_code = Column(String(64), nullable=True, comment="物品编码(冗余)")
    item_name = Column(String(200), nullable=True, comment="物品名称(冗余)")
    item_source = Column(String(32), nullable=True, comment="来源: production/manual/transfer")
    stored_at = Column(DateTime, nullable=True, comment="入库时间")
    removed_at = Column(DateTime, nullable=True, comment="出库时间")
    occupancy_status = Column(String(16), nullable=False, default="stored", comment="stored=在库/removed=已出库")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(500), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")

• Step 3: 在 backend/app/models/__init__.py 中导出新模型

在 import 块中（BizStorageFacilitySlot 附近，实际文件中没有直接导出它，但需要在 from app.models.models import (...) 中添加）：

在 from app.models.models import ( 的导入列表末尾（BizProductionTaskOutput, 之后）添加：

    BizStorageSlotOccupancy,

在 __all__ 列表末尾添加：

    "BizStorageSlotOccupancy",

---

Task 3: 后端 Schemas — storage_occupancy.py

Files:

• Create: backend/app/schemas/storage_occupancy.py

• [ ] Step 1: 创建 Pydantic schemas 文件

"""库位占用记录 schemas"""
from datetime import datetime
from typing import Optional

from pydantic import BaseModel, ConfigDict, Field


class OccupancyCreateBody(BaseModel):
    """手动入库请求"""
    model_config = ConfigDict(populate_by_name=True)

    slot_id: str = Field(..., alias="slotId")
    item_type: str = Field("other", alias="itemType")
    item_code: str = Field(..., alias="itemCode")
    item_name: str = Field(..., alias="itemName")
    item_source: str = Field("manual", alias="itemSource")
    remark: Optional[str] = None


class OccupancyListQuery(BaseModel):
    """占用记录查询"""
    model_config = ConfigDict(populate_by_name=True)

    slot_id: str = Field(..., alias="slotId")
    occupancy_status: Optional[str] = Field("stored", alias="occupancyStatus")
    item_type: Optional[str] = Field(None, alias="itemType")
    page: int = 1
    page_size: int = Field(20, alias="pageSize")


class OccupancyHistoryQuery(BaseModel):
    """出入库历史查询"""
    model_config = ConfigDict(populate_by_name=True)

    slot_id: Optional[str] = Field(None, alias="slotId")
    facility_node_id: Optional[str] = Field(None, alias="facilityNodeId")
    item_type: Optional[str] = Field(None, alias="itemType")
    start_date: Optional[str] = Field(None, alias="startDate")
    end_date: Optional[str] = Field(None, alias="endDate")
    page: int = 1
    page_size: int = Field(20, alias="pageSize")

---

Task 4: 后端 CRUD — storage_occupancy.py

Files:

• Create: backend/app/crud/storage_occupancy.py

• [ ] Step 1: 创建 CRUD 文件

"""库位占用记录 CRUD（biz_storage_slot_occupancy）"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import and_, func
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import BizStorageFacilitySlot, BizStorageSlotOccupancy


def _row_to_dict(c: BizStorageSlotOccupancy) -> Dict[str, Any]:
    return {
        "id": c.id,
        "slotId": c.slot_id,
        "itemType": c.item_type or "",
        "itemId": c.item_id or "",
        "itemCode": c.item_code or "",
        "itemName": c.item_name or "",
        "itemSource": c.item_source or "",
        "storedAt": c.stored_at.strftime("%Y-%m-%d %H:%M:%S") if c.stored_at else "",
        "removedAt": c.removed_at.strftime("%Y-%m-%d %H:%M:%S") if c.removed_at else "",
        "occupancyStatus": c.occupancy_status or "",
        "tenantNo": c.tenant_no,
        "remark": c.remark or "",
        "createBy": c.create_by or "",
        "createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "",
    }


def _refresh_slot_cache(db: Session, slot_id: str) -> None:
    """刷新 slot 的 occupied_count / last_item_type / last_stored_at / slot_status"""
    slot = db.query(BizStorageFacilitySlot).filter(
        BizStorageFacilitySlot.id == slot_id,
        BizStorageFacilitySlot.del_flag == 0,
    ).first()
    if not slot:
        return

    count = db.query(func.count(BizStorageSlotOccupancy.id)).filter(
        BizStorageSlotOccupancy.slot_id == slot_id,
        BizStorageSlotOccupancy.occupancy_status == "stored",
        BizStorageSlotOccupancy.del_flag == 0,
    ).scalar() or 0

    slot.occupied_count = count
    if count > 0:
        slot.slot_status = "占用"
        latest = db.query(BizStorageSlotOccupancy).filter(
            BizStorageSlotOccupancy.slot_id == slot_id,
            BizStorageSlotOccupancy.occupancy_status == "stored",
            BizStorageSlotOccupancy.del_flag == 0,
        ).order_by(BizStorageSlotOccupancy.stored_at.desc()).first()
        if latest:
            slot.last_item_type = latest.item_type
            slot.last_stored_at = latest.stored_at
    else:
        slot.slot_status = "空"
        slot.last_item_type = None
        slot.last_stored_at = None
    slot.update_time = datetime.now()
    db.add(slot)
    db.flush()


def create_occupancy(
    db: Session,
    *,
    slot_id: str,
    item_type: str,
    item_id: str,
    item_code: str,
    item_name: str,
    item_source: str = "manual",
    tenant_no: str = "",
    user_id: Optional[str] = None,
    remark: Optional[str] = None,
) -> BizStorageSlotOccupancy:
    """创建入库记录 + 刷新 slot 缓存"""
    now = datetime.now()
    obj = BizStorageSlotOccupancy(
        id=generate_uuid(),
        slot_id=slot_id,
        item_type=item_type,
        item_id=item_id,
        item_code=(item_code or "").strip() or None,
        item_name=(item_name or "").strip() or None,
        item_source=(item_source or "manual").strip(),
        stored_at=now,
        removed_at=None,
        occupancy_status="stored",
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    _refresh_slot_cache(db, slot_id)
    db.commit()
    db.refresh(obj)
    return obj


def batch_create_occupancies(
    db: Session,
    *,
    items: List[Dict[str, Any]],
    tenant_no: str = "",
    user_id: Optional[str] = None,
) -> List[BizStorageSlotOccupancy]:
    """批量入库"""
    now = datetime.now()
    created: List[BizStorageSlotOccupancy] = []
    slot_ids_seen: set = set()
    for it in items:
        obj = BizStorageSlotOccupancy(
            id=generate_uuid(),
            slot_id=it["slot_id"],
            item_type=it.get("item_type", "cell_product"),
            item_id=it["item_id"],
            item_code=(it.get("item_code") or "").strip() or None,
            item_name=(it.get("item_name") or "").strip() or None,
            item_source=(it.get("item_source") or "production").strip(),
            stored_at=now,
            removed_at=None,
            occupancy_status="stored",
            tenant_no=tenant_no,
            order_num=0,
            remark=(it.get("remark") or "").strip() or None,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)
        created.append(obj)
        slot_ids_seen.add(it["slot_id"])
    db.flush()
    for sid in slot_ids_seen:
        _refresh_slot_cache(db, sid)
    db.commit()
    for obj in created:
        db.refresh(obj)
    return created


def remove_occupancy(
    db: Session,
    occupancy_id: str,
    *,
    user_id: Optional[str] = None,
) -> BizStorageSlotOccupancy:
    """手动出库"""
    obj = db.query(BizStorageSlotOccupancy).filter(
        BizStorageSlotOccupancy.id == occupancy_id,
        BizStorageSlotOccupancy.del_flag == 0,
    ).first()
    if not obj:
        raise ValueError("占用记录不存在")
    if obj.occupancy_status == "removed":
        raise ValueError("该物品已出库")
    now = datetime.now()
    obj.occupancy_status = "removed"
    obj.removed_at = now
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    db.flush()
    _refresh_slot_cache(db, obj.slot_id)
    db.commit()
    db.refresh(obj)
    return obj


def remove_by_item(
    db: Session,
    item_type: str,
    item_id: str,
    *,
    user_id: Optional[str] = None,
) -> None:
    """按物品类型+ID出库（细胞产品状态变更时调用）"""
    obj = db.query(BizStorageSlotOccupancy).filter(
        BizStorageSlotOccupancy.item_type == item_type,
        BizStorageSlotOccupancy.item_id == item_id,
        BizStorageSlotOccupancy.occupancy_status == "stored",
        BizStorageSlotOccupancy.del_flag == 0,
    ).first()
    if not obj:
        return
    now = datetime.now()
    obj.occupancy_status = "removed"
    obj.removed_at = now
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    db.flush()
    _refresh_slot_cache(db, obj.slot_id)
    db.commit()


def list_occupancies(
    db: Session,
    *,
    slot_id: str,
    occupancy_status: Optional[str] = "stored",
    item_type: Optional[str] = None,
    tenant_no: str = "",
    page: int = 1,
    page_size: int = 20,
) -> Dict[str, Any]:
    """分页查询占用记录"""
    q = db.query(BizStorageSlotOccupancy).filter(
        BizStorageSlotOccupancy.slot_id == slot_id,
        BizStorageSlotOccupancy.del_flag == 0,
    )
    if tenant_no:
        q = q.filter(BizStorageSlotOccupancy.tenant_no == tenant_no)
    if occupancy_status:
        q = q.filter(BizStorageSlotOccupancy.occupancy_status == occupancy_status)
    if item_type:
        q = q.filter(BizStorageSlotOccupancy.item_type == item_type)

    total = q.order_by(None).with_entities(func.count(BizStorageSlotOccupancy.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizStorageSlotOccupancy.stored_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return {"items": [_row_to_dict(r) for r in rows], "total": int(total)}


def list_occupancy_history(
    db: Session,
    *,
    slot_id: Optional[str] = None,
    facility_node_id: Optional[str] = None,
    item_type: Optional[str] = None,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    tenant_no: str = "",
    page: int = 1,
    page_size: int = 20,
) -> Dict[str, Any]:
    """出入库历史"""
    q = db.query(BizStorageSlotOccupancy).filter(
        BizStorageSlotOccupancy.del_flag == 0,
    )
    if tenant_no:
        q = q.filter(BizStorageSlotOccupancy.tenant_no == tenant_no)
    if slot_id:
        q = q.filter(BizStorageSlotOccupancy.slot_id == slot_id)
    if item_type:
        q = q.filter(BizStorageSlotOccupancy.item_type == item_type)
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            q = q.filter(BizStorageSlotOccupancy.stored_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d")
            ed = ed.replace(hour=23, minute=59, second=59)
            q = q.filter(BizStorageSlotOccupancy.stored_at <= ed)
        except ValueError:
            pass
    if facility_node_id:
        slot_ids = [
            r[0] for r in db.query(BizStorageFacilitySlot.id).filter(
                BizStorageFacilitySlot.facility_node_id == facility_node_id,
                BizStorageFacilitySlot.del_flag == 0,
            ).all()
        ]
        if slot_ids:
            q = q.filter(BizStorageSlotOccupancy.slot_id.in_(slot_ids))
        else:
            return {"items": [], "total": 0}

    total = q.order_by(None).with_entities(func.count(BizStorageSlotOccupancy.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizStorageSlotOccupancy.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return {"items": [_row_to_dict(r) for r in rows], "total": int(total)}

---

Task 5: 后端 API 路由

Files:

• Modify: backend/app/api/operations.py

• [ ] Step 1: 在文件顶部导入中添加新依赖

在 backend/app/api/operations.py 的 import 区域，添加：

from app.crud import storage_occupancy as occupancy_svc
from app.models.models import BizStorageSlotOccupancy

（BizStorageSlotOccupancy 加到现有的 BizStorageFacility, BizStorageFacilitySlot 后面）

• Step 2: 在储位路由区域之后添加 4 个新路由

在 operations.py 文件中，找到 storage-facility-slots 相关路由的末尾（大约在 delete_storage_slot 路由之后），添加以下路由：

# ---------------------------------------------------------------------------
# 库位占用记录
# ---------------------------------------------------------------------------


class OccupancyCreateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    slot_id: str = Field(..., alias="slotId")
    item_type: str = Field("other", alias="itemType")
    item_code: str = Field(..., alias="itemCode")
    item_name: str = Field(..., alias="itemName")
    item_source: str = Field("manual", alias="itemSource")
    remark: Optional[str] = None


@router.get("/storage-slot-occupancies")
async def list_occupancies(
    slotId: str = Query(..., alias="slotId"),
    occupancyStatus: Optional[str] = Query("stored", alias="occupancyStatus"),
    itemType: Optional[str] = Query(None, alias="itemType"),
    page: int = Query(1),
    pageSize: int = Query(20, alias="pageSize"),
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    result = occupancy_svc.list_occupancies(
        db,
        slot_id=slotId,
        occupancy_status=occupancyStatus,
        item_type=itemType,
        tenant_no=tenant_no,
        page=page,
        page_size=pageSize,
    )
    return success_response(result)


@router.get("/storage-slot-occupancies/history")
async def list_occupancy_history(
    slotId: Optional[str] = Query(None, alias="slotId"),
    facilityNodeId: Optional[str] = Query(None, alias="facilityNodeId"),
    itemType: Optional[str] = Query(None, alias="itemType"),
    startDate: Optional[str] = Query(None, alias="startDate"),
    endDate: Optional[str] = Query(None, alias="endDate"),
    page: int = Query(1),
    pageSize: int = Query(20, alias="pageSize"),
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    result = occupancy_svc.list_occupancy_history(
        db,
        slot_id=slotId,
        facility_node_id=facilityNodeId,
        item_type=itemType,
        start_date=startDate,
        end_date=endDate,
        tenant_no=tenant_no,
        page=page,
        page_size=pageSize,
    )
    return success_response(result)


@router.post("/storage-slot-occupancies")
async def create_occupancy(
    body: OccupancyCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        item_id = generate_uuid()
        obj = occupancy_svc.create_occupancy(
            db,
            slot_id=body.slot_id,
            item_type=body.item_type,
            item_id=item_id,
            item_code=body.item_code,
            item_name=body.item_name,
            item_source=body.item_source,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
            remark=body.remark,
        )
        return success_response(occupancy_svc._row_to_dict(obj), message="入库成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/storage-slot-occupancies/{occupancy_id}/remove")
async def remove_occupancy(
    occupancy_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        obj = occupancy_svc.remove_occupancy(
            occupancy_id,
            user_id=user.get("user_id"),
            db=db,
        )
        return success_response(occupancy_svc._row_to_dict(obj), message="出库成功")
    except ValueError as e:
        return error_response(400, str(e))

注意：需要在文件顶部增加 generate_uuid 导入：

from app.core.utils import generate_uuid

---

Task 6: 细胞产品 CRUD 对接 — 自动同步 occupancy

Files:

• Modify: backend/app/crud/cell_product.py

• [ ] Step 1: 在文件顶部添加导入

在 from app.core.utils import generate_uuid 之后添加：

from app.crud import storage_occupancy as occupancy_svc

• Step 2: 在 create_cell_product 函数中，db.commit() 之前添加 occupancy 同步

在 create_cell_product 函数中，找到 db.add(obj) 之后、db.commit() 之前的位置。在 db.add(obj) 后面添加：

    db.flush()
    # 同步写入占用记录
    if obj.storage_slot_id:
        try:
            occupancy_svc.create_occupancy(
                db,
                slot_id=obj.storage_slot_id,
                item_type="cell_product",
                item_id=obj.id,
                item_code=obj.product_code,
                item_name=obj.product_name,
                item_source=obj.source_type or "manual",
                tenant_no=tenant_no,
                user_id=user_id,
            )
        except Exception:
            pass  # occupancy 写入失败不影响主流程

注意：原来的 db.commit() 和 db.refresh(obj) 保持不变（create_occupancy 内部已经有 db.commit()）。 但实际上 create_occupancy 有自己的 commit，所以需要调整 —— 在 create_cell_product 中改为先 db.flush()，然后调 occupancy（occupancy 内部 commit 会一起提交），然后 refresh。

实际修改策略：把 create_occupancy 改为不 commit 的内部版本，让调用方控制事务。

更好的做法：在 cell_product.py 中直接内联写 occupancy 记录，不通过 occupancy_svc.create_occupancy（避免嵌套 commit 问题）。

最终方案：在 create_cell_product 的 db.add(obj) 之后、db.commit() 之前，直接内联写入 occupancy：

    db.add(obj)
    db.flush()  # 先 flush 拿到 obj.id

    # 同步写入占用记录
    if obj.storage_slot_id:
        occ = BizStorageSlotOccupancy(
            id=generate_uuid(),
            slot_id=obj.storage_slot_id,
            item_type="cell_product",
            item_id=obj.id,
            item_code=obj.product_code,
            item_name=obj.product_name,
            item_source=obj.source_type or "manual",
            stored_at=datetime.now(),
            occupancy_status="stored",
            tenant_no=tenant_no,
            order_num=0,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        db.add(occ)
        occupancy_svc._refresh_slot_cache(db, obj.storage_slot_id)

    db.commit()
    db.refresh(obj)
    return obj

同时在文件顶部 import 中添加：

from app.models.models import BizStorageSlotOccupancy

• Step 3: 在 create_cell_products_batch 函数中添加批量 occupancy 同步

在 create_cell_products_batch 函数中，db.commit() 之前（即 for 循环结束之后），添加：

    # 批量同步占用记录
    for obj in created:
        if obj.storage_slot_id:
            occ = BizStorageSlotOccupancy(
                id=generate_uuid(),
                slot_id=obj.storage_slot_id,
                item_type="cell_product",
                item_id=obj.id,
                item_code=obj.product_code,
                item_name=obj.product_name,
                item_source=obj.source_type or "manual",
                stored_at=now,
                occupancy_status="stored",
                tenant_no=tenant_no,
                order_num=0,
                create_by=user_id,
                create_dept_id=create_dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
            db.add(occ)
    # 刷新涉及的 slot 缓存
    slot_ids_seen = {obj.storage_slot_id for obj in created if obj.storage_slot_id}
    for sid in slot_ids_seen:
        occupancy_svc._refresh_slot_cache(db, sid)

这段代码放在 db.add(obj) 循环结束之后、db.commit() 之前。

• Step 4: 在 update_cell_product 函数中，状态变更时触发出库

在 update_cell_product 函数中，找到状态变更的代码块：

    if data.product_status is not None and data.product_status != obj.product_status:
        allowed = {"frozen": {"sold", "used", "voided"}}
        from_status = obj.product_status
        to_status = data.product_status
        if from_status not in allowed or to_status not in allowed[from_status]:
            raise ValueError(f"不允许从状态 {from_status} 变更为 {to_status}")
        obj.product_status = data.product_status

在 obj.product_status = data.product_status 之后添加：

        # 状态变更时自动出库占用记录
        if obj.storage_slot_id:
            try:
                occupancy_svc.remove_by_item(
                    db, "cell_product", obj.id, user_id=user_id,
                )
            except Exception:
                pass

---

Task 7: 前端 API 函数

Files:

• Modify: web-admin/src/api/admin.ts

• [ ] Step 1: 在库位管理 API 区域末尾添加 4 个新函数

在 fetchDeleteStorageSlot 函数之后、// 展馆预约 注释之前，添加：

// 库位占用记录
export function fetchSlotOccupancies(params: {
  slotId: string
  occupancyStatus?: string
  itemType?: string
  page?: number
  pageSize?: number
}) {
  return request.get({ url: '/web/admin/storage-slot-occupancies', params })
}

export function fetchSlotOccupancyHistory(params?: {
  slotId?: string
  facilityNodeId?: string
  itemType?: string
  startDate?: string
  endDate?: string
  page?: number
  pageSize?: number
}) {
  return request.get({ url: '/web/admin/storage-slot-occupancies/history', params })
}

export function fetchCreateOccupancy(data: {
  slotId: string
  itemType?: string
  itemCode: string
  itemName: string
  itemSource?: string
  remark?: string
}) {
  return request.post({ url: '/web/admin/storage-slot-occupancies', params: data })
}

export function fetchRemoveOccupancy(id: string) {
  return request.put({ url: `/web/admin/storage-slot-occupancies/${id}/remove` })
}

---

Task 8: 前端库位管理页面 — 占用明细弹窗 + 手动出入库

Files:

• Modify: web-admin/src/views/operations/storage-facility/index.vue

• [ ] Step 1: 添加新的 API 导入

在文件 <script setup> 区域的 import 块中，添加 3 个新 API 函数：

import {
  fetchStorageFacilityList,
  fetchCreateStorageFacility,
  fetchUpdateStorageFacility,
  fetchDeleteStorageFacility,
  fetchStorageSlotList,
  fetchBatchCreateSlots,
  fetchClearStorageSlots,
  fetchInsertStorageSlotRow,
  fetchCreateStorageSlot,
  fetchUpdateStorageSlot,
  fetchDeleteStorageSlot,
  fetchSlotOccupancies,
  fetchSlotOccupancyHistory,
  fetchCreateOccupancy,
  fetchRemoveOccupancy
} from '@/api/admin'

• Step 2: 修改格子点击事件

将现有的格子点击事件从 @click="handleEditSlot(cell)" 改为弹出席占明细弹窗。

在 shelf-cell 的 click 处理中，将 handleEditSlot(cell) 替换为 handleOpenOccupancyDialog(cell)：

<div
  v-for="(cell, cellIndex) in row.cells"
  :key="cell.id"
  class="shelf-cell"
  :class="{ occupied: cell.slotStatus === '占用' }"
  @click="handleOpenOccupancyDialog(cell)"
>
  <button class="cell-delete-btn" @click.stop="handleDeleteSlot(cell)" title="删除">×</button>
  <span class="cell-code">{{ cellIndex + 1 }}</span>
  <span v-if="cell.occupiedCount > 0" class="cell-badge">{{ cell.occupiedCount }}</span>
</div>

• Step 3: 在 <template> 中添加占用明细弹窗

在现有 </section> 之前、最后一个弹窗（新增行弹窗）之后，添加：

<!-- 占用明细弹窗 -->
<el-dialog
  v-model="occupancyDialogVisible"
  :title="occupancyDialogTitle"
  width="720px"
  class="dialog-form-layout"
  align-center
  destroy-on-close
>
  <el-tabs v-model="occupancyActiveTab">
    <el-tab-pane label="在库物品" name="stored">
      <div class="occupancy-toolbar">
        <el-button type="primary" size="small" @click="handleOpenStoreDialog">
          入库
        </el-button>
      </div>
      <el-table :data="occupancyList" v-loading="occupancyLoading" size="small" max-height="360">
        <el-table-column label="物品类型" width="120">
          <template #default="{ row }">
            <el-tag :type="row.itemType === 'cell_product' ? 'primary' : 'info'" size="small">
              {{ row.itemType === 'cell_product' ? '细胞产品' : '其他' }}
            </el-tag>
          </template>
        </el-table-column>
        <el-table-column prop="itemCode" label="物品编码" min-width="140" />
        <el-table-column prop="itemName" label="物品名称" min-width="140" />
        <el-table-column label="来源" width="120">
          <template #default="{ row }">
            {{ { production: '生产入库', manual: '手动入库', transfer: '转移' }[row.itemSource] || row.itemSource }}
          </template>
        </el-table-column>
        <el-table-column prop="storedAt" label="入库时间" width="170" />
        <el-table-column label="操作" width="80" fixed="right">
          <template #default="{ row }">
            <el-button type="danger" link size="small" @click="handleRemoveOccupancy(row)">
              出库
            </el-button>
          </template>
        </el-table-column>
      </el-table>
      <div v-if="occupancyTotal > occupancyPageSize" class="occupancy-pagination">
        <el-pagination
          small
          layout="total, prev, pager, next"
          :total="occupancyTotal"
          :page-size="occupancyPageSize"
          v-model:current-page="occupancyPage"
          @current-change="loadOccupancies"
        />
      </div>
    </el-tab-pane>

    <el-tab-pane label="出入库历史" name="history">
      <el-table :data="historyList" v-loading="historyLoading" size="small" max-height="400">
        <el-table-column label="物品类型" width="120">
          <template #default="{ row }">
            <el-tag :type="row.itemType === 'cell_product' ? 'primary' : 'info'" size="small">
              {{ row.itemType === 'cell_product' ? '细胞产品' : '其他' }}
            </el-tag>
          </template>
        </el-table-column>
        <el-table-column prop="itemCode" label="物品编码" min-width="120" />
        <el-table-column prop="itemName" label="物品名称" min-width="120" />
        <el-table-column label="来源" width="100">
          <template #default="{ row }">
            {{ { production: '生产入库', manual: '手动入库', transfer: '转移' }[row.itemSource] || row.itemSource }}
          </template>
        </el-table-column>
        <el-table-column prop="storedAt" label="入库时间" width="170" />
        <el-table-column prop="removedAt" label="出库时间" width="170" />
        <el-table-column label="状态" width="100">
          <template #default="{ row }">
            <el-tag :type="row.occupancyStatus === 'stored' ? 'success' : 'warning'" size="small">
              {{ row.occupancyStatus === 'stored' ? '在库' : '已出库' }}
            </el-tag>
          </template>
        </el-table-column>
      </el-table>
      <div v-if="historyTotal > historyPageSize" class="occupancy-pagination">
        <el-pagination
          small
          layout="total, prev, pager, next"
          :total="historyTotal"
          :page-size="historyPageSize"
          v-model:current-page="historyPage"
          @current-change="loadHistory"
        />
      </div>
    </el-tab-pane>
  </el-tabs>
</el-dialog>

<!-- 手动入库弹窗 -->
<el-dialog
  v-model="storeDialogVisible"
  title="手动入库"
  width="480px"
  class="dialog-form-layout"
  align-center
  destroy-on-close
  :close-on-click-modal="false"
>
  <el-form
    ref="storeFormRef"
    :model="storeFormData"
    :rules="storeFormRules"
    label-position="top"
  >
    <el-form-item label="物品编码" prop="itemCode">
      <el-input v-model="storeFormData.itemCode" placeholder="请输入物品编码" />
    </el-form-item>
    <el-form-item label="物品名称" prop="itemName">
      <el-input v-model="storeFormData.itemName" placeholder="请输入物品名称" />
    </el-form-item>
    <el-form-item label="来源" prop="itemSource">
      <el-select v-model="storeFormData.itemSource" placeholder="请选择来源">
        <el-option label="手动入库" value="manual" />
        <el-option label="转移" value="transfer" />
      </el-select>
    </el-form-item>
    <el-form-item label="备注">
      <el-input v-model="storeFormData.remark" type="textarea" :rows="2" placeholder="可选" />
    </el-form-item>
  </el-form>
  <template #footer>
    <el-button class="btn-ghost" @click="storeDialogVisible = false">取消</el-button>
    <el-button type="primary" :loading="storeSubmitLoading" @click="handleStoreSubmit">确认入库</el-button>
  </template>
</el-dialog>

• Step 4: 在 <script setup> 中添加占用弹窗相关数据和方法

在脚本末尾（onMounted(loadData) 之前）添加：

// ==================== 占用明细弹窗 ====================
const occupancyDialogVisible = ref(false)
const occupancyDialogTitle = ref('')
const occupancyActiveTab = ref('stored')
const currentSlot = ref<any>(null)
const occupancyLoading = ref(false)
const occupancyList = ref<any[]>([])
const occupancyTotal = ref(0)
const occupancyPage = ref(1)
const occupancyPageSize = 20

const historyLoading = ref(false)
const historyList = ref<any[]>([])
const historyTotal = ref(0)
const historyPage = ref(1)
const historyPageSize = 20

// 手动入库弹窗
const storeDialogVisible = ref(false)
const storeSubmitLoading = ref(false)
const storeFormRef = ref()
const storeFormData = reactive({
  itemCode: '',
  itemName: '',
  itemSource: 'manual',
  remark: ''
})
const storeFormRules = {
  itemCode: [{ required: true, message: '请输入物品编码', trigger: 'blur' }],
  itemName: [{ required: true, message: '请输入物品名称', trigger: 'blur' }],
  itemSource: [{ required: true, message: '请选择来源', trigger: 'change' }]
}

const handleOpenOccupancyDialog = (cell: any) => {
  currentSlot.value = cell
  occupancyDialogTitle.value = `储物格明细 — ${cell.slotCode || cell.id}`
  occupancyActiveTab.value = 'stored'
  occupancyPage.value = 1
  historyPage.value = 1
  occupancyDialogVisible.value = true
  loadOccupancies()
  loadHistory()
}

const loadOccupancies = async () => {
  if (!currentSlot.value) return
  occupancyLoading.value = true
  try {
    const res: any = await fetchSlotOccupancies({
      slotId: currentSlot.value.id,
      occupancyStatus: 'stored',
      page: occupancyPage.value,
      pageSize: occupancyPageSize
    })
    const data = res?.data || res || {}
    occupancyList.value = data.items || data.list || []
    occupancyTotal.value = data.total || 0
  } catch {
    occupancyList.value = []
  } finally {
    occupancyLoading.value = false
  }
}

const loadHistory = async () => {
  if (!currentSlot.value) return
  historyLoading.value = true
  try {
    const res: any = await fetchSlotOccupancyHistory({
      slotId: currentSlot.value.id,
      page: historyPage.value,
      pageSize: historyPageSize
    })
    const data = res?.data || res || {}
    historyList.value = data.items || data.list || []
    historyTotal.value = data.total || 0
  } catch {
    historyList.value = []
  } finally {
    historyLoading.value = false
  }
}

const handleOpenStoreDialog = () => {
  Object.assign(storeFormData, { itemCode: '', itemName: '', itemSource: 'manual', remark: '' })
  storeDialogVisible.value = true
  nextTick(() => storeFormRef.value?.clearValidate?.())
}

const handleStoreSubmit = async () => {
  await storeFormRef.value?.validate?.()
  if (!currentSlot.value) return
  storeSubmitLoading.value = true
  try {
    await fetchCreateOccupancy({
      slotId: currentSlot.value.id,
      itemType: 'other',
      itemCode: storeFormData.itemCode.trim(),
      itemName: storeFormData.itemName.trim(),
      itemSource: storeFormData.itemSource,
      remark: storeFormData.remark?.trim() || undefined
    })
    ElMessage.success('入库成功')
    storeDialogVisible.value = false
    loadOccupancies()
    loadHistory()
    if (selectedNode.value?.id) loadSlots(selectedNode.value.id)
  } catch {
    /* 错误已全局处理 */
  } finally {
    storeSubmitLoading.value = false
  }
}

const handleRemoveOccupancy = async (row: any) => {
  try {
    await ElMessageBox.confirm(`确认将「${row.itemName || row.itemCode}」出库？`, '出库确认', {
      type: 'warning'
    })
    await fetchRemoveOccupancy(row.id)
    ElMessage.success('出库成功')
    loadOccupancies()
    loadHistory()
    if (selectedNode.value?.id) loadSlots(selectedNode.value.id)
  } catch {
    /* 取消或错误 */
  }
}

• Step 5: 在 <style> 中添加角标和分页样式

在 <style> 区域末尾添加：

.cell-badge {
  position: absolute;
  top: 2px;
  right: 2px;
  min-width: 16px;
  height: 16px;
  line-height: 16px;
  font-size: 10px;
  text-align: center;
  color: #fff;
  background: #f56c6c;
  border-radius: 8px;
  padding: 0 4px;
}

.occupancy-toolbar {
  margin-bottom: 12px;
}

.occupancy-pagination {
  margin-top: 12px;
  display: flex;
  justify-content: flex-end;
}

---

Task 9: 验证

• Step 1: 执行 SQL 迁移

# 在 MySQL 中执行迁移脚本
mysql -u root -p sanqiang < docs/sql/2026-05-22-storage-slot-occupancy.sql

• Step 2: 后端 diagnostics 检查

检查 backend/app/models/models.py、backend/app/crud/cell_product.py、backend/app/api/operations.py 无语法错误。

• Step 3: 前端 build 检查

cd web-admin && pnpm build

Expected: 构建成功，无错误。

• Step 4: 更新知识图谱

./scripts/update-graph.sh all