feat(server): add attachment schemas and GET /attachment/categories

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

app/api/v1/attachment.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.models.attachment import AttachmentCategory
+from app.schemas.attachment import CategoryResponse
+
+router = APIRouter(tags=["attachment"])
+
+
+@router.get("/attachment/categories", response_model=list[CategoryResponse])
+def list_categories(db: Session = Depends(get_db)):
+    """列出启用中的附件类型，按 sort_order、name 排序。"""
+    rows = (
+        db.query(AttachmentCategory)
+        .filter(AttachmentCategory.is_active == True)
+        .order_by(AttachmentCategory.sort_order, AttachmentCategory.name)
+        .all()
+    )
+    return [
+        CategoryResponse(id=r.id, name=r.name, sort_order=r.sort_order) for r in rows
+    ]

app/main.py

@@ -18,6 +18,7 @@ except Exception as e:
 
 from app.api.v1.location import router as location_router
 from app.api.v1.box import router as box_router
+from app.api.v1.attachment import router as attachment_router
 from app.services.location_service import (
     AlreadyOffShelfError,
     DuplicateLocationError,
@@ -36,6 +37,7 @@ app = FastAPI(title="CargoTrace API", version="0.1.0")
 
 app.include_router(location_router, prefix="/CargoTrace")
 app.include_router(box_router, prefix="/CargoTrace")
+app.include_router(attachment_router, prefix="/CargoTrace")
 
 
 @app.exception_handler(DuplicateLocationError)

app/schemas/attachment.py

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

tests/test_attachment_api.py

@@ -0,0 +1,30 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models.attachment import AttachmentCategory
+
+
+def _seed_categories(db: Session) -> list[int]:
+    cats = [
+        AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True),
+        AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True),
+        AttachmentCategory(name="TEST_OFF", sort_order=3, is_active=False),
+    ]
+    for c in cats:
+        db.add(c)
+    db.commit()
+    for c in cats:
+        db.refresh(c)
+    return [c.id for c in cats]
+
+
+def test_categories_lists_active_only(client: TestClient, db: Session):
+    _seed_categories(db)
+    resp = client.get("/CargoTrace/attachment/categories")
+    assert resp.status_code == 200
+    names = [c["name"] for c in resp.json()]
+    assert "TEST_CERT" in names
+    assert "TEST_EXTRA" in names
+    assert "TEST_OFF" not in names
+    # sorted by sort_order then name
+    assert names.index("TEST_CERT") < names.index("TEST_EXTRA")