feat: extend boxing module to support accessory items

- Add PendingAccessory schema and pending_accessories to BoxInfoResponse
- Add item_type field to BoxItemDetail, CurrentZongpaiBox, BoxSaveRequest, BoxSaveResponse
- Query pending (unboxed) accessories in get_box_info
- Branch save_box_record to skip ERP validation for accessory items

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

.gitignore

@@ -7,3 +7,9 @@ dist/
 build/
 .pytest_cache/
 .claude/
+.agents/
+# Config with secrets
+config/settings.yaml
+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

@@ -11,3 +11,35 @@ source .venv/bin/activate  # Linux/Mac
 pip install -r requirements.txt
 uvicorn app.main:app --reload
 ```
+
+## Configuration
+
+Configuration is managed via YAML files in the `config/` directory.
+
+### Configuration File
+
+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/accessory.py

@@ -0,0 +1,116 @@
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.schemas.accessory import (
+    AccessoryCreateRequest,
+    AccessoryListResponse,
+    AccessoryResponse,
+    AccessoryTypeCreateRequest,
+    AccessoryTypeListResponse,
+    AccessoryTypeUpdateRequest,
+    AccessoryUpdateRequest,
+    WorkOrderQueryResponse,
+)
+from app.services.accessory_service import (
+    create_accessory,
+    create_accessory_type,
+    delete_accessory,
+    delete_accessory_type,
+    list_accessories,
+    list_accessory_types,
+    update_accessory,
+    update_accessory_type,
+    query_work_orders,
+)
+
+router = APIRouter(tags=["accessory"])
+
+
+@router.get(
+    "/accessory/work-orders",
+    response_model=WorkOrderQueryResponse,
+)
+def work_orders(
+    paichan_no: str = Query(..., description="排产号"),
+    db: Session = Depends(get_db),
+):
+    """查询工单信息：根据排产号查询工令号及对应总排号。"""
+    return query_work_orders(db, paichan_no)
+
+
+@router.get(
+    "/accessory",
+    response_model=AccessoryListResponse,
+)
+def accessory_list(
+    paichan_no: str = Query(..., description="排产号"),
+    db: Session = Depends(get_db),
+):
+    """查询附件列表：根据排产号查询所有附件记录。"""
+    return list_accessories(db, paichan_no)
+
+
+@router.post(
+    "/accessory",
+    response_model=AccessoryResponse,
+)
+def accessory_create(req: AccessoryCreateRequest, db: Session = Depends(get_db)):
+    """创建附件记录。"""
+    record = create_accessory(db, req)
+    return {
+        "id": record.id,
+        "paichan_no": record.paichan_no,
+        "zongpai_no": record.zongpai_no,
+        "accessory_type": record.accessory_type,
+        "quantity": record.quantity,
+        "location_code": record.location_code,
+        "is_boxed": False,
+        "created_at": record.created_at.isoformat(),
+    }
+
+
+@router.patch(
+    "/accessory/{id}",
+    response_model=AccessoryResponse,
+)
+def accessory_update(id: int, req: AccessoryUpdateRequest, db: Session = Depends(get_db)):
+    """更新附件记录。"""
+    return update_accessory(db, id, req)
+
+
+@router.delete("/accessory/{id}")
+def accessory_delete(id: int, db: Session = Depends(get_db)):
+    """删除附件记录。"""
+    return delete_accessory(db, id)
+
+
+@router.get(
+    "/accessory-type",
+    response_model=AccessoryTypeListResponse,
+)
+def accessory_type_list(db: Session = Depends(get_db)):
+    """查询所有附件类型。"""
+    return list_accessory_types(db)
+
+
+@router.post("/accessory-type")
+def accessory_type_create(
+    req: AccessoryTypeCreateRequest, db: Session = Depends(get_db)
+):
+    """创建附件类型。"""
+    return create_accessory_type(db, req)
+
+
+@router.patch("/accessory-type/{id}")
+def accessory_type_update(
+    id: int, req: AccessoryTypeUpdateRequest, db: Session = Depends(get_db)
+):
+    """更新附件类型。"""
+    return update_accessory_type(db, id, req)
+
+
+@router.delete("/accessory-type/{id}")
+def accessory_type_delete(id: int, db: Session = Depends(get_db)):
+    """删除附件类型。"""
+    return delete_accessory_type(db, 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.schemas.box import (
-from app.services.box_service import get_box_info, save_box_record
+    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.schemas.location import (
-from app.services.location_service import register_location
+    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/core/config.py

@@ -1,32 +0,0 @@
-from sqlalchemy.engine import URL
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class Settings(BaseSettings):
-    SQL_SERVER_HOST: str
-    SQL_SERVER_PORT: int = 1433
-    SQL_SERVER_DATABASE: str
-    SQL_SERVER_USERNAME: str
-    SQL_SERVER_PASSWORD: str
-    SQL_SERVER_DRIVER: str = "{ODBC Driver 18 for SQL Server}"
-    SQL_SERVER_TRUST_SERVER_CERTIFICATE: str = "yes"
-
-    @property
-    def database_url(self) -> URL:
-        return URL.create(
-            "mssql+pyodbc",
-            username=self.SQL_SERVER_USERNAME,
-            password=self.SQL_SERVER_PASSWORD,
-            host=self.SQL_SERVER_HOST,
-            port=self.SQL_SERVER_PORT,
-            database=self.SQL_SERVER_DATABASE,
-            query={
-                "driver": self.SQL_SERVER_DRIVER.strip("{}"),
-                "TrustServerCertificate": self.SQL_SERVER_TRUST_SERVER_CERTIFICATE,
-            },
-        )
-
-    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
-
-
-settings = Settings()

app/main.py

@@ -1,16 +1,53 @@
+import sys
+
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 
+from config.settings import load_settings
+
+# 加载配置文件，失败则退出
+try:
+    settings = load_settings()
+except FileNotFoundError as e:
+    print(f"配置文件错误: {e}")
+    sys.exit(1)
+except Exception as e:
+    print(f"加载配置失败: {e}")
+    sys.exit(1)
+
 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.api.v1.accessory import router as accessory_router
-from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError
+from app.services.location_service import (
+    AlreadyOffShelfError,
+    DuplicateLocationError,
+    PaichaNotFoundError,
+)
+from app.services.box_service import (
+    BoxItemNotFoundError,
+    DuplicateBoxItemError,
+    InvalidBoxItemError,
+    InvalidQuantityError,
+    InvalidZongpaiError,
+    ZongpaiNotFoundError,
+)
+from app.services.accessory_service import (
+    AccessoryAlreadyBoxedError,
+    AccessoryAlreadyOffShelfError,
+    AccessoryDuplicateLocationError,
+    AccessoryNotFoundError,
+    AccessoryTypeNotFoundError,
+    DuplicateAccessoryTypeNameError,
+    InvalidPaichanError,
+    PaichanNotFoundError as AccessoryPaichanNotFoundError,
+)
 
 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(accessory_router, prefix="/CargoTrace")
 
 
 @app.exception_handler(DuplicateLocationError)
@@ -26,11 +63,24 @@ 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():
         msg = err.get("msg", "")
-        if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg:
+        if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg or "INVALID_PAICHAN" in msg or "INVALID_TYPE_NAME" in msg:
             error_code = msg.replace("Value error, ", "")
             return JSONResponse(
                 status_code=400,
@@ -92,6 +142,154 @@ 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(InvalidPaichanError)
+async def invalid_paichan_handler(request: Request, exc: InvalidPaichanError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_PAICHAN",
+            "message": "排产号格式不合法",
+        },
+    )
+
+
+@app.exception_handler(AccessoryPaichanNotFoundError)
+async def accessory_paichan_not_found_handler(
+    request: Request, exc: AccessoryPaichanNotFoundError
+):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "PAICHA_NOT_FOUND",
+            "message": "未找到该排产号信息",
+        },
+    )
+
+
+@app.exception_handler(AccessoryNotFoundError)
+async def accessory_not_found_handler(request: Request, exc: AccessoryNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "ACCESSORY_NOT_FOUND",
+            "message": "附件记录不存在",
+        },
+    )
+
+
+@app.exception_handler(AccessoryAlreadyBoxedError)
+async def accessory_already_boxed_handler(
+    request: Request, exc: AccessoryAlreadyBoxedError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_BOXED",
+            "message": "该附件已装箱不可操作",
+        },
+    )
+
+
+@app.exception_handler(AccessoryDuplicateLocationError)
+async def accessory_duplicate_location_handler(
+    request: Request, exc: AccessoryDuplicateLocationError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_LOCATION",
+            "message": "该附件已存在货位记录",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(AccessoryAlreadyOffShelfError)
+async def accessory_already_off_shelf_handler(
+    request: Request, exc: AccessoryAlreadyOffShelfError
+):
+    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(AccessoryTypeNotFoundError)
+async def accessory_type_not_found_handler(
+    request: Request, exc: AccessoryTypeNotFoundError
+):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "ACCESSORY_TYPE_NOT_FOUND",
+            "message": "附件类型不存在",
+        },
+    )
+
+
+@app.exception_handler(DuplicateAccessoryTypeNameError)
+async def duplicate_accessory_type_name_handler(
+    request: Request, exc: DuplicateAccessoryTypeNameError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_TYPE_NAME",
+            "message": "附件类型名称已存在",
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/models/accessory.py

@@ -0,0 +1,44 @@
+# app/models/accessory.py
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+
+SCHEMA = "CargoTrace"
+
+
+class FinishedGoodsAccessory(Base):
+    __tablename__ = "finished_goods_accessory"
+    __table_args__ = (
+        Index("idx_fga_paichan_no", "paichan_no"),
+        Index("idx_fga_zongpai_no", "zongpai_no"),
+        Index("idx_fga_location_code", "location_code"),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    paichan_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    accessory_type: Mapped[str] = mapped_column(String(64), nullable=False)
+    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
+    location_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class AccessoryType(Base):
+    __tablename__ = "accessory_type"
+    __table_args__ = (
+        Index("uk_accessory_type_name", "name", unique=True),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(64), nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

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(),
     )
 
 
@@ -52,6 +58,12 @@ class FinishedGoodsBoxItem(Base):
     box_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
     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(
+    item_type: Mapped[str] = mapped_column(
-        DateTime, nullable=False, server_default=func.getdate()
+        String(16), nullable=False, default="product", server_default="product"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
     )

app/schemas/accessory.py

@@ -0,0 +1,135 @@
+# app/schemas/accessory.py
+import re
+
+from pydantic import BaseModel, Field, field_validator
+
+PAICHAN_PATTERN = re.compile(r"^[A-Z]{1,2}\d{5}(-J)?$")
+LOCATION_NORMAL_PATTERN = re.compile(r"^[A-Z]+\d+-\d+-\d+$")
+LOCATION_TRANS_PATTERN = re.compile(r"^TRANS-\d+")
+
+
+# ── Work order query ──
+
+
+class WorkOrderGroup(BaseModel):
+    work_order_no: str
+    zongpai_nos: list[str]
+
+
+class WorkOrderQueryResponse(BaseModel):
+    paichan_no: str
+    work_orders: list[WorkOrderGroup]
+
+
+# ── Accessory CRUD ──
+
+
+class AccessoryCreateRequest(BaseModel):
+    paichan_no: str
+    zongpai_no: str
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+
+    @field_validator("paichan_no")
+    @classmethod
+    def validate_paichan_no(cls, v: str) -> str:
+        v = v.strip().upper()
+        if not PAICHAN_PATTERN.match(v):
+            raise ValueError("INVALID_PAICHAN")
+        return v
+
+    @field_validator("zongpai_no")
+    @classmethod
+    def validate_zongpai_no(cls, v: str) -> str:
+        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("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+    @field_validator("location_code")
+    @classmethod
+    def validate_location_code(cls, v: str | None) -> str | None:
+        if v is None:
+            return v
+        v = v.strip().upper()
+        if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v):
+            raise ValueError("INVALID_LOCATION")
+        return v
+
+
+class AccessoryResponse(BaseModel):
+    id: int
+    paichan_no: str
+    zongpai_no: str
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+    is_boxed: bool = False
+    created_at: str
+
+
+class AccessoryListResponse(BaseModel):
+    paichan_no: str
+    items: list[AccessoryResponse]
+
+
+class AccessoryUpdateRequest(BaseModel):
+    accessory_type: str | None = None
+    quantity: int | None = None
+    location_code: str | None = None
+
+    @field_validator("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int | None) -> int | None:
+        if v is not None and v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+    @field_validator("location_code")
+    @classmethod
+    def validate_location_code(cls, v: str | None) -> str | None:
+        if v is None:
+            return v
+        v = v.strip().upper()
+        if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v):
+            raise ValueError("INVALID_LOCATION")
+        return v
+
+
+# ── Accessory type management ──
+
+
+class AccessoryTypeItem(BaseModel):
+    id: int
+    name: str
+    sort_order: int
+
+
+class AccessoryTypeListResponse(BaseModel):
+    types: list[AccessoryTypeItem]
+
+
+class AccessoryTypeCreateRequest(BaseModel):
+    name: str
+    sort_order: int = 0
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError("INVALID_TYPE_NAME")
+        return v
+
+
+class AccessoryTypeUpdateRequest(BaseModel):
+    name: str | None = None
+    sort_order: int | None = None

app/schemas/box.py

@@ -1,37 +1,68 @@
 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 PendingAccessory(BaseModel):
+    """待装箱的配件"""
+
+    accessory_id: int
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+
+
 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
+    item_type: str = "product"
+
+
+class CurrentZongpaiBox(BaseModel):
+    """当前总排号已经分配的箱号明细"""
+
+    box_item_id: int
+    box_no: int
+    quantity: int
+    item_type: str = "product"
 
 
 class BoxDetail(BaseModel):
     """一个箱号的完整信息"""
+
     box_no: int
     items: list[BoxItemDetail]
 
 
 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]
+    pending_accessories: list[PendingAccessory] = Field(default_factory=list)
     max_box_no: int
     suggested_box_no: int
 
 
 class BoxSaveRequest(BaseModel):
     """POST /box 请求"""
+
     zongpai_no: str
     box_no: int
     quantity: int
+    item_type: str = "product"
+    accessory_id: int | None = None
 
     @field_validator("zongpai_no")
     @classmethod
@@ -58,8 +89,50 @@ class BoxSaveRequest(BaseModel):
 
 class BoxSaveResponse(BaseModel):
     """POST /box 响应"""
+
+    box_item_id: int
     paichan_no: str
     box_no: int
     zongpai_no: str
     quantity: int
+    item_type: str = "product"
     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/accessory_service.py

@@ -0,0 +1,339 @@
+import re
+from collections import defaultdict
+from datetime import datetime
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.models.accessory import AccessoryType, FinishedGoodsAccessory
+from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
+from app.schemas.accessory import (
+    AccessoryCreateRequest,
+    AccessoryTypeCreateRequest,
+    AccessoryTypeUpdateRequest,
+    AccessoryUpdateRequest,
+)
+
+PAICHAN_PATTERN = re.compile(r"^[A-Z]{1,2}\d{5}(-J)?$")
+
+
+# ── Error classes ──
+
+
+class InvalidPaichanError(Exception):
+    pass
+
+
+class PaichanNotFoundError(Exception):
+    pass
+
+
+class AccessoryNotFoundError(Exception):
+    pass
+
+
+class AccessoryAlreadyBoxedError(Exception):
+    pass
+
+
+class AccessoryDuplicateLocationError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+class AccessoryAlreadyOffShelfError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+class AccessoryTypeNotFoundError(Exception):
+    pass
+
+
+class DuplicateAccessoryTypeNameError(Exception):
+    pass
+
+
+# ── Helpers ──
+
+
+def _is_transit_location(location_code: str) -> bool:
+    return location_code.startswith("TRANS-")
+
+
+def _erp_paicha_items_sql(db: Session) -> str:
+    """Reuse the same ERP query pattern as location_service."""
+    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 _is_accessory_boxed(db: Session, accessory: FinishedGoodsAccessory) -> bool:
+    """Check whether the accessory is already boxed (EXISTS in box_item as 'accessory')."""
+    boxed = (
+        db.query(FinishedGoodsBoxItem)
+        .join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
+        .filter(
+            FinishedGoodsBoxItem.zongpai_no == accessory.zongpai_no,
+            FinishedGoodsBox.paichan_no == accessory.paichan_no,
+            FinishedGoodsBoxItem.item_type == "accessory",
+        )
+        .first()
+    )
+    return boxed is not None
+
+
+def _accessory_to_dict(
+    accessory: FinishedGoodsAccessory, is_boxed: bool = False
+) -> dict:
+    return {
+        "id": accessory.id,
+        "paichan_no": accessory.paichan_no,
+        "zongpai_no": accessory.zongpai_no,
+        "accessory_type": accessory.accessory_type,
+        "quantity": accessory.quantity,
+        "location_code": accessory.location_code,
+        "is_boxed": is_boxed,
+        "created_at": accessory.created_at.isoformat(),
+    }
+
+
+# ── Work order query ──
+
+
+def query_work_orders(db: Session, paichan_no: str) -> dict:
+    """Query ERP for work orders grouped by work_order_no for a given paichan_no."""
+    paichan_no = paichan_no.strip().upper()
+    if not PAICHAN_PATTERN.match(paichan_no):
+        raise InvalidPaichanError()
+
+    rows = db.execute(
+        text(_erp_paicha_items_sql(db)), {"paicha_no": paichan_no}
+    ).fetchall()
+
+    if not rows:
+        raise PaichanNotFoundError()
+
+    grouped: dict[str, list[str]] = defaultdict(list)
+    for row in rows:
+        zongpai_no = row[0]
+        work_order_no = row[1]
+        grouped[work_order_no].append(zongpai_no)
+
+    work_orders = [
+        {"work_order_no": wo_no, "zongpai_nos": zongpai_nos}
+        for wo_no, zongpai_nos in grouped.items()
+    ]
+
+    return {
+        "paichan_no": paichan_no,
+        "work_orders": work_orders,
+    }
+
+
+# ── Accessory CRUD ──
+
+
+def list_accessories(db: Session, paichan_no: str) -> dict:
+    """List all accessories for a given paichan_no, including is_boxed status."""
+    accessories = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.paichan_no == paichan_no)
+        .order_by(FinishedGoodsAccessory.id)
+        .all()
+    )
+
+    items = []
+    for acc in accessories:
+        boxed = _is_accessory_boxed(db, acc)
+        items.append(_accessory_to_dict(acc, is_boxed=boxed))
+
+    return {
+        "paichan_no": paichan_no,
+        "items": items,
+    }
+
+
+def create_accessory(
+    db: Session, req: AccessoryCreateRequest
+) -> FinishedGoodsAccessory:
+    """Create a new accessory record after validating against ERP data."""
+    # 1. Validate paichan_no exists in ERP
+    rows = db.execute(
+        text(_erp_paicha_items_sql(db)), {"paicha_no": req.paichan_no}
+    ).fetchall()
+
+    if not rows:
+        raise PaichanNotFoundError()
+
+    # 2. Validate zongpai_no belongs to this paichan_no in ERP
+    valid_zongpai_nos = {row[0] for row in rows}
+    if req.zongpai_no not in valid_zongpai_nos:
+        raise ValueError(
+            f"zongpai_no {req.zongpai_no} does not belong to paichan_no {req.paichan_no}"
+        )
+
+    # 3. Create the record
+    record = FinishedGoodsAccessory(
+        paichan_no=req.paichan_no,
+        zongpai_no=req.zongpai_no,
+        accessory_type=req.accessory_type,
+        quantity=req.quantity,
+        location_code=req.location_code,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return record
+
+
+def update_accessory(
+    db: Session, accessory_id: int, req: AccessoryUpdateRequest
+) -> dict:
+    """Partially update an accessory record with shelving logic."""
+    accessory = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.id == accessory_id)
+        .first()
+    )
+    if accessory is None:
+        raise AccessoryNotFoundError()
+
+    # Check if boxed
+    if _is_accessory_boxed(db, accessory):
+        raise AccessoryAlreadyBoxedError()
+
+    # Shelving logic when location_code changes
+    if req.location_code is not None and accessory.location_code is not None:
+        existing_loc = accessory.location_code
+        new_loc = req.location_code
+
+        if existing_loc == new_loc:
+            # No change needed for location
+            pass
+        elif _is_transit_location(existing_loc):
+            raise AccessoryAlreadyOffShelfError(
+                location_code=existing_loc,
+                registered_at=accessory.created_at.isoformat(),
+            )
+        elif _is_transit_location(new_loc):
+            # Off-shelf/down-shelf: normal operation
+            accessory.location_code = new_loc
+            accessory.created_at = datetime.now()
+        else:
+            # Both are normal locations → duplicate
+            raise AccessoryDuplicateLocationError(
+                location_code=existing_loc,
+                registered_at=accessory.created_at.isoformat(),
+            )
+    elif req.location_code is not None and accessory.location_code is None:
+        # Not yet shelved, just set the location
+        accessory.location_code = req.location_code
+
+    # Update other fields if provided
+    if req.accessory_type is not None:
+        accessory.accessory_type = req.accessory_type
+    if req.quantity is not None:
+        accessory.quantity = req.quantity
+
+    db.commit()
+    db.refresh(accessory)
+    return _accessory_to_dict(accessory, is_boxed=_is_accessory_boxed(db, accessory))
+
+
+def delete_accessory(db: Session, accessory_id: int) -> dict:
+    """Delete an accessory record if it is not boxed."""
+    accessory = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.id == accessory_id)
+        .first()
+    )
+    if accessory is None:
+        raise AccessoryNotFoundError()
+
+    if _is_accessory_boxed(db, accessory):
+        raise AccessoryAlreadyBoxedError()
+
+    db.delete(accessory)
+    db.commit()
+    return {"id": accessory_id, "deleted": True}
+
+
+# ── Accessory type management ──
+
+
+def list_accessory_types(db: Session) -> dict:
+    """List all accessory types ordered by sort_order."""
+    types = (
+        db.query(AccessoryType)
+        .order_by(AccessoryType.sort_order, AccessoryType.id)
+        .all()
+    )
+    return {
+        "types": [
+            {"id": t.id, "name": t.name, "sort_order": t.sort_order} for t in types
+        ]
+    }
+
+
+def create_accessory_type(db: Session, req: AccessoryTypeCreateRequest) -> dict:
+    """Create a new accessory type. Name must be unique."""
+    existing = db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
+    if existing is not None:
+        raise DuplicateAccessoryTypeNameError()
+
+    record = AccessoryType(
+        name=req.name,
+        sort_order=req.sort_order,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
+
+
+def update_accessory_type(
+    db: Session, type_id: int, req: AccessoryTypeUpdateRequest
+) -> dict:
+    """Partially update an accessory type."""
+    record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
+    if record is None:
+        raise AccessoryTypeNotFoundError()
+
+    # If name is being changed, check for duplicate
+    if req.name is not None and req.name != record.name:
+        duplicate = (
+            db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
+        )
+        if duplicate is not None:
+            raise DuplicateAccessoryTypeNameError()
+        record.name = req.name
+
+    if req.sort_order is not None:
+        record.sort_order = req.sort_order
+
+    db.commit()
+    db.refresh(record)
+    return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
+
+
+def delete_accessory_type(db: Session, type_id: int) -> dict:
+    """Delete an accessory type."""
+    record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
+    if record is None:
+        raise AccessoryTypeNotFoundError()
+
+    db.delete(record)
+    db.commit()
+    return {"id": type_id, "deleted": True}

app/services/box_service.py

@@ -1,10 +1,11 @@
 import re
 
-from sqlalchemy import text
+from sqlalchemy import bindparam, text
 from sqlalchemy.orm import Session
 
+from app.models.accessory import FinishedGoodsAccessory
 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 +24,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 视图查询总排号对应的排产号和数量。"""
+    """从 ERP 视图查询总排号对应的排产号、工令号和数量。"""
-    sql = text(
+    sql = text(_erp_info_sql(db))
-        "SELECT TOP 1 [总排号], [排产号], [数量] "
-        "FROM [ERPAuto].[vw_productionContractData] "
-        "WHERE [总排号] = :zongpai_no"
-    )
     row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone()
     if not row:
         raise ZongpaiNotFoundError()
@@ -37,6 +46,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 +113,151 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
         .all()
     )
 
-    existing_boxes = []
+    box_items = {}
-    max_box_no = 0
+    item_zongpai_nos = []
     for box in boxes:
         items = (
             db.query(FinishedGoodsBoxItem)
             .filter(FinishedGoodsBoxItem.box_id == box.id)
             .all()
         )
-        existing_boxes.append({
+        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
+    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),
+                        "item_type": item.item_type or "product",
+                    }
+                )
+        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"),
+                        "item_type": item.item_type or "product",
+                    }
                     for item in items
                 ],
-        })
+            }
+        )
         if box.box_no > max_box_no:
             max_box_no = box.box_no
 
+    # Query pending accessories for this zongpai_no
+    pending_accessories = []
+    accessory_records = (
+        db.query(FinishedGoodsAccessory)
+        .filter(
+            FinishedGoodsAccessory.zongpai_no == zongpai_no,
+            FinishedGoodsAccessory.location_code.isnot(None),
+        )
+        .all()
+    )
+    for acc in accessory_records:
+        is_boxed = (
+            db.query(FinishedGoodsBoxItem)
+            .join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
+            .filter(
+                FinishedGoodsBoxItem.zongpai_no == acc.zongpai_no,
+                FinishedGoodsBox.paichan_no == acc.paichan_no,
+                FinishedGoodsBoxItem.item_type == "accessory",
+            )
+            .first()
+        ) is not None
+        if not is_boxed:
+            pending_accessories.append(
+                {
+                    "accessory_id": acc.id,
+                    "accessory_type": acc.accessory_type,
+                    "quantity": acc.quantity,
+                    "location_code": acc.location_code,
+                }
+            )
+
     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,
+        "pending_accessories": pending_accessories,
         "max_box_no": max_box_no,
         "suggested_box_no": max_box_no + 1,
     }
 
 
+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:
-    """保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。"""
+    """保存装箱记录。同箱可凑箱，但同箱同总排不可重复。"""
+    if req.item_type == "accessory":
+        # For accessories, skip ERP quantity validation
+        accessory = (
+            db.query(FinishedGoodsAccessory)
+            .filter(FinishedGoodsAccessory.id == req.accessory_id)
+            .first()
+        )
+        if accessory is None:
+            raise BoxItemNotFoundError()
+        paichan_no = accessory.paichan_no
+    else:
         erp = query_erp_info(db, req.zongpai_no)
         paichan_no = erp["paichan_no"]
 
@@ -109,7 +281,17 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
         )
         if existing_item:
             raise DuplicateBoxItemError(paichan_no, req.box_no)
-    else:
+
+    if req.item_type != "accessory":
+        _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,
@@ -121,15 +303,134 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
         box_id=box.id,
         zongpai_no=req.zongpai_no,
         quantity=req.quantity,
+        item_type=req.item_type,
     )
     db.add(item)
     db.commit()
     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,
+        "item_type": req.item_type,
         "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,125 @@ 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 _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"
+    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, "transferred": 1, "not_shelved": 2}
+    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 +145,21 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio
         .first()
     )
     if existing:
+        if _is_transit_location(existing.location_code):
+            raise AlreadyOffShelfError(
+                location_code=existing.location_code,
+                registered_at=existing.created_at.isoformat(),
+            )
+
+        if _is_transit_location(req.location_code):
+            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

docs/plans/2026-05-24-accessory-impl.md

@@ -0,0 +1,750 @@
+# Accessory Module Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Add an independent accessory registration module to CargoTrace, allowing warehouse workers to manually register, shelve, and box order accessories that arrive without execution cards.
+
+**Architecture:** New `finished_goods_accessory` table + `accessory_type` preset table in PostgreSQL. New FastAPI service layer (`accessory_service.py`) and API routes. Flutter gets a new `AccessoryPage` on the home screen. Existing boxing module is extended to show and box accessories via a new `item_type` column on `finished_goods_box_item`.
+
+**Tech Stack:** FastAPI + SQLAlchemy 2.0 (sync) + Pydantic v2 / Flutter 3.x + Dart / PostgreSQL
+
+**Branching:** Create a `dev` branch from `master` in both `services/fastapi` and `apps/pad_scanner` submodules.
+
+---
+
+## Task 1: Create dev branches in both submodules
+
+**Step 1: Create dev branch in fastapi submodule**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+git checkout -b dev
+git push -u origin dev
+```
+
+**Step 2: Create dev branch in pad_scanner submodule**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/apps/pad_scanner
+git checkout -b dev
+git push -u origin dev
+```
+
+**Step 3: Commit plan document in fastapi**
+
+The PRD `2026-05-24-accessory-module.md` is already in the working tree.
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+git add docs/plans/2026-05-24-accessory-module.md docs/plans/2026-05-24-accessory-impl.md
+git commit -m "docs: add accessory module PRD and implementation plan"
+```
+
+---
+
+## Task 2: Database migration — new tables + ALTER existing table
+
+**Files:**
+- Create: `services/fastapi/app/models/accessory.py`
+- Modify: `services/fastapi/app/models/finished_goods.py` (add `item_type` column to `FinishedGoodsBoxItem`)
+- Modify: `services/fastapi/app/models/__init__.py` (register new models for auto-import)
+
+**Step 1: Create the accessory model file**
+
+```python
+# app/models/accessory.py
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+
+SCHEMA = "CargoTrace"
+
+
+class FinishedGoodsAccessory(Base):
+    __tablename__ = "finished_goods_accessory"
+    __table_args__ = (
+        Index("idx_fga_paichan_no", "paichan_no"),
+        Index("idx_fga_zongpai_no", "zongpai_no"),
+        Index("idx_fga_location_code", "location_code"),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    paichan_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
+    accessory_type: Mapped[str] = mapped_column(String(64), nullable=False)
+    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
+    location_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        default=func.current_timestamp(),
+        server_default=func.current_timestamp(),
+    )
+
+
+class AccessoryType(Base):
+    __tablename__ = "accessory_type"
+    __table_args__ = (
+        Index("uk_accessory_type_name", "name", unique=True),
+        {"schema": SCHEMA},
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(64), nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+```
+
+**Step 2: Add `item_type` column to `FinishedGoodsBoxItem`**
+
+In `app/models/finished_goods.py`, add to the `FinishedGoodsBoxItem` class:
+
+```python
+item_type: Mapped[str] = mapped_column(
+    String(16), nullable=False, default="product", server_default="product"
+)
+```
+
+**Step 3: Run the SQL migration manually**
+
+Execute against the PostgreSQL database:
+
+```sql
+-- 1. 附件记录表
+CREATE TABLE IF NOT EXISTS "CargoTrace".finished_goods_accessory (
+    id              SERIAL        PRIMARY KEY,
+    paichan_no      VARCHAR(64)   NOT NULL,
+    zongpai_no      VARCHAR(64)   NOT NULL,
+    accessory_type  VARCHAR(64)   NOT NULL,
+    quantity        INT           NOT NULL,
+    location_code   VARCHAR(64)   NULL,
+    created_at      TIMESTAMP     NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_fga_paichan_no    ON "CargoTrace".finished_goods_accessory (paichan_no);
+CREATE INDEX IF NOT EXISTS idx_fga_zongpai_no    ON "CargoTrace".finished_goods_accessory (zongpai_no);
+CREATE INDEX IF NOT EXISTS idx_fga_location_code ON "CargoTrace".finished_goods_accessory (location_code);
+
+-- 2. 附件类型预设表
+CREATE TABLE IF NOT EXISTS "CargoTrace".accessory_type (
+    id          SERIAL        PRIMARY KEY,
+    name        VARCHAR(64)   NOT NULL,
+    sort_order  INT           NOT NULL DEFAULT 0,
+    CONSTRAINT uk_accessory_type_name UNIQUE (name)
+);
+
+-- 3. 装箱明细表新增 item_type 字段
+ALTER TABLE "CargoTrace".finished_goods_box_item
+ADD COLUMN IF NOT EXISTS item_type VARCHAR(16) NOT NULL DEFAULT 'product';
+```
+
+**Step 4: Verify tables exist**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+# Activate venv and run a quick check
+```
+
+**Step 5: Commit**
+
+```bash
+git add app/models/accessory.py app/models/finished_goods.py
+git commit -m "feat: add accessory models and box_item item_type column"
+```
+
+---
+
+## Task 3: Pydantic schemas for accessory endpoints
+
+**Files:**
+- Create: `services/fastapi/app/schemas/accessory.py`
+
+**Step 1: Write the schema file**
+
+```python
+# app/schemas/accessory.py
+import re
+
+from pydantic import BaseModel, Field, field_validator
+
+PAICHAN_PATTERN = re.compile(r"^[A-Z]{1,2}\d{5}(-J)?$")
+LOCATION_NORMAL_PATTERN = re.compile(r"^[A-Z]+\d+-\d+-\d+$")
+LOCATION_TRANS_PATTERN = re.compile(r"^TRANS-\d+")
+
+
+# ── Work order query ──
+
+class WorkOrderGroup(BaseModel):
+    work_order_no: str
+    zongpai_nos: list[str]
+
+
+class WorkOrderQueryResponse(BaseModel):
+    paichan_no: str
+    work_orders: list[WorkOrderGroup]
+
+
+# ── Accessory CRUD ──
+
+class AccessoryCreateRequest(BaseModel):
+    paichan_no: str
+    zongpai_no: str
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+
+    @field_validator("paichan_no")
+    @classmethod
+    def validate_paichan_no(cls, v: str) -> str:
+        v = v.strip().upper()
+        if not PAICHAN_PATTERN.match(v):
+            raise ValueError("INVALID_PAICHAN")
+        return v
+
+    @field_validator("zongpai_no")
+    @classmethod
+    def validate_zongpai_no(cls, v: str) -> str:
+        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("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+    @field_validator("location_code")
+    @classmethod
+    def validate_location_code(cls, v: str | None) -> str | None:
+        if v is None:
+            return v
+        v = v.strip().upper()
+        if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v):
+            raise ValueError("INVALID_LOCATION")
+        return v
+
+
+class AccessoryResponse(BaseModel):
+    id: int
+    paichan_no: str
+    zongpai_no: str
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+    is_boxed: bool = False
+    created_at: str
+
+
+class AccessoryListResponse(BaseModel):
+    paichan_no: str
+    items: list[AccessoryResponse]
+
+
+class AccessoryUpdateRequest(BaseModel):
+    accessory_type: str | None = None
+    quantity: int | None = None
+    location_code: str | None = None
+
+    @field_validator("quantity")
+    @classmethod
+    def validate_quantity(cls, v: int | None) -> int | None:
+        if v is not None and v <= 0:
+            raise ValueError("INVALID_QUANTITY")
+        return v
+
+    @field_validator("location_code")
+    @classmethod
+    def validate_location_code(cls, v: str | None) -> str | None:
+        if v is None:
+            return v
+        v = v.strip().upper()
+        if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v):
+            raise ValueError("INVALID_LOCATION")
+        return v
+
+
+# ── Accessory type management ──
+
+class AccessoryTypeItem(BaseModel):
+    id: int
+    name: str
+    sort_order: int
+
+
+class AccessoryTypeListResponse(BaseModel):
+    types: list[AccessoryTypeItem]
+
+
+class AccessoryTypeCreateRequest(BaseModel):
+    name: str
+    sort_order: int = 0
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError("INVALID_TYPE_NAME")
+        return v
+
+
+class AccessoryTypeUpdateRequest(BaseModel):
+    name: str | None = None
+    sort_order: int | None = None
+```
+
+**Step 2: Commit**
+
+```bash
+git add app/schemas/accessory.py
+git commit -m "feat: add accessory Pydantic schemas"
+```
+
+---
+
+## Task 4: Accessory service layer — business logic
+
+**Files:**
+- Create: `services/fastapi/app/services/accessory_service.py`
+
+This service handles: work order query, accessory CRUD, shelving/off-shelf logic, and type management.
+
+**Step 1: Write the service**
+
+Key functions to implement:
+
+1. `query_work_orders(db, paichan_no)` — Query ERP for all work orders under a paichan_no, group by work_order_no, return `WorkOrderQueryResponse`.
+2. `list_accessories(db, paichan_no)` — List all accessories for a paichan_no, check `is_boxed` via `finished_goods_box_item`.
+3. `create_accessory(db, req)` — Create accessory record with shelving logic (same off-shelf rules as location_service).
+4. `update_accessory(db, accessory_id, req)` — Partial update of accessory_type/quantity/location_code.
+5. `delete_accessory(db, accessory_id)` — Delete if not boxed.
+6. `list_accessory_types(db)` / `create_accessory_type(db, req)` / `update_accessory_type(db, type_id, req)` / `delete_accessory_type(db, type_id)`
+
+**Error classes to define:**
+
+```python
+class InvalidPaichanError(Exception):
+    pass
+
+class PaichanNotFoundError(Exception):
+    pass
+
+class AccessoryNotFoundError(Exception):
+    pass
+
+class AccessoryAlreadyBoxedError(Exception):
+    pass
+
+class AccessoryDuplicateLocationError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+class AccessoryAlreadyOffShelfError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+```
+
+**Shelving logic** (in `create_accessory` when `location_code` is provided):
+
+- If the accessory row already has a `location_code`:
+  - If existing starts with `TRANS-` → raise `AccessoryAlreadyOffShelfError`
+  - If new target starts with `TRANS-` → update record (off-shelf / down-shelf)
+  - If both are normal locations → raise `AccessoryDuplicateLocationError`
+
+**is_boxed check**: Query `finished_goods_box_item` where `zongpai_no` matches and `item_type = 'accessory'` and there exists a `finished_goods_box` with `paichan_no` matching. This checks if the accessory has been packed into a box.
+
+**Step 2: Commit**
+
+```bash
+git add app/services/accessory_service.py
+git commit -m "feat: add accessory service layer with CRUD and shelving logic"
+```
+
+---
+
+## Task 5: Accessory API routes
+
+**Files:**
+- Create: `services/fastapi/app/api/v1/accessory.py`
+- Modify: `services/fastapi/app/main.py` — register new router + exception handlers
+
+**Step 1: Create the accessory router**
+
+```python
+# app/api/v1/accessory.py
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.schemas.accessory import (
+    AccessoryCreateRequest,
+    AccessoryListResponse,
+    AccessoryResponse,
+    AccessoryTypeCreateRequest,
+    AccessoryTypeItem,
+    AccessoryTypeListResponse,
+    AccessoryTypeUpdateRequest,
+    AccessoryUpdateRequest,
+    WorkOrderQueryResponse,
+)
+from app.services.accessory_service import (
+    create_accessory,
+    create_accessory_type,
+    delete_accessory,
+    delete_accessory_type,
+    list_accessories,
+    list_accessory_types,
+    update_accessory,
+    update_accessory_type,
+    query_work_orders,
+)
+
+router = APIRouter(tags=["accessory"])
+```
+
+Endpoints:
+
+| Method | Path | Handler |
+|---|---|---|
+| GET | `/accessory/work-orders` | `query_work_orders` |
+| GET | `/accessory` | `list_accessories` |
+| POST | `/accessory` | `create_accessory` |
+| PATCH | `/accessory/{id}` | `update_accessory` |
+| DELETE | `/accessory/{id}` | `delete_accessory` |
+| GET | `/accessory-type` | `list_accessory_types` |
+| POST | `/accessory-type` | `create_accessory_type` |
+| PATCH | `/accessory-type/{id}` | `update_accessory_type` |
+| DELETE | `/accessory-type/{id}` | `delete_accessory_type` |
+
+**Step 2: Register router + exception handlers in `main.py`**
+
+Add to `app/main.py`:
+
+```python
+from app.api.v1.accessory import router as accessory_router
+from app.services.accessory_service import (
+    AccessoryAlreadyBoxedError,
+    AccessoryAlreadyOffShelfError,
+    AccessoryDuplicateLocationError,
+    AccessoryNotFoundError,
+    InvalidPaichanError,
+    PaichanNotFoundError,
+)
+
+app.include_router(accessory_router, prefix="/CargoTrace")
+```
+
+Add exception handlers mirroring the existing pattern (400 for `InvalidPaichanError`, 404 for `PaichanNotFoundError`/`AccessoryNotFoundError`, 409 for duplicate/off-shelf/boxed errors).
+
+**Step 3: Commit**
+
+```bash
+git add app/api/v1/accessory.py app/main.py
+git commit -m "feat: add accessory API routes and exception handlers"
+```
+
+---
+
+## Task 6: Extend boxing module to support accessories
+
+**Files:**
+- Modify: `services/fastapi/app/services/box_service.py` — `get_box_info` returns `pending_accessories`; `save_box_record` handles `item_type=accessory`
+- Modify: `services/fastapi/app/schemas/box.py` — add `PendingAccessory`, `BoxSaveRequest.item_type`, `BoxSaveRequest.accessory_id`
+- Modify: `services/fastapi/app/api/v1/box.py` — no structural change, just forward new fields
+
+**Step 1: Extend box schemas**
+
+In `app/schemas/box.py`, add:
+
+```python
+class PendingAccessory(BaseModel):
+    accessory_id: int
+    accessory_type: str
+    quantity: int
+    location_code: str | None = None
+```
+
+Add to `BoxInfoResponse`:
+
+```python
+pending_accessories: list[PendingAccessory] = Field(default_factory=list)
+```
+
+Add to `BoxSaveRequest`:
+
+```python
+item_type: str = "product"
+accessory_id: int | None = None
+```
+
+**Step 2: Extend `get_box_info` in `box_service.py`**
+
+After building the existing response dict, query `finished_goods_accessory` for the same `zongpai_no` where `location_code IS NOT NULL` (shelved but not yet boxed). Add as `pending_accessories`.
+
+**Step 3: Extend `save_box_record` in `box_service.py`**
+
+When `item_type == "accessory"`:
+- Skip the ERP quantity validation (accessories don't have ERP quantities)
+- Use the `zongpai_no` and `paichan_no` from the `accessory_id` record
+- Create `FinishedGoodsBoxItem` with `item_type="accessory"`
+
+**Step 4: Commit**
+
+```bash
+git add app/schemas/box.py app/services/box_service.py
+git commit -m "feat: extend boxing module to support accessory items"
+```
+
+---
+
+## Task 7: FastAPI — integration test
+
+**Files:**
+- Create: `services/fastapi/tests/test_accessory.py`
+
+**Step 1: Write integration tests**
+
+Test the following scenarios against the running API:
+
+1. `GET /CargoTrace/accessory/work-orders?paichan_no=W00009` — returns work order groups
+2. `POST /CargoTrace/accessory` — create accessory with location
+3. `GET /CargoTrace/accessory?paichan_no=W00009` — list accessories
+4. `PATCH /CargoTrace/accessory/{id}` — update quantity
+5. `DELETE /CargoTrace/accessory/{id}` — delete
+6. `GET /CargoTrace/accessory-type` — list types
+7. `POST /CargoTrace/accessory-type` — create type
+8. Off-shelf scenario: create with normal location, then update to TRANS- location
+9. Error scenario: duplicate location, already off-shelf, invalid paichan
+
+**Step 2: Run tests**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+source .venv/Scripts/activate
+pytest tests/test_accessory.py -v
+```
+
+**Step 3: Commit**
+
+```bash
+git add tests/test_accessory.py
+git commit -m "test: add accessory module integration tests"
+```
+
+---
+
+## Task 8: Flutter — API service extension
+
+**Files:**
+- Modify: `apps/pad_scanner/lib/services/api_service.dart` — add accessory API methods
+
+**Step 1: Add accessory API methods to `ApiService`**
+
+Methods to add:
+
+```dart
+// Work order query
+Future<WorkOrderQueryResult> queryWorkOrders(String paichanNo)
+
+// Accessory CRUD
+Future<AccessoryResult> createAccessory(AccessoryCreateRequest req)
+Future<AccessoryListResult> listAccessories(String paichanNo)
+Future<AccessoryResult> updateAccessory(int id, AccessoryUpdateRequest req)
+Future<bool> deleteAccessory(int id)
+
+// Accessory types
+Future<List<AccessoryType>> listAccessoryTypes()
+Future<AccessoryType> createAccessoryType(String name, {int sortOrder = 0})
+Future<bool> deleteAccessoryType(int id)
+```
+
+Define corresponding result/model classes.
+
+**Step 2: Commit**
+
+```bash
+git add lib/services/api_service.dart
+git commit -m "feat: add accessory API methods to Flutter API service"
+```
+
+---
+
+## Task 9: Flutter — Accessory registration page
+
+**Files:**
+- Create: `apps/pad_scanner/lib/pages/accessory_page.dart`
+- Create: `apps/pad_scanner/lib/pages/accessory/` directory with part files for separation
+
+**Step 1: Create the main page structure**
+
+`accessory_page.dart` as StatefulWidget, following the same pattern as `registration_page.dart`:
+
+- Part files for scan handling, submit logic, status display, list management
+- State machine: `idle` → `paichanQueried` → `workOrderSelected` → `accessoryTypeSelected` → `readyToSubmit`
+
+**Step 2: Implement the step-by-step form**
+
+- Paichan_no input field + query button (hardware Enter key triggers query)
+- Work order dropdown (populated after query)
+- Zongpai_no auto-fill or dropdown (based on selected work order)
+- Accessory type selector (dropdown from presets + manual input option)
+- Quantity input field
+- Location code field (supports scanner input)
+- Submit button
+
+**Step 3: Implement the registered accessories list**
+
+Bottom half of the page, showing all accessories for the current paichan_no with inline edit/delete.
+
+**Step 4: Commit**
+
+```bash
+git add lib/pages/accessory_page.dart lib/pages/accessory/
+git commit -m "feat: add accessory registration page"
+```
+
+---
+
+## Task 10: Flutter — Update home page with accessory card
+
+**Files:**
+- Modify: `apps/pad_scanner/lib/pages/home_page.dart`
+
+**Step 1: Add accessory module card**
+
+Add a third card between "装箱编号" and "货架查询":
+
+```dart
+ModuleCard(
+  icon: Icons.build,
+  title: '附件登记',
+  description: '手动登记订单附件上架与装箱',
+  status: ModuleStatus.developing,
+  onTap: () => Navigator.pushNamed(context, '/accessory'),
+)
+```
+
+**Step 2: Register route in `main.dart`**
+
+```dart
+'/accessory': (context) => const AccessoryPage(),
+```
+
+**Step 3: Commit**
+
+```bash
+git add lib/pages/home_page.dart lib/main.dart
+git commit -m "feat: add accessory module card to home page"
+```
+
+---
+
+## Task 11: Flutter — Extend boxing page for accessories
+
+**Files:**
+- Modify: `apps/pad_scanner/lib/pages/boxing_page.dart` or relevant part files
+- Modify: `apps/pad_scanner/lib/pages/boxing/boxing_models.dart` — add `PendingAccessory` model
+
+**Step 1: Parse `pending_accessories` from box info API response**
+
+When `get_box_info` returns data, extract `pending_accessories` list.
+
+**Step 2: Add "待装箱附件" section in boxing UI**
+
+Below the existing assigned items list, show pending accessories with "加入本箱" buttons.
+
+**Step 3: Handle accessory boxing submission**
+
+When boxing an accessory, send `item_type: 'accessory'` and `accessory_id` in the request.
+
+**Step 4: Show accessory items in boxing detail page**
+
+In `boxing_detail_page.dart`, distinguish accessories from products (e.g., different row color or icon).
+
+**Step 5: Commit**
+
+```bash
+git add lib/pages/boxing_page.dart lib/pages/boxing/ lib/pages/boxing_detail_page.dart
+git commit -m "feat: extend boxing module to display and box accessories"
+```
+
+---
+
+## Task 12: Manual testing & polish
+
+**Step 1: Start FastAPI dev server**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+source .venv/Scripts/activate
+uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
+```
+
+**Step 2: Build and run Flutter app**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/apps/pad_scanner
+flutter run
+```
+
+**Step 3: Test the complete flow**
+
+1. Home page → 附件登记 card visible, tap to enter
+2. Input paichan_no → work orders load in dropdown
+3. Select work order → zongpai_no auto-fills
+4. Select accessory type → input quantity
+5. Scan location code → submit
+6. Verify accessory appears in registered list
+7. Go to boxing page → scan a zongpai_no that has accessories
+8. Verify "待装箱附件" section shows accessories
+9. Box an accessory → verify in detail page
+
+**Step 4: Commit any fixes**
+
+```bash
+git add -A
+git commit -m "fix: address issues found during manual testing"
+```
+
+---
+
+## Task 13: Final commit and merge to master
+
+**Step 1: In fastapi submodule**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/services/fastapi
+# Ensure all changes committed on dev
+git log --oneline master..dev
+# Merge dev into master
+git checkout master
+git merge dev
+git push
+```
+
+**Step 2: In pad_scanner submodule**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace/apps/pad_scanner
+git log --oneline master..dev
+git checkout master
+git merge dev
+git push
+```
+
+**Step 3: Update parent repo submodules**
+
+```bash
+cd D:/FileLib/Projects/CargoTrace
+git add apps/pad_scanner services/fastapi
+git commit -m "feat: add accessory registration module"
+git push
+```

