fastAPI/app/services/box_service.py

import re

from sqlalchemy import bindparam, text
from sqlalchemy.orm import Session

from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
from app.schemas.box import BoxSaveRequest

ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")


class InvalidZongpaiError(Exception):
    pass


class ZongpaiNotFoundError(Exception):
    pass


class DuplicateBoxItemError(Exception):
    def __init__(self, paichan_no: str, box_no: int):
        self.paichan_no = paichan_no
        self.box_no = box_no


class DuplicateZongpaiBindError(Exception):
    def __init__(self, paichan_no: str, box_no: int, zongpai_no: str):
        self.paichan_no = paichan_no
        self.box_no = box_no
        self.zongpai_no = zongpai_no


def query_erp_info(db: Session, zongpai_no: str) -> dict:
    """从 ERP 视图查询总排号对应的排产号、工令号和数量。"""
    sql = text(_erp_info_sql(db))
    row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone()
    if not row:
        raise ZongpaiNotFoundError()
    return {
        "zongpai_no": row[0],
        "paichan_no": row[1],
        "quantity": int(row[2]),
        "work_order_no": row[3],
    }


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 "总排号", "排产号", "数量", "工令号" '
            'FROM "ERPAuto"."vw_productionContractData" '
            'WHERE "总排号" = :zongpai_no '
            "LIMIT 1"
        )
    return (
        "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()
    if not ZONGPAI_PATTERN.match(zongpai_no):
        raise InvalidZongpaiError()
    erp = query_erp_info(db, zongpai_no)
    paichan_no = erp["paichan_no"]
    quantity = erp["quantity"]

    boxes = (
        db.query(FinishedGoodsBox)
        .filter(FinishedGoodsBox.paichan_no == paichan_no)
        .order_by(FinishedGoodsBox.box_no)
        .all()
    )

    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,
                    "work_order_no": work_order_map.get(item.zongpai_no),
                    "quantity": int(item.quantity or 0),
                }
                for item in items
            ],
        })
        if box.box_no > max_box_no:
            max_box_no = box.box_no

    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,
        "suggested_box_no": max_box_no + 1,
    }


def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
    """保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。"""
    erp = query_erp_info(db, req.zongpai_no)
    paichan_no = erp["paichan_no"]

    if req.box_mode == "one-to-one":
        existing_box_no = (
            db.query(FinishedGoodsBox.box_no)
            .join(FinishedGoodsBoxItem, FinishedGoodsBox.id == FinishedGoodsBoxItem.box_id)
            .filter(
                FinishedGoodsBox.paichan_no == paichan_no,
                FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
            )
            .scalar()
        )
        if existing_box_no is not None:
            raise DuplicateZongpaiBindError(paichan_no, existing_box_no, req.zongpai_no)

    box = (
        db.query(FinishedGoodsBox)
        .filter(
            FinishedGoodsBox.paichan_no == paichan_no,
            FinishedGoodsBox.box_no == req.box_no,
        )
        .first()
    )

    if box:
        existing_item = (
            db.query(FinishedGoodsBoxItem)
            .filter(
                FinishedGoodsBoxItem.box_id == box.id,
                FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
            )
            .first()
        )
        if existing_item:
            raise DuplicateBoxItemError(paichan_no, req.box_no)
    else:
        box = FinishedGoodsBox(
            paichan_no=paichan_no,
            box_no=req.box_no,
        )
        db.add(box)
        db.flush()

    item = FinishedGoodsBoxItem(
        box_id=box.id,
        zongpai_no=req.zongpai_no,
        quantity=req.quantity,
    )
    db.add(item)
    db.commit()
    db.refresh(item)

    return {
        "paichan_no": paichan_no,
        "box_no": req.box_no,
        "zongpai_no": req.zongpai_no,
        "quantity": req.quantity,
        "created_at": item.created_at.isoformat(),
    }