diff --git a/app/schemas/box.py b/app/schemas/box.py index 85bea8a..70bc844 100644 --- a/app/schemas/box.py +++ b/app/schemas/box.py @@ -8,6 +8,7 @@ ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$") class BoxItemDetail(BaseModel): """箱号下某个总排号的明细""" zongpai_no: str + work_order_no: str | None = None quantity: int @@ -21,6 +22,7 @@ class BoxInfoResponse(BaseModel): """GET /box/info 响应""" zongpai_no: str paichan_no: str + work_order_no: str | None = None quantity: int existing_boxes: list[BoxDetail] max_box_no: int diff --git a/app/services/box_service.py b/app/services/box_service.py index 11aa676..ea12962 100644 --- a/app/services/box_service.py +++ b/app/services/box_service.py @@ -1,6 +1,6 @@ import re -from sqlalchemy import text +from sqlalchemy import bindparam, text from sqlalchemy.orm import Session from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem @@ -31,7 +31,7 @@ class DuplicateZongpaiBindError(Exception): def query_erp_info(db: Session, zongpai_no: str) -> dict: - """从 ERP 视图查询总排号对应的排产号和数量。""" + """从 ERP 视图查询总排号对应的排产号、工令号和数量。""" sql = text(_erp_info_sql(db)) row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone() if not row: @@ -40,6 +40,7 @@ def query_erp_info(db: Session, zongpai_no: str) -> dict: "zongpai_no": row[0], "paichan_no": row[1], "quantity": int(row[2]), + "work_order_no": row[3], } @@ -47,18 +48,43 @@ def _erp_info_sql(db: Session) -> str: dialect_name = db.bind.dialect.name if db.bind is not None else "" if dialect_name == "postgresql": return ( - 'SELECT "总排号", "排产号", "数量" ' + 'SELECT "总排号", "排产号", "数量", "工令号" ' 'FROM "ERPAuto"."vw_productionContractData" ' 'WHERE "总排号" = :zongpai_no ' "LIMIT 1" ) return ( - "SELECT TOP 1 [总排号], [排产号], [数量] " + "SELECT TOP 1 [总排号], [排产号], [数量], [工令号] " "FROM [ERPAuto].[vw_productionContractData] " "WHERE [总排号] = :zongpai_no" ) +def query_erp_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, str | None]: + """批量查询总排号对应的工令号。""" + if not zongpai_nos: + return {} + + dialect_name = db.bind.dialect.name if db.bind is not None else "" + if dialect_name == "postgresql": + sql = text( + 'SELECT "总排号", "工令号" ' + 'FROM "ERPAuto"."vw_productionContractData" ' + 'WHERE "总排号" IN :zongpai_nos' + ) + else: + sql = text( + "SELECT [总排号], [工令号] " + "FROM [ERPAuto].[vw_productionContractData] " + "WHERE [总排号] IN :zongpai_nos" + ) + rows = db.execute( + sql.bindparams(bindparam("zongpai_nos", expanding=True)), + {"zongpai_nos": sorted(set(zongpai_nos))}, + ).fetchall() + return {row[0]: row[1] for row in rows} + + def get_box_info(db: Session, zongpai_no: str) -> dict: """获取装箱信息:排产号、数量、已有箱号明细。""" zongpai_no = zongpai_no.strip().upper() @@ -75,18 +101,31 @@ def get_box_info(db: Session, zongpai_no: str) -> dict: .all() ) - existing_boxes = [] - max_box_no = 0 + box_items = {} + item_zongpai_nos = [] for box in boxes: items = ( db.query(FinishedGoodsBoxItem) .filter(FinishedGoodsBoxItem.box_id == box.id) .all() ) + box_items[box.id] = items + item_zongpai_nos.extend(item.zongpai_no for item in items) + + work_order_map = query_erp_work_order_map(db, item_zongpai_nos) + + existing_boxes = [] + max_box_no = 0 + for box in boxes: + items = box_items[box.id] existing_boxes.append({ "box_no": box.box_no, "items": [ - {"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)} + { + "zongpai_no": item.zongpai_no, + "work_order_no": work_order_map.get(item.zongpai_no), + "quantity": int(item.quantity or 0), + } for item in items ], }) @@ -96,6 +135,7 @@ def get_box_info(db: Session, zongpai_no: str) -> dict: return { "zongpai_no": zongpai_no, "paichan_no": paichan_no, + "work_order_no": erp["work_order_no"], "quantity": quantity, "existing_boxes": existing_boxes, "max_box_no": max_box_no, diff --git a/tests/test_box_api.py b/tests/test_box_api.py index d11899e..a91d100 100644 --- a/tests/test_box_api.py +++ b/tests/test_box_api.py @@ -13,6 +13,8 @@ def test_box_info_success(client: TestClient): data = resp.json() assert data["zongpai_no"] == "26BW0011" assert data["paichan_no"] == "W00009" + assert "work_order_no" in data + assert data["work_order_no"] is None or isinstance(data["work_order_no"], str) assert data["quantity"] == 80 assert "existing_boxes" in data assert "max_box_no" in data