feat: support off-shelf location registration

app/api/v1/location.py

@@ -18,7 +18,7 @@ router = APIRouter()
             "model": ErrorResponse,
         },
         409: {
-            "description": "该总排号已存在货位记录",
+            "description": "重复上架或已下架",
             "model": ErrorResponse,
         },
     },
@@ -29,4 +29,5 @@ def create_location(req: LocationRequest, db: Session = Depends(get_db)):
         zongpai_no=record.zongpai_no,
         location_code=record.location_code,
         created_at=record.created_at.isoformat(),
+        previous_location=getattr(record, "previous_location", None),
     )

app/main.py

@@ -18,7 +18,7 @@ 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.location_service import AlreadyOffShelfError, DuplicateLocationError
 from app.services.box_service import (
     BoxItemNotFoundError,
     DuplicateBoxItemError,
@@ -47,6 +47,19 @@ async def duplicate_location_handler(request: Request, exc: DuplicateLocationErr
     )
 
 
+@app.exception_handler(AlreadyOffShelfError)
+async def already_off_shelf_handler(request: Request, exc: AlreadyOffShelfError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_OFF_SHELF",
+            "message": "该总排号已下架至转运区域，不可重新上架",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
 @app.exception_handler(RequestValidationError)
 async def validation_error_handler(request: Request, exc: RequestValidationError):
     for err in exc.errors():

app/schemas/location.py

@@ -32,6 +32,7 @@ class LocationResponse(BaseModel):
     zongpai_no: str
     location_code: str
     created_at: str
+    previous_location: str | None = None
 
 
 class DuplicateLocationDetail(BaseModel):

app/services/location_service.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsLocation
@@ -12,6 +14,18 @@ class DuplicateLocationError(Exception):
         self.registered_at = registered_at
 
 
+class AlreadyOffShelfError(Exception):
+    """总排号已下架至转运区域。"""
+
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+def _is_transit_location(location_code: str) -> bool:
+    return location_code.startswith("TRANS-")
+
+
 def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation:
     existing = (
         db.query(FinishedGoodsLocation)
@@ -19,6 +33,21 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio
         .first()
     )
     if existing:
+        if _is_transit_location(existing.location_code):
+            raise AlreadyOffShelfError(
+                location_code=existing.location_code,
+                registered_at=existing.created_at.isoformat(),
+            )
+
+        if _is_transit_location(req.location_code):
+            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 DuplicateLocationError(
             location_code=existing.location_code,
             registered_at=existing.created_at.isoformat(),

tests/conftest.py

@@ -22,7 +22,15 @@ def db():
 @pytest.fixture(autouse=True)
 def cleanup_test_data(db: Session):
     yield
-    test_zongpai_nos = ["26B999", "26C999", "26BW0999", "26T999"]
+    test_zongpai_nos = [
+        "26B999",
+        "26C999",
+        "26BW0999",
+        "26T999",
+        "26T998",
+        "26B998",
+        "26C998",
+    ]
     db.query(FinishedGoodsLocation).filter(
         FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos)
     ).delete(synchronize_session=False)

tests/test_location_api.py

@@ -1,5 +1,8 @@
 from fastapi.testclient import TestClient
 
+from app.core.database import SessionLocal
+from app.models.finished_goods import FinishedGoodsLocation
+
 
 def test_register_location_success(client: TestClient):
     """正常上架 — 应返回 200"""
@@ -13,8 +16,8 @@ def test_register_location_success(client: TestClient):
     assert data["location_code"] == "A01-02-03"
 
 
-def test_register_location_duplicate(client: TestClient):
-    """重复上架 — 应返回 409，包含统一错误格式"""
+def test_register_location_duplicate_normal_to_normal(client: TestClient):
+    """普通货位重复上架到普通货位 — 应返回 DUPLICATE_LOCATION"""
     client.post(
         "/CargoTrace/location",
         json={"zongpai_no": "26C999", "location_code": "A01-02-03"},
@@ -30,6 +33,69 @@ def test_register_location_duplicate(client: TestClient):
     assert "registered_at" in data
 
 
+def test_register_location_normal_to_transit_updates_record(client: TestClient):
+    """普通货位下架到转运货位 — 应更新记录并返回 previous_location"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26T998", "location_code": "A01-02-03"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26T998", "location_code": "TRANS-01"},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["zongpai_no"] == "26T998"
+    assert data["location_code"] == "TRANS-01"
+    assert data["previous_location"] == "A01-02-03"
+
+    db = SessionLocal()
+    try:
+        record = (
+            db.query(FinishedGoodsLocation)
+            .filter(FinishedGoodsLocation.zongpai_no == "26T998")
+            .first()
+        )
+        assert record is not None
+        assert record.location_code == "TRANS-01"
+    finally:
+        db.close()
+
+
+def test_register_location_transit_to_normal_rejected(client: TestClient):
+    """已下架后再上架到普通货位 — 应返回 ALREADY_OFF_SHELF"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26B998", "location_code": "TRANS-01"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26B998", "location_code": "A01-02-03"},
+    )
+    assert resp.status_code == 409
+    data = resp.json()
+    assert data["error_code"] == "ALREADY_OFF_SHELF"
+    assert data["message"] == "该总排号已下架至转运区域，不可重新上架"
+    assert data["location_code"] == "TRANS-01"
+    assert "registered_at" in data
+
+
+def test_register_location_transit_to_transit_rejected(client: TestClient):
+    """已下架后再提交转运货位 — 应返回 ALREADY_OFF_SHELF"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26C998", "location_code": "TRANS-01"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26C998", "location_code": "TRANS-02"},
+    )
+    assert resp.status_code == 409
+    data = resp.json()
+    assert data["error_code"] == "ALREADY_OFF_SHELF"
+    assert data["location_code"] == "TRANS-01"
+
+
 def test_register_location_invalid_zongpai(client: TestClient):
     """无效总排号 — 应返回 400"""
     resp = client.post(