feat(server): extend GET /box/info with mixed items and attachment summary

Co-Authored-By: Claude <noreply@anthropic.com>

.gitignore

@@ -7,6 +7,9 @@ dist/
 build/
 .pytest_cache/
 .claude/
+.agents/
+# Config with secrets
+config/settings.yaml
+config/settings.local.yaml
 
-# Local configuration overrides
-config/settings.local.yaml
+.runtime

CLAUDE.md

@@ -0,0 +1,7 @@
+# fastapi CLAUDE.md
+
+## Branch Strategy
+
+- The main branch is `master`, always kept in a releasable state
+- All new feature development must be done on a `dev` branch, created from `master`
+- After verification, merge `dev` back into `master`

README.md

@@ -22,14 +22,24 @@ Edit `config/settings.yaml` to configure your environment:
 
 ```yaml
 database:
+  active: sql_server  # sql_server or postgresql
   sql_server:
     host: your-host
     port: 1433
     database: your-database
     username: your-username
     password: your-password
+  postgresql:
+    host: your-postgres-host
+    port: 5432
+    database: your-database
+    username: your-username
+    password: your-password
 ```
 
+Set `database.active` to choose the database backend. Both backends expect the
+same database name, schemas, and table structure.
+
 ### Local Overrides
 
 For local development, create `config/settings.local.yaml` to override specific values without committing them.

app/api/v1/attachment.py

@@ -0,0 +1,115 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.models.attachment import AttachmentCategory
+from app.schemas.attachment import (
+    AttachmentBoxDeleteResponse,
+    AttachmentBoxSaveRequest,
+    AttachmentBoxSaveResponse,
+    AttachmentBoxUpdateRequest,
+    AttachmentBoxUpdateResponse,
+    AttachmentConfigRequest,
+    AttachmentLocationRequest,
+    AttachmentLocationResponse,
+    AttachmentStatusResponse,
+    CategoryResponse,
+)
+from app.schemas.common import ErrorResponse
+from app.services.attachment_service import (
+    configure_attachment,
+    delete_attachment_box,
+    get_attachment_status,
+    register_attachment_location,
+    save_attachment_box,
+    update_attachment_box,
+)
+
+router = APIRouter(tags=["attachment"])
+
+
+@router.get("/attachment/categories", response_model=list[CategoryResponse])
+def list_categories(db: Session = Depends(get_db)):
+    """列出启用中的附件类型，按 sort_order、name 排序。"""
+    rows = (
+        db.query(AttachmentCategory)
+        .filter(AttachmentCategory.is_active == True)
+        .order_by(AttachmentCategory.sort_order, AttachmentCategory.name)
+        .all()
+    )
+    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)
+
+
+@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),
+    )
+
+
+@router.post(
+    "/attachment/box",
+    response_model=AttachmentBoxSaveResponse,
+    responses={
+        400: {"description": "参数非法或超量", "model": ErrorResponse},
+        404: {"description": "总排号/类型不存在", "model": ErrorResponse},
+        409: {"description": "同箱同类型重复", "model": ErrorResponse},
+    },
+)
+def post_attachment_box(
+    req: AttachmentBoxSaveRequest, db: Session = Depends(get_db)
+):
+    return save_attachment_box(db, req)
+
+
+@router.patch(
+    "/attachment/box/{item_id}",
+    response_model=AttachmentBoxUpdateResponse,
+    responses={
+        400: {"description": "参数非法或超量", "model": ErrorResponse},
+        404: {"description": "明细不存在", "model": ErrorResponse},
+        409: {"description": "同箱同类型重复", "model": ErrorResponse},
+    },
+)
+def patch_attachment_box(
+    item_id: int, req: AttachmentBoxUpdateRequest, db: Session = Depends(get_db)
+):
+    return update_attachment_box(db, item_id, req)
+
+
+@router.delete(
+    "/attachment/box/{item_id}",
+    response_model=AttachmentBoxDeleteResponse,
+    responses={404: {"description": "明细不存在", "model": ErrorResponse}},
+)
+def delete_attachment_box_endpoint(
+    item_id: int, db: Session = Depends(get_db)
+):
+    return delete_attachment_box(db, item_id)

app/api/v1/box.py

@@ -2,8 +2,20 @@ 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
+from app.schemas.box import (
+    BoxDeleteResponse,
+    BoxInfoResponse,
+    BoxSaveRequest,
+    BoxSaveResponse,
+    BoxUpdateRequest,
+    BoxUpdateResponse,
+)
+from app.services.box_service import (
+    delete_box_item,
+    get_box_info,
+    save_box_record,
+    update_box_item,
+)
 
 router = APIRouter(tags=["boxing"])
 
@@ -21,3 +33,15 @@ def box_info(
 def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
     """保存装箱记录：将总排号记录到指定箱号中。"""
     return save_box_record(db, req)
+
+
+@router.patch("/box/{box_item_id}", response_model=BoxUpdateResponse)
+def update_box(box_item_id: int, req: BoxUpdateRequest, db: Session = Depends(get_db)):
+    """更新已分配装箱明细。"""
+    return update_box_item(db, box_item_id, req)
+
+
+@router.delete("/box/{box_item_id}", response_model=BoxDeleteResponse)
+def delete_box(box_item_id: int, db: Session = Depends(get_db)):
+    """删除误录的装箱明细。"""
+    return delete_box_item(db, box_item_id)

app/api/v1/location.py

@@ -3,8 +3,12 @@ from sqlalchemy.orm import Session
 
 from app.core.database import get_db
 from app.schemas.common import ErrorResponse
-from app.schemas.location import LocationRequest, LocationResponse
-from app.services.location_service import register_location
+from app.schemas.location import (
+    LocationRequest,
+    LocationResponse,
+    PaichaOverviewResponse,
+)
+from app.services.location_service import get_paicha_overview, register_location
 
 router = APIRouter()
 
@@ -18,7 +22,7 @@ router = APIRouter()
             "model": ErrorResponse,
         },
         409: {
-            "description": "该总排号已存在货位记录",
+            "description": "重复上架或已下架",
             "model": ErrorResponse,
         },
     },
@@ -29,4 +33,23 @@ def create_location(req: LocationRequest, db: Session = Depends(get_db)):
         zongpai_no=record.zongpai_no,
         location_code=record.location_code,
         created_at=record.created_at.isoformat(),
+        previous_location=getattr(record, "previous_location", None),
     )
+
+
+@router.get(
+    "/location/paicha-overview",
+    response_model=PaichaOverviewResponse,
+    responses={
+        400: {
+            "description": "总排号格式不合法",
+            "model": ErrorResponse,
+        },
+        404: {
+            "description": "该总排号未找到对应排产号",
+            "model": ErrorResponse,
+        },
+    },
+)
+def paicha_overview(zongpai_no: str, db: Session = Depends(get_db)):
+    return get_paicha_overview(db, zongpai_no)

app/main.py

@@ -18,13 +18,34 @@ except Exception as e:
 
 from app.api.v1.location import router as location_router
 from app.api.v1.box import router as box_router
-from app.services.location_service import DuplicateLocationError
-from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError
+from app.api.v1.attachment import router as attachment_router
+from app.services.location_service import (
+    AlreadyOffShelfError,
+    DuplicateLocationError,
+    PaichaNotFoundError,
+)
+from app.services.box_service import (
+    BoxItemNotFoundError,
+    DuplicateBoxItemError,
+    InvalidBoxItemError,
+    InvalidQuantityError,
+    InvalidZongpaiError,
+    ZongpaiNotFoundError,
+)
+from app.services.attachment_service import (
+    AlreadyOffShelfAttachmentError,
+    AttachmentItemNotFoundError,
+    CategoryHasBoxesError,
+    CategoryNotFoundError,
+    DuplicateAttachmentBoxItemError,
+    DuplicateAttachmentLocationError,
+)
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
 app.include_router(location_router, prefix="/CargoTrace")
 app.include_router(box_router, prefix="/CargoTrace")
+app.include_router(attachment_router, prefix="/CargoTrace")
 
 
 @app.exception_handler(DuplicateLocationError)
@@ -40,6 +61,19 @@ async def duplicate_location_handler(request: Request, exc: DuplicateLocationErr
     )
 
 
+@app.exception_handler(AlreadyOffShelfError)
+async def already_off_shelf_handler(request: Request, exc: AlreadyOffShelfError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_OFF_SHELF",
+            "message": "该总排号已下架至转运区域，不可重新上架",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
 @app.exception_handler(RequestValidationError)
 async def validation_error_handler(request: Request, exc: RequestValidationError):
     for err in exc.errors():
@@ -106,6 +140,124 @@ async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
     )
 
 
