feat(server): add attachment ORM models and table provisioning

Adds 5 tables in the CargoTrace schema: attachment_category,
finished_goods_attachment, finished_goods_attachment_item,
finished_goods_attachment_location, finished_goods_box_attachment_item.
Includes a one-off create_all script and extends test cleanup.

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

app/models/__init__.py

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

app/models/attachment.py

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

scripts/create_attachment_tables.py

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

tests/conftest.py

@@ -34,7 +34,36 @@ def cleanup_test_data(db: Session):
         "26T998",
         "26B998",
         "26C998",
+        "26BW0011",
+        "26BW0012",
     ]
+
+    # Attachment tables (order: children before parents / before shared box)
+    from app.models.attachment import (
+        AttachmentCategory,
+        FinishedGoodsAttachment,
+        FinishedGoodsAttachmentItem,
+        FinishedGoodsAttachmentLocation,
+        FinishedGoodsBoxAttachmentItem,
+    )
+
+    db.query(FinishedGoodsBoxAttachmentItem).filter(
+        FinishedGoodsBoxAttachmentItem.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachmentLocation).filter(
+        FinishedGoodsAttachmentLocation.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachmentItem).filter(
+        FinishedGoodsAttachmentItem.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(FinishedGoodsAttachment).filter(
+        FinishedGoodsAttachment.zongpai_no.in_(test_zongpai_nos)
+    ).delete(synchronize_session=False)
+    db.query(AttachmentCategory).filter(
+        AttachmentCategory.name.like("TEST_%")
+    ).delete(synchronize_session=False)
+
+    # Existing product cleanup (unchanged) — attachment box items for these boxes:
     db.query(FinishedGoodsLocation).filter(
         FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos)
     ).delete(synchronize_session=False)
@@ -48,6 +77,9 @@ def cleanup_test_data(db: Session):
     )
     test_box_ids = [box.id for box in test_boxes]
     if test_box_ids:
+        db.query(FinishedGoodsBoxAttachmentItem).filter(
+            FinishedGoodsBoxAttachmentItem.box_id.in_(test_box_ids)
+        ).delete(synchronize_session=False)
         db.query(FinishedGoodsBoxItem).filter(
             FinishedGoodsBoxItem.box_id.in_(test_box_ids)
         ).delete(synchronize_session=False)

tests/test_attachment_models_smoke.py

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