From d9e6d384667a6520b6f81263b2b6a793b759988b Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 15 May 2026 08:45:18 +0800 Subject: [PATCH] feat(location): add paicha overview endpoint --- app/api/v1/location.py | 26 ++++++- app/main.py | 17 ++++- app/schemas/location.py | 14 ++++ app/services/location_service.py | 112 +++++++++++++++++++++++++++++++ tests/test_location_overview.py | 106 +++++++++++++++++++++++++++++ 5 files changed, 272 insertions(+), 3 deletions(-) create mode 100644 tests/test_location_overview.py diff --git a/app/api/v1/location.py b/app/api/v1/location.py index f068e0a..9923960 100644 --- a/app/api/v1/location.py +++ b/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.services.location_service import register_location +from app.schemas.location import ( + 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) diff --git a/app/main.py b/app/main.py index c7b4b6c..52b71a3 100644 --- a/app/main.py +++ b/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( diff --git a/app/schemas/location.py b/app/schemas/location.py index 4b9af67..a43514e 100644 --- a/app/schemas/location.py +++ b/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 diff --git a/app/services/location_service.py b/app/services/location_service.py index bd9f330..d7402c3 100644 --- a/app/services/location_service.py +++ b/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) diff --git a/tests/test_location_overview.py b/tests/test_location_overview.py new file mode 100644 index 0000000..d60ef1b --- /dev/null +++ b/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"