+@app.exception_handler(BoxItemNotFoundError)
+async def box_item_not_found_handler(request: Request, exc: BoxItemNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "BOX_ITEM_NOT_FOUND",
+            "message": "指定装箱明细不存在",
+        },
+    )
+
+
+@app.exception_handler(PaichaNotFoundError)
+async def paicha_not_found_handler(request: Request, exc: PaichaNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "PAICHA_NOT_FOUND",
+            "message": "暂无排产信息",
+        },
+    )
+
+
+@app.exception_handler(InvalidQuantityError)
+async def invalid_quantity_handler(request: Request, exc: InvalidQuantityError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_QUANTITY",
+            "message": "装箱数量超出可装数量上限",
+        },
+    )
+
+
+@app.exception_handler(InvalidBoxItemError)
+async def invalid_box_item_handler(request: Request, exc: InvalidBoxItemError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_BOX_ITEM",
+            "message": "装箱明细记录不合法或不允许修改",
+        },
+    )
+
+
+@app.exception_handler(CategoryNotFoundError)
+async def category_not_found_handler(request: Request, exc: CategoryNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={"error_code": "CATEGORY_NOT_FOUND", "message": "附件类型不存在"},
+    )
+
+
+@app.exception_handler(CategoryHasBoxesError)
+async def category_has_boxes_handler(request: Request, exc: CategoryHasBoxesError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "CATEGORY_HAS_BOXES",
+            "message": f"类型 {exc.category_name} 已有装箱记录，不可删除，可改应到数量",
+        },
+    )
+
+
+@app.exception_handler(DuplicateAttachmentLocationError)
+async def duplicate_attachment_location_handler(
+    request: Request, exc: DuplicateAttachmentLocationError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_LOCATION",
+            "message": "该总排号该类型已存在货位记录",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(AlreadyOffShelfAttachmentError)
+async def already_off_shelf_attachment_handler(
+    request: Request, exc: AlreadyOffShelfAttachmentError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_OFF_SHELF",
+            "message": "该总排号该类型已下架至转运区域，不可重新上架",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(DuplicateAttachmentBoxItemError)
+async def duplicate_attachment_box_handler(
+    request: Request, exc: DuplicateAttachmentBoxItemError
+):
+    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,
+        },
+    )
+
+
+@app.exception_handler(AttachmentItemNotFoundError)
+async def attachment_item_not_found_handler(
+    request: Request, exc: AttachmentItemNotFoundError
+):
+    return JSONResponse(
+        status_code=404,
+        content={"error_code": "BOX_ITEM_NOT_FOUND", "message": "指定装箱明细不存在"},
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/models/__init__.py

@@ -1,11 +1,23 @@
-from app.models.finished_goods import (
+from app.models.finished_goods import (  # noqa: F401
     FinishedGoodsBox,
     FinishedGoodsBoxItem,
     FinishedGoodsLocation,
 )
+from app.models.attachment import (  # noqa: F401
+    AttachmentCategory,
+    FinishedGoodsAttachment,
+    FinishedGoodsAttachmentItem,
+    FinishedGoodsAttachmentLocation,
+    FinishedGoodsBoxAttachmentItem,
+)
 
 __all__ = [
     "FinishedGoodsLocation",
     "FinishedGoodsBox",
     "FinishedGoodsBoxItem",
+    "AttachmentCategory",
+    "FinishedGoodsAttachment",
+    "FinishedGoodsAttachmentItem",
+    "FinishedGoodsAttachmentLocation",
+    "FinishedGoodsBoxAttachmentItem",
 ]

app/models/attachment.py

@@ -0,0 +1,142 @@
+from datetime import datetime
+
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    DateTime,
+    Index,
+    Integer,
+    Numeric,
+    String,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+
+SCHEMA = "CargoTrace"
+
+
+class AttachmentCategory(Base):
+    """附件类型字典。本期由业务方直接维护表数据，无管理界面。"""
+
+    __tablename__ = "attachment_category"
+    __table_args__ = ({"schema": SCHEMA},)
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class FinishedGoodsAttachment(Base):
+    """订单级附件判定，每个总排号一行。"""
+
+    __tablename__ = "finished_goods_attachment"
+    __table_args__ = (
+        Index("uk_fga_zongpai_no", "zongpai_no", unique=True),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    # undetermined / none / has
+    determination: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="undetermined"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class FinishedGoodsAttachmentItem(Base):
+    """类型级计划明细：每个 总排号 × 类型 一行。已到数量由箱明细派生，不落库。"""
+
+    __tablename__ = "finished_goods_attachment_item"
+    __table_args__ = (
+        Index(
+            "uk_fgai_zongpai_category", "zongpai_no", "category_id", unique=True
+        ),
+        Index("idx_fgai_zongpai_no", "zongpai_no"),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    category_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    expected_qty: Mapped[int] = mapped_column(Integer, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class FinishedGoodsAttachmentLocation(Base):
+    """附件上架，镜像 finished_goods_location。每种类型一个货位。"""
+
+    __tablename__ = "finished_goods_attachment_location"
+    __table_args__ = (
+        Index(
+            "uk_fgal_zongpai_category", "zongpai_no", "category_id", unique=True
+        ),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    category_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    location_code: Mapped[str] = mapped_column(String(64), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class FinishedGoodsBoxAttachmentItem(Base):
+    """附件装箱明细，共用 finished_goods_box（排产号级箱）。"""
+
+    __tablename__ = "finished_goods_box_attachment_item"
+    __table_args__ = (
+        Index("idx_fgbai_box_id", "box_id"),
+        Index("idx_fgbai_zongpai_category", "zongpai_no", "category_id"),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    box_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    category_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    quantity: Mapped[float] = mapped_column(Numeric(18, 3), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )

app/models/finished_goods.py

@@ -20,7 +20,10 @@ class FinishedGoodsLocation(Base):
     zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
     location_code: Mapped[str] = mapped_column(String(64), nullable=False)
     created_at: Mapped[datetime] = mapped_column(
-        DateTime, nullable=False, server_default=func.getdate()
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
     )
 
 
@@ -36,7 +39,10 @@ class FinishedGoodsBox(Base):
     paichan_no: Mapped[str] = mapped_column(String(64), nullable=False)
     box_no: Mapped[int] = mapped_column(nullable=False)
     created_at: Mapped[datetime] = mapped_column(
-        DateTime, nullable=False, server_default=func.getdate()
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
     )
 
 
@@ -53,5 +59,8 @@ class FinishedGoodsBoxItem(Base):
     zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
     quantity: Mapped[float | None] = mapped_column(Numeric(18, 3), nullable=True)
     created_at: Mapped[datetime] = mapped_column(
-        DateTime, nullable=False, server_default=func.getdate()
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
     )

app/schemas/attachment.py

@@ -0,0 +1,169 @@
+from pydantic import BaseModel, Field, field_validator
+
+
+class CategoryResponse(BaseModel):
+    id: int
+    name: str
+    sort_order: int
+
+
+class AttachmentStatusItem(BaseModel):
+    category_id: int
+    name: str
+    expected_qty: int
+    boxed_qty: int
+    location_code: str | None = None
+    complete: bool
+
+
+class AttachmentStatusResponse(BaseModel):
+    zongpai_no: str
+    determination: str
+    all_complete: bool
+    items: list[AttachmentStatusItem] = Field(default_factory=list)
+
+
+class AttachmentConfigItem(BaseModel):
+    category_id: int
+    expected_qty: int
+
+    @field_validator("expected_qty")
+    @classmethod
+    def _positive(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+
+class AttachmentConfigRequest(BaseModel):
+    zongpai_no: str
+    determination: str
+    items: list[AttachmentConfigItem] = Field(default_factory=list)
+
+    @field_validator("zongpai_no")
+    @classmethod
+    def _zongpai(cls, v: str) -> str:
+        import re
+
+        v = v.strip().upper()
+        if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v):
+            raise ValueError("INVALID_ZONGPAI")
+        return v
+
+    @field_validator("determination")
+    @classmethod
+    def _det(cls, v: str) -> str:
+        if v not in ("none", "has"):
+            raise ValueError("INVALID_DETERMINATION")
+        return v
+
+
+class AttachmentLocationRequest(BaseModel):
+    zongpai_no: str
+    category_id: int
+    location_code: str
+
+    @field_validator("zongpai_no")
+    @classmethod
+    def _zongpai(cls, v: str) -> str:
+        import re
+
+        v = v.strip().upper()
+        if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v):
+            raise ValueError("INVALID_ZONGPAI")
+        return v
+
+    @field_validator("location_code")
+    @classmethod
+    def _location(cls, v: str) -> str:
+        import re
+
+        v = v.strip().upper()
+        normal = re.match(r"^[A-Z]+\d+-\d+-\d+$", v)
+        transit = re.match(r"^TRANS-\d+", v)
+        if not normal and not transit:
+            raise ValueError("INVALID_LOCATION")
+        return v
+
+
+class AttachmentLocationResponse(BaseModel):
+    zongpai_no: str
+    category_id: int
+    location_code: str
+    created_at: str
+    previous_location: str | None = None
+
+
+class AttachmentBoxSaveRequest(BaseModel):
+    zongpai_no: str
+    category_id: int
+    box_no: int
+    quantity: int
+
+    @field_validator("zongpai_no")
+    @classmethod
+    def _zongpai(cls, v: str) -> str:
+        import re
+
+        v = v.strip().upper()
+        if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v):
+            raise ValueError("INVALID_ZONGPAI")
+        return v
+
+    @field_validator("box_no")
+    @classmethod
+    def _box_no(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_BOX_NO")
+        return v
+
+    @field_validator("quantity")
+    @classmethod
+    def _qty(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+
+class AttachmentBoxSaveResponse(BaseModel):
+    box_item_id: int
+    paichan_no: str
+    box_no: int
+    zongpai_no: str
+    category_id: int
+    quantity: int
+    created_at: str
+
+
+class AttachmentBoxUpdateRequest(BaseModel):
+    box_no: int
+    quantity: int
+
+    @field_validator("box_no")
+    @classmethod
+    def _box_no(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_BOX_NO")
+        return v
+
+    @field_validator("quantity")
+    @classmethod
+    def _qty(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+
+class AttachmentBoxUpdateResponse(BaseModel):
+    box_item_id: int
+    paichan_no: str
+    box_no: int
+    zongpai_no: str
+    category_id: int
+    quantity: int
+    updated_at: str
+
+
+class AttachmentBoxDeleteResponse(BaseModel):
+    box_item_id: int
+    deleted: bool

app/schemas/box.py

@@ -1,13 +1,26 @@
 import re
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
 
 class BoxItemDetail(BaseModel):
-    """箱号下某个总排号的明细"""
+    """箱号下某个总排号的明细（产品或附件）"""
+    box_item_id: int | None = None
     zongpai_no: str
+    work_order_no: str | None = None
+    quantity: int
+    total_quantity: int | None = None
+    kind: str = "product"
+    category_id: int | None = None
+    category_name: str | None = None
+
+
+class CurrentZongpaiBox(BaseModel):
+    """当前总排号已经分配的箱号明细"""
+    box_item_id: int
+    box_no: int
     quantity: int
 
 
@@ -17,14 +30,23 @@ class BoxDetail(BaseModel):
     items: list[BoxItemDetail]
 
 
+class AttachmentSummary(BaseModel):
+    determination: str
+    all_complete: bool
+    pending_count: int
+
+
 class BoxInfoResponse(BaseModel):
     """GET /box/info 响应"""
     zongpai_no: str
     paichan_no: str
+    work_order_no: str | None = None
     quantity: int
+    current_zongpai_boxes: list[CurrentZongpaiBox] = Field(default_factory=list)
     existing_boxes: list[BoxDetail]
     max_box_no: int
     suggested_box_no: int
+    attachment_summary: AttachmentSummary
 
 
 class BoxSaveRequest(BaseModel):
@@ -58,8 +80,45 @@ class BoxSaveRequest(BaseModel):
 
 class BoxSaveResponse(BaseModel):
     """POST /box 响应"""
+    box_item_id: int
     paichan_no: str
     box_no: int
     zongpai_no: str
     quantity: int
     created_at: str
+
+
+class BoxUpdateRequest(BaseModel):
+    """PATCH /box/{box_item_id} 请求"""
+    box_no: int
+    quantity: int
+
+    @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 BoxUpdateResponse(BaseModel):
+    """PATCH /box/{box_item_id} 响应"""
+    box_item_id: int
+    paichan_no: str
+    box_no: int
+    zongpai_no: str
+    quantity: int
+    updated_at: str
+
+
+class BoxDeleteResponse(BaseModel):
+    """DELETE /box/{box_item_id} 响应"""
+    box_item_id: int
+    deleted: bool

app/schemas/location.py

@@ -32,6 +32,21 @@ class LocationResponse(BaseModel):
     zongpai_no: str
     location_code: str
     created_at: str
+    previous_location: str | None = None
+
+
+class PaichaOverviewItem(BaseModel):
+    zongpai_no: str
+    work_order_no: str | None = None
+    quantity: int
+    location_code: str | None = None
+    status: str
+
+
+class PaichaOverviewResponse(BaseModel):
+    paicha_no: str
+    total_count: int
+    items: list[PaichaOverviewItem]
 
 
 class DuplicateLocationDetail(BaseModel):

app/services/attachment_service.py

@@ -0,0 +1,489 @@
+import re
+from datetime import datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models.attachment import (
+    AttachmentCategory,
+    FinishedGoodsAttachment,
+    FinishedGoodsAttachmentItem,
+    FinishedGoodsAttachmentLocation,
+    FinishedGoodsBoxAttachmentItem,
+)
+from app.models.finished_goods import FinishedGoodsBox
+from app.schemas.attachment import AttachmentConfigRequest
+from app.services.box_service import (
+    InvalidQuantityError,
+    InvalidZongpaiError,
+    query_erp_info,
+)
+
+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,
+    }
+
+
+def _ensure_categories_exist(db: Session, category_ids: list[int]) -> dict[int, str]:
+    names = _category_name_map(db, category_ids)
+    missing = [c for c in category_ids if c not in names]
+    if missing:
+        raise CategoryNotFoundError(missing[0])
+    return names
+
+
+def configure_attachment(db: Session, req: AttachmentConfigRequest) -> dict:
+    zongpai_no = _validate_zongpai(req.zongpai_no)
+
+    # Validate categories referenced in the submitted list
+    submitted_ids = [it.category_id for it in req.items]
+    _ensure_categories_exist(db, submitted_ids)
+
+    # Upsert order-level determination row
+    att = (
+        db.query(FinishedGoodsAttachment)
+        .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
+        .first()
+    )
+    if att is None:
+        att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination=req.determination)
+        db.add(att)
+    else:
+        att.determination = req.determination
+        att.updated_at = datetime.now()
+    db.flush()
+
+    # Load existing items
+    existing = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no)
+        .all()
+    )
+    existing_by_cat = {it.category_id: it for it in existing}
+    submitted_set = set(submitted_ids)
+    boxed = _boxed_qty_map(db, zongpai_no)
+    names = _category_name_map(db, list(existing_by_cat.keys()) + submitted_ids)
+
+    # Remove items not in the submitted list; block those with boxed > 0
+    for cat_id, it in existing_by_cat.items():
+        if cat_id not in submitted_set:
+            if boxed.get(cat_id, 0) > 0:
+                raise CategoryHasBoxesError(names.get(cat_id, str(cat_id)))
+            db.delete(it)
+
+    # Upsert submitted items
+    for submitted in req.items:
+        it = existing_by_cat.get(submitted.category_id)
+        if it is None:
+            db.add(
+                FinishedGoodsAttachmentItem(
+                    zongpai_no=zongpai_no,
+                    category_id=submitted.category_id,
+                    expected_qty=submitted.expected_qty,
+                )
+            )
+        else:
+            it.expected_qty = submitted.expected_qty
+            it.updated_at = datetime.now()
+
+    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
+
+
+def _boxed_for_category(
+    db: Session, zongpai_no: str, category_id: int, exclude_item_id: int | None = None
+) -> int:
+    q = db.query(FinishedGoodsBoxAttachmentItem).filter(
+        FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
+        FinishedGoodsBoxAttachmentItem.category_id == category_id,
+    )
+    if exclude_item_id is not None:
+        q = q.filter(FinishedGoodsBoxAttachmentItem.id != exclude_item_id)
+    return sum(int(r.quantity or 0) for r in q.all())
+
+
+def _ensure_attachment_item(db: Session, zongpai_no: str, category_id: int, erp_qty: int):
+    """实物证据：不存在则建 item（expected 兜底 = ERP 数量），并保证 determination=has。"""
+    if not _category_name_map(db, [category_id]):
+        raise CategoryNotFoundError(category_id)
+
+    att = (
+        db.query(FinishedGoodsAttachment)
+        .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
+        .first()
+    )
+    if att is None:
+        att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination="has")
+        db.add(att)
+    elif att.determination in ("undetermined", "none"):
+        att.determination = "has"
+        att.updated_at = datetime.now()
+    db.flush()
+
+    item = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(
+            FinishedGoodsAttachmentItem.zongpai_no == zongpai_no,
+            FinishedGoodsAttachmentItem.category_id == category_id,
+        )
+        .first()
+    )
+    if item is None:
+        item = FinishedGoodsAttachmentItem(
+            zongpai_no=zongpai_no,
+            category_id=category_id,
+            expected_qty=erp_qty,
+        )
+        db.add(item)
+        db.flush()
+    return item
+
+
+def save_attachment_box(db: Session, req) -> dict:
+    zongpai_no = _validate_zongpai(req.zongpai_no)
+
+    erp = query_erp_info(db, zongpai_no)  # raises ZongpaiNotFoundError if missing
+    paichan_no = erp["paichan_no"]
+    erp_qty = erp["quantity"]
+
+    item = _ensure_attachment_item(db, zongpai_no, req.category_id, erp_qty)
+
+    box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+    if box is not None:
+        existing_in_box = (
+            db.query(FinishedGoodsBoxAttachmentItem)
+            .filter(
+                FinishedGoodsBoxAttachmentItem.box_id == box.id,
+                FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
+                FinishedGoodsBoxAttachmentItem.category_id == req.category_id,
+            )
+            .first()
+        )
+        if existing_in_box is not None:
+            raise DuplicateAttachmentBoxItemError(paichan_no, req.box_no)
+    else:
+        box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(box)
+        db.flush()
+
+    # Cap check against expected_qty (mirrors product _ensure_quantity_within_erp_total)
+    already_boxed = _boxed_for_category(db, zongpai_no, req.category_id)
+    if already_boxed + req.quantity > item.expected_qty:
+        raise InvalidQuantityError()
+
+    record = FinishedGoodsBoxAttachmentItem(
+        box_id=box.id,
+        zongpai_no=zongpai_no,
+        category_id=req.category_id,
+        quantity=req.quantity,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+
+    return {
+        "box_item_id": record.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": zongpai_no,
+        "category_id": req.category_id,
+        "quantity": req.quantity,
+        "created_at": record.created_at.isoformat(),
+    }
+
+
+def update_attachment_box(db: Session, item_id: int, req) -> dict:
+    record = (
+        db.query(FinishedGoodsBoxAttachmentItem)
+        .filter(FinishedGoodsBoxAttachmentItem.id == item_id)
+        .first()
+    )
+    if record is None:
+        raise AttachmentItemNotFoundError()
+
+    current_box = (
+        db.query(FinishedGoodsBox)
+        .filter(FinishedGoodsBox.id == record.box_id)
+        .first()
+    )
+    if current_box is None:
+        raise AttachmentItemNotFoundError()
+    paichan_no = current_box.paichan_no
+
+    target_box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+    if target_box is None:
+        target_box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(target_box)
+        db.flush()
+    else:
+        dup = (
+            db.query(FinishedGoodsBoxAttachmentItem)
+            .filter(
+                FinishedGoodsBoxAttachmentItem.box_id == target_box.id,
+                FinishedGoodsBoxAttachmentItem.zongpai_no == record.zongpai_no,
+                FinishedGoodsBoxAttachmentItem.category_id == record.category_id,
+                FinishedGoodsBoxAttachmentItem.id != record.id,
+            )
+            .first()
+        )
+        if dup is not None:
+            raise DuplicateAttachmentBoxItemError(paichan_no, req.box_no)
+
+    item = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(
+            FinishedGoodsAttachmentItem.zongpai_no == record.zongpai_no,
+            FinishedGoodsAttachmentItem.category_id == record.category_id,
+        )
+        .first()
+    )
+    expected_qty = item.expected_qty if item is not None else 0
+    already_boxed = _boxed_for_category(
+        db, record.zongpai_no, record.category_id, exclude_item_id=record.id
+    )
+    if already_boxed + req.quantity > expected_qty:
+        raise InvalidQuantityError()
+
+    old_box_id = record.box_id
+    record.box_id = target_box.id
+    record.quantity = req.quantity
+
+    if old_box_id != target_box.id:
+        _cleanup_empty_box(db, old_box_id, exclude_item_id=record.id)
+
+    db.commit()
+    db.refresh(record)
+    return {
+        "box_item_id": record.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": record.zongpai_no,
+        "category_id": record.category_id,
+        "quantity": int(record.quantity or 0),
+        "updated_at": record.created_at.isoformat(),
+    }
+
+
+def delete_attachment_box(db: Session, item_id: int) -> dict:
+    record = (
+        db.query(FinishedGoodsBoxAttachmentItem)
+        .filter(FinishedGoodsBoxAttachmentItem.id == item_id)
+        .first()
+    )
+    if record is None:
+        raise AttachmentItemNotFoundError()
+    box_id = record.box_id
+    db.delete(record)
+    db.flush()
+    _cleanup_empty_box(db, box_id)
+    db.commit()
+    return {"box_item_id": item_id, "deleted": True}
+
+
+def _cleanup_empty_box(db: Session, box_id: int, exclude_item_id: int | None = None) -> None:
+    """If a box has no product items and no attachment items, delete it."""
+    from app.models.finished_goods import FinishedGoodsBoxItem
+
+    product_q = db.query(FinishedGoodsBoxItem.id).filter(
+        FinishedGoodsBoxItem.box_id == box_id
+    )
+    if product_q.first() is not None:
+        return
+    att_q = db.query(FinishedGoodsBoxAttachmentItem.id).filter(
+        FinishedGoodsBoxAttachmentItem.box_id == box_id
+    )
+    if exclude_item_id is not None:
+        att_q = att_q.filter(FinishedGoodsBoxAttachmentItem.id != exclude_item_id)
+    if att_q.first() is not None:
+        return
+    box = db.query(FinishedGoodsBox).filter(FinishedGoodsBox.id == box_id).first()
+    if box is not None:
+        db.delete(box)

app/services/box_service.py

@@ -1,10 +1,16 @@
 import re
 
-from sqlalchemy import text
+from sqlalchemy import bindparam, text
 from sqlalchemy.orm import Session
 
+from app.models.attachment import (
+    AttachmentCategory,
+    FinishedGoodsAttachment,
+    FinishedGoodsAttachmentItem,
+    FinishedGoodsBoxAttachmentItem,
+)
 from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
-from app.schemas.box import BoxSaveRequest
+from app.schemas.box import BoxSaveRequest, BoxUpdateRequest
 
 ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 
@@ -23,13 +29,21 @@ class DuplicateBoxItemError(Exception):
         self.box_no = box_no
 
 
+class BoxItemNotFoundError(Exception):
+    pass
+
+
+class InvalidBoxItemError(Exception):
+    pass
+
+
+class InvalidQuantityError(Exception):
+    pass
+
+
 def query_erp_info(db: Session, zongpai_no: str) -> dict:
-    """从 ERP 视图查询总排号对应的排产号和数量。"""
-    sql = text(
-        "SELECT TOP 1 [总排号], [排产号], [数量] "
-        "FROM [ERPAuto].[vw_productionContractData] "
-        "WHERE [总排号] = :zongpai_no"
-    )
+    """从 ERP 视图查询总排号对应的排产号、工令号和数量。"""
+    sql = text(_erp_info_sql(db))
     row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone()
     if not row:
         raise ZongpaiNotFoundError()
@@ -37,6 +51,54 @@ 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],
+    }
+
+
+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_item_info_map(db: Session, zongpai_nos: list[str]) -> dict[str, dict]:
+    """批量查询总排号对应的工令号和 ERP 总数量。"""
+    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]: {
+            "work_order_no": row[1],
+            "total_quantity": int(row[2]) if row[2] is not None else None,
+        }
+        for row in rows
     }
 
 
