90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
|
from sqlalchemy.orm import Session
|
|||
|
|
from sqlalchemy import func
|
|||
|
|
from typing import List
|
|||
|
|
import logging
|
|||
|
|
|
|||
|
|
from app.models.location import TAppLocation
|
|||
|
|
from app.schemas.location import LocationQueryRequest
|
|||
|
|
from app.schemas.common import BaseWmsApiResponse, WmsApiResponse
|
|||
|
|
from app.utils.string_utils import pad_left
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class LocationService:
|
|||
|
|
|
|||
|
|
def __init__(self, db: Session):
|
|||
|
|
self.db = db
|
|||
|
|
|
|||
|
|
def gen_locations(self, location_query: LocationQueryRequest) -> BaseWmsApiResponse:
|
|||
|
|
try:
|
|||
|
|
new_location_list: List[TAppLocation] = []
|
|||
|
|
|
|||
|
|
for row in range(1, location_query.l_row + 1):
|
|||
|
|
for col in range(1, location_query.l_col + 1):
|
|||
|
|
for layer in range(1, location_query.l_layer + 1):
|
|||
|
|
for depth in range(1, location_query.l_depth + 1):
|
|||
|
|
# 生成库位ID，格式：子区域 + 行(2位) + "-" + 列(2位) + "-" + 层(2位) + "-" + 深度(2位)
|
|||
|
|
location_id = (
|
|||
|
|
f"{location_query.sub_area}"
|
|||
|
|
f"{pad_left(str(row), '0', 2)}-"
|
|||
|
|
f"{pad_left(str(col), '0', 2)}-"
|
|||
|
|
f"{pad_left(str(layer), '0', 2)}-"
|
|||
|
|
f"{pad_left(str(depth), '0', 2)}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
new_location = TAppLocation(
|
|||
|
|
location_id=location_id,
|
|||
|
|
location_name="",
|
|||
|
|
location_type=1,
|
|||
|
|
is_occupy=0,
|
|||
|
|
is_enable=0,
|
|||
|
|
equipment_id=0,
|
|||
|
|
aisle_num_left=(row + 1) // 2,
|
|||
|
|
aisle_num_right=(row + 1) // 2,
|
|||
|
|
l_row=row,
|
|||
|
|
l_col=col,
|
|||
|
|
l_layer=layer,
|
|||
|
|
l_depth=depth,
|
|||
|
|
remark="",
|
|||
|
|
area_id=location_query.area_id if location_query.area_id is not None else 1,
|
|||
|
|
sub_area=location_query.sub_area
|
|||
|
|
)
|
|||
|
|
new_location_list.append(new_location)
|
|||
|
|
|
|||
|
|
self.db.add_all(new_location_list)
|
|||
|
|
self.db.commit()
|
|||
|
|
|
|||
|
|
logger.info(f"成功创建了{len(new_location_list)}个库位")
|
|||
|
|
return BaseWmsApiResponse.success(f"成功创建了{len(new_location_list)}个库位。")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.db.rollback()
|
|||
|
|
logger.error(f"创建库位失败: {str(e)}")
|
|||
|
|
return BaseWmsApiResponse.error(f"创建库位失败: {str(e)}")
|
|||
|
|
|
|||
|
|
def get_used_locations(self, equipment_id: int, location_type: int) -> WmsApiResponse:
|
|||
|
|
try:
|
|||
|
|
count = self.db.query(func.count(TAppLocation.location_id)).filter(
|
|||
|
|
TAppLocation.equipment_id == equipment_id,
|
|||
|
|
TAppLocation.location_type == location_type,
|
|||
|
|
TAppLocation.is_occupy == 1
|
|||
|
|
).scalar()
|
|||
|
|
|
|||
|
|
# 确保count不为None
|
|||
|
|
if count is None:
|
|||
|
|
count = 0
|
|||
|
|
|
|||
|
|
return WmsApiResponse.success(data=count, message="查询成功")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
# 打印完整异常信息（包含堆栈）
|
|||
|
|
logger.exception("查询已使用库位时发生异常")
|
|||
|
|
|
|||
|
|
# 返回错误响应 - 也使用直接实例化
|
|||
|
|
return WmsApiResponse.error(
|
|||
|
|
code=500,
|
|||
|
|
message=f"查询失败: {type(e).__name__}",
|
|||
|
|
data=None
|
|||
|
|
)
|