feat: add accessory service layer with CRUD and shelving logic
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Misaka
339
app/services/accessory_service.py
Normal file
339
app/services/accessory_service.py
Normal file
@@ -0,0 +1,339 @@
|
||||
import re
|
||||
from collections import defaultdict
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.accessory import AccessoryType, FinishedGoodsAccessory
|
||||
from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
|
||||
from app.schemas.accessory import (
|
||||
AccessoryCreateRequest,
|
||||
AccessoryTypeCreateRequest,
|
||||
AccessoryTypeUpdateRequest,
|
||||
AccessoryUpdateRequest,
|
||||
)
|
||||
|
||||
PAICHAN_PATTERN = re.compile(r"^[A-Z]{1,2}\d{5}(-J)?$")
|
||||
|
||||
|
||||
# ── Error classes ──
|
||||
|
||||
|
||||
class InvalidPaichanError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class PaichanNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class AccessoryNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class AccessoryAlreadyBoxedError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class AccessoryDuplicateLocationError(Exception):
|
||||
def __init__(self, location_code: str, registered_at: str):
|
||||
self.location_code = location_code
|
||||
self.registered_at = registered_at
|
||||
|
||||
|
||||
class AccessoryAlreadyOffShelfError(Exception):
|
||||
def __init__(self, location_code: str, registered_at: str):
|
||||
self.location_code = location_code
|
||||
self.registered_at = registered_at
|
||||
|
||||
|
||||
class AccessoryTypeNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class DuplicateAccessoryTypeNameError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
# ── Helpers ──
|
||||
|
||||
|
||||
def _is_transit_location(location_code: str) -> bool:
|
||||
return location_code.startswith("TRANS-")
|
||||
|
||||
|
||||
def _erp_paicha_items_sql(db: Session) -> str:
|
||||
"""Reuse the same ERP query pattern as location_service."""
|
||||
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 _is_accessory_boxed(db: Session, accessory: FinishedGoodsAccessory) -> bool:
|
||||
"""Check whether the accessory is already boxed (EXISTS in box_item as 'accessory')."""
|
||||
boxed = (
|
||||
db.query(FinishedGoodsBoxItem)
|
||||
.join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
|
||||
.filter(
|
||||
FinishedGoodsBoxItem.zongpai_no == accessory.zongpai_no,
|
||||
FinishedGoodsBox.paichan_no == accessory.paichan_no,
|
||||
FinishedGoodsBoxItem.item_type == "accessory",
|
||||
)
|
||||
.first()
|
||||
)
|
||||
return boxed is not None
|
||||
|
||||
|
||||
def _accessory_to_dict(
|
||||
accessory: FinishedGoodsAccessory, is_boxed: bool = False
|
||||
) -> dict:
|
||||
return {
|
||||
"id": accessory.id,
|
||||
"paichan_no": accessory.paichan_no,
|
||||
"zongpai_no": accessory.zongpai_no,
|
||||
"accessory_type": accessory.accessory_type,
|
||||
"quantity": accessory.quantity,
|
||||
"location_code": accessory.location_code,
|
||||
"is_boxed": is_boxed,
|
||||
"created_at": accessory.created_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
# ── Work order query ──
|
||||
|
||||
|
||||
def query_work_orders(db: Session, paichan_no: str) -> dict:
|
||||
"""Query ERP for work orders grouped by work_order_no for a given paichan_no."""
|
||||
paichan_no = paichan_no.strip().upper()
|
||||
if not PAICHAN_PATTERN.match(paichan_no):
|
||||
raise InvalidPaichanError()
|
||||
|
||||
rows = db.execute(
|
||||
text(_erp_paicha_items_sql(db)), {"paicha_no": paichan_no}
|
||||
).fetchall()
|
||||
|
||||
if not rows:
|
||||
raise PaichanNotFoundError()
|
||||
|
||||
grouped: dict[str, list[str]] = defaultdict(list)
|
||||
for row in rows:
|
||||
zongpai_no = row[0]
|
||||
work_order_no = row[1]
|
||||
grouped[work_order_no].append(zongpai_no)
|
||||
|
||||
work_orders = [
|
||||
{"work_order_no": wo_no, "zongpai_nos": zongpai_nos}
|
||||
for wo_no, zongpai_nos in grouped.items()
|
||||
]
|
||||
|
||||
return {
|
||||
"paichan_no": paichan_no,
|
||||
"work_orders": work_orders,
|
||||
}
|
||||
|
||||
|
||||
# ── Accessory CRUD ──
|
||||
|
||||
|
||||
def list_accessories(db: Session, paichan_no: str) -> dict:
|
||||
"""List all accessories for a given paichan_no, including is_boxed status."""
|
||||
accessories = (
|
||||
db.query(FinishedGoodsAccessory)
|
||||
.filter(FinishedGoodsAccessory.paichan_no == paichan_no)
|
||||
.order_by(FinishedGoodsAccessory.id)
|
||||
.all()
|
||||
)
|
||||
|
||||
items = []
|
||||
for acc in accessories:
|
||||
boxed = _is_accessory_boxed(db, acc)
|
||||
items.append(_accessory_to_dict(acc, is_boxed=boxed))
|
||||
|
||||
return {
|
||||
"paichan_no": paichan_no,
|
||||
"items": items,
|
||||
}
|
||||
|
||||
|
||||
def create_accessory(
|
||||
db: Session, req: AccessoryCreateRequest
|
||||
) -> FinishedGoodsAccessory:
|
||||
"""Create a new accessory record after validating against ERP data."""
|
||||
# 1. Validate paichan_no exists in ERP
|
||||
rows = db.execute(
|
||||
text(_erp_paicha_items_sql(db)), {"paicha_no": req.paichan_no}
|
||||
).fetchall()
|
||||
|
||||
if not rows:
|
||||
raise PaichanNotFoundError()
|
||||
|
||||
# 2. Validate zongpai_no belongs to this paichan_no in ERP
|
||||
valid_zongpai_nos = {row[0] for row in rows}
|
||||
if req.zongpai_no not in valid_zongpai_nos:
|
||||
raise ValueError(
|
||||
f"zongpai_no {req.zongpai_no} does not belong to paichan_no {req.paichan_no}"
|
||||
)
|
||||
|
||||
# 3. Create the record
|
||||
record = FinishedGoodsAccessory(
|
||||
paichan_no=req.paichan_no,
|
||||
zongpai_no=req.zongpai_no,
|
||||
accessory_type=req.accessory_type,
|
||||
quantity=req.quantity,
|
||||
location_code=req.location_code,
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
db.refresh(record)
|
||||
return record
|
||||
|
||||
|
||||
def update_accessory(
|
||||
db: Session, accessory_id: int, req: AccessoryUpdateRequest
|
||||
) -> dict:
|
||||
"""Partially update an accessory record with shelving logic."""
|
||||
accessory = (
|
||||
db.query(FinishedGoodsAccessory)
|
||||
.filter(FinishedGoodsAccessory.id == accessory_id)
|
||||
.first()
|
||||
)
|
||||
if accessory is None:
|
||||
raise AccessoryNotFoundError()
|
||||
|
||||
# Check if boxed
|
||||
if _is_accessory_boxed(db, accessory):
|
||||
raise AccessoryAlreadyBoxedError()
|
||||
|
||||
# Shelving logic when location_code changes
|
||||
if req.location_code is not None and accessory.location_code is not None:
|
||||
existing_loc = accessory.location_code
|
||||
new_loc = req.location_code
|
||||
|
||||
if existing_loc == new_loc:
|
||||
# No change needed for location
|
||||
pass
|
||||
elif _is_transit_location(existing_loc):
|
||||
raise AccessoryAlreadyOffShelfError(
|
||||
location_code=existing_loc,
|
||||
registered_at=accessory.created_at.isoformat(),
|
||||
)
|
||||
elif _is_transit_location(new_loc):
|
||||
# Off-shelf/down-shelf: normal operation
|
||||
accessory.location_code = new_loc
|
||||
accessory.created_at = datetime.now()
|
||||
else:
|
||||
# Both are normal locations → duplicate
|
||||
raise AccessoryDuplicateLocationError(
|
||||
location_code=existing_loc,
|
||||
registered_at=accessory.created_at.isoformat(),
|
||||
)
|
||||
elif req.location_code is not None and accessory.location_code is None:
|
||||
# Not yet shelved, just set the location
|
||||
accessory.location_code = req.location_code
|
||||
|
||||
# Update other fields if provided
|
||||
if req.accessory_type is not None:
|
||||
accessory.accessory_type = req.accessory_type
|
||||
if req.quantity is not None:
|
||||
accessory.quantity = req.quantity
|
||||
|
||||
db.commit()
|
||||
db.refresh(accessory)
|
||||
return _accessory_to_dict(accessory, is_boxed=_is_accessory_boxed(db, accessory))
|
||||
|
||||
|
||||
def delete_accessory(db: Session, accessory_id: int) -> dict:
|
||||
"""Delete an accessory record if it is not boxed."""
|
||||
accessory = (
|
||||
db.query(FinishedGoodsAccessory)
|
||||
.filter(FinishedGoodsAccessory.id == accessory_id)
|
||||
.first()
|
||||
)
|
||||
if accessory is None:
|
||||
raise AccessoryNotFoundError()
|
||||
|
||||
if _is_accessory_boxed(db, accessory):
|
||||
raise AccessoryAlreadyBoxedError()
|
||||
|
||||
db.delete(accessory)
|
||||
db.commit()
|
||||
return {"id": accessory_id, "deleted": True}
|
||||
|
||||
|
||||
# ── Accessory type management ──
|
||||
|
||||
|
||||
def list_accessory_types(db: Session) -> dict:
|
||||
"""List all accessory types ordered by sort_order."""
|
||||
types = (
|
||||
db.query(AccessoryType)
|
||||
.order_by(AccessoryType.sort_order, AccessoryType.id)
|
||||
.all()
|
||||
)
|
||||
return {
|
||||
"types": [
|
||||
{"id": t.id, "name": t.name, "sort_order": t.sort_order} for t in types
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def create_accessory_type(db: Session, req: AccessoryTypeCreateRequest) -> dict:
|
||||
"""Create a new accessory type. Name must be unique."""
|
||||
existing = db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
|
||||
if existing is not None:
|
||||
raise DuplicateAccessoryTypeNameError()
|
||||
|
||||
record = AccessoryType(
|
||||
name=req.name,
|
||||
sort_order=req.sort_order,
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
db.refresh(record)
|
||||
return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
|
||||
|
||||
|
||||
def update_accessory_type(
|
||||
db: Session, type_id: int, req: AccessoryTypeUpdateRequest
|
||||
) -> dict:
|
||||
"""Partially update an accessory type."""
|
||||
record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
|
||||
if record is None:
|
||||
raise AccessoryTypeNotFoundError()
|
||||
|
||||
# If name is being changed, check for duplicate
|
||||
if req.name is not None and req.name != record.name:
|
||||
duplicate = (
|
||||
db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
|
||||
)
|
||||
if duplicate is not None:
|
||||
raise DuplicateAccessoryTypeNameError()
|
||||
record.name = req.name
|
||||
|
||||
if req.sort_order is not None:
|
||||
record.sort_order = req.sort_order
|
||||
|
||||
db.commit()
|
||||
db.refresh(record)
|
||||
return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
|
||||
|
||||
|
||||
def delete_accessory_type(db: Session, type_id: int) -> dict:
|
||||
"""Delete an accessory type."""
|
||||
record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
|
||||
if record is None:
|
||||
raise AccessoryTypeNotFoundError()
|
||||
|
||||
db.delete(record)
|
||||
db.commit()
|
||||
return {"id": type_id, "deleted": True}
|
||||
Reference in New Issue
Block a user