@@ -56,36 +118,175 @@ 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)
+
+    erp_item_info_map = query_erp_item_info_map(db, item_zongpai_nos)
+
+    existing_boxes = []
+    current_zongpai_boxes = []
+    max_box_no = 0
+    paichan_box_rows = (
+        db.query(FinishedGoodsBox)
+        .filter(FinishedGoodsBox.paichan_no == paichan_no)
+        .all()
+    )
+    att_rows = (
+        db.query(FinishedGoodsBoxAttachmentItem)
+        .filter(
+            FinishedGoodsBoxAttachmentItem.box_id.in_(
+                [b.id for b in paichan_box_rows]
+            )
+        )
+        .all()
+    )
+    att_category_ids = {r.category_id for r in att_rows}
+    att_names = {
+        c.id: c.name
+        for c in db.query(AttachmentCategory)
+        .filter(AttachmentCategory.id.in_(list(att_category_ids)))
+        .all()
+    } if att_category_ids else {}
+    items_by_box: dict[int, list[dict]] = {}
+    for r in att_rows:
+        items_by_box.setdefault(r.box_id, []).append(
+            {
+                "box_item_id": r.id,
+                "zongpai_no": r.zongpai_no,
+                "work_order_no": None,
+                "quantity": int(r.quantity or 0),
+                "total_quantity": None,
+                "kind": "attachment",
+                "category_id": r.category_id,
+                "category_name": att_names.get(r.category_id),
+            }
+        )
+    for box in boxes:
+        items = box_items[box.id]
+        for item in items:
+            if item.zongpai_no == zongpai_no:
+                current_zongpai_boxes.append({
+                    "box_item_id": item.id,
+                    "box_no": box.box_no,
+                    "quantity": int(item.quantity or 0),
+                })
         existing_boxes.append({
             "box_no": box.box_no,
             "items": [
-                {"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)}
+                {
+                    "box_item_id": item.id,
+                    "zongpai_no": item.zongpai_no,
+                    "work_order_no": erp_item_info_map.get(
+                        item.zongpai_no, {}
+                    ).get("work_order_no"),
+                    "quantity": int(item.quantity or 0),
+                    "total_quantity": erp_item_info_map.get(
+                        item.zongpai_no, {}
+                    ).get("total_quantity"),
+                    "kind": "product",
+                }
                 for item in items
-            ],
+            ] + items_by_box.get(box.id, []),
         })
         if box.box_no > max_box_no:
             max_box_no = box.box_no
 
