64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.finished_goods import FinishedGoodsLocation
|
|
from app.schemas.location import LocationRequest
|
|
|
|
|
|
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
|
|
|
|
|
|
def _is_transit_location(location_code: str) -> bool:
|
|
return location_code.startswith("TRANS-")
|
|
|
|
|
|
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
|