#!/usr/bin/env python # -*- coding: utf-8 -*- """ 库位处理器模块 提供与库区操作相关的各种处理器 """ import logging import json import asyncio import aiohttp import uuid from typing import Dict, Any, List, Optional from sqlalchemy import and_, or_, desc, asc, select, update from sqlalchemy.orm import Session from services.execution.task_context import TaskContext from .base import BlockHandler, register_handler from config.settings import settings from utils.logger import get_logger from .model.block_name import StorageBlockName from data.session import get_async_session from data.models.operate_point_layer import OperatePointLayer from data.models.storage_area import StorageArea from data.models.extended_property import ExtendedProperty, ExtendedPropertyTypeEnum # 获取日志记录器 logger = get_logger("services.execution.handlers.storage_location") # 创建一个基础库位处理器类，包含通用方法 class StorageBlockHandler(BlockHandler): pass # 批量设置库位处理器 @register_handler(StorageBlockName.BATCH_SETTING_SITE) class BatchSettingSiteBlockHandler(StorageBlockHandler): """批量设置库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """执行批量设置库位操作""" try: # 获取关键参数用于验证 site_ids = input_params.get("siteIds", []) group_names = input_params.get("groupNames", []) filled = input_params.get("filled") content = input_params.get("content") # 获取地图ID map_id = context.map_id # print(f"批量设置库位处理器参数>>>>>: {input_params}<<<<<<=======================") # 校验 siteIds 参数格式（可以为空或null，但有值时必须是列表） if site_ids is not None and site_ids != "" and site_ids != []: if not isinstance(site_ids, list): # 如果是字符串，尝试解析为列表 if isinstance(site_ids, str): # 检查是否为空字符串 if site_ids.strip() == "": site_ids = [] input_params["siteIds"] = site_ids else: try: import json site_ids = json.loads(site_ids) if not isinstance(site_ids, list): raise ValueError("解析后不是列表格式") input_params["siteIds"] = site_ids except (json.JSONDecodeError, ValueError): result = { "success": False, "message": "siteIds参数必须是列表格式或能解析为列表的字符串" } await self._record_task_log(block, result, context) return result else: result = { "success": False, "message": "siteIds参数必须是列表格式" } await self._record_task_log(block, result, context) return result # 校验 groupNames 参数格式（可以为空或null，但有值时必须是列表） if group_names is not None and group_names != "" and group_names != []: if not isinstance(group_names, list): # 如果是字符串，尝试解析为列表 if isinstance(group_names, str): # 检查是否为空字符串 if group_names.strip() == "": group_names = [] input_params["groupNames"] = group_names else: try: import json group_names = json.loads(group_names) if not isinstance(group_names, list): raise ValueError("解析后不是列表格式") input_params["groupNames"] = group_names except (json.JSONDecodeError, ValueError): result = { "success": False, "message": "groupNames参数必须是列表格式或能解析为列表的字符串" } await self._record_task_log(block, result, context) return result else: result = { "success": False, "message": "groupNames参数必须是列表格式" } await self._record_task_log(block, result, context) return result # 参数检查 if not site_ids and not group_names: result = { "success": False, "message": "库位ID和库区集不能同时为空" } await self._record_task_log(block, result, context) return result if not map_id: result = { "success": False, "message": "地图ID不能为空" } await self._record_task_log(block, result, context) return result if filled is None: result = { "success": False, "message": "占用参数不能为空" } await self._record_task_log(block, result, context) return result # 校验并转换 filled 参数为布尔类型 if isinstance(filled, str): filled_lower = filled.lower() if filled_lower in ('true', '1', 'yes', 'on'): filled = True elif filled_lower in ('false', '0', 'no', 'off'): filled = False else: result = { "success": False, "message": "filled参数必须是布尔类型或可转换为布尔类型的字符串" } await self._record_task_log(block, result, context) return result input_params["filled"] = filled elif not isinstance(filled, bool): result = { "success": False, "message": "filled参数必须是布尔类型" } await self._record_task_log(block, result, context) return result # 直接操作数据库执行批量设置 result = await self._batch_setting_site_in_db(site_ids, group_names, filled, content, map_id) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"批量设置库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _batch_setting_site_in_db(self, site_ids: List[str], group_names: List[str], filled: bool, content: str, map_id: str) -> Dict[str, Any]: """在数据库中批量设置库位状态""" try: async with get_async_session() as session: # 收集所有需要更新的库位ID target_site_ids = [] # 1. 校验并收集指定的库位ID if site_ids: site_check_query = select(OperatePointLayer.layer_name).where( OperatePointLayer.layer_name.in_(site_ids), OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False, OperatePointLayer.is_deleted == False ) site_check_result = await session.execute(site_check_query) existing_site_ids = [row[0] for row in site_check_result.fetchall()] # 检查是否有不存在的库位 missing_sites = set(site_ids) - set(existing_site_ids) if missing_sites: return { "success": False, "message": f"以下库位不存在或已禁用: {', '.join(str(site) for site in missing_sites)}" } target_site_ids.extend(existing_site_ids) # 2. 校验并收集库区下的所有库位 if group_names: # 校验库区是否存在 area_check_query = select(StorageArea.area_name).where( StorageArea.area_name.in_(group_names), StorageArea.scene_id == map_id, StorageArea.is_deleted == False ) area_check_result = await session.execute(area_check_query) existing_area_names = [row[0] for row in area_check_result.fetchall()] # 检查是否有不存在的库区 missing_areas = set(group_names) - set(existing_area_names) if missing_areas: return { "success": False, "message": f"以下库区不存在: {', '.join(str(area) for area in missing_areas)}" } # 获取库区下的所有库位 area_sites_query = select(OperatePointLayer.layer_name).where( OperatePointLayer.area_name.in_(existing_area_names), OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False, OperatePointLayer.is_deleted == False ) area_sites_result = await session.execute(area_sites_query) area_site_ids = [row[0] for row in area_sites_result.fetchall()] target_site_ids.extend(area_site_ids) # 去重 target_site_ids = list(set(target_site_ids)) if not target_site_ids: return { "success": False, "message": "没有找到符合条件的库位" } # 3. 根据filled参数决定更新逻辑 import datetime now = datetime.datetime.now() if filled: # 设置为占用状态 # 先检查哪些库位已经被占用 occupied_check_query = select(OperatePointLayer.layer_name).where( OperatePointLayer.layer_name.in_(target_site_ids), OperatePointLayer.is_occupied == True ) occupied_check_result = await session.execute(occupied_check_query) already_occupied = [row[0] for row in occupied_check_result.fetchall()] # 更新未占用的库位 update_values = { 'is_occupied': True, 'last_access_at': now } # 如果提供了content，则设置货物内容 if content is not None: update_values['goods_content'] = content update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name.in_(target_site_ids), OperatePointLayer.is_occupied == False # 只更新未占用的库位 ).values(update_values) result = await session.execute(update_stmt) await session.commit() updated_count = result.rowcount already_occupied_count = len(already_occupied) if already_occupied_count > 0: message = f"批量设置库位为占用成功，更新了 {updated_count} 个库位，{already_occupied_count} 个库位已是占用状态" else: message = f"批量设置库位为占用成功，更新了 {updated_count} 个库位" else: # 设置为空闲状态 # 先检查哪些库位已经是空闲状态 empty_check_query = select(OperatePointLayer.layer_name).where( OperatePointLayer.layer_name.in_(target_site_ids), OperatePointLayer.is_occupied == False ) empty_check_result = await session.execute(empty_check_query) already_empty = [row[0] for row in empty_check_result.fetchall()] # 更新占用的库位 update_values = { 'is_occupied': False, 'goods_content': '', # 清空货物内容 'goods_retrieved_at': now, 'last_access_at': now } update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name.in_(target_site_ids), OperatePointLayer.is_occupied == True # 只更新占用的库位 ).values(update_values) result = await session.execute(update_stmt) await session.commit() updated_count = result.rowcount already_empty_count = len(already_empty) if already_empty_count > 0: message = f"批量设置库位为空闲成功，更新了 {updated_count} 个库位，{already_empty_count} 个库位已是空闲状态" else: message = f"批量设置库位为空闲成功，更新了 {updated_count} 个库位" logger.info(f"批量设置库位完成: {message}") return { "success": True, "message": message, "data": { "totalSites": len(target_site_ids), "updatedSites": updated_count } } except Exception as e: logger.error(f"批量设置库位异常: {str(e)}") return { "success": False, "message": f"批量设置库位异常: {str(e)}" } # 从密集库区中获取库位处理器 @register_handler(StorageBlockName.GET_IDLE_CROWDED_SITE) class GetIdleCrowdedSiteBlockHandler(StorageBlockHandler): """从密集库区中获取库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """执行获取密集库位操作""" try: print(f"从密集库区中获取库位处理器参数>>>>>: {input_params}<<<<<<=======================") # 获取关键参数用于验证 # {'groupName': '["a", "b"]', 'filled': 'true', 'content': '1321', 'lock': 'true', 'retry': 'true', 'retryPeriod': '1000', 'retryNum': '3'} group_name = input_params.get("groupName", []) filled = input_params.get("filled") content = input_params.get("content") lock = input_params.get("lock", False) retry = input_params.get("retry", False) retry_num = input_params.get("retryNum", 1) # 获取地图ID map_id = context.map_id # 校验并转换 groupName 参数格式（必须是列表） if group_name is not None and group_name != "" and group_name != []: if not isinstance(group_name, list): # 如果是字符串，尝试解析为列表 if isinstance(group_name, str): # 检查是否为空字符串 if group_name.strip() == "": group_name = [] else: try: import json group_name = json.loads(group_name) if not isinstance(group_name, list): raise ValueError("解析后不是列表格式") input_params["groupName"] = group_name except (json.JSONDecodeError, ValueError): result = { "success": False, "message": "groupName参数必须是列表格式或能解析为列表的字符串" } await self._record_task_log(block, result, context) return result else: result = { "success": False, "message": "groupName参数必须是列表格式" } await self._record_task_log(block, result, context) return result # 参数检查 if not group_name: result = { "success": False, "message": "库区集不能为空" } await self._record_task_log(block, result, context) return result if filled is None: result = { "success": False, "message": "取/放参数不能为空" } await self._record_task_log(block, result, context) return result # 校验并转换 filled 参数为布尔类型 if isinstance(filled, str): filled_lower = filled.lower() if filled_lower in ('true', '1', 'yes', 'on'): filled = True elif filled_lower in ('false', '0', 'no', 'off'): filled = False else: result = { "success": False, "message": "filled参数必须是布尔类型或可转换为布尔类型的字符串" } await self._record_task_log(block, result, context) return result input_params["filled"] = filled elif not isinstance(filled, bool): result = { "success": False, "message": "filled参数必须是布尔类型" } await self._record_task_log(block, result, context) return result # 校验并转换 lock 参数为布尔类型 if isinstance(lock, str): # 如果是空字符串，映射为 false if lock.strip() == "": lock = False else: lock_lower = lock.lower() if lock_lower in ('true', '1', 'yes', 'on'): lock = True elif lock_lower in ('false', '0', 'no', 'off'): lock = False else: result = { "success": False, "message": "lock参数必须是布尔类型或可转换为布尔类型的字符串" } await self._record_task_log(block, result, context) return result input_params["lock"] = lock elif not isinstance(lock, bool): result = { "success": False, "message": "lock参数必须是布尔类型" } await self._record_task_log(block, result, context) return result # 校验并转换 retry 参数为布尔类型 if isinstance(retry, str): retry_lower = retry.lower() if retry_lower in ('true', '1', 'yes', 'on'): retry = True elif retry_lower in ('false', '0', 'no', 'off'): retry = False else: result = { "success": False, "message": "retry参数必须是布尔类型或可转换为布尔类型的字符串" } await self._record_task_log(block, result, context) return result input_params["retry"] = retry elif not isinstance(retry, bool): result = { "success": False, "message": "retry参数必须是布尔类型" } await self._record_task_log(block, result, context) return result # 确保retry_num是整数类型 try: retry_num = int(retry_num) except (ValueError, TypeError): logger.warning(f"retry_num参数类型错误或无法转换为整数: {retry_num}, 将使用默认值1") retry_num = 1 input_params["retryNum"] = retry_num # 确保retry_period是数字 retry_period = input_params.get("retryPeriod") if retry_period is not None: try: retry_sleep = float(retry_period) / 1000 except (ValueError, TypeError): logger.warning(f"retry_period参数类型错误或无法转换为数字: {retry_period}, 使用默认值1秒") retry_sleep = 1 else: retry_sleep = 1 # 调用外部API执行获取密集库位 attempts = 1 # 确保max_attempts是整数 max_attempts = retry_num if retry and retry_num is not None else 1 result = None while attempts <= max_attempts: # 调用数据库查询获取密集库位 result = await self._get_idle_crowded_site_from_db(group_name, filled, content, lock, map_id, context.task_record_id) # print(f"从密集库区中获取库位结果>>>>>: {result}<<<<<<=======================") if result.get("success", False) and result.get("data", {}).get("siteId"): # 获取成功 site_id = result.get("data", {}).get("siteId") # 设置上下文变量 context.set_variable("siteId", site_id) context.set_block_output(block.get("name"), {"siteId": site_id}) if attempts == 1: result["message"] = f"从密集库区中获取库位成功，库位ID: {site_id}" else: result["message"] = f"第{attempts}次重试从密集库区中获取库位成功，库位ID: {site_id}" break else: # 获取失败，判断是否需要重试 if retry and attempts < max_attempts: logger.info(f"从密集库区中获取库位失败，第 {attempts} 次重试，等待 {retry_sleep} 秒后重试") await asyncio.sleep(retry_sleep) attempts += 1 else: result["message"] = f"从密集库区中获取库位失败: {result.get('message', '未找到合适的库位')}" break # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"从密集库区中获取库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _get_idle_crowded_site_from_db(self, group_names: List[str], filled: bool, content: str, lock: bool, map_id: str, task_record_id: str) -> Dict[str, Any]: """从密集库区中获取库位""" try: async with get_async_session() as session: # 按库区优先级逐个查询 for area_name in group_names: logger.info(f"正在查询库区: {area_name}") # 先获取库区信息，确定选择库位的逻辑 area_query = select(StorageArea).where( StorageArea.area_name == area_name, StorageArea.scene_id == map_id, StorageArea.is_deleted == False ) area_result = await session.execute(area_query) area = area_result.scalar_one_or_none() if not area: logger.warning(f"库区 {area_name} 不存在，跳过") continue # 检查库区是否有锁定的库位 locked_count_query = select(OperatePointLayer).where( OperatePointLayer.area_name == area_name, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_locked == True, OperatePointLayer.is_disabled == False, OperatePointLayer.is_deleted == False ) locked_count_result = await session.execute(locked_count_query) locked_layers = locked_count_result.fetchall() if locked_layers: logger.info(f"库区 {area_name} 有 {len(locked_layers)} 个锁定的库位，跳过此库区") continue # 获取库区内所有库位，用于确定禁用库位的影响范围 all_layers_query = select(OperatePointLayer).where( OperatePointLayer.area_name == area_name, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_deleted == False ) all_layers_result = await session.execute(all_layers_query) all_layers = all_layers_result.scalars().all() # 自然排序函数，适配包含数字的库位名称 def natural_sort_key(layer_name): """自然排序键函数，将字符串中的数字部分转换为整数进行排序""" import re def convert(text): return int(text) if text.isdigit() else text.lower() return [convert(c) for c in re.split('([0-9]+)', layer_name)] import datetime # 对所有库位进行自然排序 all_layers.sort(key=lambda x: natural_sort_key(x.layer_name)) # 找出禁用的库位 disabled_layers = [layer for layer in all_layers if layer.is_disabled] # 构建查询条件 - 查找符合条件的库位 query = select(OperatePointLayer).where( OperatePointLayer.area_name == area_name, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False, OperatePointLayer.is_deleted == False, OperatePointLayer.is_locked == False, ) # 如果有禁用的库位，只能使用第一个禁用库位之后的库位 if disabled_layers: first_disabled_layer = disabled_layers[0] # 获取第一个禁用库位对象 logger.info(f"库区 {area_name} 存在禁用库位 {first_disabled_layer.layer_name}，只能使用其后的库位") # 获取第一个禁用库位在所有库位中的位置 disabled_position = None for i, layer in enumerate(all_layers): if layer.layer_name == first_disabled_layer.layer_name: disabled_position = i break if disabled_position is not None: # 获取禁用库位之后的所有库位名称 valid_layer_names = [layer.layer_name for layer in all_layers[disabled_position + 1:]] if valid_layer_names: # 添加条件：只能使用禁用库位之后的库位 query = query.where(OperatePointLayer.layer_name.in_(valid_layer_names)) logger.info(f"库区 {area_name} 可用库位范围: {valid_layer_names}") else: # 禁用库位之后没有库位了，跳过这个库区 logger.info(f"库区 {area_name} 禁用库位之后没有可用库位，跳过") continue else: logger.warning(f"库区 {area_name} 无法确定禁用库位位置，跳过") continue else: logger.info(f"库区 {area_name} 没有禁用库位，可以使用所有符合条件的库位") # 根据取放货需求添加条件 if filled: # 取货：需要有货的库位 query = query.where(OperatePointLayer.is_occupied == True) else: # 放货：需要没货的库位 query = query.where(OperatePointLayer.is_occupied == False) # 注意：不在数据库层面限制结果数量，因为需要在Python中进行自然排序 # 执行查询 result = await session.execute(query) candidate_layers = result.scalars().all() # 对候选库位进行自然排序，确保正确的顺序 if candidate_layers: if filled: # 取货：根据库区的选择逻辑进行排序，需要考虑时间 if area.select_logic == 1: # 先进先出：按存放时间升序，时间相同时按库位名称自然排序 candidate_layers.sort(key=lambda x: ( x.goods_stored_at or datetime.datetime.min, natural_sort_key(x.layer_name) )) elif area.select_logic == 2: # 先进后出：按存放时间降序，时间相同时按库位名称自然排序 candidate_layers.sort(key=lambda x: ( -(x.goods_stored_at or datetime.datetime.min).timestamp(), natural_sort_key(x.layer_name) )) else: # 默认按库位名称自然排序 candidate_layers.sort(key=lambda x: natural_sort_key(x.layer_name)) else: # 放货：只按库位名称自然排序，不考虑时间 candidate_layers.sort(key=lambda x: natural_sort_key(x.layer_name)) layer = candidate_layers[0] # 取第一个 else: layer = None if layer: site_id = layer.layer_name logger.info(f"在库区 {area_name} 中找到合适的库位: {site_id}") # 如果需要锁定库位 if lock: try: # 锁定库位 import datetime lock_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_locked == False # 乐观锁 ).values( is_locked=True, locked_by=f"task_{task_record_id}", last_access_at=datetime.datetime.now() ) lock_result = await session.execute(lock_stmt) if lock_result.rowcount > 0: # 如果是放货且指定了货物内容，更新货物内容 if not filled and content: content_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( goods_content=content, goods_stored_at=datetime.datetime.now(), last_access_at=datetime.datetime.now() ) await session.execute(content_stmt) await session.commit() return { "success": True, "message": f"从密集库区 {area_name} 获取库位成功并已锁定，库位ID: {site_id}", "data": { "siteId": site_id, "areaName": area_name, "locked": True } } else: # 锁定失败，可能被其他进程抢占 logger.warning(f"库位 {site_id} 锁定失败，可能被其他进程抢占") continue except Exception as e: logger.error(f"锁定库位 {site_id} 失败: {str(e)}") continue else: # 不需要锁定，直接返回 # 如果是放货且指定了货物内容，更新货物内容 if not filled and content: try: import datetime content_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( goods_content=content, goods_stored_at=datetime.datetime.now(), last_access_at=datetime.datetime.now() ) await session.execute(content_stmt) await session.commit() except Exception as e: logger.error(f"更新库位 {site_id} 货物内容失败: {str(e)}") # 即使更新失败，也返回库位信息 return { "success": True, "message": f"从密集库区 {area_name} 获取库位成功，库位ID: {site_id}", "data": { "siteId": site_id, "areaName": area_name, "locked": False } } else: logger.info(f"库区 {area_name} 中没有找到合适的库位") continue # 所有库区都没有找到合适的库位 return { "success": False, "message": f"在指定的库区集 {group_names} 中没有找到合适的库位" } except Exception as e: logger.error(f"从密集库区中获取库位异常: {str(e)}") return { "success": False, "message": f"从密集库区中获取库位异常: {str(e)}" } # 获取库位处理器 @register_handler(StorageBlockName.GET_IDLE_SITE) class GetIdleSiteBlockHandler(StorageBlockHandler): """获取库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """执行获取库位操作""" try: # 获取关键参数用于验证 locked = input_params.get("locked") retry_period = input_params.get("retryPeriod") # 获取地图ID map_id = context.map_id logger.info(f"获取库位处理器参数: {input_params}") # 必填参数检查 if locked is None: result = { "success": False, "message": "是否已锁定参数不能为空" } await self._record_task_log(block, result, context) return result # 确保retry_period是数字类型 if retry_period is not None: try: retry_period = float(retry_period) retry_sleep = retry_period / 1000 except (ValueError, TypeError): logger.warning(f"retry_period参数类型错误或无法转换为数字: {retry_period}, 将使用默认值1000") retry_sleep = 1 else: retry_sleep = 1 # 设置重试计数器（用于日志记录） retry_count = 0 while True: # 无限循环，直到找到库位 # 尝试获取库位 result = await self._query_idle_site_from_db(input_params, context.task_record_id, map_id) logger.info(f"获取库位处理器结果: {result}") if result.get("success", False) and result.get("data", {}).get("siteId"): # 获取成功 site_id = result.get("data", {}).get("siteId") # 设置上下文变量 context.set_variable("siteId", site_id) context.set_block_output(block.get("name"), {"siteId": site_id}) # 根据重试次数设置不同的成功消息 if retry_count == 0: result["message"] = f"获取库位成功，库位ID: {site_id}" else: result["message"] = f"第{retry_count}次重试获取库位成功，库位ID: {site_id}" # 记录执行结果并退出循环 await self._record_task_log(block, result, context) return result else: # 获取失败，记录日志并继续重试 retry_count += 1 logger.info(f"获取库位失败，第{retry_count}次重试，等待 {retry_sleep} 秒后继续") # 更新临时结果消息（仅用于日志记录） temp_result = { "success": False, "message": f"第{retry_count}次尝试获取库位失败: {result.get('message', '未找到合适的库位')}，将继续重试" } await self._record_task_log(block, temp_result, context) # 等待指定时间后继续重试 await asyncio.sleep(retry_sleep) except Exception as e: result = { "success": False, "message": f"获取库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _query_idle_site_from_db(self, input_params: Dict[str, Any], task_record_id: str, map_id:str) -> Dict[str, Any]: """从数据库查询空闲库位""" try: # 解析输入参数 site_id = input_params.get("siteId") # 库位ID content = input_params.get("content") # 货物 filled = input_params.get("filled") # 是否有货物 locked = input_params.get("locked") # 是否已锁定（必填） group_name = input_params.get("groupName") # 库区名 order_desc = input_params.get("orderDesc", True) # 是否为降序 lock_after_get = input_params.get("lock") # 获取库位后是否锁定 # 转换字符串类型的布尔值 if isinstance(filled, str): filled = filled.lower() in ('true', '1', 'yes') if isinstance(locked, str): locked = locked.lower() in ('true', '1', 'yes') if isinstance(order_desc, str): order_desc = order_desc.lower() in ('true', '1', 'yes') if isinstance(lock_after_get, str): # 如果是空字符串，映射为 false if lock_after_get.strip() == "": lock_after_get = False else: lock_after_get = lock_after_get.lower() in ('true', '1', 'yes') logger.info(f"查询库位参数: site_id={site_id}, content={content}, filled={filled}, locked={locked}, group_name={group_name}, order_desc={order_desc}, lock_after_get={lock_after_get}") # 自然排序函数，适配包含数字的库位名称 def natural_sort_key(layer_name): """自然排序键函数，将字符串中的数字部分转换为整数进行排序""" import re def convert(text): return int(text) if text.isdigit() else text.lower() return [convert(c) for c in re.split('([0-9]+)', layer_name)] async with get_async_session() as session: # 构建查询条件 query = select(OperatePointLayer) conditions = [] # 必填条件：是否已锁定 conditions.append(OperatePointLayer.is_locked == locked) conditions.append(OperatePointLayer.is_disabled == False) conditions.append(OperatePointLayer.scene_id == map_id) # 可选条件：库位ID if site_id: conditions.append(OperatePointLayer.layer_name == site_id) # 可选条件：货物内容 if content: conditions.append(OperatePointLayer.goods_content == content) # 可选条件：是否有货物 if filled: conditions.append(OperatePointLayer.is_occupied == filled) # 可选条件：库区名 if group_name: conditions.append(OperatePointLayer.area_name == group_name) # 应用所有条件 if conditions: query = query.where(and_(*conditions)) # 不在数据库层面排序，改为在Python中使用自然排序 # 执行查询获取所有候选库位 result = await session.execute(query) candidate_layers = result.scalars().all() # 如果有候选库位，使用自然排序 if candidate_layers: # 使用自然排序，确保正确的顺序 candidate_layers.sort(key=lambda x: natural_sort_key(x.layer_name)) # 根据order_desc参数决定排序方向 if order_desc: candidate_layers.reverse() # 取第一个库位 layer = candidate_layers[0] else: layer = None if layer: found_site_id = layer.layer_name # 如果需要在获取后锁定库位 if lock_after_get: try: # 更新库位锁定状态 from sqlalchemy import update update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == found_site_id ).values( is_locked=True, locked_by=f"task_{task_record_id}" # 生成一个锁定者ID ) await session.execute(update_stmt) await session.commit() logger.info(f"库位 {found_site_id} 已成功锁定") return { "success": True, "message": f"查询库位成功并已锁定，库位ID: {found_site_id}", "data": { "siteId": found_site_id } } except Exception as lock_error: logger.error(f"锁定库位失败: {str(lock_error)}") # 即使锁定失败，也返回查询成功的结果 return { "success": True, "message": f"查询库位成功，但锁定失败: {str(lock_error)}，库位ID: {found_site_id}", "data": { "siteId": found_site_id } } else: # 不需要锁定，直接返回查询结果 return { "success": True, "message": f"查询库位成功，库位ID: {found_site_id}", "data": { "siteId": found_site_id } } else: # 未找到匹配的库位 return { "success": False, "message": "未找到符合条件的库位" } except Exception as e: logger.error(f"查询库位异常: {str(e)}") return { "success": False, "message": f"查询库位异常: {str(e)}" } # 获取任务实例加锁库位处理器 @register_handler(StorageBlockName.GET_LOCKED_SITES_BY_TASK_RECORD_ID) class GetLockedSitesByTaskRecordIdBlockHandler(StorageBlockHandler): """根据任务实例ID获取所有加锁库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """获取任务实例关联的所有已锁库位""" try: # 获取参数进行验证 task_record_id = input_params.get("taskRecordId") # 必填参数检查 if not task_record_id: result = { "success": False, "message": "任务实例ID不能为空" } await self._record_task_log(block, result, context) return result # 直接调用外部API获取库位列表 result = await self._call_external_api("get_locked_sites_by_task_record_id", input_params) if result.get("success", False): # 获取成功，设置上下文变量 locked_site_list = result.get("data", {}).get("lockedSiteIdList", []) context.set_variable("lockedSiteIdList", locked_site_list) context.set_block_output(block.get("name"), {"lockedSiteIdList": locked_site_list}) result["message"] = f"获取任务锁定库位成功，共 {len(locked_site_list)} 个库位" else: result["message"] = f"获取任务锁定库位失败: {result.get('message', '未知错误')}" # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"获取任务锁定库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result # 获取库位扩展属性处理器 @register_handler(StorageBlockName.GET_SITE_ATTR) class GetSiteAttrBlockHandler(StorageBlockHandler): """获取库位扩展属性处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """获取库位扩展属性值""" try: # 获取参数进行验证 site_id = input_params.get("siteId") attr_name = input_params.get("attrName") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result if not attr_name: result = { "success": False, "message": "属性名称不能为空" } await self._record_task_log(block, result, context) return result # 直接从数据库获取扩展属性值 result = await self._get_site_attr_from_db(site_id, attr_name, map_id) if result.get("success", False): # 获取成功，设置上下文变量 attr_value = result.get("data", {}).get("attrValue") context.set_variable("attrValue", attr_value) context.set_block_output(block.get("name"), {"attrValue": attr_value}) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"获取库位属性值执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _get_site_attr_from_db(self, site_id: str, attr_name: str, map_id: str) -> Dict[str, Any]: """从数据库获取库位扩展属性值""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.is_deleted == False, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查扩展属性是否已定义 extended_property_query = select(ExtendedProperty).where( ExtendedProperty.property_name == attr_name, ExtendedProperty.is_deleted == False ) extended_property_result = await session.execute(extended_property_query) extended_property = extended_property_result.scalar_one_or_none() if not extended_property: return { "success": False, "message": f"扩展属性 '{attr_name}' 不存在" } # 解析config_json获取扩展属性值 import json attr_value = None attr_type = extended_property.property_type.value is_required = extended_property.is_required if existing_layer.config_json: try: config = json.loads(existing_layer.config_json) if 'extended_fields' in config and attr_name in config['extended_fields']: attr_info = config['extended_fields'][attr_name] attr_value = attr_info.get('value') # 如果config中有类型信息，使用config中的类型 if 'type' in attr_info: attr_type = attr_info['type'] # 如果config中有required信息，使用config中的required if 'is_required' in attr_info: is_required = attr_info['is_required'] except Exception as e: logger.error(f"解析库位层 {site_id} 的config_json失败: {str(e)}") # 如果没有找到属性值，使用默认值 if attr_value is None: attr_value = extended_property.default_value logger.info(f"获取库位 {site_id} 扩展属性 {attr_name} 成功: {attr_value}") return { "success": True, "message": f"获取库位扩展属性成功，{attr_name}: {attr_value}", "data": { "attrValue": attr_value, "attrType": attr_type, "isRequired": is_required, "propertyName": attr_name } } except Exception as e: logger.error(f"获取库位扩展属性异常: {str(e)}") return { "success": False, "message": f"获取库位扩展属性异常: {str(e)}" } # 查询库位处理器 @register_handler(StorageBlockName.QUERY_IDLE_SITE) class QueryIdleSiteBlockHandler(StorageBlockHandler): """查询库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """查询库位""" try: # print(f"查询库位处理器参数>>>>>: {input_params}<<<<<<=======================") # 获取地图ID map_id = context.map_id # 获取参数 site_id = input_params.get("siteId") content = input_params.get("content") filled = input_params.get("filled") locked = input_params.get("locked") group_name = input_params.get("groupName") order_desc = input_params.get("orderDesc", True) # 参数类型转换 if filled is not None: if isinstance(filled, str): filled = filled.lower() in ('true', '1', 'yes', 'on') if locked is not None: if isinstance(locked, str): locked = locked.lower() in ('true', '1', 'yes', 'on') if isinstance(order_desc, str): order_desc = order_desc.lower() in ('true', '1', 'yes', 'on') # 执行数据库查询 result = await self._query_site_from_db(input_params, map_id, site_id, content, filled, locked, group_name, order_desc) if result.get("success", False) and result.get("data", {}).get("site"): # 查询成功，设置上下文变量 site = result.get("data", {}).get("site") context.set_variable("site", site) context.set_block_output(block.get("name"), {"site": site}) site_id = site.get("layer_name", "未知") result["message"] = f"查询库位成功，库位ID: {site_id}" else: result["message"] = f"查询库位失败: {result.get('message', '未找到符合条件的库位')}" # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"查询库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _query_site_from_db(self, input_params: Dict[str, Any], map_id: str, site_id: str, content: str, filled: bool, locked: bool, group_name: str, order_desc: bool) -> Dict[str, Any]: """从数据库查询库位""" try: async with get_async_session() as session: # 构建查询条件 query = select(OperatePointLayer) conditions = [] # 基础条件 conditions.append(OperatePointLayer.is_deleted == False) conditions.append(OperatePointLayer.scene_id == map_id) conditions.append(OperatePointLayer.is_disabled == False) # 可选条件：库位ID if site_id: conditions.append(OperatePointLayer.layer_name == site_id) # 可选条件：货物内容 if content: conditions.append(OperatePointLayer.goods_content == content) # 可选条件：是否有货物/占用状态 if filled is not None: conditions.append(OperatePointLayer.is_occupied == filled) # 可选条件：是否锁定 if locked is not None: conditions.append(OperatePointLayer.is_locked == locked) # 可选条件：库区名 if group_name: conditions.append(OperatePointLayer.area_name == group_name) # 应用所有条件 if conditions: query = query.where(and_(*conditions)) # 排序：按station_name排序 if order_desc: query = query.order_by(desc(OperatePointLayer.layer_name)) else: query = query.order_by(asc(OperatePointLayer.layer_name)) # 只获取第一个匹配的库位 query = query.limit(1) logger.info(f"查询库位SQL条件: site_id={site_id}, content={content}, filled={filled}, locked={locked}, group_name={group_name}, order_desc={order_desc}") # 执行查询 result = await session.execute(query) layer = result.scalar_one_or_none() if layer: # 构造返回的库位信息 site_info = { "layer_name": layer.layer_name, } return { "success": True, "message": f"查询库位成功，库位ID: {layer.layer_name}", "data": { "site": site_info } } else: # 未找到匹配的库位 return { "success": False, "message": "未找到符合条件的库位" } except Exception as e: logger.error(f"查询库位异常: {str(e)}") return { "success": False, "message": f"查询库位异常: {str(e)}" } # 设置库位扩展属性处理器 @register_handler(StorageBlockName.SET_SITE_ATTR) class SetSiteAttrBlockHandler(StorageBlockHandler): """设置库位扩展属性处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """设置库位扩展属性值""" try: # 获取参数进行验证 site_id = input_params.get("siteId") attr_name = input_params.get("attrName") attr_value = input_params.get("attrValue") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result if not attr_name: result = { "success": False, "message": "属性名称不能为空" } await self._record_task_log(block, result, context) return result # 直接操作数据库设置扩展属性值 result = await self._set_site_attr_in_db(site_id, attr_name, attr_value, map_id) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"设置库位属性值执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_attr_in_db(self, site_id: str, attr_name: str, attr_value: Any, map_id:str) -> Dict[str, Any]: """在数据库中设置库位扩展属性值""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.is_deleted == False, OperatePointLayer.scene_id == map_id ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查扩展属性是否已定义 extended_property_query = select(ExtendedProperty).where( ExtendedProperty.property_name == attr_name, ExtendedProperty.is_deleted == False ) extended_property_result = await session.execute(extended_property_query) extended_property = extended_property_result.scalar_one_or_none() # 如果扩展属性不存在，则创建新的扩展属性 if not extended_property: try: # 推断属性类型 # property_type = self._infer_property_type(attr_value) # 创建新的扩展属性 extended_property = ExtendedProperty( property_key=attr_name, property_name=attr_name, property_type=ExtendedPropertyTypeEnum.STRING, ) session.add(extended_property) await session.commit() await session.refresh(extended_property) logger.info(f"自动创建扩展属性: {attr_name}") except Exception as e: logger.error(f"创建扩展属性失败: {str(e)}") return { "success": False, "message": f"创建扩展属性失败: {str(e)}" } # 如果扩展属性存在但被禁用，启用它 elif not extended_property.is_enabled: try: # 启用扩展属性 enable_stmt = update(ExtendedProperty).where( ExtendedProperty.id == extended_property.id ).values( is_enabled=True, updated_at=datetime.datetime.now() ) await session.execute(enable_stmt) await session.commit() # 刷新对象状态 await session.refresh(extended_property) logger.info(f"启用扩展属性: {attr_name}") except Exception as e: logger.error(f"启用扩展属性失败: {str(e)}") return { "success": False, "message": f"启用扩展属性失败: {str(e)}" } # 解析现有的config_json import json import datetime config = {} if existing_layer.config_json: try: config = json.loads(existing_layer.config_json) except Exception as e: logger.error(f"解析库位层 {site_id} 的config_json失败: {str(e)}") config = {} # 确保extended_fields字段存在 if 'extended_fields' not in config: config['extended_fields'] = {} # 更新扩展属性值 config['extended_fields'][attr_name] = { 'value': attr_value, 'type': extended_property.property_type.value, 'is_required': extended_property.is_required, 'updated_at': datetime.datetime.now().isoformat() } # 更新config_json try: updated_config_json = json.dumps(config, ensure_ascii=False, indent=2) # 更新数据库 update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id ).values( config_json=updated_config_json, updated_at=datetime.datetime.now() ) await session.execute(update_stmt) await session.commit() value_display = attr_value if attr_value is not None else "null" logger.info(f"库位 {site_id} 扩展属性 {attr_name} 已成功设置为: {value_display}") return { "success": True, "message": f"设置库位扩展属性成功，{site_id}.{attr_name} = {value_display}" } except Exception as e: logger.error(f"序列化库位层 {site_id} 的config_json失败: {str(e)}") return { "success": False, "message": f"更新扩展属性失败: {str(e)}" } except Exception as e: logger.error(f"设置库位扩展属性异常: {str(e)}") return { "success": False, "message": f"设置库位扩展属性异常: {str(e)}" } # 设置库位货物处理器 @register_handler(StorageBlockName.SET_SITE_CONTENT) class SetSiteContentBlockHandler(StorageBlockHandler): """设置库位货物处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """设置库位货物""" try: # 获取参数进行验证 site_id = input_params.get("siteId") content = input_params.get("content") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result if content is None: result = { "success": False, "message": "货物内容不能为空" } await self._record_task_log(block, result, context) return result # 直接操作数据库设置库位货物 result = await self._set_site_content_in_db(site_id, content, map_id) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"设置库位货物执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_content_in_db(self, site_id: str, content: str, map_id:str) -> Dict[str, Any]: """在数据库中设置库位货物内容""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 库位存在，设置货物内容 from sqlalchemy import update import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( goods_content=content, is_occupied=True, goods_stored_at=datetime.datetime.now(), # 记录货物存放时间 last_access_at=datetime.datetime.now() # 更新最后访问时间 ) await session.execute(update_stmt) await session.commit() logger.info(f"库位 {site_id} 货物内容已成功设置为: {content}") return { "success": True, "message": f"设置库位货物成功，库位ID: {site_id}，货物内容: {content}" } except Exception as e: logger.error(f"设置库位货物异常: {str(e)}") return { "success": False, "message": f"设置库位货物异常: {str(e)}" } # 设置库位为空处理器 @register_handler(StorageBlockName.SET_SITE_EMPTY) class SetSiteEmptyBlockHandler(StorageBlockHandler): """设置库位为空处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """设置库位为空""" try: # 获取参数进行验证 site_id = input_params.get("siteId") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result result = await self._set_site_empty_in_db(site_id, map_id) if result.get("success", False): result["message"] = f"设置库位为空成功，库位ID: {site_id}" else: result["message"] = f"设置库位为空失败: {result.get('message', '未知错误')}" # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"设置库位为空执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_empty_in_db(self, site_id: str, map_id:str) -> Dict[str, Any]: """在数据库中设置库位为非占用状态""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查库位是否已经被占用 if not existing_layer.is_occupied: return { "success": False, "message": f"库位 {site_id} 已处于非占用状态，无法重复设置" } # 库位存在且未被占用，设置为占用状态 from sqlalchemy import update import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( is_occupied=False, goods_content='', # 清空货物内容 goods_retrieved_at=datetime.datetime.now(), # 记录货物取出时间 last_access_at=datetime.datetime.now() ) await session.execute(update_stmt) await session.commit() logger.info(f"库位 {site_id} 已成功设置为空闲状态") return { "success": True, "message": f"设置库位为空成功，库位ID: {site_id}" } except Exception as e: logger.error(f"设置库位为非占用异常: {str(e)}") return { "success": False, "message": f"设置库位为非占用异常: {str(e)}" } # 设置库位为占用处理器 @register_handler(StorageBlockName.SET_SITE_FILLED) class SetSiteFilledBlockHandler(StorageBlockHandler): """设置库位为占用处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """设置库位为占用""" try: # 获取参数进行验证 site_id = input_params.get("siteId") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result # 直接操作数据库设置库位为占用 result = await self._set_site_filled_in_db(site_id, map_id) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"设置库位为占用执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_filled_in_db(self, site_id: str, map_id:str) -> Dict[str, Any]: """在数据库中设置库位为占用状态""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查库位是否已经被占用 if existing_layer.is_occupied: return { "success": False, "message": f"库位 {site_id} 已处于占用状态，无法重复设置" } # 库位存在且未被占用，设置为占用状态 from sqlalchemy import update import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( is_occupied=True, goods_content='', goods_stored_at=datetime.datetime.now(), last_access_at=datetime.datetime.now(), ) await session.execute(update_stmt) await session.commit() logger.info(f"库位 {site_id} 已成功设置为占用状态") return { "success": True, "message": f"设置库位为占用成功，库位ID: {site_id}" } except Exception as e: logger.error(f"设置库位为占用异常: {str(e)}") return { "success": False, "message": f"设置库位为占用异常: {str(e)}" } # 锁定库位处理器 @register_handler(StorageBlockName.SET_SITE_LOCKED) class SetSiteLockedBlockHandler(StorageBlockHandler): """锁定库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """锁定库位""" try: # 获取参数进行验证 site_id = input_params.get("siteId") locked_id = input_params.get("lockedId") if_fair = input_params.get("ifFair", False) # 是否为公平锁 retry_times = input_params.get("retryTimes", 0) # 重试次数，默认为0表示不重试 map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result # 如果未指定锁定者ID，则使用当前任务ID if not locked_id: locked_id = context.block_record_id input_params["lockedId"] = locked_id # 确保retry_times是整数类型 try: retry_times = int(retry_times) except (ValueError, TypeError): logger.warning(f"retryTimes参数类型错误或无法转换为整数: {retry_times}, 将使用默认值0") retry_times = 0 logger.info(f"锁定库位参数: site_id={site_id}, locked_id={locked_id}, if_fair={if_fair}, retry_times={retry_times}") # 直接从数据库锁定库位 result = await self._set_site_locked_in_db(site_id, locked_id, if_fair, retry_times, map_id) if result.get("success", False): # 设置上下文变量 context.set_variable("success", True) context.set_block_output(block.get("name"), {"success": True}) else: # 设置上下文变量 context.set_variable("success", False) context.set_block_output(block.get("name"), {"success": False}) # 记录最终执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"锁定库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_locked_in_db(self, site_id: str, locked_id: str, if_fair: bool, retry_times: int, map_id:str) -> Dict[str, Any]: """在数据库中锁定库位""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 实现重试逻辑 attempts = 0 max_attempts = retry_times + 1 # 包括首次尝试 retry_sleep = 1 # 默认重试间隔为1秒 while attempts < max_attempts: attempts += 1 # 重新查询库位当前状态 current_site_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False, # OperatePointLayer.is_empty_tray == False ) current_site_result = await session.execute(current_site_query) current_layer = current_site_result.scalar_one_or_none() if not current_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查库位是否已被锁定 if current_layer.is_locked: # 如果是同一个锁定者，视为已锁定成功 if current_layer.locked_by == locked_id: logger.info(f"库位 {site_id} 已被同一锁定者 {locked_id} 锁定") return { "success": True, "message": f"库位已被同一锁定者锁定，库位ID: {site_id}" } # 被其他锁定者锁定，需要重试 if attempts < max_attempts: logger.info(f"库位 {site_id} 已被其他锁定者 {current_layer.locked_by} 锁定，第 {attempts} 次尝试，等待 {retry_sleep} 秒后重试") await asyncio.sleep(retry_sleep) continue else: # 达到最大重试次数，返回失败 return { "success": False, "message": f"库位 {site_id} 已被其他锁定者 {current_layer.locked_by} 锁定，重试 {retry_times} 次后仍然失败" } # 库位未被锁定，尝试锁定 try: import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_locked == False # 乐观锁：只有未锁定时才能锁定 ).values( is_locked=True, locked_by=locked_id, last_access_at=datetime.datetime.now() ) result = await session.execute(update_stmt) # 检查是否成功更新 if result.rowcount > 0: await session.commit() if attempts == 1: success_msg = f"锁定库位成功，库位ID: {site_id}" else: success_msg = f"第{attempts}次尝试锁定库位成功，库位ID: {site_id}" logger.info(success_msg) return { "success": True, "message": success_msg } else: # 更新失败，可能在更新期间被其他进程锁定 if attempts < max_attempts: logger.info(f"库位 {site_id} 锁定失败（可能被其他进程抢占），第 {attempts} 次尝试，等待 {retry_sleep} 秒后重试") await asyncio.sleep(retry_sleep) continue else: return { "success": False, "message": f"库位 {site_id} 锁定失败，可能被其他进程抢占，重试 {retry_times} 次后仍然失败" } except Exception as e: logger.error(f"锁定库位时发生异常: {str(e)}") if attempts < max_attempts: logger.info(f"第 {attempts} 次尝试发生异常，等待 {retry_sleep} 秒后重试") await asyncio.sleep(retry_sleep) continue else: return { "success": False, "message": f"锁定库位时发生异常: {str(e)}" } # 不应该到达这里 return { "success": False, "message": f"锁定库位失败，未知错误" } except Exception as e: logger.error(f"锁定库位异常: {str(e)}") return { "success": False, "message": f"锁定库位异常: {str(e)}" } # 设置库位标签处理器 @register_handler(StorageBlockName.SET_SITE_TAGS) class SetSiteTagsBlockHandler(StorageBlockHandler): """设置库位标签处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """设置库位标签""" try: # 获取参数进行验证 site_id = input_params.get("siteId") tags = input_params.get("tags") replace_mode = input_params.get("replaceMode", False) # 是否替换模式，默认为追加模式 map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result # 转换字符串类型的布尔值 if isinstance(replace_mode, str): replace_mode = replace_mode.lower() in ('true', '1', 'yes') logger.info(f"设置库位标签参数: site_id={site_id}, tags={tags}, replace_mode={replace_mode}") # 直接从数据库设置库位标签 result = await self._set_site_tags_in_db(site_id, tags, replace_mode, map_id) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"设置库位标签执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_tags_in_db(self, site_id: str, tags: str, replace_mode: bool = False, map_id:str=None) -> Dict[str, Any]: """在数据库中设置库位标签""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 处理标签数据 new_tags_set = set() # 解析新输入的标签 if tags: # 支持多种分隔符：逗号、分号、空格等 import re tag_list = re.split(r'[,，;；\s]+', str(tags).strip()) new_tags_set.update(tag.strip() for tag in tag_list if tag.strip()) # 处理现有标签 final_tags_set = set() if not replace_mode and existing_layer.tags: # 追加模式：合并现有标签 existing_tag_list = re.split(r'[,，;；\s]+', existing_layer.tags.strip()) final_tags_set.update(tag.strip() for tag in existing_tag_list if tag.strip()) # 合并新标签 final_tags_set.update(new_tags_set) # 去重并用逗号连接 final_tags = ','.join(sorted(final_tags_set)) if final_tags_set else '' # 执行更新操作 try: import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id ).values( tags=final_tags, last_access_at=datetime.datetime.now() ) result = await session.execute(update_stmt) # 检查是否成功更新 if result.rowcount > 0: await session.commit() tags_display = final_tags if final_tags else "空" operation_mode = "替换" if replace_mode else "追加" success_msg = f"设置库位标签成功（{operation_mode}模式），库位ID: {site_id}，标签: {tags_display}" logger.info(success_msg) return { "success": True, "message": success_msg } else: # 更新失败 return { "success": False, "message": f"设置库位标签失败，库位 {site_id} 可能不存在或已被删除" } except Exception as e: logger.error(f"设置库位标签时发生异常: {str(e)}") return { "success": False, "message": f"设置库位标签时发生异常: {str(e)}" } except Exception as e: logger.error(f"设置库位标签异常: {str(e)}") return { "success": False, "message": f"设置库位标签异常: {str(e)}" } # 解锁库位处理器 @register_handler(StorageBlockName.SET_SITE_UNLOCKED) class SetSiteUnlockedBlockHandler(StorageBlockHandler): """解锁库位处理器""" async def execute( self, block: Dict[str, Any], input_params: Dict[str, Any], context: TaskContext ) -> Dict[str, Any]: """解锁库位""" try: # 获取参数进行验证 site_id = input_params.get("siteId") un_locked_id = input_params.get("unLockedId") map_id = context.map_id # 必填参数检查 if not site_id: result = { "success": False, "message": "库位ID不能为空" } await self._record_task_log(block, result, context) return result # 如果未指定解锁者ID，则使用当前任务ID if not un_locked_id: un_locked_id = context.get_task_record_id() input_params["unLockedId"] = un_locked_id logger.info(f"未指定解锁者ID，使用当前任务ID: {un_locked_id}") logger.info(f"解锁库位参数: site_id={site_id}, un_locked_id={un_locked_id}") # 直接从数据库解锁库位 result = await self._set_site_unlocked_in_db(site_id, un_locked_id, map_id) if result.get("success", False): # 设置上下文变量 context.set_variable("unlockSuccess", True) context.set_block_output(block.get("name"), {"unlockSuccess": True}) else: # 设置上下文变量 context.set_variable("unlockSuccess", False) context.set_block_output(block.get("name"), {"unlockSuccess": False}) # 记录执行结果 await self._record_task_log(block, result, context) return result except Exception as e: result = { "success": False, "message": f"解锁库位执行异常: {str(e)}" } # 记录异常 await self._record_task_log(block, result, context) return result async def _set_site_unlocked_in_db(self, site_id: str, un_locked_id: str, map_id:str) -> Dict[str, Any]: """在数据库中解锁库位""" try: async with get_async_session() as session: # 先检查库位是否存在 site_check_query = select(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_disabled == False ) site_check_result = await session.execute(site_check_query) existing_layer = site_check_result.scalar_one_or_none() if not existing_layer: return { "success": False, "message": f"库位不存在: {site_id}" } # 检查库位是否已被锁定 if not existing_layer.is_locked: # 库位未被锁定，直接返回成功 logger.info(f"库位 {site_id} 未被锁定，解锁操作视为成功") return { "success": True, "message": f"库位未被锁定，解锁操作成功，库位ID: {site_id}" } # 检查解锁者权限 if existing_layer.locked_by != un_locked_id: # 解锁者不是当前锁定者，执行失败 logger.error(f"解锁者 {un_locked_id} 不是库位 {site_id} 的锁定者 {existing_layer.locked_by}，解锁失败") return { "success": False, "message": f"解锁失败，解锁者 {un_locked_id} 不是库位 {site_id} 的锁定者 {existing_layer.locked_by}" } # 执行解锁操作 try: import datetime update_stmt = update(OperatePointLayer).where( OperatePointLayer.layer_name == site_id, OperatePointLayer.scene_id == map_id, OperatePointLayer.is_locked == True # 只有锁定的库位才能解锁 ).values( is_locked=False, locked_by=None, # 清空锁定者 last_access_at=datetime.datetime.now() ) result = await session.execute(update_stmt) # 检查是否成功更新 if result.rowcount > 0: await session.commit() success_msg = f"解锁库位成功，库位ID: {site_id}" logger.info(success_msg) return { "success": True, "message": success_msg } else: # 更新失败，可能在更新期间库位状态发生变化 return { "success": False, "message": f"解锁库位失败，库位 {site_id} 可能已被其他进程解锁" } except Exception as e: logger.error(f"解锁库位时发生异常: {str(e)}") return { "success": False, "message": f"解锁库位时发生异常: {str(e)}" } except Exception as e: logger.error(f"解锁库位异常: {str(e)}") return { "success": False, "message": f"解锁库位异常: {str(e)}" }