+    # Attachment summary for the scanned总排号
+    att_row = (
+        db.query(FinishedGoodsAttachment)
+        .filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
+        .first()
+    )
+    determination = att_row.determination if att_row else "undetermined"
+    plan_items = (
+        db.query(FinishedGoodsAttachmentItem)
+        .filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no)
+        .all()
+    )
+    pending_count = 0
+    for it in plan_items:
+        boxed = sum(
+            int(r.quantity or 0)
+            for r in db.query(FinishedGoodsBoxAttachmentItem)
+            .filter(
+                FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
+                FinishedGoodsBoxAttachmentItem.category_id == it.category_id,
+            )
+            .all()
+        )
+        if boxed < it.expected_qty:
+            pending_count += 1
+    all_complete = (
+        determination == "has"
+        and len(plan_items) > 0
+        and pending_count == 0
+    )
+    attachment_summary = {
+        "determination": determination,
+        "all_complete": all_complete,
+        "pending_count": pending_count,
+    }
+
     return {
         "zongpai_no": zongpai_no,
         "paichan_no": paichan_no,
+        "work_order_no": erp["work_order_no"],
         "quantity": quantity,
+        "current_zongpai_boxes": current_zongpai_boxes,
         "existing_boxes": existing_boxes,
         "max_box_no": max_box_no,
         "suggested_box_no": max_box_no + 1,
+        "attachment_summary": attachment_summary,
     }
 
 
