fastAPI/docs/plans/2026-05-11-boxing-module.md

装箱编号模块 (Boxing) Implementation Plan — FastAPI

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 实现 PRD v1.3 中定义的装箱编号模块后端接口：GET /CargoTrace/box/info 和 POST /CargoTrace/box，支持总排号查询排产号信息、已有箱号查询、以及装箱记录保存。

Architecture: 在现有三层架构（API → Service → Model）基础上扩展。ERP 视图数据通过 SQLAlchemy raw SQL 查询（跨 schema 访问 [ERPAuto].[vw_productionContractData]），CargoTrace 业务表通过 ORM 操作。

Tech Stack: Python 3.12+, FastAPI, SQLAlchemy 2.x (sync), pyodbc, pydantic

---

数据流

Flutter App
    │
    ├── GET /CargoTrace/box/info?zongpai_no=xxx
    │   └── 查询 ERP 视图 → 排产号 + 数量
    │   └── 查询 CargoTrace 表 → 已有箱号 + 明细
    │   └── 返回: paichan_no, quantity, existing_boxes, max_box_no, suggested_box_no
    │
    └── POST /CargoTrace/box
        └── 查询 ERP 视图 → 排产号
        └── 查找或创建 CargoTrace.finished_goods_box
        └── 创建 CargoTrace.finished_goods_box_item
        └── 返回: paichan_no, box_no, zongpai_no, quantity, created_at

关键设计决策

多码一箱的 POST 逻辑：

• 若 (paichan_no, box_no) 不存在 → 创建 box 记录 + 创建 box_item
• 若 (paichan_no, box_no) 已存在且该 zongpai_no 不在此箱中 → 仅创建 box_item（多码一箱场景）
• 若 (paichan_no, box_no) 已存在且该 zongpai_no 已在此箱中 → 返回 409 DUPLICATE_BOX_NO

---

Task 1: Schema 层 — 装箱请求/响应模型

Files:

• Create: app/schemas/box.py

Step 1: 创建 box.py

import re

from pydantic import BaseModel, field_validator

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


class BoxItemDetail(BaseModel):
    """箱号下某个总排号的明细"""
    zongpai_no: str
    quantity: int


class BoxDetail(BaseModel):
    """一个箱号的完整信息"""
    box_no: int
    items: list[BoxItemDetail]


class BoxInfoResponse(BaseModel):
    """GET /box/info 响应"""
    zongpai_no: str
    paichan_no: str
    quantity: int
    existing_boxes: list[BoxDetail]
    max_box_no: int
    suggested_box_no: int


class BoxSaveRequest(BaseModel):
    """POST /box 请求"""
    zongpai_no: str
    box_no: int
    quantity: int

    @field_validator("zongpai_no")
    @classmethod
    def validate_zongpai_no(cls, v: str) -> str:
        v = v.strip().upper()
        if not ZONGPAI_PATTERN.match(v):
            raise ValueError("INVALID_ZONGPAI")
        return v

    @field_validator("box_no")
    @classmethod
    def validate_box_no(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("INVALID_BOX_NO")
        return v

    @field_validator("quantity")
    @classmethod
    def validate_quantity(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("INVALID_QUANTITY")
        return v


class BoxSaveResponse(BaseModel):
    """POST /box 响应"""
    paichan_no: str
    box_no: int
    zongpai_no: str
    quantity: int
    created_at: str

Step 2: 验证导入

Run: cd apps/fastAPI && .venv/Scripts/python -c "from app.schemas.box import BoxInfoResponse, BoxSaveRequest; print('Schemas OK')"

Step 3: Commit

git add app/schemas/box.py
git commit -m "feat: add box schemas for boxing module API"

---

Task 2: Service 层 — 装箱业务逻辑

Files:

• Create: app/services/box_service.py

Step 1: 创建 box_service.py

from sqlalchemy import text
from sqlalchemy.orm import Session

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


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


def query_erp_info(db: Session, zongpai_no: str) -> dict:
    """从 ERP 视图查询总排号对应的排产号和数量。"""
    sql = text(
        "SELECT TOP 1 [总排号], [排产号], [数量] "
        "FROM [ERPAuto].[vw_productionContractData] "
        "WHERE [总排号] = :zongpai_no"
    )
    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]),
    }


