feat: support editable boxing assignments

app/api/v1/box.py

@@ -2,8 +2,20 @@ from fastapi import APIRouter, Depends, Query
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
-from app.schemas.box import BoxInfoResponse, BoxSaveRequest, BoxSaveResponse
+from app.schemas.box import (
-from app.services.box_service import get_box_info, save_box_record
+    BoxDeleteResponse,
+    BoxInfoResponse,
+    BoxSaveRequest,
+    BoxSaveResponse,
+    BoxUpdateRequest,
+    BoxUpdateResponse,
+)
+from app.services.box_service import (
+    delete_box_item,
+    get_box_info,
+    save_box_record,
+    update_box_item,
+)
 
 router = APIRouter(tags=["boxing"])
 
@@ -21,3 +33,15 @@ def box_info(
 def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
     """保存装箱记录：将总排号记录到指定箱号中。"""
     return save_box_record(db, req)
+
+
+@router.patch("/box/{box_item_id}", response_model=BoxUpdateResponse)
+def update_box(box_item_id: int, req: BoxUpdateRequest, db: Session = Depends(get_db)):
+    """更新一码多箱已分配装箱明细。"""
+    return update_box_item(db, box_item_id, req)
+
+
+@router.delete("/box/{box_item_id}", response_model=BoxDeleteResponse)
+def delete_box(box_item_id: int, db: Session = Depends(get_db)):
+    """删除误录的装箱明细。"""
+    return delete_box_item(db, box_item_id)

app/main.py

@@ -19,7 +19,14 @@ except Exception as e:
 from app.api.v1.location import router as location_router
 from app.api.v1.box import router as box_router
 from app.services.location_service import DuplicateLocationError
-from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError, DuplicateZongpaiBindError
+from app.services.box_service import (
+    BoxItemNotFoundError,
+    DuplicateBoxItemError,
+    DuplicateZongpaiBindError,
+    InvalidBoxItemError,
+    InvalidZongpaiError,
+    ZongpaiNotFoundError,
+)
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
@@ -120,6 +127,28 @@ async def duplicate_zongpai_bind_handler(request: Request, exc: DuplicateZongpai
     )
 
 
+@app.exception_handler(BoxItemNotFoundError)
+async def box_item_not_found_handler(request: Request, exc: BoxItemNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "BOX_ITEM_NOT_FOUND",
+            "message": "指定装箱明细不存在",
+        },
+    )
+
+
+@app.exception_handler(InvalidBoxItemError)
+async def invalid_box_item_handler(request: Request, exc: InvalidBoxItemError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_BOX_ITEM",
+            "message": "装箱明细记录不合法或不允许修改",
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/schemas/box.py

@@ -1,17 +1,25 @@
 import re
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
 
 class BoxItemDetail(BaseModel):
     """箱号下某个总排号的明细"""
+    box_item_id: int | None = None
     zongpai_no: str
     work_order_no: str | None = None
     quantity: int
 
 
+class CurrentZongpaiBox(BaseModel):
+    """当前总排号已经分配的箱号明细"""
+    box_item_id: int
+    box_no: int
+    quantity: int
+
+
 class BoxDetail(BaseModel):
     """一个箱号的完整信息"""
     box_no: int
@@ -24,6 +32,7 @@ class BoxInfoResponse(BaseModel):
     paichan_no: str
     work_order_no: str | None = None
     quantity: int
+    current_zongpai_boxes: list[CurrentZongpaiBox] = Field(default_factory=list)
     existing_boxes: list[BoxDetail]
     max_box_no: int
     suggested_box_no: int
@@ -61,8 +70,46 @@ class BoxSaveRequest(BaseModel):
 
 class BoxSaveResponse(BaseModel):
     """POST /box 响应"""
+    box_item_id: int
     paichan_no: str
     box_no: int
     zongpai_no: str
     quantity: int
     created_at: str
+
+
+class BoxUpdateRequest(BaseModel):
+    """PATCH /box/{box_item_id} 请求"""
+    box_no: int
+    quantity: int
+    box_mode: str = "one-to-many"
+
+    @field_validator("box_no")
+    @classmethod
+    def validate_box_no(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_BOX_NO")
+        return v
+
+    @field_validator("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+
+class BoxUpdateResponse(BaseModel):
+    """PATCH /box/{box_item_id} 响应"""
+    box_item_id: int
+    paichan_no: str
+    box_no: int
+    zongpai_no: str
+    quantity: int
+    updated_at: str
+
+
+class BoxDeleteResponse(BaseModel):
+    """DELETE /box/{box_item_id} 响应"""
+    box_item_id: int
+    deleted: bool

app/services/box_service.py

@@ -4,7 +4,7 @@ from sqlalchemy import bindparam, text
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
-from app.schemas.box import BoxSaveRequest
+from app.schemas.box import BoxSaveRequest, BoxUpdateRequest
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
@@ -30,6 +30,14 @@ class DuplicateZongpaiBindError(Exception):
         self.zongpai_no = zongpai_no
 
 
+class BoxItemNotFoundError(Exception):
+    pass
+
+
+class InvalidBoxItemError(Exception):
+    pass
+
+
 def query_erp_info(db: Session, zongpai_no: str) -> dict:
     """从 ERP 视图查询总排号对应的排产号、工令号和数量。"""
     sql = text(_erp_info_sql(db))
@@ -115,13 +123,22 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
     work_order_map = query_erp_work_order_map(db, item_zongpai_nos)
 
     existing_boxes = []
+    current_zongpai_boxes = []
     max_box_no = 0
     for box in boxes:
         items = box_items[box.id]
+        for item in items:
+            if item.zongpai_no == zongpai_no:
+                current_zongpai_boxes.append({
+                    "box_item_id": item.id,
+                    "box_no": box.box_no,
+                    "quantity": int(item.quantity or 0),
+                })
         existing_boxes.append({
             "box_no": box.box_no,
             "items": [
                 {
+                    "box_item_id": item.id,
                     "zongpai_no": item.zongpai_no,
                     "work_order_no": work_order_map.get(item.zongpai_no),
                     "quantity": int(item.quantity or 0),
@@ -137,6 +154,7 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
         "paichan_no": paichan_no,
         "work_order_no": erp["work_order_no"],
         "quantity": quantity,
+        "current_zongpai_boxes": current_zongpai_boxes,
         "existing_boxes": existing_boxes,
         "max_box_no": max_box_no,
         "suggested_box_no": max_box_no + 1,
@@ -171,6 +189,8 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
     )
 
     if box:
+        if req.box_mode in ("one-to-one", "one-to-many"):
+            raise DuplicateBoxItemError(paichan_no, req.box_no)
         existing_item = (
             db.query(FinishedGoodsBoxItem)
             .filter(
@@ -199,9 +219,109 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
     db.refresh(item)
 
     return {
+        "box_item_id": item.id,
         "paichan_no": paichan_no,
         "box_no": req.box_no,
         "zongpai_no": req.zongpai_no,
         "quantity": req.quantity,
         "created_at": item.created_at.isoformat(),
     }
+
+
+def update_box_item(db: Session, box_item_id: int, req: BoxUpdateRequest) -> dict:
+    """更新一码多箱已分配明细的箱号和数量。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    current_box = (
+        db.query(FinishedGoodsBox)
+        .filter(FinishedGoodsBox.id == item.box_id)
+        .first()
+    )
+    if current_box is None:
+        raise InvalidBoxItemError()
+
+    paichan_no = current_box.paichan_no
+    target_box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+
+    if target_box is not None and target_box.id != current_box.id:
+        raise DuplicateBoxItemError(paichan_no, req.box_no)
+
+    if target_box is None:
+        target_box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(target_box)
+        db.flush()
+
+    old_box_id = item.box_id
+    item.box_id = target_box.id
+    item.quantity = req.quantity
+
+    if old_box_id != target_box.id:
+        old_box_has_items = (
+            db.query(FinishedGoodsBoxItem.id)
+            .filter(
+                FinishedGoodsBoxItem.box_id == old_box_id,
+                FinishedGoodsBoxItem.id != item.id,
+            )
+            .first()
+        )
+        if old_box_has_items is None:
+            old_box = (
+                db.query(FinishedGoodsBox)
+                .filter(FinishedGoodsBox.id == old_box_id)
+                .first()
+            )
+            if old_box is not None:
+                db.delete(old_box)
+
+    db.commit()
+    db.refresh(item)
+
+    return {
+        "box_item_id": item.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": item.zongpai_no,
+        "quantity": int(item.quantity or 0),
+        "updated_at": item.created_at.isoformat(),
+    }
+
+
+def delete_box_item(db: Session, box_item_id: int) -> dict:
+    """删除误录的装箱明细；若箱号下无其他明细，同步删除空箱号。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    box_id = item.box_id
+    db.delete(item)
+    db.flush()
+
+    remaining = (
+        db.query(FinishedGoodsBoxItem.id)
+        .filter(FinishedGoodsBoxItem.box_id == box_id)
+        .first()
+    )
+    if remaining is None:
+        box = db.query(FinishedGoodsBox).filter(FinishedGoodsBox.id == box_id).first()
+        if box is not None:
+            db.delete(box)
+
+    db.commit()
+    return {"box_item_id": box_item_id, "deleted": True}