chore: ignore agent workspace

Add box_mode field to BoxSaveRequest and DuplicateZongpaiBindError
exception. In one-to-one mode, prevent the same zongpai from being
packed into multiple boxes, returning 409 with existing box info.

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

.gitignore

@@ -6,4 +6,10 @@ __pycache__/
 dist/
 build/
 .pytest_cache/
-.claude/
+.claude/
+.agents/
+# Config with secrets
+config/settings.yaml
+config/settings.local.yaml
+
+.runtime

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/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,11 +1,25 @@
+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.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError
+from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError, DuplicateZongpaiBindError
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
@@ -92,6 +106,20 @@ async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
     )
 
 
+@app.exception_handler(DuplicateZongpaiBindError)
+async def duplicate_zongpai_bind_handler(request: Request, exc: DuplicateZongpaiBindError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_ZONGPAI_BIND",
+            "message": f"总排号 {exc.zongpai_no} 已绑定箱号 {exc.box_no}，请勿重复装箱",
+            "paichan_no": exc.paichan_no,
+            "box_no": exc.box_no,
+            "zongpai_no": exc.zongpai_no,
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}

app/models/finished_goods.py

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

app/schemas/box.py

@@ -8,6 +8,7 @@ ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
 class BoxItemDetail(BaseModel):
     """箱号下某个总排号的明细"""
     zongpai_no: str
+    work_order_no: str | None = None
     quantity: int
 
 
@@ -21,6 +22,7 @@ class BoxInfoResponse(BaseModel):
     """GET /box/info 响应"""
     zongpai_no: str
     paichan_no: str
+    work_order_no: str | None = None
     quantity: int
     existing_boxes: list[BoxDetail]
     max_box_no: int
@@ -32,6 +34,7 @@ class BoxSaveRequest(BaseModel):
     zongpai_no: str
     box_no: int
     quantity: int
+    box_mode: str | None = None  # one-to-one | one-to-many | many-to-one
 
     @field_validator("zongpai_no")
     @classmethod

app/services/box_service.py

@@ -1,6 +1,6 @@
 import re
 
-from sqlalchemy import text
+from sqlalchemy import bindparam, text
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
@@ -23,13 +23,16 @@ class DuplicateBoxItemError(Exception):
         self.box_no = box_no
 
 
+class DuplicateZongpaiBindError(Exception):
+    def __init__(self, paichan_no: str, box_no: int, zongpai_no: str):
+        self.paichan_no = paichan_no
+        self.box_no = box_no
+        self.zongpai_no = zongpai_no
+
+
 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,9 +40,51 @@ 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_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, str | None]:
+    """批量查询总排号对应的工令号。"""
+    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]: row[1] for row in rows}
+
+
 def get_box_info(db: Session, zongpai_no: str) -> dict:
     """获取装箱信息：排产号、数量、已有箱号明细。"""
     zongpai_no = zongpai_no.strip().upper()
@@ -56,18 +101,31 @@ 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()
         )
+        box_items[box.id] = items
+        item_zongpai_nos.extend(item.zongpai_no for item in items)
+
+    work_order_map = query_erp_work_order_map(db, item_zongpai_nos)
+
+    existing_boxes = []
+    max_box_no = 0
+    for box in boxes:
+        items = box_items[box.id]
         existing_boxes.append({
             "box_no": box.box_no,
             "items": [
-                {"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)}
+                {
+                    "zongpai_no": item.zongpai_no,
+                    "work_order_no": work_order_map.get(item.zongpai_no),
+                    "quantity": int(item.quantity or 0),
+                }
                 for item in items
             ],
         })
@@ -77,6 +135,7 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
     return {
         "zongpai_no": zongpai_no,
         "paichan_no": paichan_no,
+        "work_order_no": erp["work_order_no"],
         "quantity": quantity,
         "existing_boxes": existing_boxes,
         "max_box_no": max_box_no,
@@ -89,6 +148,19 @@ def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
     erp = query_erp_info(db, req.zongpai_no)
     paichan_no = erp["paichan_no"]
 
+    if req.box_mode == "one-to-one":
+        existing_box_no = (
+            db.query(FinishedGoodsBox.box_no)
+            .join(FinishedGoodsBoxItem, FinishedGoodsBox.id == FinishedGoodsBoxItem.box_id)
+            .filter(
+                FinishedGoodsBox.paichan_no == paichan_no,
+                FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
+            )
+            .scalar()
+        )
+        if existing_box_no is not None:
+            raise DuplicateZongpaiBindError(paichan_no, existing_box_no, req.zongpai_no)
+
     box = (
         db.query(FinishedGoodsBox)
         .filter(

config/settings.py

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

config/settings.yaml

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

config/settings.yaml.example

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

requirements.txt

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

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,6 +13,8 @@ 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 "existing_boxes" in data
     assert "max_box_no" in data

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"