def get_box_info(db: Session, zongpai_no: str) -> dict:
    """获取装箱信息：排产号、数量、已有箱号明细。"""
    erp = query_erp_info(db, zongpai_no)
    paichan_no = erp["paichan_no"]
    quantity = erp["quantity"]

    # 查询该排产号下的所有箱（按 box_no 升序）
    boxes = (
        db.query(FinishedGoodsBox)
        .filter(FinishedGoodsBox.paichan_no == paichan_no)
        .order_by(FinishedGoodsBox.box_no)
        .all()
    )

    existing_boxes = []
    max_box_no = 0
    for box in boxes:
        items = (
            db.query(FinishedGoodsBoxItem)
            .filter(FinishedGoodsBoxItem.box_id == box.id)
            .all()
        )
        existing_boxes.append({
            "box_no": box.box_no,
            "items": [
                {"zongpai_no": 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,
        "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:
    """保存装箱记录。处理多码一箱场景：箱已存在时仅追加明细。"""
    # 1. 查询 ERP 获取排产号
    erp = query_erp_info(db, req.zongpai_no)
    paichan_no = erp["paichan_no"]

    # 2. 查找已有箱记录
    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()  # 获取 box.id

    # 3. 创建箱明细
    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(),
    }

Step 2: 验证导入

Run: cd apps/fastAPI && .venv/Scripts/python -c "from app.services.box_service import get_box_info, save_box_record; print('Service OK')"

Step 3: Commit

git add app/services/box_service.py
git commit -m "feat: add box service with ERP query and save logic"

---

Task 3: API 路由层 — GET /box/info 和 POST /box

Files:

• Create: app/api/v1/box.py
• Modify: app/main.py

Step 1: 创建 api/v1/box.py

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.schemas.box import BoxInfoResponse, BoxSaveRequest, BoxSaveResponse
from app.services.box_service import get_box_info, save_box_record

router = APIRouter(tags=["boxing"])


@router.get("/box/info", response_model=BoxInfoResponse)
def box_info(
    zongpai_no: str = Query(..., description="总排号"),
    db: Session = Depends(get_db),
):
    """查询装箱信息：根据总排号查询排产号、数量和已有箱号明细。"""
    return get_box_info(db, zongpai_no)


@router.post("/box", response_model=BoxSaveResponse)
def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
    """保存装箱记录：将总排号记录到指定箱号中。"""
    return save_box_record(db, req)

Step 2: 更新 main.py — 注册路由 + 异常处理

在现有 main.py 中添加：

# 新增导入（文件顶部）
from app.api.v1.box import router as box_router
from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError

# 注册路由（在 location_router 之后）
app.include_router(box_router, prefix="/CargoTrace")

# 新增异常处理（在现有 handler 之后）
@app.exception_handler(ZongpaiNotFoundError)
async def zongpai_not_found_handler(request: Request, exc: ZongpaiNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error_code": "ZONGPAI_NOT_FOUND",
            "message": "未找到该总排号对应的排产号信息",
        },
    )


@app.exception_handler(DuplicateBoxItemError)
async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
    return JSONResponse(
        status_code=409,
        content={
            "error_code": "DUPLICATE_BOX_NO",
            "message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在",
            "paichan_no": exc.paichan_no,
            "box_no": exc.box_no,
        },
    )

Step 3: 启动验证

Run: cd apps/fastAPI && .venv/Scripts/python -m uvicorn app.main:app --reload

访问 http://127.0.0.1:8000/docs，确认 Swagger UI 中出现 GET /CargoTrace/box/info 和 POST /CargoTrace/box 两个端点。

Step 4: 快速功能验证

使用 Swagger UI 或 curl 测试：

• GET /CargoTrace/box/info?zongpai_no=26BW0011 → 应返回 W00009 的箱号信息
• POST /CargoTrace/box body: {"zongpai_no": "26BW0011", "box_no": 99, "quantity": 5} → 应返回 200

Step 5: Commit

git add app/api/v1/box.py app/main.py
git commit -m "feat: add GET /box/info and POST /box endpoints for boxing module"

---

Task 4: 测试 — 接口集成测试

Files:

• Modify: tests/conftest.py
• Create: tests/test_box_api.py

Step 1: 更新 conftest.py — 添加装箱测试数据清理

在现有 cleanup_test_data fixture 中追加装箱表的测试数据清理：

# 在 conftest.py 的 cleanup_test_data 中追加
from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem

# 清理装箱测试数据
test_box_items = db.query(FinishedGoodsBoxItem).filter(
    FinishedGoodsBoxItem.zongpai_no.in_(["26B997", "26B998", "26B999"])
).all()
for item in test_box_items:
    db.delete(item)

db.query(FinishedGoodsBox).filter(
    FinishedGoodsBox.paichan_no.in_(["TEST_PAICHAN_999"])
).delete(synchronize_session=False)
db.commit()

注意：装箱测试使用 ERP 中真实存在的总排号 26BW0011（排产号 W00009）做查询测试，保存测试则使用无效的测试总排号（会被 404 拦截），避免污染真实数据。如需完整保存测试，使用临时箱号（如 999），测试后立即清理。

Step 2: 创建 test_box_api.py

from fastapi.testclient import TestClient


# --- GET /box/info 测试 ---


def test_box_info_success(client: TestClient):
    """查询真实存在的总排号 → 200"""
    resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
    assert resp.status_code == 200
    data = resp.json()
    assert data["zongpai_no"] == "26BW0011"
    assert data["paichan_no"] == "W00009"
    assert data["quantity"] == 80
    assert "existing_boxes" in data
    assert "max_box_no" in data
    assert data["suggested_box_no"] == data["max_box_no"] + 1


def test_box_info_invalid_zongpai(client: TestClient):
    """无效总排号格式 → 400"""
    resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "INVALID"})
    assert resp.status_code == 400
    assert resp.json()["error_code"] == "INVALID_ZONGPAI"