docs/plans/2026-05-24-accessory-module.md

@@ -0,0 +1,646 @@
+# PRD · 模块三：附件登记
+
+**版本：** v1.0
+**状态：** 待实现
+**适用端：** 安卓扫码枪 Flutter 应用 + FastAPI 后端
+
+---
+
+## 1. 功能概述
+
+附件登记模块是为成品库房中**无执行卡的订单附件**提供信息化管理的独立模块。
+
+在实际作业中，产品的附件（安装配件、说明书、包装材料等）可能在主产品生产完成之前就运抵库房，需要临时存放。这些附件不携带执行卡，无法通过扫码获取总排号，因此需要通过手动输入排产号、选择工令号的方式定位到具体总排号，完成上架和装箱登记。
+
+本模块覆盖附件的**上架登记、下架操作、装箱处理**全流程，与现有的上架登记和装箱编号模块并行存在于首页导航。
+
+---
+
+## 2. 用户与使用场景
+
+**使用人员：** 成品库房工人
+
+**典型场景：**
+
+| 场景 | 描述 |
+|---|---|
+| 附件到货上架 | 附件先于产品到达库房，工人根据送货单上的排产号，手动输入后查询工令号，选择后登记到货位 |
+| 附件下架 | 已上架的附件需要转移到转运区域，与产品下架逻辑一致 |
+| 附件与产品混装 | 装箱时将附件与主产品装入同一箱号 |
+| 附件单独装箱 | 装箱时将附件单独装箱，关联到同一排产号 |
+| 附件类型管理 | 管理员在系统中预设附件类型列表，工人操作时从中点选 |
+
+---
+
+## 3. 设备能力约定
+
+与现有模块一致：
+
+| 项目 | 说明 |
+|---|---|
+| 扫码输出方式 | 广播输出（Android Intent） |
+| 屏幕尺寸 | 3.5 英寸，宽度 640px，高度 960px |
+| 键盘 | 31 键实体键盘 |
+| 网络 | WiFi / 蓝牙 / 蜂窝 |
+
+---
+
+## 4. 码值识别规则
+
+本模块中，货位号仍通过扫码获取，复用现有码值解析规则。排产号和工令号通过手动输入和选择操作。
+
+| 码值类型 | 获取方式 | 识别规则 |
+|---|---|---|
+| 排产号 | 手动输入 | 正则 `^[A-Z]{1,2}\d{5}(-J)?$` |
+| 工令号 | 从列表选择 | 由 ERP 查询返回 |
+| 总排号 | 系统自动关联 | 由 排产号+工令号 从 ERP 关联得出 |
+| 普通货位号 | 扫码或手动输入 | 正则 `^[A-Z]+\d+-\d+-\d+$` |
+| 转运特殊货位号 | 扫码或手动输入 | 以 `TRANS-` 开头 |
+| 附件类型 | 预设点选或自由输入 | 字符串，最长 64 字符 |
+
+---
+
+## 5. 数据模型
+
+### 5.1 新增表：附件记录表
+
+```sql
+CREATE TABLE "CargoTrace".finished_goods_accessory (
+    id              SERIAL        PRIMARY KEY,
+    paichan_no      VARCHAR(64)   NOT NULL,
+    zongpai_no      VARCHAR(64)   NOT NULL,
+    accessory_type  VARCHAR(64)   NOT NULL,
+    quantity        INT           NOT NULL,
+    location_code   VARCHAR(64)   NULL,
+    created_at      TIMESTAMP     NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX idx_fga_paichan_no     ON "CargoTrace".finished_goods_accessory (paichan_no);
+CREATE INDEX idx_fga_zongpai_no     ON "CargoTrace".finished_goods_accessory (zongpai_no);
+CREATE INDEX idx_fga_location_code  ON "CargoTrace".finished_goods_accessory (location_code);
+```
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| id | SERIAL | 主键 |
+| paichan_no | VARCHAR(64) | 排产号，工人手动输入 |
+| zongpai_no | VARCHAR(64) | 总排号，由排产号+工令号从 ERP 关联得出 |
+| accessory_type | VARCHAR(64) | 附件类型名称，来自预设列表或自由输入 |
+| quantity | INT | 附件数量，精确计量 |
+| location_code | VARCHAR(64) | 货位号，上架后填入；未上架时为 NULL |
+| created_at | TIMESTAMP | 创建时间 |
+
+**设计说明：**
+- 无唯一约束：同一总排号可登记多种附件，每种附件也可分多次登记。
+- `location_code` 为 NULL 表示未上架，有值表示已上架或已下架到转运区域。
+- 下架逻辑复用现有模式：`location_code` 以 `TRANS-` 开头即为已下架。
+
+### 5.2 新增表：附件类型预设表
+
+```sql
+CREATE TABLE "CargoTrace".accessory_type (
+    id          SERIAL        PRIMARY KEY,
+    name        VARCHAR(64)   NOT NULL,
+    sort_order  INT           NOT NULL DEFAULT 0,
+
+    CONSTRAINT uk_accessory_type_name UNIQUE (name)
+);
+```
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| id | SERIAL | 主键 |
+| name | VARCHAR(64) | 类型名称，唯一 |
+| sort_order | INT | 排序权重，值越小越靠前 |
+
+### 5.3 现有表变更：装箱明细表
+
+在 `finished_goods_box_item` 表新增 `item_type` 字段：
+
+```sql
+ALTER TABLE "CargoTrace".finished_goods_box_item
+ADD COLUMN item_type VARCHAR(16) NOT NULL DEFAULT 'product';
+
+COMMENT ON COLUMN "CargoTrace".finished_goods_box_item.item_type
+IS '条目类型：product = 产品，accessory = 附件';
+```
+
+---
+
+## 6. 页面设计
+
+### 6.1 首页更新
+
+在现有首页功能卡片列表中，在"装箱编号"卡片下方新增"附件登记"卡片：
+
+```
+┌──────────────────────────────┐
+│  📦 上架登记                 │  ← 已上线
+│  ────────────────────────    │
+│  📋 装箱编号                 │  ← 已上线
+│  ────────────────────────    │
+│  🔧 附件登记                 │  ← 新增，开发中
+│  扫码或手动登记订单附件       │
+│                   开发中 ○    │
+│  ────────────────────────    │
+│  🔍 货架查询                 │  ← 规划中（已并入总览）
+│                   规划中 ○    │
+└──────────────────────────────┘
+```
+
+### 6.2 附件登记主页面
+
+页面分为上下两个区域：上方为操作区，下方为已登记附件列表。整体可滚动。
+
+**初始状态（等待输入）：**
+
+```
+┌──────────────────────────────────┐
+│ 附件登记                          │
+├──────────────────────────────────┤
+│ ▌ 操作区                         │
+│                                  │
+│  排产号 [            ] [查询]     │  ← 手动输入框 + 查询按钮
+│                                  │
+│  工令号  （请先查询排产号）        │  ← 置灰，查询后填充下拉列表
+│  总排号  （请先选择工令号）        │  ← 置灰，选择工令号后自动填入
+│  附件类型 ▼ [                  ]  │  ← 置灰，选择总排号后激活
+│  数量   [                  ]      │  ← 置灰，选择附件类型后激活
+│  货位号 [                  ]      │  ← 置灰，输入数量后激活，支持扫码
+│                                  │
+│  ┌────────────────────────┐      │
+│  │      确 认 上 架        │      │  ← 置灰，全部填入后激活
+│  └────────────────────────┘      │
+│                                  │
+├ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┤
+│ ▌ 已登记附件                     │  ← 初始隐藏，查询排产号后展示
+│ （请先查询排产号）                │
+│                                  │
+├──────────────────────────────────┤
+│  ● 等待输入排产号…               │
+└──────────────────────────────────┘
+```
+
+**查询排产号后的状态：**
+
+```
+┌──────────────────────────────────┐
+│ 附件登记                          │
+├──────────────────────────────────┤
+│ ▌ 操作区                         │
+│                                  │
+│  排产号 [ W00009        ] [查询]  │
+│                                  │
+│  工令号 ▼ [ 6-1(7)           ]   │  ← 下拉展示该排产号下所有工令号
+│                                  │  │  可点选
+│  总排号  26BW0011                 │  ← 自动填入（工令号选定后关联）
+│                                  │
+│  附件类型 ▼ [ 安装配件        ]   │  ← 下拉预设 + 自由输入
+│  数量   [ 5                ]      │
+│  货位号 [ A01-02-03       ]      │  ← 扫码或手动输入
+│                                  │
+│  ┌────────────────────────┐      │
+│  │      确 认 上 架        │      │
+│  └────────────────────────┘      │
+│                                  │
+├ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┤
+│ ▌ 已登记附件 · W00009            │
+│ ┌────────┬──────────┬────┬──────┬──────────┐
+│ │总排号  │附件类型   │数量│货位号│操作      │
+│ ├────────┼──────────┼────┼──────┼──────────┤
+│ │26BW0011│安装配件   │ 5  │A01-02│ [改][删] │
+│ │26BW0011│说明书     │ 10 │A01-02│ [改][删] │
+│ │26BW0010│包装材料   │ 2  │—     │ [改][删] │  ← 未上架，货位显示 —
+│ └────────┴──────────┴────┴──────┴──────────┘
+│                                  │
+├──────────────────────────────────┤
+│  ● 已填入，请确认或继续编辑       │
+└──────────────────────────────────┘
+```
+
+### 6.3 字段联动规则
+
+操作区采用**逐步解锁**模式，每一步完成后才激活下一步：
+
+| 步骤 | 操作 | 激活条件 | 说明 |
+|---|---|---|---|
+| 1 | 输入排产号并查询 | 无 | 手动输入排产号，点击查询或按回车 |
+| 2 | 选择工令号 | 排产号查询成功 | 下拉列表展示该排产号下所有去重工令号 |
+| 3 | 总排号自动填入 | 工令号选定 | 系统根据排产号+工令号从 ERP 关联总排号；若只有1个则自动填入，若多个则展示下拉供选择 |
+| 4 | 选择附件类型 | 总排号已填入 | 下拉展示预设类型列表，支持手动输入自定义类型 |
+| 5 | 输入数量 | 附件类型已填入 | 手动输入正整数 |
+| 6 | 输入货位号 | 数量已填入 | 扫码或手动输入，复用现有货位号格式校验 |
+| 7 | 确认上架 | 全部字段有效 | 提交到后端 |
+
+### 6.4 已登记列表
+
+已登记列表在首次查询排产号成功后展示，展示该排产号下所有已登记的附件记录。
+
+**列定义：**
+
+| 列 | 内容 | 说明 |
+|---|---|---|
+| 总排号 | zongpai_no | 该附件关联的总排号 |
+| 附件类型 | accessory_type | 类型名称 |
+| 数量 | quantity | 登记的附件数量 |
+| 货位号 | location_code | 已上架显示货位号，未上架显示 "—"，转运显示 TRANS-xx |
+| 操作 | [改] [删] | 行内编辑和删除 |
+
+**行内编辑（点击 [改]）：**
+
+```
+│ 26BW0011 │ [安装配件] │ [5] │ A01-02 │ [✓][✗] │
+```
+
+- 仅附件类型和数量可编辑，总排号和货位号不可修改
+- 点 [✓] 调用 PATCH 接口保存
+- 点 [✗] 取消恢复原值
+
+**行内删除（点击 [删]）：**
+
+```
+│ 确认删除此条附件记录？ [是] [否]
+```
+
+- 确认后调用 DELETE 接口删除
+
+**已装箱附件标记：** 已完成装箱的附件记录在列表中显示 `[已装箱]` 标签，不支持编辑和删除。
+
+---
+
+## 7. 功能详细说明
+
+### 7.1 操作流程图
+
+```mermaid
+flowchart TD
+    A([进入附件登记页]) --> B[等待输入排产号]
+    B --> C[手动输入排产号\n点击查询或回车]
+    C --> D{排产号格式校验}
+    D -->|无效| E[底部红条提示：排产号格式无效]
+    E --> B
+    D -->|有效| F[调用后端查询工令号列表]
+    F --> G{查询结果}
+    G -->|未找到| H[底部红条提示：未找到该排产号信息]
+    H --> B
+    G -->|成功| I[工令号下拉列表填充\n已登记列表刷新]
+
+    I --> J[选择工令号]
+    J --> K{工令号关联的总排号}
+    K -->|仅1个| L[自动填入总排号]
+    K -->|多个| M[总排号下拉供选择]
+
+    L --> N[选择/输入附件类型]
+    M --> N
+    N --> O[输入数量]
+    O --> P[扫描或输入货位号]
+    P --> Q{全部字段有效?}
+    Q -->|否| R[相应字段标红提示]
+    Q -->|是| S[确认按钮激活]
+
+    S --> T[点击确认或回车提交]
+    T --> U[调用后端保存接口]
+    U --> V{结果}
+    V -->|成功| W[成功反馈\n已登记列表刷新\n操作区部分重置\n保留排产号和工令号选择]
+    V -->|失败| X[保留数据\n底部状态栏显示错误]
+
+    W --> Y{是否继续登记?}
+    Y -->|同排产号继续| N
+    Y -->|换排产号| B
+
+    P --> P2{货位号类型判断}
+    P2 -->|现有记录在普通货架\n且目标为转运货架| DOWN[下架操作\n后端更新记录]
+    P2 -->|现有记录在转运货架| ERR[错误：已下架不可重新上架]
+```
+
+### 7.2 下架逻辑
+
+附件的下架逻辑与产品完全一致：
+
+| 现有状态 | 目标货位 | 结果 |
+|---|---|---|
+| 未上架（location_code 为 NULL） | 普通货架 | 上架成功，新建 location_code |
+| 未上架 | 转运货架 | 直接转运成功 |
+| 普通货架 | 普通货架 | 错误：该附件已有货位记录 |
+| 普通货架 | 转运货架 | 下架成功，更新 location_code |
+| 转运货架 | 任何 | 错误：已下架不可重新上架 |
+
+### 7.3 已登记列表刷新策略
+
+| 时机 | 行为 |
+|---|---|
+| 排产号查询成功 | 加载该排产号下全部附件记录 |
+| 上架/下架提交成功 | 刷新列表 |
+| 编辑/删除操作完成 | 刷新列表 |
+| 切换排产号 | 替换为新排产号的记录 |
+
+---
+
+## 8. 装箱集成
+
+附件的装箱操作在**现有装箱编号模块**中完成，附件登记模块只负责上架和下架。
+
+### 8.1 装箱模块改动
+
+#### 8.1.1 接口响应扩展
+
+`GET /CargoTrace/box/info` 接口的响应新增 `pending_accessories` 字段：
+
+```json
+{
+  "zongpai_no": "26BW0011",
+  "paichan_no": "W00009",
+  "work_order_no": "6-1(7)",
+  "quantity": 80,
+  "current_zongpai_boxes": [...],
+  "existing_boxes": [...],
+  "max_box_no": 5,
+  "suggested_box_no": 6,
+  "pending_accessories": [
+    {
+      "accessory_id": 1,
+      "accessory_type": "安装配件",
+      "quantity": 5,
+      "location_code": "A01-02-03"
+    },
+    {
+      "accessory_id": 2,
+      "accessory_type": "说明书",
+      "quantity": 10,
+      "location_code": "TRANS-01"
+    }
+  ]
+}
+```
+
+#### 8.1.2 装箱页面展示
+
+在装箱页面已分配列表下方，新增"待装箱附件"区域：
+
+```
+┌──────────────────────────────┐
+│ ...已分配产品列表...          │
+├──────────────────────────────┤
+│ 待装箱附件（2 项）            │
+│ 安装配件 × 5     [加入本箱]   │
+│ 说明书     × 10   [加入本箱]   │
+└──────────────────────────────┘
+```
+
+- 工人点击 [加入本箱]，将附件作为一条装箱明细写入 `finished_goods_box_item`，`item_type` 为 `accessory`
+- 附件在装箱详情页中通过不同的行样式或标签区分
+
+#### 8.1.3 装箱记录保存
+
+`POST /CargoTrace/box` 接口的请求体新增可选字段：
+
+```json
+{
+  "zongpai_no": "26BW0011",
+  "box_no": 3,
+  "quantity": 80,
+  "item_type": "accessory",
+  "accessory_id": 1
+}
+```
+
+- `item_type`：默认为 `product`，附件装箱时传 `accessory`
+- `accessory_id`：当 `item_type` 为 `accessory` 时必填，关联附件记录
+
+---
+
+## 9. 接口依赖
+
+### 9.1 工令号查询接口（新增）
+
+**GET** `/CargoTrace/accessory/work-orders?paichan_no={paichan_no}`
+
+根据排产号查询该排产号下所有去重的工令号及其对应的总排号。
+
+**请求参数：**
+
+| 参数 | 类型 | 位置 | 说明 |
+|---|---|---|---|
+| paichan_no | string | Query | 排产号 |
+
+**成功响应 200：**
+
+```json
+{
+  "paichan_no": "W00009",
+  "work_orders": [
+    {
+      "work_order_no": "6-1(7)",
+      "zongpai_nos": ["26BW0010", "26BW0011", "26BW0012"]
+    },
+    {
+      "work_order_no": "6-2(3)",
+      "zongpai_nos": ["26BW0013", "26BW0014"]
+    }
+  ]
+}
+```
+
+**失败响应：**
+
+| HTTP 状态码 | 错误码 | 含义 |
+|---|---|---|
+| 400 | INVALID_PAICHAN | 排产号格式不合法 |
+| 404 | PAICHA_NOT_FOUND | 未找到该排产号 |
+
+### 9.2 附件登记接口（新增）
+
+**POST** `/CargoTrace/accessory`
+
+**请求体：**
+
+```json
+{
+  "paichan_no": "W00009",
+  "zongpai_no": "26BW0011",
+  "accessory_type": "安装配件",
+  "quantity": 5,
+  "location_code": "A01-02-03"
+}
+```
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| paichan_no | string | 是 | 排产号 |
+| zongpai_no | string | 是 | 总排号 |
+| accessory_type | string | 是 | 附件类型名称 |
+| quantity | int | 是 | 数量，必须为正整数 |
+| location_code | string | 否 | 货位号，为空表示仅登记不上架 |
+
+**成功响应 200：**
+
+```json
+{
+  "id": 1,
+  "paichan_no": "W00009",
+  "zongpai_no": "26BW0011",
+  "accessory_type": "安装配件",
+  "quantity": 5,
+  "location_code": "A01-02-03",
+  "created_at": "2026-05-24T10:30:00"
+}
+```
+
+### 9.3 附件记录查询接口（新增）
+
+**GET** `/CargoTrace/accessory?paichan_no={paichan_no}`
+
+查询指定排产号下所有附件记录。
+
+**成功响应 200：**
+
+```json
+{
+  "paichan_no": "W00009",
+  "items": [
+    {
+      "id": 1,
+      "zongpai_no": "26BW0011",
+      "accessory_type": "安装配件",
+      "quantity": 5,
+      "location_code": "A01-02-03",
+      "is_boxed": false,
+      "created_at": "2026-05-24T10:30:00"
+    }
+  ]
+}
+```
+
+### 9.4 附件记录更新接口（新增）
+
+**PATCH** `/CargoTrace/accessory/{id}`
+
+**请求体：**
+
+```json
+{
+  "accessory_type": "安装配件",
+  "quantity": 8,
+  "location_code": "A01-02-04"
+}
+```
+
+支持部分更新，仅传需要修改的字段。
+
+### 9.5 附件记录删除接口（新增）
+
+**DELETE** `/CargoTrace/accessory/{id}`
+
+已装箱的附件不可删除，返回 `409 ALREADY_BOXED`。
+
+### 9.6 附件类型管理接口（新增）
+
+**GET** `/CargoTrace/accessory-type**
+
+返回所有预设附件类型，按 sort_order 排序。
+
+```json
+{
+  "types": [
+    {"id": 1, "name": "安装配件", "sort_order": 0},
+    {"id": 2, "name": "说明书", "sort_order": 1},
+    {"id": 3, "name": "包装材料", "sort_order": 2}
+  ]
+}
+```
+
+**POST** `/CargoTrace/accessory-type` — 新增类型
+
+**PATCH** `/CargoTrace/accessory-type/{id}` — 修改类型名称或排序
+
+**DELETE** `/CargoTrace/accessory-type/{id}` — 删除类型
+
+---
+
+## 10. 异常与错误处理
+
+| 异常情况 | 触发条件 | 系统表现 |
+|---|---|---|
+| 排产号格式无效 | 不符合正则 | 底部红条提示"排产号格式无效" |
+| 排产号未找到 | ERP 无记录 | 底部红条提示"未找到该排产号信息" |
+| 附件类型为空 | 未选择或输入 | 提示"请选择或输入附件类型" |
+| 数量无效 | 非正整数 | 提示"数量必须为正整数" |
+| 货位号格式无效 | 不符合货位号规则 | 底部红条提示"无效的货位号" |
+| 附件已有货位 | 重复上架普通货架 | 错误弹窗，显示已有货位信息 |
+| 已下架附件 | 转运货架→任何 | 底部红条提示"该附件已下架不可重新上架" |
+| 已装箱附件 | 尝试编辑/删除已装箱记录 | 底部红条提示"该附件已装箱不可操作" |
+| 网络异常 | 超时或断网 | 底部黄条提示"网络异常，请检查网络连接" |
+
+---
+
+## 11. 反馈机制
+
+复用现有统一反馈标准：
+
+| 事件 | 屏幕 | 声音 | 振动 |
+|---|---|---|---|
+| 排产号查询成功 | 工令号下拉填充，绿色高亮 | 短促提示音 | 短震 |
+| 上架成功 | 底部状态栏绿色提示 1.5 秒 | 成功音 | 长震 |
+| 下架成功 | 底部状态栏绿色提示"下架成功" | 成功音 | 长震 |
+| 操作失败 | 底部状态栏红色提示 2 秒 | 错误音 | 短震 |
+| 网络异常 | 底部状态栏黄色提示 | 失败音 | 短震 |
+
+---
+
+## 12. 非功能需求
+
+| 项目 | 要求 |
+|---|---|
+| 工令号查询响应时间 | ≤ 500ms |
+| 附件保存响应时间 | ≤ 500ms |
+| 扫码到货位号填入延迟 | ≤ 100ms |
+| 离线处理策略 | 断网时提示检查网络，数据保留在页面 |
+| 软键盘控制 | 附件类型选择弹出选择面板时使用应用内 UI，不依赖系统软键盘 |
+
+---
+
+## 13. 项目结构变更
+
+### 13.1 Flutter 端
+
+```
+lib/
+├── pages/
+│   ├── accessory_page.dart          # 新增：附件登记主页面
+│   ├── boxing_page.dart             # 修改：增加附件装箱展示
+│   └── boxing_detail_page.dart      # 修改：附件装箱明细展示
+├── services/
+│   └── api_service.dart             # 修改：新增附件相关 API 调用
+└── widgets/
+    └── accessory_type_selector.dart # 新增：附件类型选择器组件
+```
+
+### 13.2 FastAPI 端
+
+```
+app/
+├── api/v1/
+│   ├── accessory.py                 # 新增：附件登记 API
+│   ├── accessory_type.py            # 新增：附件类型管理 API
+│   └── box.py                       # 修改：装箱接口支持附件
+├── models/
+│   ├── finished_goods.py            # 修改：新增附件模型
+│   └── accessory_type.py            # 新增：附件类型模型
+├── schemas/
+│   ├── accessory.py                 # 新增：附件请求/响应 Schema
+│   └── box.py                       # 修改：装箱 Schema 支持附件
+└── services/
+    ├── accessory_service.py         # 新增：附件业务逻辑
+    └── box_service.py               # 修改：装箱逻辑支持附件
+```
+
+---
+
+## 14. 超出当前版本范围
+
+以下内容在当前版本中不实现：
+
+- 附件到货数量与 ERP BOM 对比校验（附件数量是否与订单需求匹配）
+- 附件条码标签生成（为附件打印临时标签）
+- 附件库存盘点功能
+- 附件出入库历史查询

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

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,32 @@ 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",
+    ]
     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(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_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):
+def test_register_location_duplicate_normal_to_normal(client: TestClient):
-    """重复上架 — 应返回 409，包含统一错误格式"""
+    """普通货位重复上架到普通货位 — 应返回 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"