730 lines
23 KiB
Python
730 lines
23 KiB
Python
"""存储设施（树结构）+ 小格 CRUD（biz_storage_facility / biz_storage_facility_slot）"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from sqlalchemy import and_, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.models import BizStorageFacility, BizStorageFacilitySlot
|
||
from app.services.storage_facility_code import (
|
||
assert_sibling_node_code_unique,
|
||
build_full_slot_code,
|
||
cascade_refresh_slot_name_snapshots_for_nodes,
|
||
cascade_refresh_slot_name_snapshots_for_slots,
|
||
decode_slot_layout_order,
|
||
encode_slot_layout_order,
|
||
load_facility_map,
|
||
validate_node_code,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Facility node helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _row_to_dict(f: BizStorageFacility) -> Dict[str, Any]:
|
||
return {
|
||
"id": f.id,
|
||
"parentId": f.parent_id,
|
||
"nodeName": f.node_name,
|
||
"nodeCode": f.node_code or "",
|
||
"nodeType": f.node_type or "",
|
||
"isLeaf": 1 if f.is_leaf else 0,
|
||
"tenantNo": f.tenant_no,
|
||
"orderNum": f.order_num if f.order_num is not None else 0,
|
||
"remark": f.remark or "",
|
||
"createTime": f.create_time.strftime("%Y-%m-%d %H:%M:%S") if f.create_time else "",
|
||
"updateTime": f.update_time.strftime("%Y-%m-%d %H:%M:%S") if f.update_time else "",
|
||
}
|
||
|
||
|
||
def list_all_facilities(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
keyword: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""Return FLAT list of all non-deleted facility nodes (no pagination, no tree building)."""
|
||
q = db.query(BizStorageFacility).filter(BizStorageFacility.del_flag == 0)
|
||
if tenant_no:
|
||
q = q.filter(BizStorageFacility.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q = q.filter(
|
||
or_(
|
||
BizStorageFacility.node_name.like(like),
|
||
BizStorageFacility.node_code.like(like),
|
||
)
|
||
)
|
||
rows = q.order_by(BizStorageFacility.order_num.asc(), BizStorageFacility.create_time.asc()).all()
|
||
return [_row_to_dict(x) for x in rows]
|
||
|
||
|
||
def get_facility(db: Session, node_id: str, tenant_no: Optional[str] = None) -> Optional[BizStorageFacility]:
|
||
q = db.query(BizStorageFacility).filter(
|
||
and_(BizStorageFacility.id == node_id, BizStorageFacility.del_flag == 0)
|
||
)
|
||
if tenant_no:
|
||
q = q.filter(BizStorageFacility.tenant_no == tenant_no)
|
||
return q.first()
|
||
|
||
|
||
def get_children(db: Session, parent_id: str, tenant_no: str) -> List[BizStorageFacility]:
|
||
"""Direct children of a node."""
|
||
return (
|
||
db.query(BizStorageFacility)
|
||
.filter(
|
||
and_(
|
||
BizStorageFacility.parent_id == parent_id,
|
||
BizStorageFacility.del_flag == 0,
|
||
BizStorageFacility.tenant_no == tenant_no,
|
||
)
|
||
)
|
||
.order_by(BizStorageFacility.order_num.asc(), BizStorageFacility.create_time.asc())
|
||
.all()
|
||
)
|
||
|
||
|
||
def collect_descendant_ids(db: Session, node_id: str, tenant_no: str) -> List[str]:
|
||
"""Iteratively collect ALL descendant node IDs (stack-based, no recursion)."""
|
||
ids: List[str] = []
|
||
stack: List[str] = [node_id]
|
||
while stack:
|
||
current = stack.pop()
|
||
children = get_children(db, current, tenant_no)
|
||
for child in children:
|
||
ids.append(child.id)
|
||
stack.append(child.id)
|
||
return ids
|
||
|
||
|
||
def create_facility(
|
||
db: Session,
|
||
*,
|
||
parent_id: str = "0",
|
||
node_name: str,
|
||
node_code: str,
|
||
node_type: Optional[str] = None,
|
||
is_leaf: int = 0,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizStorageFacility:
|
||
name = (node_name or "").strip()
|
||
if not name:
|
||
raise ValueError("节点名称不能为空")
|
||
code = validate_node_code(node_code)
|
||
pid = parent_id or "0"
|
||
assert_sibling_node_code_unique(
|
||
db, parent_id=pid, node_code=code, tenant_no=tenant_no
|
||
)
|
||
obj = BizStorageFacility(
|
||
id=generate_uuid(),
|
||
parent_id=pid,
|
||
node_name=name,
|
||
node_code=code,
|
||
node_type=(node_type or "").strip() or None,
|
||
is_leaf=is_leaf,
|
||
tenant_no=tenant_no,
|
||
order_num=order_num if order_num is not None else 0,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_facility(
|
||
db: Session,
|
||
node_id: str,
|
||
*,
|
||
node_name: Optional[str] = None,
|
||
node_code: Optional[str] = None,
|
||
node_type: Optional[str] = None,
|
||
is_leaf: Optional[int] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> BizStorageFacility:
|
||
obj = get_facility(db, node_id, tenant_no=tenant_no)
|
||
if not obj:
|
||
raise ValueError("存储设施节点不存在")
|
||
old_node_code = (obj.node_code or "").strip()
|
||
if node_name is not None:
|
||
name = (node_name or "").strip()
|
||
if not name:
|
||
raise ValueError("节点名称不能为空")
|
||
obj.node_name = name
|
||
if node_code is not None:
|
||
code = validate_node_code(node_code)
|
||
assert_sibling_node_code_unique(
|
||
db,
|
||
parent_id=obj.parent_id or "0",
|
||
node_code=code,
|
||
tenant_no=obj.tenant_no,
|
||
exclude_id=obj.id,
|
||
)
|
||
obj.node_code = code
|
||
if node_type is not None:
|
||
obj.node_type = (node_type or "").strip() or None
|
||
if is_leaf is not None:
|
||
obj.is_leaf = is_leaf
|
||
if order_num is not None:
|
||
obj.order_num = order_num
|
||
if remark is not None:
|
||
obj.remark = (remark or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
|
||
node_code_changed = node_code is not None and (obj.node_code or "").strip() != old_node_code
|
||
if node_code_changed:
|
||
cascade_refresh_slot_name_snapshots_for_nodes(
|
||
db,
|
||
[node_id],
|
||
obj.tenant_no,
|
||
collect_descendant_ids_fn=collect_descendant_ids,
|
||
)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_facility(db: Session, node_id: str, tenant_no: Optional[str] = None) -> None:
|
||
"""Cascade soft-delete: node + all descendants + their slots."""
|
||
node = get_facility(db, node_id, tenant_no=tenant_no)
|
||
if not node:
|
||
raise ValueError("存储设施节点不存在")
|
||
# Collect all descendant IDs
|
||
all_ids = [node_id] + collect_descendant_ids(db, node_id, node.tenant_no)
|
||
now = datetime.now()
|
||
# Soft-delete all slots belonging to these nodes
|
||
db.query(BizStorageFacilitySlot).filter(
|
||
BizStorageFacilitySlot.facility_node_id.in_(all_ids),
|
||
BizStorageFacilitySlot.del_flag == 0,
|
||
).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
|
||
# Soft-delete all facility nodes
|
||
db.query(BizStorageFacility).filter(
|
||
BizStorageFacility.id.in_(all_ids),
|
||
BizStorageFacility.del_flag == 0,
|
||
).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
|
||
db.commit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Slot helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _slot_to_dict(
|
||
s: BizStorageFacilitySlot,
|
||
facility_by_id: Optional[Dict[str, BizStorageFacility]] = None,
|
||
) -> Dict[str, Any]:
|
||
occupied_count = s.occupied_count if s.occupied_count is not None else 0
|
||
full_slot_code = ""
|
||
if facility_by_id is not None:
|
||
full_slot_code = build_full_slot_code(s.facility_node_id, s.slot_code, facility_by_id)
|
||
return {
|
||
"id": s.id,
|
||
"facilityNodeId": s.facility_node_id,
|
||
"slotCode": s.slot_code or "",
|
||
"fullSlotCode": full_slot_code,
|
||
"slotStatus": s.slot_status or "",
|
||
"occupiedCount": int(occupied_count),
|
||
"tenantNo": s.tenant_no,
|
||
"orderNum": s.order_num if s.order_num is not None else 0,
|
||
"remark": s.remark or "",
|
||
"createTime": s.create_time.strftime("%Y-%m-%d %H:%M:%S") if s.create_time else "",
|
||
"updateTime": s.update_time.strftime("%Y-%m-%d %H:%M:%S") if s.update_time else "",
|
||
}
|
||
|
||
|
||
def list_slots(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
tenant_no: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
q = db.query(BizStorageFacilitySlot).filter(
|
||
and_(
|
||
BizStorageFacilitySlot.facility_node_id == facility_node_id,
|
||
BizStorageFacilitySlot.del_flag == 0,
|
||
)
|
||
)
|
||
if tenant_no:
|
||
q = q.filter(BizStorageFacilitySlot.tenant_no == tenant_no)
|
||
rows = q.order_by(BizStorageFacilitySlot.order_num.asc(), BizStorageFacilitySlot.create_time.asc()).all()
|
||
facility_by_id = load_facility_map(db, tenant_no or "") if tenant_no else {}
|
||
return [_slot_to_dict(x, facility_by_id) for x in rows]
|
||
|
||
|
||
def _is_slot_available_row(row: Dict[str, Any]) -> bool:
|
||
status = row.get("slotStatus") or ""
|
||
occupied = int(row.get("occupiedCount") or 0)
|
||
return status == "空" or occupied == 0
|
||
|
||
|
||
def _slot_select_display_name(row: Dict[str, Any]) -> str:
|
||
full = (row.get("fullSlotCode") or "").strip()
|
||
code = (row.get("slotCode") or "").strip()
|
||
return full or code or str(row.get("id") or "")
|
||
|
||
|
||
def list_slot_select_options(
|
||
db: Session,
|
||
*,
|
||
query,
|
||
tenant_no: str,
|
||
keyword: Optional[str] = None,
|
||
available_only: bool = False,
|
||
include_slot_ids: Optional[List[str]] = None,
|
||
page: int = 1,
|
||
page_size: int = 100,
|
||
) -> tuple[List[Dict[str, Any]], int]:
|
||
"""Paginated storage slot options for remote select; keyword matches code/path/name."""
|
||
include_set = {
|
||
str(x).strip()
|
||
for x in (include_slot_ids or [])
|
||
if str(x).strip()
|
||
}
|
||
kw = (keyword or "").strip().lower()
|
||
facility_by_id = load_facility_map(db, tenant_no)
|
||
|
||
rows = query.order_by(
|
||
BizStorageFacilitySlot.order_num.asc(),
|
||
BizStorageFacilitySlot.create_time.asc(),
|
||
).all()
|
||
|
||
enriched: List[Dict[str, Any]] = []
|
||
for slot_row in rows:
|
||
row = _slot_to_dict(slot_row, facility_by_id)
|
||
slot_id = str(row.get("id") or "").strip()
|
||
if not slot_id:
|
||
continue
|
||
if available_only and slot_id not in include_set and not _is_slot_available_row(row):
|
||
continue
|
||
if kw:
|
||
node = facility_by_id.get(str(row.get("facilityNodeId") or "").strip())
|
||
haystack = " ".join(
|
||
[
|
||
str(row.get("slotCode") or ""),
|
||
str(row.get("fullSlotCode") or ""),
|
||
str(row.get("remark") or ""),
|
||
str(getattr(node, "node_name", "") or "") if node else "",
|
||
str(getattr(node, "node_code", "") or "") if node else "",
|
||
]
|
||
).lower()
|
||
if kw not in haystack:
|
||
continue
|
||
row["name"] = _slot_select_display_name(row)
|
||
enriched.append(row)
|
||
|
||
enriched.sort(
|
||
key=lambda x: (
|
||
str(x.get("fullSlotCode") or x.get("slotCode") or ""),
|
||
str(x.get("id") or ""),
|
||
)
|
||
)
|
||
|
||
total = len(enriched)
|
||
page = max(int(page or 1), 1)
|
||
page_size = max(min(int(page_size or 100), 200), 1)
|
||
start = (page - 1) * page_size
|
||
page_rows = enriched[start : start + page_size]
|
||
|
||
result_list = [
|
||
{
|
||
"id": row.get("id"),
|
||
"name": row.get("name") or _slot_select_display_name(row),
|
||
"fullSlotCode": row.get("fullSlotCode") or "",
|
||
"slotCode": row.get("slotCode") or "",
|
||
"facilityNodeId": row.get("facilityNodeId") or "",
|
||
"slotStatus": row.get("slotStatus") or "",
|
||
"occupiedCount": int(row.get("occupiedCount") or 0),
|
||
}
|
||
for row in page_rows
|
||
]
|
||
return result_list, total
|
||
|
||
|
||
def get_slot(db: Session, slot_id: str, tenant_no: Optional[str] = None) -> Optional[BizStorageFacilitySlot]:
|
||
q = db.query(BizStorageFacilitySlot).filter(
|
||
and_(BizStorageFacilitySlot.id == slot_id, BizStorageFacilitySlot.del_flag == 0)
|
||
)
|
||
if tenant_no:
|
||
q = q.filter(BizStorageFacilitySlot.tenant_no == tenant_no)
|
||
return q.first()
|
||
|
||
|
||
def _slot_to_dict_for_row(db: Session, s: BizStorageFacilitySlot) -> Dict[str, Any]:
|
||
facility_by_id = load_facility_map(db, s.tenant_no)
|
||
return _slot_to_dict(s, facility_by_id)
|
||
|
||
|
||
def create_slot(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
slot_code: Optional[str] = None,
|
||
slot_status: Optional[str] = None,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
at_row_num: Optional[int] = None,
|
||
) -> BizStorageFacilitySlot:
|
||
fid = (facility_node_id or "").strip()
|
||
if not fid:
|
||
raise ValueError("所属设施节点不能为空")
|
||
if at_row_num is not None:
|
||
return insert_slot_at_row_end(
|
||
db,
|
||
facility_node_id=fid,
|
||
row_num=at_row_num,
|
||
slot_status=slot_status,
|
||
tenant_no=tenant_no,
|
||
remark=remark,
|
||
user_id=user_id,
|
||
)
|
||
obj = BizStorageFacilitySlot(
|
||
id=generate_uuid(),
|
||
facility_node_id=fid,
|
||
slot_code=(slot_code or "").strip() or None,
|
||
slot_status=(slot_status or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=order_num if order_num is not None else 0,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def insert_slot_at_row_end(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
row_num: int,
|
||
slot_status: Optional[str] = None,
|
||
tenant_no: str = "zkxw",
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizStorageFacilitySlot:
|
||
fid = (facility_node_id or "").strip()
|
||
row = int(row_num)
|
||
if row < 1:
|
||
raise ValueError("行号参数不合法")
|
||
|
||
node = db.query(BizStorageFacility).filter(
|
||
and_(BizStorageFacility.id == fid, BizStorageFacility.del_flag == 0)
|
||
).first()
|
||
if not node:
|
||
raise ValueError("所属设施节点不存在")
|
||
if not node.is_leaf:
|
||
raise ValueError("只能为叶子节点创建格子")
|
||
|
||
max_col = 0
|
||
for slot in _list_facility_slots(db, fid, tenant_no):
|
||
slot_row, slot_col = _resolve_slot_layout_position(slot)
|
||
if slot_row == row:
|
||
max_col = max(max_col, slot_col)
|
||
_ensure_slot_layout_order(slot, user_id=user_id)
|
||
|
||
now = datetime.now()
|
||
obj = BizStorageFacilitySlot(
|
||
id=generate_uuid(),
|
||
facility_node_id=fid,
|
||
slot_code=None,
|
||
slot_status=(slot_status or "空").strip() or "空",
|
||
tenant_no=tenant_no,
|
||
order_num=encode_slot_layout_order(row, max_col + 1),
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_slot(
|
||
db: Session,
|
||
slot_id: str,
|
||
*,
|
||
slot_code: Optional[str] = None,
|
||
slot_status: Optional[str] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> BizStorageFacilitySlot:
|
||
obj = get_slot(db, slot_id, tenant_no=tenant_no)
|
||
if not obj:
|
||
raise ValueError("存储设施小格不存在")
|
||
old_slot_code = (obj.slot_code or "").strip()
|
||
if slot_code is not None:
|
||
obj.slot_code = (slot_code or "").strip() or None
|
||
if slot_status is not None:
|
||
obj.slot_status = (slot_status or "").strip() or None
|
||
if order_num is not None:
|
||
obj.order_num = order_num
|
||
if remark is not None:
|
||
obj.remark = (remark or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
|
||
slot_code_changed = slot_code is not None and (obj.slot_code or "").strip() != old_slot_code
|
||
if slot_code_changed:
|
||
cascade_refresh_slot_name_snapshots_for_slots(db, [slot_id], obj.tenant_no)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_slot(db: Session, slot_id: str, tenant_no: Optional[str] = None, user_id: Optional[str] = None) -> None:
|
||
obj = get_slot(db, slot_id, tenant_no=tenant_no)
|
||
if not obj:
|
||
raise ValueError("存储设施小格不存在")
|
||
obj.del_flag = 1
|
||
obj.update_time = datetime.now()
|
||
obj.update_by = user_id
|
||
db.add(obj)
|
||
db.commit()
|
||
|
||
|
||
def clear_slots_by_facility_node(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str] = None,
|
||
) -> int:
|
||
fid = (facility_node_id or "").strip()
|
||
if not fid:
|
||
raise ValueError("所属设施节点不能为空")
|
||
node = db.query(BizStorageFacility).filter(
|
||
and_(
|
||
BizStorageFacility.id == fid,
|
||
BizStorageFacility.del_flag == 0,
|
||
BizStorageFacility.tenant_no == tenant_no,
|
||
)
|
||
).first()
|
||
if not node:
|
||
raise ValueError("所属设施节点不存在")
|
||
now = datetime.now()
|
||
updated = db.query(BizStorageFacilitySlot).filter(
|
||
BizStorageFacilitySlot.facility_node_id == fid,
|
||
BizStorageFacilitySlot.del_flag == 0,
|
||
BizStorageFacilitySlot.tenant_no == tenant_no,
|
||
).update(
|
||
{"del_flag": 1, "update_time": now, "update_by": user_id},
|
||
synchronize_session=False
|
||
)
|
||
db.commit()
|
||
return int(updated or 0)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Slot layout & sequential code helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _resolve_slot_layout_position(slot: BizStorageFacilitySlot) -> tuple[int, int]:
|
||
order = int(slot.order_num or 0)
|
||
row, col = decode_slot_layout_order(order)
|
||
if row > 0:
|
||
return row, col
|
||
match = re.match(r"^(.*?)(\d+)-(\d+)$", (slot.slot_code or "").strip())
|
||
if match:
|
||
return int(match.group(2)), int(match.group(3))
|
||
return 0, 0
|
||
|
||
|
||
def _ensure_slot_layout_order(
|
||
slot: BizStorageFacilitySlot,
|
||
*,
|
||
user_id: Optional[str] = None,
|
||
) -> tuple[int, int]:
|
||
row, col = _resolve_slot_layout_position(slot)
|
||
if row > 0:
|
||
expected = encode_slot_layout_order(row, col)
|
||
if int(slot.order_num or 0) != expected:
|
||
slot.order_num = expected
|
||
if user_id:
|
||
slot.update_by = user_id
|
||
slot.update_time = datetime.now()
|
||
return row, col
|
||
return 0, 0
|
||
|
||
|
||
def _list_facility_slots(
|
||
db: Session,
|
||
facility_node_id: str,
|
||
tenant_no: str,
|
||
) -> List[BizStorageFacilitySlot]:
|
||
return (
|
||
db.query(BizStorageFacilitySlot)
|
||
.filter(
|
||
BizStorageFacilitySlot.facility_node_id == facility_node_id,
|
||
BizStorageFacilitySlot.del_flag == 0,
|
||
BizStorageFacilitySlot.tenant_no == tenant_no,
|
||
)
|
||
.order_by(
|
||
BizStorageFacilitySlot.order_num.asc(),
|
||
BizStorageFacilitySlot.create_time.asc(),
|
||
)
|
||
.all()
|
||
)
|
||
|
||
|
||
def batch_create_slots(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
rows: int,
|
||
cols: int,
|
||
tenant_no: str = "zkxw",
|
||
user_id: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
fid = (facility_node_id or "").strip()
|
||
if not fid:
|
||
raise ValueError("所属设施节点不能为空")
|
||
if rows < 1 or cols < 1:
|
||
raise ValueError("行数和列数必须大于 0")
|
||
if rows > 50 or cols > 50:
|
||
raise ValueError("单次最多生成 50×50 = 2500 个格子")
|
||
node = db.query(BizStorageFacility).filter(
|
||
and_(BizStorageFacility.id == fid, BizStorageFacility.del_flag == 0)
|
||
).first()
|
||
if not node:
|
||
raise ValueError("所属设施节点不存在")
|
||
if not node.is_leaf:
|
||
raise ValueError("只能为叶子节点创建格子")
|
||
now = datetime.now()
|
||
slots: List[BizStorageFacilitySlot] = []
|
||
seq = 0
|
||
for r in range(1, rows + 1):
|
||
for c in range(1, cols + 1):
|
||
seq += 1
|
||
obj = BizStorageFacilitySlot(
|
||
id=generate_uuid(),
|
||
facility_node_id=fid,
|
||
slot_code=str(seq),
|
||
slot_status="空",
|
||
tenant_no=tenant_no,
|
||
order_num=encode_slot_layout_order(r, c),
|
||
remark=None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
slots.append(obj)
|
||
db.commit()
|
||
facility_by_id = load_facility_map(db, tenant_no)
|
||
for s in slots:
|
||
db.refresh(s)
|
||
return [_slot_to_dict(s, facility_by_id) for s in slots]
|
||
|
||
|
||
def insert_row_and_shift_slots(
|
||
db: Session,
|
||
*,
|
||
facility_node_id: str,
|
||
after_row_num: int,
|
||
cols: int,
|
||
tenant_no: str = "zkxw",
|
||
user_id: Optional[str] = None,
|
||
) -> Dict[str, int]:
|
||
fid = (facility_node_id or "").strip()
|
||
if not fid:
|
||
raise ValueError("所属设施节点不能为空")
|
||
if after_row_num < 0:
|
||
raise ValueError("插入行号参数不合法")
|
||
if cols < 1:
|
||
raise ValueError("列数必须大于 0")
|
||
|
||
node = db.query(BizStorageFacility).filter(
|
||
and_(
|
||
BizStorageFacility.id == fid,
|
||
BizStorageFacility.del_flag == 0,
|
||
BizStorageFacility.tenant_no == tenant_no,
|
||
)
|
||
).first()
|
||
if not node:
|
||
raise ValueError("所属设施节点不存在")
|
||
if not node.is_leaf:
|
||
raise ValueError("只能为叶子节点新增格子行")
|
||
|
||
slots = _list_facility_slots(db, fid, tenant_no)
|
||
|
||
now = datetime.now()
|
||
shifted_count = 0
|
||
for slot in slots:
|
||
row_no, col_no = _ensure_slot_layout_order(slot, user_id=user_id)
|
||
if row_no <= 0:
|
||
continue
|
||
if row_no > after_row_num:
|
||
slot.order_num = encode_slot_layout_order(row_no + 1, col_no)
|
||
slot.update_by = user_id
|
||
slot.update_time = now
|
||
shifted_count += 1
|
||
|
||
insert_row_num = after_row_num + 1
|
||
created_slots: List[BizStorageFacilitySlot] = []
|
||
for col in range(1, cols + 1):
|
||
created_slots.append(
|
||
BizStorageFacilitySlot(
|
||
id=generate_uuid(),
|
||
facility_node_id=fid,
|
||
slot_code=None,
|
||
slot_status="空",
|
||
tenant_no=tenant_no,
|
||
order_num=encode_slot_layout_order(insert_row_num, col),
|
||
remark=None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
)
|
||
if created_slots:
|
||
db.add_all(created_slots)
|
||
db.commit()
|
||
return {"shiftedCount": shifted_count, "createdCount": len(created_slots)}
|