1 Commits
dev ... master

Author SHA1 Message Date
Misaka
eec0fd3189 feat: add temp storage shelf (B prefix) support
Add _is_temp_storage_location() helper and temp_stored status to
_location_status(). Refactor register_location() state transitions to
allow normal→temp shelf changes while blocking temp→normal and temp→temp.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-05-25 22:48:28 +08:00

View File

@@ -35,6 +35,10 @@ def _is_transit_location(location_code: str) -> bool:
return location_code.startswith("TRANS-")
def _is_temp_storage_location(location_code: str) -> bool:
return location_code.startswith("B")
def _erp_info_sql(db: Session) -> str:
dialect_name = db.bind.dialect.name if db.bind is not None else ""
if dialect_name == "postgresql":
@@ -74,7 +78,9 @@ def _query_paicha_no(db: Session, zongpai_no: str) -> str:
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()
rows = db.execute(
text(_erp_paicha_items_sql(db)), {"paicha_no": paicha_no}
).fetchall()
return [
{
"zongpai_no": row[0],
@@ -90,6 +96,8 @@ def _location_status(location_code: str | None) -> str:
return "not_shelved"
if _is_transit_location(location_code):
return "transferred"
if _is_temp_storage_location(location_code):
return "temp_stored"
return "on_shelf"
@@ -111,18 +119,20 @@ def get_paicha_overview(db: Session, zongpai_no: str) -> dict:
)
location_map = {item.zongpai_no: item.location_code for item in locations}
status_order = {"on_shelf": 0, "transferred": 1, "not_shelved": 2}
status_order = {"on_shelf": 0, "temp_stored": 1, "transferred": 2, "not_shelved": 3}
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.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: (
@@ -145,13 +155,19 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio
.first()
)
if existing:
# Terminal state: transit location → any operation raises error
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):
target_is_transit = _is_transit_location(req.location_code)
target_is_temp = _is_temp_storage_location(req.location_code)
current_is_temp = _is_temp_storage_location(existing.location_code)
# Allowed: target is transit (off-shelf) or normal→temp (shelf change)
if target_is_transit or (not current_is_temp and target_is_temp):
previous_location = existing.location_code
existing.location_code = req.location_code
existing.created_at = datetime.now()