feat(location): add paicha overview endpoint

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()
 
@@ -31,3 +35,21 @@ def create_location(req: LocationRequest, db: Session = Depends(get_db)):
         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/main.py

@@ -18,7 +18,11 @@ 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.services.location_service import AlreadyOffShelfError, DuplicateLocationError
+from app.services.location_service import (
+    AlreadyOffShelfError,
+    DuplicateLocationError,
+    PaichaNotFoundError,
+)
 from app.services.box_service import (
     BoxItemNotFoundError,
     DuplicateBoxItemError,
@@ -137,6 +141,17 @@ async def box_item_not_found_handler(request: Request, exc: BoxItemNotFoundError
     )
 
 
+@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(

app/schemas/location.py

@@ -35,6 +35,20 @@ class LocationResponse(BaseModel):
     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):
     existing_location: str
     created_at: str

app/services/location_service.py

@@ -1,9 +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):
@@ -22,10 +27,117 @@ class AlreadyOffShelfError(Exception):
         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)

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"