#!/usr/bin/env python # -*- coding: utf-8 -*- """ 天风任务服务层 处理天风任务相关的业务逻辑 """ import json import uuid import datetime from sqlalchemy.orm import Session from typing import List, Dict, Any, Optional from sqlalchemy import and_, or_, desc, asc from data.models.taskdef import VWEDTaskDef from data.models.taskrecord import VWEDTaskRecord from data.models.tasktemplate import VWEDTaskTemplate from data.enum.task_def_enum import TaskTypeEnum, TaskStatusEnum, EnableStatus, PeriodicTaskStatus from utils.logger import get_logger from data.session import get_async_session from sqlalchemy import select, update from data.enum.task_record_enum import TaskStatus # 设置日志 logger = get_logger("services.task_service") # 自定义异常类 class TaskNameExistsError(Exception): """任务名称已存在异常""" pass class TaskService: """ 天风任务服务类 提供天风任务的相关业务操作 """ @staticmethod def get_task_list( db: Session, page_num: int = 1, page_size: int = 10, keyword: Optional[str] = None, status: Optional[int] = None, sort_field: Optional[str] = None, sort_order: Optional[str] = "desc" ) -> Dict[str, Any]: """ 获取任务列表 Args: db: 数据库会话 page_num: 页码，从1开始 page_size: 每页记录数 keyword: 关键字(任务名称)，模糊匹配 status: 状态过滤 sort_field: 排序字段 sort_order: 排序方式：asc（升序）、desc（降序） Returns: Dict[str, Any]: 包含任务列表和分页信息 """ try: # 构建查询条件 query = db.query(VWEDTaskDef) # 添加筛选条件 if keyword: query = query.filter(VWEDTaskDef.label.like(f'%{keyword}%')) if status is not None: query = query.filter(VWEDTaskDef.status == status) # 计算总数 total = query.count() # 添加排序 if sort_field and sort_field in VWEDTaskDef.__table__.columns: direction = desc if sort_order == "desc" else asc query = query.order_by(direction(getattr(VWEDTaskDef, sort_field))) else: # 默认按创建时间倒序 query = query.order_by(desc(VWEDTaskDef.created_at)) # 分页处理 offset = (page_num - 1) * page_size tasks = query.offset(offset).limit(page_size).all() # 格式化返回结果 task_list = [] for task in tasks: task_list.append({ "id": task.id, "label": task.label, "version": task.version, "detail": task.detail, "templateName": task.template_name, "periodicTask": bool(task.periodic_task), "ifEnable": task.if_enable, "status": task.status, "createDate": task.created_at, "remark": task.remark, "tenant_id": task.tenant_id, "updated_at": task.updated_at }) return { "total": total, "list": task_list, "pageNum": page_num, "pageSize": page_size, "totalPages": (total + page_size - 1) // page_size } except Exception as e: logger.error(f"获取任务列表失败: {str(e)}") raise Exception(f"获取任务列表失败: {str(e)}") @staticmethod def create_task( db: Session, label: str, task_type: int, remark: Optional[str] = None, period: Optional[int] = None, delay: Optional[int] = 3000, release_sites: Optional[bool] = True, token: Optional[str] = None, tenant_id: str = "default", map_id: Optional[str] = None ) -> Dict[str, Any]: """ 创建任务 Args: db: 数据库会话 label: 任务名称 task_type: 任务类型：1-普通任务，2-定时任务 remark: 任务备注 period: 周期时间（毫秒），定时任务必填 delay: 延迟时间（毫秒），默认3000 release_sites: 是否释放站点，默认true token: 用户token值，用于认证 tenant_id: 租户ID，用于多租户隔离，默认为"default" map_id: 相关地图ID Returns: Dict[str, Any]: 新建任务信息 Raises: ValueError: 参数验证失败时 TaskNameExistsError: 任务名称已存在时 """ try: # 验证任务名称 if not label or label.strip() == "": raise ValueError("任务名称不能为空") # 验证任务名称长度 if len(label.strip()) < 2: raise ValueError("任务名称至少需要2个字符") # 检查任务名称是否已存在 existing_task = db.query(VWEDTaskDef).filter(VWEDTaskDef.label == label.strip()).first() if existing_task: raise TaskNameExistsError(f"任务名称 '{label}' 已存在，请使用其他名称") # 验证任务类型 if task_type not in [TaskTypeEnum.NORMAL, TaskTypeEnum.PERIODIC]: raise ValueError("任务类型无效，应为1(普通任务)或2(定时任务)") # 验证定时任务参数 if task_type == TaskTypeEnum.PERIODIC and not period: raise ValueError("定时任务必须指定周期时间") # 获取当前启用的模板 template = db.query(VWEDTaskTemplate).filter( VWEDTaskTemplate.template_if_enable == 1 ).first() if not template: raise ValueError("没有可用的任务模板，请先启用一个模板") # 创建任务定义 task_id = str(uuid.uuid4()) new_task = VWEDTaskDef( id=task_id, label=label.strip(), # 去除首尾空白字符 version=1, detail=json.dumps({ "inputParams": [], "outputParams": [], "rootBlock": None }, ensure_ascii=False), template_name=template.template_name, period=period if period else 3000, periodic_task=PeriodicTaskStatus.PERIODIC if task_type == TaskTypeEnum.PERIODIC else PeriodicTaskStatus.NON_PERIODIC, status=TaskStatusEnum.PENDING, if_enable=EnableStatus.DISABLED, delay=delay, release_sites=release_sites, remark=remark, tenant_id=tenant_id, # 使用传入的租户ID user_token=token, # 保存用户token map_id=map_id # 保存地图ID ) db.add(new_task) db.commit() db.refresh(new_task) # 返回任务信息 return { "id": new_task.id, "label": new_task.label, "version": new_task.version, "detail": new_task.detail, "templateName": new_task.template_name, "periodicTask": bool(new_task.periodic_task), "status": new_task.status, "createDate": new_task.created_at, "mapId": new_task.map_id # 返回地图ID } except TaskNameExistsError as e: db.rollback() logger.warning(f"创建任务失败: {str(e)}") raise # 重新抛出异常，不包装 except Exception as e: db.rollback() logger.error(f"创建任务失败: {str(e)}") raise Exception(f"创建任务失败: {str(e)}") @staticmethod def delete_tasks(db: Session, ids: List[str]) -> bool: """ 删除任务 Args: db: 数据库会话 ids: 要删除的任务ID列表 Returns: bool: 是否删除成功 """ try: # 检查任务是否存在 tasks = db.query(VWEDTaskDef).filter(VWEDTaskDef.id.in_(ids)).all() if len(tasks) != len(ids): missing_ids = set(ids) - {task.id for task in tasks} raise ValueError(f"以下任务不存在: {', '.join(missing_ids)}") # 删除任务 for task in tasks: db.delete(task) db.commit() return True except Exception as e: db.rollback() logger.error(f"删除任务失败: {str(e)}") raise Exception(f"删除任务失败: {str(e)}") @staticmethod def get_task_by_id(db: Session, task_id: str) -> Optional[Dict[str, Any]]: """ 根据ID获取任务 Args: db: 数据库会话 task_id: 任务ID Returns: Optional[Dict[str, Any]]: 任务信息，如果不存在则返回None """ try: # 显式启动事务 task = db.query(VWEDTaskDef).filter(VWEDTaskDef.id == task_id).first() if not task: return None # 格式化任务信息 task_info = { "id": task.id, "label": task.label, "version": task.version, "detail": json.loads(task.detail) if task.detail else {}, "templateName": task.template_name, "period": task.period, "periodicTask": bool(task.periodic_task), "status": task.status, "ifEnable": task.if_enable, "delay": task.delay, "releaseSites": bool(task.release_sites) if task.release_sites is not None else None, "createDate": task.created_at, "remark": task.remark, } # 提交事务以释放数据库锁 db.commit() return task_info except Exception as e: # 发生异常时回滚事务 db.rollback() logger.error(f"获取任务失败: {str(e)}") raise Exception(f"获取任务失败: {str(e)}") @staticmethod def export_tasks(db: Session, ids: List[str]) -> List[Dict[str, Any]]: """ 导出任务 Args: db: 数据库会话 ids: 任务ID列表 Returns: List[Dict[str, Any]]: 任务配置列表 """ try: tasks = db.query(VWEDTaskDef).filter(VWEDTaskDef.id.in_(ids)).all() if not tasks: raise ValueError(f"未找到指定的任务") # 格式化任务信息 tasks_export = [] for task in tasks: task_export = { "id": task.id, "label": task.label, "version": task.version, "detail": task.detail, "templateName": task.template_name, "templateDescription": db.query(VWEDTaskTemplate.template_description) .filter(VWEDTaskTemplate.template_name == task.template_name) .scalar() or "", "period": task.period, "periodicTask": task.periodic_task, "status": task.status, "ifEnable": task.if_enable, "delay": task.delay, "remark": task.remark, "windcategoryId": 0 # 默认分类ID } tasks_export.append(task_export) return tasks_export except Exception as e: logger.error(f"导出任务失败: {str(e)}") raise Exception(f"导出任务失败: {str(e)}") @staticmethod def import_task( db: Session, task_data: Dict[str, Any], task_name: Optional[str] = None, token: Optional[str] = None, tenant_id: str = "default" ) -> Dict[str, Any]: """ 导入任务 Args: db: 数据库会话 task_data: 任务配置数据 task_name: 导入后的任务名称，不填则使用原名称 token: 用户token，用于认证 tenant_id: 租户ID，默认为"default" Returns: Dict[str, Any]: 导入的任务信息 Raises: ValueError: 参数验证失败时 TaskNameExistsError: 任务名称已存在时 """ try: # 创建新ID避免冲突 new_id = str(uuid.uuid4()) # 设置任务名称 label = task_name or task_data.get("label", "导入的任务") # 获取detail字段，这是唯一必需的字段 detail = task_data.get("detail") if not detail: raise ValueError("任务数据缺少detail字段，无法导入") # 检查任务名称是否已存在 existing_task = db.query(VWEDTaskDef).filter( VWEDTaskDef.label == label.strip(), VWEDTaskDef.tenant_id == tenant_id ).first() if existing_task: raise TaskNameExistsError(f"任务名称 '{label}' 已存在，请使用其他名称") # 获取当前启用的模板 template = db.query(VWEDTaskTemplate).filter( VWEDTaskTemplate.template_if_enable == 1 ).first() if not template: template_name = task_data.get("templateName", "默认模板") else: template_name = template.template_name # 创建新任务 new_task = VWEDTaskDef( id=new_id, label=label, version=1, # 新任务从版本1开始 detail=detail, template_name=template_name, period=task_data.get("period", 3000), periodic_task=task_data.get("periodicTask", PeriodicTaskStatus.NON_PERIODIC), status=TaskStatusEnum.PENDING, # 初始状态为0 if_enable=EnableStatus.DISABLED, # 初始未启用 delay=task_data.get("delay", 3000), release_sites=task_data.get("releaseSites", True), # 默认释放站点 remark=task_data.get("remark", ""), tenant_id=tenant_id, user_token=token ) db.add(new_task) db.commit() db.refresh(new_task) # 返回任务信息 return { "id": new_task.id, "label": new_task.label, "version": new_task.version, "status": new_task.status, "createDate": new_task.created_at } except TaskNameExistsError as e: db.rollback() logger.warning(f"导入任务失败: {str(e)}") raise # 重新抛出异常，不包装 except Exception as e: db.rollback() logger.error(f"导入任务失败: {str(e)}") raise Exception(f"导入任务失败: {str(e)}") @staticmethod async def stop_task_def(task_def_id: str) -> Dict[str, Any]: """ 停止指定任务定义下的所有运行任务实例，同时禁用定时任务 Args: task_def_id: 任务定义ID Returns: 字典包含停止结果信息： - success: 是否操作成功 - message: 操作结果消息 - is_periodic: 是否为定时任务 - total_running: 运行中的任务总数 - stopped_count: 成功停止的任务数量 - failed_count: 停止失败的任务数量 - failed_tasks: 停止失败的任务记录ID列表 """ # 导入增强版调度器 from services.enhanced_scheduler import scheduler try: async with get_async_session() as session: # 查询任务定义 result = await session.execute( select(VWEDTaskDef).where(VWEDTaskDef.id == task_def_id) ) task_def = result.scalars().first() if not task_def: return { "success": False, "message": f"未找到任务定义: {task_def_id}" } is_periodic = task_def.periodic_task == PeriodicTaskStatus.PERIODIC # 初始化计数器 total_running = 0 stopped_count = 0 failed_count = 0 failed_tasks = [] # 如果是定时任务则禁用 if is_periodic and task_def.if_enable == EnableStatus.ENABLED: # 更新任务定义状态为禁用 await session.execute( update(VWEDTaskDef) .where(VWEDTaskDef.id == task_def_id) .values(if_enable=EnableStatus.DISABLED) ) await session.commit() # 通知调度器 update_result = await scheduler.update_periodic_task(task_def_id, enable=False) # 将定时任务的禁用也计入停止任务的数量 total_running += 1 if update_result.get("success", True): # 假设通知调度器成功，除非明确返回失败 stopped_count += 1 else: failed_count += 1 failed_tasks.append({ "taskRecordId": "periodic_" + task_def_id, "reason": update_result.get("message", "禁用定时任务失败") }) # 查找所有正在运行的任务记录 running_tasks_query = await session.execute( select(VWEDTaskRecord) .where( VWEDTaskRecord.def_id == task_def_id, VWEDTaskRecord.status == TaskStatus.RUNNING # 执行中状态码 ) ) running_tasks = running_tasks_query.scalars().all() # 更新总计数 total_running += len(running_tasks) # 取消所有运行中的任务 for task_record in running_tasks: cancel_result = await scheduler.cancel_task(task_record.id) if cancel_result.get("success", False): stopped_count += 1 else: failed_count += 1 failed_tasks.append({ "taskRecordId": task_record.id, "reason": cancel_result.get("message", "未知原因") }) # 更新任务定义状态为已结束(0) await session.execute( update(VWEDTaskDef) .where(VWEDTaskDef.id == task_def_id) .values(status=TaskStatusEnum.PENDING) ) await session.commit() return { "success": True, "message": "操作完成", "is_periodic": is_periodic, "total_running": total_running, "stopped_count": stopped_count, "failed_count": failed_count, "failed_tasks": failed_tasks } except Exception as e: return { "success": False, "message": f"停止任务失败: {str(e)}" }