diff --git a/app/api/v1/attachment.py b/app/api/v1/attachment.py index 77d726b..31d4ea0 100644 --- a/app/api/v1/attachment.py +++ b/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), + ) diff --git a/app/services/attachment_service.py b/app/services/attachment_service.py index a1780b8..197555a 100644 --- a/app/services/attachment_service.py +++ b/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 diff --git a/tests/test_attachment_api.py b/tests/test_attachment_api.py index ac882c1..991ab6d 100644 --- a/tests/test_attachment_api.py +++ b/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"