+def _packed_quantity_for_zongpai(
+    db: Session,
+    paichan_no: str,
+    zongpai_no: str,
+    exclude_box_item_id: int | None = None,
+) -> int:
+    query = (
+        db.query(FinishedGoodsBoxItem)
+        .join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBoxItem.zongpai_no == zongpai_no,
+        )
+    )
+    if exclude_box_item_id is not None:
+        query = query.filter(FinishedGoodsBoxItem.id != exclude_box_item_id)
+    return sum(int(item.quantity or 0) for item in query.all())
+
+
+def _ensure_quantity_within_erp_total(
+    db: Session,
+    paichan_no: str,
+    zongpai_no: str,
+    erp_quantity: int,
+    submitted_quantity: int,
+    exclude_box_item_id: int | None = None,
+) -> None:
+    packed_quantity = _packed_quantity_for_zongpai(
+        db,
+        paichan_no,
+        zongpai_no,
+        exclude_box_item_id=exclude_box_item_id,
+    )
+    if packed_quantity + submitted_quantity > erp_quantity:
+        raise InvalidQuantityError()
+
+
 def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
-    """保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。"""
+    """保存装箱记录。同箱可凑箱，但同箱同总排不可重复。"""
     erp = query_erp_info(db, req.zongpai_no)
     paichan_no = erp["paichan_no"]
 
@@ -109,7 +310,16 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
         )
         if existing_item:
             raise DuplicateBoxItemError(paichan_no, req.box_no)
