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
|
||
)
|