From f155dc671b7d73a171d01145a70eaf6177594b4c Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 12 Aug 2026 13:42:53 +0800 Subject: [PATCH] feat(server): add GET /attachment/status with derived boxed qty Co-Authored-By: Claude --- app/api/v1/attachment.py | 8 +- app/services/attachment_service.py | 132 +++++++++++++++++++++++++++++ tests/test_attachment_api.py | 56 ++++++++++++ 3 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 app/services/attachment_service.py diff --git a/app/api/v1/attachment.py b/app/api/v1/attachment.py index 0c7962e..66731a4 100644 --- a/app/api/v1/attachment.py +++ b/app/api/v1/attachment.py @@ -3,7 +3,8 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.models.attachment import AttachmentCategory -from app.schemas.attachment import CategoryResponse +from app.schemas.attachment import AttachmentStatusResponse, CategoryResponse +from app.services.attachment_service import get_attachment_status router = APIRouter(tags=["attachment"]) @@ -20,3 +21,8 @@ def list_categories(db: Session = Depends(get_db)): return [ CategoryResponse(id=r.id, name=r.name, sort_order=r.sort_order) for r in rows ] + + +@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) diff --git a/app/services/attachment_service.py b/app/services/attachment_service.py new file mode 100644 index 0000000..ddf1839 --- /dev/null +++ b/app/services/attachment_service.py @@ -0,0 +1,132 @@ +import re + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models.attachment import ( + AttachmentCategory, + FinishedGoodsAttachment, + FinishedGoodsAttachmentItem, + FinishedGoodsAttachmentLocation, + FinishedGoodsBoxAttachmentItem, +) +from app.services.box_service import InvalidZongpaiError + +ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") + + +class CategoryNotFoundError(Exception): + pass + + +class AttachmentItemNotFoundError(Exception): + pass + + +class DuplicateAttachmentLocationError(Exception): + def __init__(self, location_code: str, registered_at: str): + self.location_code = location_code + self.registered_at = registered_at + + +class AlreadyOffShelfAttachmentError(Exception): + def __init__(self, location_code: str, registered_at: str): + self.location_code = location_code + self.registered_at = registered_at + + +class DuplicateAttachmentBoxItemError(Exception): + def __init__(self, paichan_no: str, box_no: int): + self.paichan_no = paichan_no + self.box_no = box_no + + +class CategoryHasBoxesError(Exception): + """配置中删除已有装箱记录的类型时抛出。""" + + def __init__(self, category_name: str): + self.category_name = category_name + + +def _validate_zongpai(zongpai_no: str) -> str: + zongpai_no = zongpai_no.strip().upper() + if not ZONGPAI_PATTERN.match(zongpai_no): + raise InvalidZongpaiError() + return zongpai_no + + +def _boxed_qty_map(db: Session, zongpai_no: str) -> dict[int, int]: + """按 category_id 汇总已装数量。""" + rows = db.execute( + select( + FinishedGoodsBoxAttachmentItem.category_id, + func.sum(FinishedGoodsBoxAttachmentItem.quantity), + ) + .where(FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no) + .group_by(FinishedGoodsBoxAttachmentItem.category_id) + ).all() + return {row[0]: int(row[1] or 0) for row in rows} + + +def _location_map(db: Session, zongpai_no: str) -> dict[int, str]: + rows = ( + db.query(FinishedGoodsAttachmentLocation) + .filter(FinishedGoodsAttachmentLocation.zongpai_no == zongpai_no) + .all() + ) + return {r.category_id: r.location_code for r in rows} + + +def _category_name_map(db: Session, category_ids: list[int]) -> dict[int, str]: + if not category_ids: + return {} + rows = ( + db.query(AttachmentCategory) + .filter(AttachmentCategory.id.in_(category_ids)) + .all() + ) + return {r.id: r.name for r in rows} + + +def get_attachment_status(db: Session, zongpai_no: str) -> dict: + zongpai_no = _validate_zongpai(zongpai_no) + + att = ( + db.query(FinishedGoodsAttachment) + .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no) + .first() + ) + determination = att.determination if att else "undetermined" + + items_db = ( + db.query(FinishedGoodsAttachmentItem) + .filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no) + .all() + ) + category_ids = [i.category_id for i in items_db] + names = _category_name_map(db, category_ids) + boxed = _boxed_qty_map(db, zongpai_no) + locations = _location_map(db, zongpai_no) + + items = [] + for it in items_db: + boxed_qty = boxed.get(it.category_id, 0) + items.append( + { + "category_id": it.category_id, + "name": names.get(it.category_id, "(已删除类型)"), + "expected_qty": it.expected_qty, + "boxed_qty": boxed_qty, + "location_code": locations.get(it.category_id), + "complete": boxed_qty >= it.expected_qty, + } + ) + + items.sort(key=lambda x: x["category_id"]) + all_complete = determination == "has" and bool(items) and all(i["complete"] for i in items) + return { + "zongpai_no": zongpai_no, + "determination": determination, + "all_complete": all_complete, + "items": items, + } diff --git a/tests/test_attachment_api.py b/tests/test_attachment_api.py index 1057765..02f1baa 100644 --- a/tests/test_attachment_api.py +++ b/tests/test_attachment_api.py @@ -28,3 +28,59 @@ def test_categories_lists_active_only(client: TestClient, db: Session): assert "TEST_OFF" not in names # sorted by sort_order then name assert names.index("TEST_CERT") < names.index("TEST_EXTRA") + + +def test_status_unknown_zongpai_is_undetermined(client: TestClient): + resp = client.get( + "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"} + ) + assert resp.status_code == 200 + data = resp.json() + assert data["zongpai_no"] == "26BW0011" + assert data["determination"] == "undetermined" + assert data["items"] == [] + assert data["all_complete"] is False + + +def test_status_invalid_zongpai(client: TestClient): + resp = client.get( + "/CargoTrace/attachment/status", params={"zongpai_no": "BAD"} + ) + assert resp.status_code == 400 + assert resp.json()["error_code"] == "INVALID_ZONGPAI" + + +def test_status_reflects_config_and_boxed(client: TestClient, db: Session): + cat_ids = _seed_categories(db) + cert_id = cat_ids[0] + # Plan: 检验证书 expected=80 + client.put( + "/CargoTrace/attachment/config", + json={ + "zongpai_no": "26BW0011", + "determination": "has", + "items": [{"category_id": cert_id, "expected_qty": 80}], + }, + ) + # Box 30 of them + client.post( + "/CargoTrace/attachment/box", + json={ + "zongpai_no": "26BW0011", + "category_id": cert_id, + "box_no": 921, + "quantity": 30, + }, + ) + resp = client.get( + "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"} + ) + assert resp.status_code == 200 + data = resp.json() + assert data["determination"] == "has" + assert data["all_complete"] is False + item = data["items"][0] + assert item["category_id"] == cert_id + assert item["expected_qty"] == 80 + assert item["boxed_qty"] == 30 + assert item["complete"] is False