-    else:
+
+    _ensure_quantity_within_erp_total(
+        db,
+        paichan_no,
+        req.zongpai_no,
+        erp["quantity"],
+        req.quantity,
+    )
+
+    if box is None:
         box = FinishedGoodsBox(
             paichan_no=paichan_no,
             box_no=req.box_no,
@@ -127,9 +337,128 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
     db.refresh(item)
 
     return {
+        "box_item_id": item.id,
         "paichan_no": paichan_no,
         "box_no": req.box_no,
         "zongpai_no": req.zongpai_no,
         "quantity": req.quantity,
         "created_at": item.created_at.isoformat(),
     }
+
+
+def update_box_item(db: Session, box_item_id: int, req: BoxUpdateRequest) -> dict:
+    """更新已分配明细的箱号和数量。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    current_box = (
+        db.query(FinishedGoodsBox)
+        .filter(FinishedGoodsBox.id == item.box_id)
+        .first()
+    )
+    if current_box is None:
+        raise InvalidBoxItemError()
+
+    paichan_no = current_box.paichan_no
+    target_box = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == paichan_no,
+            FinishedGoodsBox.box_no == req.box_no,
+        )
+        .first()
+    )
+
+    if target_box is None:
+        target_box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
+        db.add(target_box)
+        db.flush()
+    else:
+        duplicate_item = (
+            db.query(FinishedGoodsBoxItem)
+            .filter(
+                FinishedGoodsBoxItem.box_id == target_box.id,
+                FinishedGoodsBoxItem.zongpai_no == item.zongpai_no,
+                FinishedGoodsBoxItem.id != item.id,
+            )
+            .first()
+        )
+        if duplicate_item is not None:
+            raise DuplicateBoxItemError(paichan_no, req.box_no)
+
+    erp = query_erp_info(db, item.zongpai_no)
+    _ensure_quantity_within_erp_total(
+        db,
+        paichan_no,
+        item.zongpai_no,
+        erp["quantity"],
+        req.quantity,
+        exclude_box_item_id=box_item_id,
+    )
+
+    old_box_id = item.box_id
+    item.box_id = target_box.id
+    item.quantity = req.quantity
+
+    if old_box_id != target_box.id:
+        old_box_has_items = (
+            db.query(FinishedGoodsBoxItem.id)
+            .filter(
+                FinishedGoodsBoxItem.box_id == old_box_id,
+                FinishedGoodsBoxItem.id != item.id,
+            )
+            .first()
+        )
+        if old_box_has_items is None:
+            old_box = (
+                db.query(FinishedGoodsBox)
+                .filter(FinishedGoodsBox.id == old_box_id)
+                .first()
+            )
+            if old_box is not None:
+                db.delete(old_box)
+
+    db.commit()
+    db.refresh(item)
+
+    return {
+        "box_item_id": item.id,
+        "paichan_no": paichan_no,
+        "box_no": req.box_no,
+        "zongpai_no": item.zongpai_no,
+        "quantity": int(item.quantity or 0),
+        "updated_at": item.created_at.isoformat(),
+    }
+
+
+def delete_box_item(db: Session, box_item_id: int) -> dict:
+    """删除误录的装箱明细；若箱号下无其他明细，同步删除空箱号。"""
+    item = (
+        db.query(FinishedGoodsBoxItem)
+        .filter(FinishedGoodsBoxItem.id == box_item_id)
+        .first()
+    )
+    if item is None:
+        raise BoxItemNotFoundError()
+
+    box_id = item.box_id
+    db.delete(item)
+    db.flush()
+
+    remaining = (
+        db.query(FinishedGoodsBoxItem.id)
+        .filter(FinishedGoodsBoxItem.box_id == box_id)
+        .first()
+    )
+    if remaining is None:
+        box = db.query(FinishedGoodsBox).filter(FinishedGoodsBox.id == box_id).first()
+        if box is not None:
+            db.delete(box)
+
+    db.commit()
+    return {"box_item_id": box_item_id, "deleted": True}

app/services/location_service.py

@@ -1,7 +1,14 @@
+import re
+from datetime import datetime
+
+from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsLocation
 from app.schemas.location import LocationRequest
+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 DuplicateLocationError(Exception):
@@ -12,6 +19,135 @@ class DuplicateLocationError(Exception):
         self.registered_at = registered_at
 
 
+class AlreadyOffShelfError(Exception):
+    """总排号已下架至转运区域。"""
+
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+class PaichaNotFoundError(Exception):
+    pass
+
+
+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 _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 _erp_paicha_items_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 "排产号" = :paicha_no'
+        )
+    return (
+        "SELECT [总排号], [工令号], [数量] "
+        "FROM [ERPAuto].[vw_productionContractData] "
+        "WHERE [排产号] = :paicha_no"
+    )
+
+
+def _query_paicha_no(db: Session, zongpai_no: str) -> str:
+    row = db.execute(text(_erp_info_sql(db)), {"zongpai_no": zongpai_no}).fetchone()
+    if not row:
+        raise PaichaNotFoundError()
+    return row[1]
+
+
+def _query_paicha_items(db: Session, paicha_no: str) -> list[dict]:
+    rows = db.execute(
+        text(_erp_paicha_items_sql(db)), {"paicha_no": paicha_no}
+    ).fetchall()
+    return [
+        {
+            "zongpai_no": row[0],
+            "work_order_no": row[1],
+            "quantity": int(row[2]) if row[2] is not None else 0,
+        }
+        for row in rows
+    ]
+
+
+def _location_status(location_code: str | None) -> str:
+    if location_code is None:
+        return "not_shelved"
+    if _is_transit_location(location_code):
+        return "transferred"
+    if _is_temp_storage_location(location_code):
+        return "temp_stored"
+    return "on_shelf"
+
+
+def get_paicha_overview(db: Session, zongpai_no: str) -> dict:
+    zongpai_no = zongpai_no.strip().upper()
+    if not ZONGPAI_PATTERN.match(zongpai_no):
+        raise InvalidZongpaiError()
+
+    paicha_no = _query_paicha_no(db, zongpai_no)
+    erp_items = _query_paicha_items(db, paicha_no)
+    if not erp_items:
+        raise PaichaNotFoundError()
+
+    zongpai_nos = [item["zongpai_no"] for item in erp_items]
+    locations = (
+        db.query(FinishedGoodsLocation)
+        .filter(FinishedGoodsLocation.zongpai_no.in_(zongpai_nos))
+        .all()
+    )
+    location_map = {item.zongpai_no: item.location_code for item in locations}
+
+    status_order = {"on_shelf": 0, "temp_stored": 1, "transferred": 2, "not_shelved": 3}
+    overview_items = []
+    for item in erp_items:
+        location_code = location_map.get(item["zongpai_no"])
+        status = _location_status(location_code)
+        overview_items.append(
+            {
+                "zongpai_no": item["zongpai_no"],
+                "work_order_no": item["work_order_no"],
+                "quantity": item["quantity"],
+                "location_code": location_code,
+                "status": status,
+            }
+        )
+
+    overview_items.sort(
+        key=lambda item: (
+            status_order.get(item["status"], 99),
+            item["zongpai_no"],
+        )
+    )
+
+    return {
+        "paicha_no": paicha_no,
+        "total_count": len(overview_items),
+        "items": overview_items,
+    }
+
+
 def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation:
     existing = (
         db.query(FinishedGoodsLocation)
@@ -19,6 +155,27 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio
         .first()
     )
     if existing:
+        # Terminal state: transit location → any operation raises error
+        if _is_transit_location(existing.location_code):
+            raise AlreadyOffShelfError(
+                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)
+
+        # Allowed: target is transit (off-shelf) or normal→temp (shelf change)
+        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 DuplicateLocationError(
             location_code=existing.location_code,
             registered_at=existing.created_at.isoformat(),

config/settings.py

@@ -1,7 +1,7 @@
 from pathlib import Path
-from typing import Optional
+from typing import Literal
 from sqlalchemy.engine import URL
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 from pydantic_yaml import parse_yaml_raw_as
 
 
@@ -16,9 +16,20 @@ class SqlServerConfig(BaseModel):
     trust_server_certificate: str = "yes"
 
 
+class PostgreSqlConfig(BaseModel):
+    """PostgreSQL 连接配置"""
+    host: str
+    port: int = 5432
+    database: str
+    username: str
+    password: str
+
+
 class DatabaseConfig(BaseModel):
     """数据库配置"""
+    active: Literal["sql_server", "postgresql"] = "sql_server"
     sql_server: SqlServerConfig
+    postgresql: PostgreSqlConfig | None = None
 
 
 class Settings(BaseModel):
@@ -28,6 +39,11 @@ class Settings(BaseModel):
     @property
     def database_url(self) -> URL:
         """构建数据库连接 URL"""
+        if self.database.active == "postgresql":
+            return self._postgresql_url()
+        return self._sql_server_url()
+
+    def _sql_server_url(self) -> URL:
         conf = self.database.sql_server
         return URL.create(
             "mssql+pyodbc",
@@ -42,6 +58,19 @@ class Settings(BaseModel):
             },
         )
 
+    def _postgresql_url(self) -> URL:
+        conf = self.database.postgresql
+        if conf is None:
+            raise ValueError("已选择 postgresql，但未配置 database.postgresql")
+        return URL.create(
+            "postgresql+psycopg",
+            username=conf.username,
+            password=conf.password,
+            host=conf.host,
+            port=conf.port,
+            database=conf.database,
+        )
+
 
 def load_settings(config_path: str = "config/settings.yaml") -> Settings:
     """加载 YAML 配置文件"""

config/settings.yaml

@@ -1,10 +0,0 @@
-# 数据库配置
-database:
-  sql_server:
-    host: 192.168.110.114
-    port: 1433
-    database: CompanyDB
-    username: peng
-    password: Cqbld123456.
-    driver: "{ODBC Driver 18 for SQL Server}"
-    trust_server_certificate: yes

config/settings.yaml.example

@@ -0,0 +1,17 @@
+# 数据库配置
+database:
+  active: sql_server  # 可选: sql_server, postgresql
+  sql_server:
+    host: YOUR_DB_HOST
+    port: 1433
+    database: YOUR_DB_NAME
+    username: YOUR_USERNAME
+    password: YOUR_PASSWORD
+    driver: "{ODBC Driver 18 for SQL Server}"
+    trust_server_certificate: yes
+  postgresql:
+    host: YOUR_POSTGRES_HOST
+    port: 5432
+    database: YOUR_POSTGRES_DB_NAME
+    username: YOUR_POSTGRES_USERNAME
+    password: YOUR_POSTGRES_PASSWORD

requirements.txt

@@ -2,6 +2,7 @@ fastapi>=0.115.0
 uvicorn[standard]>=0.34.0
 sqlalchemy>=2.0.0
 pyodbc>=5.2.0
+psycopg[binary]>=3.2.0
 pydantic-settings>=2.0.0
 python-dotenv>=1.0.0
 pytest>=8.0.0

scripts/create_attachment_tables.py

@@ -0,0 +1,39 @@
+"""One-off: create the 5 attachment tables in the configured DB.
+
+Run from services/fastapi/ with the venv active:
+    python scripts/create_attachment_tables.py
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.core.database import Base, engine  # noqa: E402
+from app.models.attachment import (  # noqa: E402
+    AttachmentCategory,
+    FinishedGoodsAttachment,
+    FinishedGoodsAttachmentItem,
+    FinishedGoodsAttachmentLocation,
+    FinishedGoodsBoxAttachmentItem,
+)
+
+NEW_TABLES = [
+    AttachmentCategory,
+    FinishedGoodsAttachment,
+    FinishedGoodsAttachmentItem,
+    FinishedGoodsAttachmentLocation,
+    FinishedGoodsBoxAttachmentItem,
+]
+
+
+def main() -> None:
+    Base.metadata.create_all(
+        engine,
+        tables=[t.__table__ for t in NEW_TABLES],
+    )
+    print(f"Created {len(NEW_TABLES)} attachment tables.")
+
+
+if __name__ == "__main__":
+    main()

tests/conftest.py

@@ -4,7 +4,11 @@ from sqlalchemy.orm import Session
 
 from app.core.database import SessionLocal
 from app.main import app
-from app.models.finished_goods import FinishedGoodsLocation
+from app.models.finished_goods import (
+    FinishedGoodsBox,
+    FinishedGoodsBoxItem,
+    FinishedGoodsLocation,
+)
 
 
 @pytest.fixture
@@ -22,8 +26,64 @@ def db():
 @pytest.fixture(autouse=True)
 def cleanup_test_data(db: Session):
     yield
-    test_zongpai_nos = ["26B999", "26C999", "26BW0999", "26T999"]
+    test_zongpai_nos = [
+        "26B999",
+        "26C999",
+        "26BW0999",
+        "26T999",
+        "26T998",
+        "26B998",
+        "26C998",
+        "26BW0011",
+        "26BW0012",
+    ]
+
+    # Attachment tables (order: children before parents / before shared box)
+    from app.models.attachment import (
+        AttachmentCategory,
+        FinishedGoodsAttachment,
+        FinishedGoodsAttachmentItem,
+        FinishedGoodsAttachmentLocation,
+        FinishedGoodsBoxAttachmentItem,
+    )
+
+    db.query(FinishedGoodsBoxAttachmentItem).filter(
+        FinishedGoodsBoxAttachmentItem.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachmentLocation).filter(
+        FinishedGoodsAttachmentLocation.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachmentItem).filter(
+        FinishedGoodsAttachmentItem.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachment).filter(
+        FinishedGoodsAttachment.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(AttachmentCategory).filter(
+        AttachmentCategory.name.like("TEST_%")
+    ).delete(synchronize_session=False)
+
+    # Existing product cleanup (unchanged) — attachment box items for these boxes:
     db.query(FinishedGoodsLocation).filter(
         FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos)
     ).delete(synchronize_session=False)
+    test_boxes = (
+        db.query(FinishedGoodsBox)
+        .filter(
+            FinishedGoodsBox.paichan_no == "W00009",
+            FinishedGoodsBox.box_no >= 900,
+        )
+        .all()
+    )
+    test_box_ids = [box.id for box in test_boxes]
+    if test_box_ids:
+        db.query(FinishedGoodsBoxAttachmentItem).filter(
+            FinishedGoodsBoxAttachmentItem.box_id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
+        db.query(FinishedGoodsBoxItem).filter(
+            FinishedGoodsBoxItem.box_id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
+        db.query(FinishedGoodsBox).filter(
+            FinishedGoodsBox.id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
     db.commit()

tests/fixtures/test_config.yaml

@@ -1,5 +1,6 @@
 # 测试配置
 database:
+  active: sql_server
   sql_server:
     host: localhost
     port: 1433
@@ -8,3 +9,9 @@ database:
     password: test_pass
     driver: "{ODBC Driver 18 for SQL Server}"
     trust_server_certificate: yes
+  postgresql:
+    host: localhost
+    port: 5432
+    database: TestDB
+    username: test_user
+    password: test_pass

tests/test_attachment_api.py

@@ -0,0 +1,296 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models.attachment import AttachmentCategory
+
+
+def _seed_categories(db: Session) -> list[int]:
+    cats = [
+        AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True),
+        AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True),
+        AttachmentCategory(name="TEST_OFF", sort_order=3, is_active=False),
+    ]
+    for c in cats:
+        db.add(c)
+    db.commit()
+    for c in cats:
+        db.refresh(c)
+    return [c.id for c in cats]
+
+
+def test_categories_lists_active_only(client: TestClient, db: Session):
+    _seed_categories(db)
+    resp = client.get("/CargoTrace/attachment/categories")
+    assert resp.status_code == 200
+    names = [c["name"] for c in resp.json()]
+    assert "TEST_CERT" in names
+    assert "TEST_EXTRA" in names
+    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
+
+
+def test_config_sets_none(client: TestClient):
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={"zongpai_no": "26BW0011", "determination": "none", "items": []},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["determination"] == "none"
+    assert resp.json()["items"] == []
+
+
+def test_config_creates_and_updates_items(client: TestClient, db: Session):
+    cert_id, extra_id, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 5},
+            ],
+        },
+    )
+    # Update: change extra expected, drop nothing
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 8},
+            ],
+        },
+    )
+    assert resp.status_code == 200
+    extra = next(i for i in resp.json()["items"] if i["category_id"] == extra_id)
+    assert extra["expected_qty"] == 8
+
+
+def test_config_rejects_removing_type_with_boxes(client: TestClient, db: Session):
+    cert_id, _, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 922,
+            "quantity": 10,
+        },
+    )
+    # Try to drop the type that now has boxes
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [],
+        },
+    )
+    assert resp.status_code == 409
+    assert resp.json()["error_code"] == "CATEGORY_HAS_BOXES"
+
+
+def test_config_allows_removing_type_with_zero_boxes(client: TestClient, db: Session):
+    cert_id, extra_id, _ = _seed_categories(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 5},
+            ],
+        },
+    )
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    assert resp.status_code == 200
+    assert {i["category_id"] for i in resp.json()["items"]} == {cert_id}
+
+
+def test_config_rejects_unknown_category(client: TestClient):
+    resp = client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": 9999999, "expected_qty": 1}],
+        },
+    )
+    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"

