fastAPI/app/services/location_service.py

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):
    """总排号已存在货位记录。"""

    def __init__(self, location_code: str, registered_at: str):
        self.location_code = location_code
        self.registered_at = registered_at


class AlreadyOffShelfError(Exception):
    """总排号已下架至转运区域。"""

    def __init__(self, location_code: str, registered_at: str):
        self.location_code = location_code
        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)
        .filter(FinishedGoodsLocation.zongpai_no == req.zongpai_no)
        .first()
    )
    if existing:
        if _is_transit_location(existing.location_code):
            raise AlreadyOffShelfError(
                location_code=existing.location_code,
                registered_at=existing.created_at.isoformat(),
            )

        if _is_transit_location(req.location_code):
            previous_location = existing.location_code
            existing.location_code = req.location_code
            existing.created_at = datetime.now()
            db.commit()
            db.refresh(existing)
            existing.previous_location = previous_location
            return existing

        raise DuplicateLocationError(
            location_code=existing.location_code,
            registered_at=existing.created_at.isoformat(),
        )

    record = FinishedGoodsLocation(
        zongpai_no=req.zongpai_no,
        location_code=req.location_code,
    )
    db.add(record)
    db.commit()
    db.refresh(record)
    return record