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

# 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
```