tests/test_attachment_boxing.py

@@ -0,0 +1,258 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models.attachment import AttachmentCategory
+
+
+def _seed(db: Session) -> list[int]:
+    cats = [
+        AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True),
+        AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True),
+    ]
+    for c in cats:
+        db.add(c)
+    db.commit()
+    for c in cats:
+        db.refresh(c)
+    return [c.id for c in cats]
+
+
+def test_attachment_box_basic_and_flip(client: TestClient, db: Session):
+    """扫附件码即到货：未判定 → 自动 has，item 不存在则按 ERP 数量建。"""
+    cert_id, _ = _seed(db)
+    resp = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 931,
+            "quantity": 30,
+        },
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["paichan_no"] == "W00009"
+    assert body["box_no"] == 931
+    assert body["quantity"] == 30
+
+    status = client.get(
+        "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
+    ).json()
+    assert status["determination"] == "has"
+    item = status["items"][0]
+    assert item["expected_qty"] == 80  # ERP 数量兜底
+    assert item["boxed_qty"] == 30
+
+
+def test_attachment_box_rejects_over_expected(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    first = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 932,
+            "quantity": 70,
+        },
+    )
+    assert first.status_code == 200
+    over = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 933,
+            "quantity": 11,
+        },
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_attachment_box_duplicate_in_same_box(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 934,
+            "quantity": 10,
+        },
+    )
+    dup = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 934,
+            "quantity": 5,
+        },
+    )
+    assert dup.status_code == 409
+    assert dup.json()["error_code"] == "DUPLICATE_BOX_NO"
+
+
+def test_attachment_box_mixed_with_product(client: TestClient, db: Session):
+    """附件与产品同箱：先装产品，再装附件，同 box_no。"""
+    cert_id, _ = _seed(db)
+    client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 935, "quantity": 10},
+    )
+    resp = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 935,
+            "quantity": 5,
+        },
+    )
+    assert resp.status_code == 200
+    info = client.get(
+        "/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
+    ).json()
+    box = next(b for b in info["existing_boxes"] if b["box_no"] == 935)
+    kinds = {i["kind"] for i in box["items"]}
+    assert kinds == {"product", "attachment"}
+
+
+def test_attachment_box_completion(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 936,
+            "quantity": 80,
+        },
+    )
+    status = client.get(
+        "/CargoTrace/attachment/status", params={"zongpai_no": "26BW0011"}
+    ).json()
+    assert status["all_complete"] is True
+
+
+def test_attachment_box_update_quantity(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    created = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 941,
+            "quantity": 30,
+        },
+    ).json()
+    item_id = created["box_item_id"]
+    updated = client.patch(
+        f"/CargoTrace/attachment/box/{item_id}",
+        json={"box_no": 941, "quantity": 25},
+    )
+    assert updated.status_code == 200
+    assert updated.json()["quantity"] == 25
+
+
+def test_attachment_box_update_rejects_over_expected(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [{"category_id": cert_id, "expected_qty": 80}],
+        },
+    )
+    created = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 942,
+            "quantity": 70,
+        },
+    ).json()
+    over = client.patch(
+        f"/CargoTrace/attachment/box/{created['box_item_id']}",
+        json={"box_no": 942, "quantity": 81},
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_attachment_box_delete_cleans_empty_box(client: TestClient, db: Session):
+    cert_id, _ = _seed(db)
+    created = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 943,
+            "quantity": 10,
+        },
+    ).json()
+    item_id = created["box_item_id"]
+    deleted = client.delete(f"/CargoTrace/attachment/box/{item_id}")
+    assert deleted.status_code == 200
+    assert deleted.json() == {"box_item_id": item_id, "deleted": True}
+
+    # Box was attachment-only; deleting it should also remove the empty box
+    # so the same box_no can be recreated for a different总排号.
+    recreated = client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0012",
+            "category_id": cert_id,
+            "box_no": 943,
+            "quantity": 5,
+        },
+    )
+    assert recreated.status_code == 200
+
+
+def test_box_info_includes_attachment_summary(client: TestClient, db: Session):
+    cert_id, extra_id = _seed(db)
+    client.put(
+        "/CargoTrace/attachment/config",
+        json={
+            "zongpai_no": "26BW0011",
+            "determination": "has",
+            "items": [
+                {"category_id": cert_id, "expected_qty": 80},
+                {"category_id": extra_id, "expected_qty": 5},
+            ],
+        },
+    )
+    client.post(
+        "/CargoTrace/attachment/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "category_id": cert_id,
+            "box_no": 951,
+            "quantity": 80,
+        },
+    )
+    info = client.get(
+        "/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
+    ).json()
+    assert info["attachment_summary"]["determination"] == "has"
+    assert info["attachment_summary"]["all_complete"] is False
+    assert info["attachment_summary"]["pending_count"] == 1  # extra not yet boxed

tests/test_attachment_models_smoke.py

@@ -0,0 +1,22 @@
+from sqlalchemy.orm import Session
+
+from app.models.attachment import AttachmentCategory, FinishedGoodsAttachment
+
+
+def test_category_round_trip(db: Session):
+    cat = AttachmentCategory(name="TEST_SMOKE", sort_order=1, is_active=True)
+    db.add(cat)
+    db.commit()
+    db.refresh(cat)
+    assert cat.id is not None
+    fetched = db.query(AttachmentCategory).filter_by(name="TEST_SMOKE").first()
+    assert fetched is not None
+    assert fetched.is_active is True
+
+
+def test_attachment_defaults_to_undetermined(db: Session):
+    att = FinishedGoodsAttachment(zongpai_no="26BW0011")
+    db.add(att)
+    db.commit()
+    db.refresh(att)
+    assert att.determination == "undetermined"

tests/test_box_api.py

@@ -13,8 +13,14 @@ 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 "current_zongpai_boxes" in data
     assert "existing_boxes" in data
+    for box in data["existing_boxes"]:
+        for item in box["items"]:
+            assert "box_item_id" in item
     assert "max_box_no" in data
     assert data["suggested_box_no"] == data["max_box_no"] + 1
 
@@ -76,3 +82,182 @@ def test_box_save_zongpai_not_found(client: TestClient):
     )
     assert resp.status_code == 404
     assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
+
+
+def test_box_save_multi_code_appends_different_zongpais(client: TestClient):
+    """多码凑箱：同一排产号同一箱号可追加不同总排号。"""
+    box_no = 901
+    first = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert first.status_code == 200
+    assert first.json()["box_item_id"] is not None
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0012",
+            "box_no": box_no,
+            "quantity": 8,
+        },
+    )
+    assert second.status_code == 200
+    assert second.json()["box_no"] == box_no
+    assert second.json()["zongpai_no"] == "26BW0012"
+
+    info = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
+    assert info.status_code == 200
+    target_box = next(
+        box for box in info.json()["existing_boxes"] if box["box_no"] == box_no
+    )
+    assert {item["zongpai_no"] for item in target_box["items"]} == {
+        "26BW0011",
+        "26BW0012",
+    }
+    for item in target_box["items"]:
+        assert "work_order_no" in item
+        assert isinstance(item["total_quantity"], int)
+
+
+def test_box_save_multi_code_duplicate_same_zongpai(client: TestClient):
+    """多码凑箱：同箱同总排重复录入返回 DUPLICATE_BOX_NO。"""
+    box_no = 902
+    first = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert first.status_code == 200
+
+    duplicate = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 5,
+        },
+    )
+    assert duplicate.status_code == 409
+    assert duplicate.json()["error_code"] == "DUPLICATE_BOX_NO"
+
+
+def test_box_save_split_same_zongpai_up_to_total_quantity(client: TestClient):
+    """单码装箱：同一总排号可分箱提交，累计等于 ERP 数量时成功。"""
+    first = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 905, "quantity": 30},
+    )
+    assert first.status_code == 200
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 906, "quantity": 50},
+    )
+    assert second.status_code == 200
+
+    info = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
+    assert info.status_code == 200
+    current_boxes = [
+        item
+        for item in info.json()["current_zongpai_boxes"]
+        if item["box_no"] in (905, 906)
+    ]
+    assert sum(item["quantity"] for item in current_boxes) == 80
+
+
+def test_box_save_rejects_quantity_over_remaining(client: TestClient):
+    """保存后累计数量超过 ERP 总数量时返回 INVALID_QUANTITY。"""
+    created = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 907, "quantity": 70},
+    )
+    assert created.status_code == 200
+
+    over = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 908, "quantity": 11},
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_box_update_same_box_no_does_not_conflict(client: TestClient):
+    """PATCH 原箱号未变时，应排除当前 box_item_id，避免误判重复。"""
+    box_no = 903
+    created = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert created.status_code == 200
+    box_item_id = created.json()["box_item_id"]
+
+    updated = client.patch(
+        f"/CargoTrace/box/{box_item_id}",
+        json={"box_no": box_no, "quantity": 7},
+    )
+    assert updated.status_code == 200
+    assert updated.json()["quantity"] == 7
+
+
+def test_box_update_rejects_quantity_over_remaining(client: TestClient):
+    """PATCH 调整数量导致累计超过 ERP 总数量时返回 INVALID_QUANTITY。"""
+    first = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 909, "quantity": 70},
+    )
+    assert first.status_code == 200
+
+    second = client.post(
+        "/CargoTrace/box",
+        json={"zongpai_no": "26BW0011", "box_no": 910, "quantity": 5},
+    )
+    assert second.status_code == 200
+    box_item_id = second.json()["box_item_id"]
+
+    over = client.patch(
+        f"/CargoTrace/box/{box_item_id}",
+        json={"box_no": 910, "quantity": 11},
+    )
+    assert over.status_code == 400
+    assert over.json()["error_code"] == "INVALID_QUANTITY"
+
+
+def test_box_delete_removes_empty_box(client: TestClient):
+    """DELETE 删除最后一条明细后，同步清理空箱号。"""
+    box_no = 904
+    created = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0011",
+            "box_no": box_no,
+            "quantity": 10,
+        },
+    )
+    assert created.status_code == 200
+    box_item_id = created.json()["box_item_id"]
+
+    deleted = client.delete(f"/CargoTrace/box/{box_item_id}")
+    assert deleted.status_code == 200
+    assert deleted.json() == {"box_item_id": box_item_id, "deleted": True}
+
+    recreated = client.post(
+        "/CargoTrace/box",
+        json={
+            "zongpai_no": "26BW0012",
+            "box_no": box_no,
+            "quantity": 8,
+        },
+    )
+    assert recreated.status_code == 200

