feat(server): add POST /attachment/box with real-evidence flip + cap

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

app/api/v1/attachment.py

@@ -4,6 +4,8 @@ from sqlalchemy.orm import Session
 from app.core.database import get_db
 from app.models.attachment import AttachmentCategory
 from app.schemas.attachment import (
+    AttachmentBoxSaveRequest,
+    AttachmentBoxSaveResponse,
     AttachmentConfigRequest,
     AttachmentLocationRequest,
     AttachmentLocationResponse,
@@ -15,6 +17,7 @@ from app.services.attachment_service import (
     configure_attachment,
     get_attachment_status,
     register_attachment_location,
+    save_attachment_box,
 )
 
 router = APIRouter(tags=["attachment"])
@@ -64,3 +67,18 @@ def post_attachment_location(
         created_at=record.created_at.isoformat(),
         previous_location=getattr(record, "previous_location", None),
     )
+
+
+@router.post(
+    "/attachment/box",
+    response_model=AttachmentBoxSaveResponse,
+    responses={
+        400: {"description": "参数非法或超量", "model": ErrorResponse},
+        404: {"description": "总排号/类型不存在", "model": ErrorResponse},
+        409: {"description": "同箱同类型重复", "model": ErrorResponse},
+    },
+)
+def post_attachment_box(
+    req: AttachmentBoxSaveRequest, db: Session = Depends(get_db)
+):
+    return save_attachment_box(db, req)

app/main.py

@@ -36,6 +36,7 @@ from app.services.attachment_service import (
     AlreadyOffShelfAttachmentError,
     CategoryHasBoxesError,
     CategoryNotFoundError,
+    DuplicateAttachmentBoxItemError,
     DuplicateAttachmentLocationError,
 )
 
@@ -231,6 +232,21 @@ async def already_off_shelf_attachment_handler(
     )
 
 
+@app.exception_handler(DuplicateAttachmentBoxItemError)
+async def duplicate_attachment_box_handler(
+    request: Request, exc: DuplicateAttachmentBoxItemError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_BOX_NO",
+            "message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在",
+            "paichan_no": exc.paichan_no,
+            "box_no": exc.box_no,
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/services/attachment_service.py

@@ -11,8 +11,13 @@ from app.models.attachment import (
     FinishedGoodsAttachmentLocation,
     FinishedGoodsBoxAttachmentItem,
 )
+from app.models.finished_goods import FinishedGoodsBox
 from app.schemas.attachment import AttachmentConfigRequest
-from app.services.box_service import InvalidZongpaiError
+from app.services.box_service import (
+    InvalidQuantityError,
+    InvalidZongpaiError,
+    query_erp_info,
+)
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
@@ -257,3 +262,112 @@ def register_attachment_location(db: Session, req) -> FinishedGoodsAttachmentLoc
     db.commit()
     db.refresh(record)
     return record
+
+
+def _boxed_for_category(
+    db: Session, zongpai_no: str, category_id: int, exclude_item_id: int | None = None
+) -> int:
+    q = db.query(FinishedGoodsBoxAttachmentItem).filter(
+        FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
+        FinishedGoodsBoxAttachmentItem.category_id == category_id,
+    )
+    if exclude_item_id is not None:
+        q = q.filter(FinishedGoodsBoxAttachmentItem.id != exclude_item_id)
+    return sum(int(r.quantity or 0) for r in q.all())
+
+
+def _ensure_attachment_item(db: Session, zongpai_no: str, category_id: int, erp_qty: int):
+    """实物证据：不存在则建 item（expected 兜底 = ERP 数量），并保证 determination=has。"""
+    if not _category_name_map(db, [category_id]):
+        raise CategoryNotFoundError(category_id)
+
+    att = (
+        db.query(FinishedGoodsAttachment)
+        .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
+        .first()
+    )
+    if att is None:
+        att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination="has")
+        db.add(att)
+    elif att.determination in ("undetermined", "none"):
+        att.determination = "has"
+        att.updated_at = datetime.now()
+    db.flush()
+
+    item = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(
+            FinishedGoodsAttachmentItem.zongpai_no == zongpai_no,
+            FinishedGoodsAttachmentItem.category_id == category_id,
+        )
+        .first()
+    )
+    if item is None:
+        item = FinishedGoodsAttachmentItem(
+            zongpai_no=zongpai_no,
+            category_id=category_id,
+            expected_qty=erp_qty,
+        )
+        db.add(item)
+        db.flush()
+    return item
+
+
+def save_attachment_box(db: Session, req) -> dict:
+    zongpai_no = _validate_zongpai(req.zongpai_no)
+
+    erp = query_erp_info(db, zongpai_no)  # raises ZongpaiNotFoundError if missing
+    paichan_no = erp["paichan_no"]
+    erp_qty = erp["quantity"]
+
+    item = _ensure_attachment_item(db, zongpai_no, req.category_id, erp_qty)
+
+    box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+    if box is not None:
+        existing_in_box = (
+            db.query(FinishedGoodsBoxAttachmentItem)
+            .filter(
+                FinishedGoodsBoxAttachmentItem.box_id == box.id,
+                FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
+                FinishedGoodsBoxAttachmentItem.category_id == req.category_id,
+            )
+            .first()
+        )
+        if existing_in_box is not None:
+            raise DuplicateAttachmentBoxItemError(paichan_no, req.box_no)
+    else:
+        box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(box)
+        db.flush()
+
+    # Cap check against expected_qty (mirrors product _ensure_quantity_within_erp_total)
+    already_boxed = _boxed_for_category(db, zongpai_no, req.category_id)
+    if already_boxed + req.quantity > item.expected_qty:
+        raise InvalidQuantityError()
+
+    record = FinishedGoodsBoxAttachmentItem(
+        box_id=box.id,
+        zongpai_no=zongpai_no,
+        category_id=req.category_id,
+        quantity=req.quantity,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+
+    return {
+        "box_item_id": record.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": zongpai_no,
+        "category_id": req.category_id,
+        "quantity": req.quantity,
+        "created_at": record.created_at.isoformat(),
+    }

tests/test_attachment_boxing.py

@@ -0,0 +1,151 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models.attachment import AttachmentCategory
+
+
+def _seed(db: Session) -> list[int]:
+    cats = [
+        AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True),
+        AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True),
+    ]
+    for c in cats:
+        db.add(c)
+    db.commit()
+    for c in cats:
+        db.refresh(c)
+    return [c.id for c in cats]
+
+
+def test_attachment_box_basic_and_flip(client: TestClient, db: Session):
+    """扫附件码即到货：未判定 → 自动 has，item 不存在则按 ERP 数量建。"""
+    cert_id, _ = _seed(db)
+    resp = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 931,
+            "quantity": 30,
+        },
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["paichan_no"] == "W00009"
+    assert body["box_no"] == 931
+    assert body["quantity"] == 30
+
+    status = client.get(
+        "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
+    ).json()
+    assert status["determination"] == "has"
+    item = status["items"][0]
+    assert item["expected_qty"] == 80  # ERP 数量兜底
+    assert item["boxed_qty"] == 30
+
+
+def test_attachment_box_rejects_over_expected(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    first = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 932,
+            "quantity": 70,
+        },
+    )
+    assert first.status_code == 200
+    over = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 933,
+            "quantity": 11,
+        },
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_attachment_box_duplicate_in_same_box(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 934,
+            "quantity": 10,
+        },
+    )
+    dup = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 934,
+            "quantity": 5,
+        },
+    )
+    assert dup.status_code == 409
+    assert dup.json()["error_code"] == "DUPLICATE_BOX_NO"
+
+
+def test_attachment_box_mixed_with_product(client: TestClient, db: Session):
+    """附件与产品同箱：先装产品，再装附件，同 box_no。"""
+    cert_id, _ = _seed(db)
+    client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 935, "quantity": 10},
+    )
+    resp = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 935,
+            "quantity": 5,
+        },
+    )
+    assert resp.status_code == 200
+    info = client.get(
+        "/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
+    ).json()
+    box = next(b for b in info["existing_boxes"] if b["box_no"] == 935)
+    kinds = {i["kind"] for i in box["items"]}
+    assert kinds == {"product", "attachment"}
+
+
+def test_attachment_box_completion(client: TestClient, db: Session):
+    cert_id, _ = _seed(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": 936,
+            "quantity": 80,
+        },
+    )
+    status = client.get(
+        "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
+    ).json()
+    assert status["all_complete"] is True