#!/usr/bin/env python # -*- coding: utf-8 -*- """ 任务编辑服务模块 提供任务编辑相关的业务逻辑 """ import json import uuid from typing import Dict, List, Any, Optional from datetime import datetime from sqlalchemy import select, and_, func, update from routes.model.task_edit_model import ( TaskBasicInfo, TaskBackupRequest, SubTaskListParams, TaskEditRunRequest ) from data.models.taskdef import VWEDTaskDef from data.models.taskrecord import VWEDTaskRecord from data.session import get_async_session from utils.component_manager import component_manager from data.enum.task_def_enum import EnableStatus, PeriodicTaskStatus, TaskStatusEnum from data.enum.task_record_enum import TaskStatus from utils.logger import get_logger # 获取日志记录器 logger = get_logger("services.task_edit_service") class TaskEditService: """任务编辑服务类""" @staticmethod async def get_blocks() -> List[Dict[str, Any]]: """ 获取系统中所有可用的块数据 用于任务编辑页面左侧的组件面板显示 Returns: List[Dict[str, Any]]: 所有组件分类及其组件 """ try: # 通过组件管理器获取所有块数据 blocks = component_manager.get_all_blocks() return blocks except Exception as e: logger.error(f"获取块数据失败: {str(e)}") raise @staticmethod async def get_task_source(task_id: str) -> Optional[Dict[str, Any]]: """ 获取指定任务的源码详情数据 Args: task_id: 任务ID Returns: Optional[Dict[str, Any]]: 任务详情数据 """ try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.warning(f"任务不存在: {task_id}") return None # 解析详情JSON detail = json.loads(task_def.detail) if task_def.detail else {} # 构建响应数据 response = { "id": task_def.id, "label": task_def.label, "version": task_def.version, "detail": detail, "delay": task_def.delay, "ifEnable": task_def.if_enable, "period": task_def.period, "periodicTask": task_def.periodic_task, "remark": task_def.remark, "status": task_def.status, "templateName": task_def.template_name, "templateDescription": task_def.template_name, # 使用模板名称作为描述 "createDate": task_def.created_at, "createdBy": task_def.created_by } return response except Exception as e: logger.log_error_with_trace("获取任务源码详情失败", e) raise @staticmethod async def save_task_edit(task_def_data: Dict[str, Any]) -> Dict[str, Any]: """ 保存任务编辑数据 Args: task_def_data: 任务定义数据，已通过TaskSaveRequest模型验证 必须包含id和detail字段 label字段是可选的 detail字段已被验证包含inputParams、outputParams和rootBlock结构 Returns: Dict[str, Any]: 保存结果 """ try: task_id = task_def_data.get("id") if not task_id: logger.error("缺少任务ID") return {"code": 400, "message": "缺少任务ID", "success": False} # 在API层已经调用check_task_changes，此处不再重复检查 async with get_async_session() as session: # 查询任务是否存在 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {task_id}") return {"code": 404, "message": "任务不存在", "success": False} # 更新版本号 new_version = task_def.version + 1 # 只有当提供了label且非空时，才更新任务名称 label = task_def_data.get("label") if label and label.strip(): task_def.label = label # 转换detail为JSON字符串 - 使用ensure_ascii=False确保中文直接存储而不被编码 detail = task_def_data.get("detail") if detail: # detail字段已通过TaskDetailSave模型验证，可以直接转换为JSON字符串 task_def.detail = json.dumps(detail, ensure_ascii=False) task_def.version = new_version # 保存到数据库 await session.commit() # 获取当前时间作为更新时间 update_time = datetime.now() return { "code": 200, "message": "保存成功", "success": True, "id": task_def.id, "version": new_version, "updateTime": update_time } except Exception as e: logger.error(f"保存任务编辑数据失败: {str(e)}") raise @staticmethod async def backup_task(task_id: str, backup_request: TaskBackupRequest) -> Optional[Dict[str, Any]]: """ 备份任务 Args: task_id: 源任务ID backup_request: 备份请求 Returns: Optional[Dict[str, Any]]: 备份结果 """ try: async with get_async_session() as session: # 查询源任务 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) source_task = result.scalars().first() if not source_task: logger.error(f"源任务不存在: {task_id}") return None # 创建新任务ID new_task_id = str(uuid.uuid4()) # 设置备份任务名称 backup_name = backup_request.backupName # backup_reques if not backup_name: backup_name = f"{source_task.label}-备份" # 创建备份任务 backup_task = VWEDTaskDef( id=new_task_id, label=backup_name, version=1, detail=source_task.detail, delay=source_task.delay, if_enable=EnableStatus.DISABLED, # 默认禁用 period=source_task.period, periodic_task=source_task.periodic_task, remark=backup_request.remark, status=TaskStatusEnum.PENDING, template_name=source_task.template_name, created_by=source_task.created_by, tenant_id=source_task.tenant_id, release_sites=source_task.release_sites, map_id=source_task.map_id, user_token=source_task.user_token ) session.add(backup_task) await session.commit() return { "id": backup_task.id, "label": backup_task.label, "version": backup_task.version, "templateName": backup_task.template_name, "periodicTask": backup_task.periodic_task, "ifEnable": backup_task.if_enable, "status": backup_task.status, "createDate": backup_task.created_at, "sourceTaskId": source_task.id, "remark": backup_task.remark } except Exception as e: logger.error(f"备份任务失败: {str(e)}") raise @staticmethod async def get_subtasks_list(params: SubTaskListParams) -> Dict[str, Any]: """ 获取子任务列表 Args: params: 查询参数 Returns: Dict[str, Any]: 子任务列表数据 """ try: async with get_async_session() as session: # 构建查询条件 query = select(VWEDTaskDef) # 添加条件 conditions = [] # 排除指定ID if params.exclude_id: conditions.append(VWEDTaskDef.id != params.exclude_id) # 关键词搜索 if params.keyword: conditions.append(VWEDTaskDef.label.like(f"%{params.keyword}%")) if conditions: query = query.where(and_(*conditions)) # 计算总数 count_result = await session.execute(select( func.count(VWEDTaskDef.id)).where(query.whereclause)) total = count_result.scalar() or 0 # 分页 query = query.order_by(VWEDTaskDef.created_at.desc()) query = query.offset((params.pageNum - 1) * params.pageSize).limit(params.pageSize) # 执行查询 result = await session.execute(query) tasks = result.scalars().all() # 构建返回数据 task_list = [] for task in tasks: try: # 解析输入参数 detail = json.loads(task.detail) if task.detail else {} input_params = detail.get("inputParams", []) task_list.append({ "id": task.id, "label": task.label, "version": task.version, "templateName": task.template_name, "remark": task.remark, "createDate": task.created_at, "createdBy": task.created_by, "status": task.status, "inputParams": input_params }) except Exception as e: logger.error(f"解析子任务数据失败: {str(e)}") return { "total": total, "list": task_list } except Exception as e: logger.error(f"获取子任务列表失败: {str(e)}") raise @staticmethod async def run_task(run_request: TaskEditRunRequest, client_ip: str = None, client_info: str = None, tf_api_token: str = None) -> Optional[Dict[str, Any]]: """ 运行任务 Args: run_request: 运行任务请求 client_ip: 客户端IP地址 client_info: 客户端设备信息 tf_api_token: 系统任务令牌 Returns: Optional[Dict[str, Any]]: 运行结果 """ try: # 获取任务定义，从数据库加载任务 async with get_async_session() as session: # 查询任务是否存在 from sqlalchemy import select from data.models import VWEDTaskDef result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == run_request.taskId) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {run_request.taskId}") return None # 导入增强版任务调度器 from services.enhanced_scheduler import scheduler # 转换参数对象为字典格式，解决JSON序列化问题 params = [] if run_request.params: # 使用字典推导式将参数列表转换为字典 params = [param.model_dump() for param in run_request.params] # 获取当前时间作为任务运行时间 now = datetime.now() # 设置任务来源信息 source_type = run_request.source_type source_system = run_request.source_system source_device = run_request.source_device map_id = task_def.map_id source_ip = client_ip # 获取IP地址 source_time = now # 始终使用当前时间 # 记录任务请求信息 logger.info(f"准备启动任务: {run_request.taskId}, 来源: {source_system}, 设备: {source_device}") logger.debug(f"任务参数: {params}") # 区分定时任务和普通任务处理 if task_def.periodic_task == PeriodicTaskStatus.PERIODIC: # 定时任务处理流程 logger.info(f"启动定时任务: {run_request.taskId}") # 1. 确保定时任务启用 if task_def.if_enable != EnableStatus.ENABLED: # 如果定时任务未启用，先启用它 async with get_async_session() as session: task_def.if_enable = EnableStatus.ENABLED await session.commit() logger.info(f"已启用定时任务: {run_request.taskId}") # 2. 通知调度器更新定时任务状态 await scheduler.update_periodic_task(run_request.taskId, True) # 4. 更新任务定义状态为运行中(1) async with get_async_session() as session: from sqlalchemy import update await session.execute( update(VWEDTaskDef) .where(VWEDTaskDef.id == run_request.taskId) .values(status=TaskStatusEnum.RUNNING, user_token=tf_api_token) ) await session.commit() logger.info(f"更新任务定义状态为运行中: {run_request.taskId}") # 3. 立即执行一次任务 result = await scheduler.run_task( task_def_id=run_request.taskId, params=params, source_type=source_type, source_system=source_system, source_device=source_device, source_ip=source_ip, source_time=source_time, source_client_info=client_info, tf_api_token=tf_api_token, map_id=map_id ) if not result.get("success", False): logger.error(f"启动定时任务失败: {result.get('message')}") return None logger.info(f"定时任务启动成功: {run_request.taskId}, 记录ID: {result.get('taskRecordId')}") # 返回定时任务记录信息 return { "success": True, "taskRecordId": result.get("taskRecordId"), "status": result.get("status"), "createTime": result.get("createTime"), "isPeriodic": True, "period": task_def.period, "message": "定时任务已启动，将按照设定的周期自动执行" } else: # 普通任务处理流程 logger.info(f"启动普通任务: {run_request.taskId}") async with get_async_session() as session: from sqlalchemy import update await session.execute( update(VWEDTaskDef) .where(VWEDTaskDef.id == run_request.taskId) .values(user_token=tf_api_token) ) await session.commit() result = await scheduler.run_task( task_def_id=run_request.taskId, params=params, source_type=source_type, source_system=source_system, source_device=source_device, source_ip=source_ip, source_time=source_time, source_client_info=client_info, tf_api_token=tf_api_token, map_id=map_id ) if not result.get("success", False): logger.error(f"启动任务失败: {result.get('message')}") return result logger.info(f"普通任务启动成功: {run_request.taskId}, 记录ID: {result.get('taskRecordId')}") # 返回任务记录信息 return { "success": True, "message": "任务启动成功", "taskRecordId": result.get("taskRecordId"), "status": result.get("status"), "createTime": result.get("createTime"), "isPeriodic": False } except Exception as e: logger.log_error_with_trace("运行任务失败", e) raise @staticmethod async def save_input_params(task_id: str, input_params: List[Dict[str, Any]]) -> Dict[str, Any]: """ 保存任务输入参数配置 Args: task_id: 任务ID input_params: 输入参数列表 Returns: Dict[str, Any]: 保存结果，包含是否成功、是否有变化的信息 """ try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {task_id}") return {"success": False, "changed": False, "message": "任务不存在"} # 解析详情JSON detail = json.loads(task_def.detail) if task_def.detail else {} # 获取现有输入参数 current_params = detail.get("inputParams", []) # 比较是否有变化 - 将两个列表转为JSON字符串比较 current_params_str = json.dumps(current_params, sort_keys=True, ensure_ascii=False) new_params_str = json.dumps(input_params, sort_keys=True, ensure_ascii=False) has_changes = current_params_str != new_params_str if not has_changes: return {"success": True, "changed": False, "message": "数据未发生变化"} # 更新输入参数 detail["inputParams"] = input_params # 更新任务定义 - 使用ensure_ascii=False确保中文直接存储而不被编码 task_def.detail = json.dumps(detail, ensure_ascii=False) task_def.version += 1 # 保存到数据库 await session.commit() # 获取当前时间作为更新时间 update_time = datetime.now() return { "success": True, "changed": True, "message": "保存成功", "version": task_def.version, "updateTime": update_time } except Exception as e: logger.error(f"保存输入参数失败: {str(e)}") raise @staticmethod async def save_basic_settings(task_id: str, basic_info: TaskBasicInfo) -> Dict[str, Any]: """ 保存任务基本设置 Args: task_id: 任务ID basic_info: 基本信息 Returns: Dict[str, Any]: 保存结果，包含是否成功、是否有变化的信息 """ try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {task_id}") return {"success": False, "changed": False, "message": "任务不存在"} # 检查是否有变化 release_sites_new = 1 if basic_info.releaseSites else 0 has_changes = ( task_def.label != basic_info.label or task_def.remark != basic_info.remark or task_def.release_sites != release_sites_new ) if not has_changes: return { "success": True, "changed": False, "message": "数据未发生变化", "id": task_def.id, "label": task_def.label, "remark": task_def.remark, "releaseSites": bool(task_def.release_sites) } # 更新基本信息 task_def.label = basic_info.label task_def.remark = basic_info.remark task_def.release_sites = release_sites_new # 保存到数据库 await session.commit() # 获取当前时间作为更新时间 update_time = datetime.now() return { "success": True, "changed": True, "message": "保存成功", "id": task_def.id, "label": task_def.label, "remark": task_def.remark, "releaseSites": bool(task_def.release_sites), "updateTime": update_time } except Exception as e: logger.error(f"保存基本设置失败: {str(e)}") raise @staticmethod async def get_common_params() -> List[Dict[str, Any]]: """ 获取常用参数字段 Returns: List[Dict[str, Any]]: 常用参数字段列表 """ try: # 导入内置函数管理模块 from utils.built_in_functions import get_function_list # 获取所有内置函数 built_in_functions = get_function_list() # 常用参数字段列表 common_params = [ { "id": "robotId", "label": "机器人id", "type": "String", "description": "机器人唯一标识", "order": 1, "values": [ {"id": "robot_001", "name": "AGV-001", "description": "1号AGV机器人"}, {"id": "robot_002", "name": "AGV-002", "description": "2号AGV机器人"}, {"id": "robot_003", "name": "AGV-003", "description": "3号AGV机器人"} ] }, { "id": "robotGroup", "label": "机器人组", "type": "Array", "description": "机器人分组信息", "order": 2, "values": [ {"id": "group1", "name": "分拣组", "description": "负责物品分拣的机器人组"}, {"id": "group2", "name": "运输组", "description": "负责物品运输的机器人组"}, {"id": "group3", "name": "装配组", "description": "负责物品装配的机器人组"} ] }, { "id": "siteId", "label": "库位id", "type": "String", "description": "库位唯一标识", "order": 3, "values": [ {"id": "site_001", "name": "A01", "description": "A区1号库位"}, {"id": "site_002", "name": "A02", "description": "A区2号库位"}, {"id": "site_003", "name": "B01", "description": "B区1号库位"} ] }, { "id": "area", "label": "库区", "type": "String", "description": "库区信息", "order": 4, "values": [ {"id": "area_A", "name": "A区", "description": "A区储存区"}, {"id": "area_B", "name": "B区", "description": "B区储存区"}, {"id": "area_C", "name": "C区", "description": "C区装配区"} ] }, { "id": "binTask", "label": "binTask", "type": "String", "description": "动作类型", "order": 5, "values": [ {"id": "QuickLoad", "name": "入库", "description": "物品入库"}, {"id": "QuickUnLoad", "name": "出库", "description": "物品出库"}, ] }, { "id": "user", "label": "用户", "type": "String", "description": "用户信息", "order": 6, "values": [ {"id": "user_001", "name": "张三", "description": "张三用户"}, {"id": "user_002", "name": "李四", "description": "李四用户"}, ] }, { "id": "cache", "label": "缓存", "type": "String", "description": "缓存信息", "order": 7, "values": [ {"id": "cache_001", "name": "cache1", "description": "cache1"}, {"id": "cache_002", "name": "cache2", "description": "cache2"}, ] }, { "id": "builtInFunction", "label": "内置函数", "type": "String", "description": "系统内置函数", "order": 8, "values": [func for func in built_in_functions] } ] return common_params except Exception as e: logger.error(f"获取常用参数字段失败: {str(e)}") raise @staticmethod async def get_task_input_params(task_id: str) -> Dict[str, Any]: """ 获取指定任务的输入参数配置 Args: task_id: 任务ID Returns: Dict[str, Any]: 包含输入参数配置和系统参数类型信息 """ try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskRecord).where(VWEDTaskRecord.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {task_id}") return {"success": False, "message": "任务不存在"} # 解析详情JSON input_params = json.loads(task_def.input_params) if task_def.input_params else {} # 返回完整信息 return { "success": True, "message": "获取任务输入参数配置成功", "data": { "inputParams": input_params } } except Exception as e: logger.error(f"获取任务输入参数配置失败: {str(e)}") raise @staticmethod async def get_basic_settings(task_id: str) -> Dict[str, Any]: """ 获取任务的基本设置信息 Args: task_id: 任务ID Returns: Dict[str, Any]: 任务基本信息 """ try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: logger.error(f"任务不存在: {task_id}") return {"success": False, "message": "任务不存在"} # 返回基本信息 return { "success": True, "data": { "id": task_def.id, "label": task_def.label, "remark": task_def.remark, "releaseSites": bool(task_def.release_sites), "version": task_def.version, "createDate": task_def.created_at } } except Exception as e: logger.error(f"获取任务基本设置失败: {str(e)}") raise @staticmethod async def check_task_changes(task_id: str, task_def_data: Dict[str, Any]) -> Dict[str, Any]: """ 检查任务数据是否有变化 Args: task_id: 任务ID task_def_data: 待保存的任务数据 Returns: Dict[str, Any]: 检查结果，包含是否发生变化的标志 """ try: if not task_id: return {"changed": True} # 无法检查，默认为已变化 async with get_async_session() as session: # 查询现有任务数据 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_id) ) task_def = result.scalars().first() if not task_def: return {"changed": True} # 任务不存在，视为有变化 # 检查label是否有变化 label = task_def_data.get("label") if label and label.strip() and label.strip() != task_def.label: return {"changed": True, "reason": "label变化"} # 检查detail是否有变化 detail = task_def_data.get("detail") if not detail: # 没有提供detail，可能只更新了label return {"changed": label and label.strip() and label.strip() != task_def.label} # 比较detail current_detail = json.loads(task_def.detail) if task_def.detail else {} # 将提供的detail转为JSON字符串，排序键以确保一致的比较结果 current_detail_str = json.dumps(current_detail, sort_keys=True, ensure_ascii=False) new_detail_str = json.dumps(detail, sort_keys=True, ensure_ascii=False) if current_detail_str != new_detail_str: return {"changed": True, "reason": "detail变化"} # 数据未变化 return { "changed": False, "id": task_def.id, "version": task_def.version } except Exception as e: logger.error(f"检查任务数据变化失败: {str(e)}") # 出错时默认为有变化，以确保数据安全 return {"changed": True, "error": str(e)} @staticmethod async def check_running_task_for_device(task_id: str, device_id: str) -> Dict[str, Any]: """ 检查指定设备是否有正在运行的相同任务 Args: task_id: 任务定义ID device_id: 设备ID Returns: Dict[str, Any]: 检查结果，包含是否允许启动任务的信息 { "has_running_task": bool, # 是否有运行中的任务 "allow_restart": bool, # 是否允许重启 "message": str # 提示消息 } """ try: from data.enum.task_record_enum import TaskStatus async with get_async_session() as session: # 查询是否存在相同任务ID且相同设备且正在运行的任务 result = await session.execute( select(VWEDTaskRecord).where( VWEDTaskRecord.def_id == task_id, VWEDTaskRecord.source_device == device_id, VWEDTaskRecord.status == TaskStatus.RUNNING # 状态为"进行中" ) ) running_tasks = result.scalars().all() # 如果没有运行中的任务，直接允许启动 if not running_tasks: return { "has_running_task": False, "allow_restart": True, "message": "没有运行中的任务，可以启动" } # 如果存在运行中的任务，检查allow_restart_same_location字段 allow_restart = True for task in running_tasks: # 如果任何一个任务不允许重启，则整体不允许 if not task.allow_restart_same_location: allow_restart = False break return { "has_running_task": True, "allow_restart": allow_restart, "message": "相同设备已有此任务正在运行中" + ("，但允许再次启动" if allow_restart else "，请等待任务完成后再次启动") } except Exception as e: logger.error(f"检查运行中任务失败: {str(e)}") raise