feat(server): add POST /attachment/location mirroring product shelving

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

app/api/v1/attachment.py

@@ -5,12 +5,16 @@ from app.core.database import get_db
 from app.models.attachment import AttachmentCategory
 from app.schemas.attachment import (
     AttachmentConfigRequest,
+    AttachmentLocationRequest,
+    AttachmentLocationResponse,
     AttachmentStatusResponse,
     CategoryResponse,
 )
+from app.schemas.common import ErrorResponse
 from app.services.attachment_service import (
     configure_attachment,
     get_attachment_status,
+    register_attachment_location,
 )
 
 router = APIRouter(tags=["attachment"])
@@ -38,3 +42,25 @@ def attachment_status(zongpai_no: str, db: Session = Depends(get_db)):
 @router.put("/attachment/config", response_model=AttachmentStatusResponse)
 def put_attachment_config(req: AttachmentConfigRequest, db: Session = Depends(get_db)):
     return configure_attachment(db, req)
+
+
+@router.post(
+    "/attachment/location",
+    response_model=AttachmentLocationResponse,
+    responses={
+        400: {"description": "参数非法", "model": ErrorResponse},
+        404: {"description": "类型不存在", "model": ErrorResponse},
+        409: {"description": "重复上架或已下架", "model": ErrorResponse},
+    },
+)
+def post_attachment_location(
+    req: AttachmentLocationRequest, db: Session = Depends(get_db)
+):
+    record = register_attachment_location(db, req)
+    return AttachmentLocationResponse(
+        zongpai_no=record.zongpai_no,
+        category_id=record.category_id,
+        location_code=record.location_code,
+        created_at=record.created_at.isoformat(),
+        previous_location=getattr(record, "previous_location", None),
+    )

app/services/attachment_service.py

@@ -198,3 +198,62 @@ def configure_attachment(db: Session, req: AttachmentConfigRequest) -> dict:
 
     db.commit()
     return get_attachment_status(db, zongpai_no)
+
+
+def _is_transit_location(location_code: str) -> bool:
+    return location_code.startswith("TRANS-")
+
+
+def _is_temp_storage_location(location_code: str) -> bool:
+    return location_code.startswith("B")
+
+
+def register_attachment_location(db: Session, req) -> FinishedGoodsAttachmentLocation:
+    zongpai_no = _validate_zongpai(req.zongpai_no)
+
+    # Category must exist
+    if not _category_name_map(db, [req.category_id]):
+        raise CategoryNotFoundError(req.category_id)
+
+    existing = (
+        db.query(FinishedGoodsAttachmentLocation)
+        .filter(
+            FinishedGoodsAttachmentLocation.zongpai_no == zongpai_no,
+            FinishedGoodsAttachmentLocation.category_id == req.category_id,
+        )
+        .first()
+    )
+    if existing:
+        if _is_transit_location(existing.location_code):
+            raise AlreadyOffShelfAttachmentError(
+                location_code=existing.location_code,
+                registered_at=existing.created_at.isoformat(),
+            )
+
+        target_is_transit = _is_transit_location(req.location_code)
+        target_is_temp = _is_temp_storage_location(req.location_code)
+        current_is_temp = _is_temp_storage_location(existing.location_code)
+
+        if target_is_transit or (not current_is_temp and target_is_temp):
+            previous_location = existing.location_code
+            existing.location_code = req.location_code
+            existing.created_at = datetime.now()
+            db.commit()
+            db.refresh(existing)
+            existing.previous_location = previous_location
+            return existing
+
+        raise DuplicateAttachmentLocationError(
+            location_code=existing.location_code,
+            registered_at=existing.created_at.isoformat(),
+        )
+
+    record = FinishedGoodsAttachmentLocation(
+        zongpai_no=zongpai_no,
+        category_id=req.category_id,
+        location_code=req.location_code,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return record

tests/test_attachment_api.py

@@ -194,3 +194,103 @@ def test_config_rejects_unknown_category(client: TestClient):
     )
     assert resp.status_code == 404
     assert resp.json()["error_code"] == "CATEGORY_NOT_FOUND"
+
+
+def _config_one_type(client: TestClient, cert_id: int):
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+
+
+def test_attachment_location_create_and_duplicate(client: TestClient, db: Session):
+    cert_id = _seed_categories(db)[0]
+    _config_one_type(client, cert_id)
+
+    created = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "A01-01-01",
+        },
+    )
+    assert created.status_code == 200
+    assert created.json()["location_code"] == "A01-01-01"
+    assert created.json()["previous_location"] is None
+
+    dup = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "A02-02-02",
+        },
+    )
+    assert dup.status_code == 409
+    assert dup.json()["error_code"] == "DUPLICATE_LOCATION"
+
+
+def test_attachment_location_normal_to_temp_then_transit(client: TestClient, db: Session):
+    cert_id = _seed_categories(db)[0]
+    _config_one_type(client, cert_id)
+
+    client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "A01-01-01",
+        },
+    )
+    # normal -> temp storage (B prefix) allowed
+    to_temp = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "B01-01-01",
+        },
+    )
+    assert to_temp.status_code == 200
+    assert to_temp.json()["previous_location"] == "A01-01-01"
+
+    # -> transit (terminal)
+    to_transit = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "TRANS-001",
+        },
+    )
+    assert to_transit.status_code == 200
+
+    # already transit -> any op blocked
+    again = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "location_code": "A03-03-03",
+        },
+    )
+    assert again.status_code == 409
+    assert again.json()["error_code"] == "ALREADY_OFF_SHELF"
+
+
+def test_attachment_location_rejects_unknown_category(client: TestClient):
+    resp = client.post(
+        "/CargoTrace/attachment/location",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": 9999999,
+            "location_code": "A01-01-01",
+        },
+    )
+    assert resp.status_code == 404
+    assert resp.json()["error_code"] == "CATEGORY_NOT_FOUND"