feat(server): add PUT /attachment/config with removal guard

Co-Authored-By: Claude <noreply@anthropic.com>

app/api/v1/attachment.py

@@ -3,8 +3,15 @@ from sqlalchemy.orm import Session
 
 from app.core.database import get_db
 from app.models.attachment import AttachmentCategory
-from app.schemas.attachment import AttachmentStatusResponse, CategoryResponse
+from app.schemas.attachment import (
-from app.services.attachment_service import get_attachment_status
+    AttachmentConfigRequest,
+    AttachmentStatusResponse,
+    CategoryResponse,
+)
+from app.services.attachment_service import (
+    configure_attachment,
+    get_attachment_status,
+)
 
 router = APIRouter(tags=["attachment"])
 
@@ -26,3 +33,8 @@ def list_categories(db: Session = Depends(get_db)):
 @router.get("/attachment/status", response_model=AttachmentStatusResponse)
 def attachment_status(zongpai_no: str, db: Session = Depends(get_db)):
     return get_attachment_status(db, zongpai_no)
+
+
+@router.put("/attachment/config", response_model=AttachmentStatusResponse)
+def put_attachment_config(req: AttachmentConfigRequest, db: Session = Depends(get_db)):
+    return configure_attachment(db, req)

app/main.py

@@ -32,6 +32,12 @@ from app.services.box_service import (
     InvalidZongpaiError,
     ZongpaiNotFoundError,
 )
+from app.services.attachment_service import (
+    AlreadyOffShelfAttachmentError,
+    CategoryHasBoxesError,
+    CategoryNotFoundError,
+    DuplicateAttachmentLocationError,
+)
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
@@ -176,6 +182,55 @@ async def invalid_box_item_handler(request: Request, exc: InvalidBoxItemError):
     )
 
 
+@app.exception_handler(CategoryNotFoundError)
+async def category_not_found_handler(request: Request, exc: CategoryNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={"error_code": "CATEGORY_NOT_FOUND", "message": "附件类型不存在"},
+    )
+
+
+@app.exception_handler(CategoryHasBoxesError)
+async def category_has_boxes_handler(request: Request, exc: CategoryHasBoxesError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "CATEGORY_HAS_BOXES",
+            "message": f"类型 {exc.category_name} 已有装箱记录，不可删除，可改应到数量",
+        },
+    )
+
+
+@app.exception_handler(DuplicateAttachmentLocationError)
+async def duplicate_attachment_location_handler(
+    request: Request, exc: DuplicateAttachmentLocationError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_LOCATION",
+            "message": "该总排号该类型已存在货位记录",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(AlreadyOffShelfAttachmentError)
+async def already_off_shelf_attachment_handler(
+    request: Request, exc: AlreadyOffShelfAttachmentError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_OFF_SHELF",
+            "message": "该总排号该类型已下架至转运区域，不可重新上架",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/services/attachment_service.py

@@ -1,4 +1,5 @@
 import re
+from datetime import datetime
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -10,6 +11,7 @@ from app.models.attachment import (
     FinishedGoodsAttachmentLocation,
     FinishedGoodsBoxAttachmentItem,
 )
+from app.schemas.attachment import AttachmentConfigRequest
 from app.services.box_service import InvalidZongpaiError
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
@@ -130,3 +132,69 @@ def get_attachment_status(db: Session, zongpai_no: str) -> dict:
         "all_complete": all_complete,
         "items": items,
     }
+
+
+def _ensure_categories_exist(db: Session, category_ids: list[int]) -> dict[int, str]:
+    names = _category_name_map(db, category_ids)
+    missing = [c for c in category_ids if c not in names]
+    if missing:
+        raise CategoryNotFoundError(missing[0])
+    return names
+
+
+def configure_attachment(db: Session, req: AttachmentConfigRequest) -> dict:
+    zongpai_no = _validate_zongpai(req.zongpai_no)
+
+    # Validate categories referenced in the submitted list
+    submitted_ids = [it.category_id for it in req.items]
+    _ensure_categories_exist(db, submitted_ids)
+
+    # Upsert order-level determination row
+    att = (
+        db.query(FinishedGoodsAttachment)
+        .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
+        .first()
+    )
+    if att is None:
+        att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination=req.determination)
+        db.add(att)
+    else:
+        att.determination = req.determination
+        att.updated_at = datetime.now()
+    db.flush()
+
+    # Load existing items
+    existing = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no)
+        .all()
+    )
+    existing_by_cat = {it.category_id: it for it in existing}
+    submitted_set = set(submitted_ids)
+    boxed = _boxed_qty_map(db, zongpai_no)
+    names = _category_name_map(db, list(existing_by_cat.keys()) + submitted_ids)
+
+    # Remove items not in the submitted list; block those with boxed > 0
+    for cat_id, it in existing_by_cat.items():
+        if cat_id not in submitted_set:
+            if boxed.get(cat_id, 0) > 0:
+                raise CategoryHasBoxesError(names.get(cat_id, str(cat_id)))
+            db.delete(it)
+
+    # Upsert submitted items
+    for submitted in req.items:
+        it = existing_by_cat.get(submitted.category_id)
+        if it is None:
+            db.add(
+                FinishedGoodsAttachmentItem(
+                    zongpai_no=zongpai_no,
+                    category_id=submitted.category_id,
+                    expected_qty=submitted.expected_qty,
+                )
+            )
+        else:
+            it.expected_qty = submitted.expected_qty
+            it.updated_at = datetime.now()
+
+    db.commit()
+    return get_attachment_status(db, zongpai_no)

tests/test_attachment_api.py

@@ -84,3 +84,113 @@ def test_status_reflects_config_and_boxed(client: TestClient, db: Session):
     assert item["expected_qty"] == 80
     assert item["boxed_qty"] == 30
     assert item["complete"] is False
+
+
+def test_config_sets_none(client: TestClient):
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={"zongpai_no": "26BW0011", "determination": "none", "items": []},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["determination"] == "none"
+    assert resp.json()["items"] == []
+
+
+def test_config_creates_and_updates_items(client: TestClient, db: Session):
+    cert_id, extra_id, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 5},
+            ],
+        },
+    )
+    # Update: change extra expected, drop nothing
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 8},
+            ],
+        },
+    )
+    assert resp.status_code == 200
+    extra = next(i for i in resp.json()["items"] if i["category_id"] == extra_id)
+    assert extra["expected_qty"] == 8
+
+
+def test_config_rejects_removing_type_with_boxes(client: TestClient, db: Session):
+    cert_id, _, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 922,
+            "quantity": 10,
+        },
+    )
+    # Try to drop the type that now has boxes
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [],
+        },
+    )
+    assert resp.status_code == 409
+    assert resp.json()["error_code"] == "CATEGORY_HAS_BOXES"
+
+
+def test_config_allows_removing_type_with_zero_boxes(client: TestClient, db: Session):
+    cert_id, extra_id, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 5},
+            ],
+        },
+    )
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    assert resp.status_code == 200
+    assert {i["category_id"] for i in resp.json()["items"]} == {cert_id}
+
+
+def test_config_rejects_unknown_category(client: TestClient):
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": 9999999, "expected_qty": 1}],
+        },
+    )
+    assert resp.status_code == 404
+    assert resp.json()["error_code"] == "CATEGORY_NOT_FOUND"