tests/test_config.py

@@ -5,6 +5,7 @@ from config.settings import load_settings
 def test_load_settings_success():
     """测试成功加载配置文件"""
     settings = load_settings("tests/fixtures/test_config.yaml")
+    assert settings.database.active == "sql_server"
     assert settings.database.sql_server.port == 1433
     assert settings.database.sql_server.host == "localhost"
     assert settings.database.sql_server.database == "TestDB"
@@ -23,3 +24,14 @@ def test_database_url_property():
     assert "mssql+pyodbc" in str(url)
     assert "test_user" in str(url)
     assert "TestDB" in str(url)
+
+
+def test_postgresql_database_url_property():
+    """测试 PostgreSQL database_url 属性生成"""
+    settings = load_settings("tests/fixtures/test_config.yaml")
+    settings.database.active = "postgresql"
+    url = settings.database_url
+    assert url.drivername == "postgresql+psycopg"
+    assert url.host == "localhost"
+    assert url.port == 5432
+    assert url.database == "TestDB"

tests/test_location_api.py

@@ -1,5 +1,8 @@
 from fastapi.testclient import TestClient
 
+from app.core.database import SessionLocal
+from app.models.finished_goods import FinishedGoodsLocation
+
 
 def test_register_location_success(client: TestClient):
     """正常上架 — 应返回 200"""
@@ -13,8 +16,8 @@ def test_register_location_success(client: TestClient):
     assert data["location_code"] == "A01-02-03"
 
 
-def test_register_location_duplicate(client: TestClient):
-    """重复上架 — 应返回 409，包含统一错误格式"""
+def test_register_location_duplicate_normal_to_normal(client: TestClient):
+    """普通货位重复上架到普通货位 — 应返回 DUPLICATE_LOCATION"""
     client.post(
         "/CargoTrace/location",
         json={"zongpai_no": "26C999", "location_code": "A01-02-03"},
@@ -30,6 +33,69 @@ def test_register_location_duplicate(client: TestClient):
     assert "registered_at" in data
 
 
+def test_register_location_normal_to_transit_updates_record(client: TestClient):
+    """普通货位下架到转运货位 — 应更新记录并返回 previous_location"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26T998", "location_code": "A01-02-03"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26T998", "location_code": "TRANS-01"},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["zongpai_no"] == "26T998"
+    assert data["location_code"] == "TRANS-01"
+    assert data["previous_location"] == "A01-02-03"
+
+    db = SessionLocal()
+    try:
+        record = (
+            db.query(FinishedGoodsLocation)
+            .filter(FinishedGoodsLocation.zongpai_no == "26T998")
+            .first()
+        )
+        assert record is not None
+        assert record.location_code == "TRANS-01"
+    finally:
+        db.close()
+
+
+def test_register_location_transit_to_normal_rejected(client: TestClient):
+    """已下架后再上架到普通货位 — 应返回 ALREADY_OFF_SHELF"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26B998", "location_code": "TRANS-01"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26B998", "location_code": "A01-02-03"},
+    )
+    assert resp.status_code == 409
+    data = resp.json()
+    assert data["error_code"] == "ALREADY_OFF_SHELF"
+    assert data["message"] == "该总排号已下架至转运区域，不可重新上架"
+    assert data["location_code"] == "TRANS-01"
+    assert "registered_at" in data
+
+
+def test_register_location_transit_to_transit_rejected(client: TestClient):
+    """已下架后再提交转运货位 — 应返回 ALREADY_OFF_SHELF"""
+    client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26C998", "location_code": "TRANS-01"},
+    )
+    resp = client.post(
+        "/CargoTrace/location",
+        json={"zongpai_no": "26C998", "location_code": "TRANS-02"},
+    )
+    assert resp.status_code == 409
+    data = resp.json()
+    assert data["error_code"] == "ALREADY_OFF_SHELF"
+    assert data["location_code"] == "TRANS-01"
+
+
 def test_register_location_invalid_zongpai(client: TestClient):
     """无效总排号 — 应返回 400"""
     resp = client.post(

tests/test_location_overview.py

@@ -0,0 +1,106 @@
+import pytest
+from fastapi.testclient import TestClient
+
+from app.api.v1 import location as location_api
+from app.services.box_service import InvalidZongpaiError
+from app.services.location_service import PaichaNotFoundError, get_paicha_overview
+
+
+class _Bind:
+    class Dialect:
+        name = "postgresql"
+
+    dialect = Dialect()
+
+
+class _Rows:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def fetchone(self):
+        return self._rows[0] if self._rows else None
+
+    def fetchall(self):
+        return self._rows
+
+
+class _Location:
+    def __init__(self, zongpai_no, location_code):
+        self.zongpai_no = zongpai_no
+        self.location_code = location_code
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def filter(self, *_args, **_kwargs):
+        return self
+
+    def all(self):
+        return self._rows
+
+
+class _Session:
+    bind = _Bind()
+
+    def execute(self, _sql, params):
+        if "zongpai_no" in params:
+            if params["zongpai_no"] == "26B404":
+                return _Rows([])
+            return _Rows([("26B1", "R00001", 80, "6-1(7)")])
+        return _Rows([
+            ("26B2", "6-1(7)", 34),
+            ("26B3", "6-2(3)", 60),
+            ("26B1", "6-1(7)", 80),
+            ("26B4", "6-2(3)", 45),
+        ])
+
+    def query(self, _model):
+        return _Query([
+            _Location("26B1", "A01-02-03"),
+            _Location("26B3", "TRANS-01"),
+        ])
+
+
+def test_paicha_overview_maps_statuses_and_sorts():
+    data = get_paicha_overview(_Session(), "26B1")
+
+    assert data["paicha_no"] == "R00001"
+    assert data["total_count"] == 4
+    assert [item["zongpai_no"] for item in data["items"]] == [
+        "26B1",
+        "26B3",
+        "26B2",
+        "26B4",
+    ]
+    assert data["items"][0]["status"] == "on_shelf"
+    assert data["items"][0]["location_code"] == "A01-02-03"
+    assert data["items"][1]["status"] == "transferred"
+    assert data["items"][1]["location_code"] == "TRANS-01"
+    assert data["items"][2]["status"] == "not_shelved"
+    assert data["items"][2]["location_code"] is None
+
+
+def test_paicha_overview_invalid_zongpai():
+    with pytest.raises(InvalidZongpaiError):
+        get_paicha_overview(_Session(), "INVALID")
+
+
+def test_paicha_overview_not_found():
+    with pytest.raises(PaichaNotFoundError):
+        get_paicha_overview(_Session(), "26B404")
+
+
+def test_paicha_overview_api_not_found(client: TestClient, monkeypatch):
+    def raise_not_found(_db, _zongpai_no):
+        raise PaichaNotFoundError()
+
+    monkeypatch.setattr(location_api, "get_paicha_overview", raise_not_found)
+    resp = client.get(
+        "/CargoTrace/location/paicha-overview",
+        params={"zongpai_no": "26B404"},
+    )
+
+    assert resp.status_code == 404
+    assert resp.json()["error_code"] == "PAICHA_NOT_FOUND"