def test_box_info_not_found(client: TestClient):
    """格式合法但 ERP 中不存在 → 404"""
    resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26B99999"})
    assert resp.status_code == 404
    assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"


# --- POST /box 测试 ---


def test_box_save_invalid_zongpai(client: TestClient):
    """无效总排号 → 400"""
    resp = client.post(
        "/CargoTrace/box",
        json={"zongpai_no": "INVALID", "box_no": 1, "quantity": 10},
    )
    assert resp.status_code == 400
    assert resp.json()["error_code"] == "INVALID_ZONGPAI"


def test_box_save_invalid_box_no(client: TestClient):
    """箱号 <= 0 → 400"""
    resp = client.post(
        "/CargoTrace/box",
        json={"zongpai_no": "26BW0011", "box_no": 0, "quantity": 10},
    )
    assert resp.status_code == 400


def test_box_save_invalid_quantity(client: TestClient):
    """数量 <= 0 → 400"""
    resp = client.post(
        "/CargoTrace/box",
        json={"zongpai_no": "26BW0011", "box_no": 1, "quantity": 0},
    )
    assert resp.status_code == 400


def test_box_save_zongpai_not_found(client: TestClient):
    """ERP 中不存在的总排号 → 404"""
    resp = client.post(
        "/CargoTrace/box",
        json={"zongpai_no": "26B99999", "box_no": 1, "quantity": 10},
    )
    assert resp.status_code == 404
    assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"

Step 3: 运行测试

Run: cd apps/fastAPI && .venv/Scripts/python -m pytest tests/test_box_api.py -v

Step 4: Commit

git add tests/
git commit -m "test: add integration tests for boxing API"

---

API 接口汇总

方法	路径	描述
GET	/CargoTrace/box/info?zongpai_no=xxx	查询总排号对应的装箱信息
POST	/CargoTrace/box	保存装箱记录

错误码

HTTP 状态码	error_code	含义
400	INVALID_ZONGPAI	总排号格式不合法
400	INVALID_BOX_NO	箱号不合法（≤0）
400	INVALID_QUANTITY	数量不合法（≤0）
404	ZONGPAI_NOT_FOUND	总排号在 ERP 中未找到对应排产号
409	DUPLICATE_BOX_NO	该排产号下该箱号中已存在该总排号
500	SERVER_ERROR	服务器内部错误