优化

fs-ai-call-task/src/main/java/com/fs/app/service/CallTaskService.java

@@ -1,8 +1,13 @@
 package com.fs.app.service;
 
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.fs.common.core.redis.RedisCache;
 import com.fs.common.core.redis.RedisCacheT;
+import com.fs.company.domain.CompanyVoiceRobotic;
+import com.fs.company.mapper.CompanyVoiceRoboticMapper;
 import com.fs.company.param.ExecutionContext;
+import com.fs.company.param.PauseRoboticActiveParam;
 import com.fs.company.service.*;
 import com.fs.company.service.impl.call.node.AiCallTaskNode;
 import lombok.RequiredArgsConstructor;
@@ -12,6 +17,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import java.time.LocalTime;
 import java.util.*;
 import java.util.concurrent.*;
 
@@ -29,6 +35,7 @@ public class CallTaskService {
     private final RedisCache redisCache2;
     private final CompanyWorkflowEngine companyWorkflowEngine;
     private final ICompanyVoiceRoboticService companyVoiceRoboticService;
+    private final CompanyVoiceRoboticMapper companyVoiceRoboticMapper;
     @Autowired
     @Qualifier("cidWorkFlowExecutor")
     private Executor cidWorkFlowExecutor;
@@ -81,4 +88,114 @@ public class CallTaskService {
         });
         log.info("===========工作流延时任务扫描结束===========");
     }
+
+    /**
+     * 检查任务状态来暂停外呼任务的外呼运行
+     */
+    public void taskStatusChange(){
+        LocalTime now = LocalTime.now();
+        log.info("===========任务状态变更扫描开始, 当前时间: {}, 分组: {}===========", now, cidGroupNo);
+        pauseOutOfRangeTasks(now);
+        resumeInRangeTasks(now);
+        log.info("===========任务状态变更扫描结束===========");
+    }
+
+    private void pauseOutOfRangeTasks(LocalTime now) {
+        List<CompanyVoiceRobotic> activeTasks = companyVoiceRoboticMapper.selectList(
+                new QueryWrapper<CompanyVoiceRobotic>()
+                        .eq("task_status", 1)
+                        .eq("task_type", 1)
+                        .eq("cid_group_no", cidGroupNo)
+                        .eq("del_flag", 0)
+        );
+        log.info("扫描到执行中的普通任务 {} 个", activeTasks.size());
+        if (activeTasks.isEmpty()) {
+            return;
+        }
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        for (CompanyVoiceRobotic task : activeTasks) {
+            if (isOutOfRuntimeRange(task, now)) {
+                log.info("任务[{}]运行时间范围[{},{})，当前{}超出范围，准备自动暂停", task.getId(), task.getRuntimeRangeStart(), task.getRuntimeRangeEnd(), now);
+                futures.add(CompletableFuture.runAsync(() -> {
+                    try {
+                        PauseRoboticActiveParam param = new PauseRoboticActiveParam();
+                        param.setTaskId(task.getId());
+                        param.setActiveType(1);
+                        companyVoiceRoboticService.pauseRoboticActive(param);
+                        companyVoiceRoboticMapper.update(null,
+                                new LambdaUpdateWrapper<CompanyVoiceRobotic>()
+                                        .eq(CompanyVoiceRobotic::getId, task.getId())
+                                        .set(CompanyVoiceRobotic::getPauseSource, "auto")
+                        );
+                        log.info("任务[{}]超出运行时间范围，已自动暂停", task.getId());
+                    } catch (Exception e) {
+                        log.error("自动暂停任务[{}]异常", task.getId(), e);
+                    }
+                }, cidWorkFlowExecutor));
+            }
+        }
+        if (!futures.isEmpty()) {
+            log.info("本次需自动暂停 {} 个任务，等待并行执行完成", futures.size());
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+        }
+    }
+
+    private void resumeInRangeTasks(LocalTime now) {
+        List<CompanyVoiceRobotic> pausedTasks = companyVoiceRoboticMapper.selectList(
+                new QueryWrapper<CompanyVoiceRobotic>()
+                        .eq("task_status", 2)
+                        .eq("task_type", 1)
+                        .eq("pause_source", "auto")
+                        .eq("cid_group_no", cidGroupNo)
+                        .eq("del_flag", 0)
+        );
+        log.info("扫描到自动暂停中的普通任务 {} 个", pausedTasks.size());
+        if (pausedTasks.isEmpty()) {
+            return;
+        }
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
+        for (CompanyVoiceRobotic task : pausedTasks) {
+            if (isWithinRuntimeRange(task, now)) {
+                log.info("任务[{}]运行时间范围[{},{})，当前{}在范围内，准备自动恢复", task.getId(), task.getRuntimeRangeStart(), task.getRuntimeRangeEnd(), now);
+                futures.add(CompletableFuture.runAsync(() -> {
+                    try {
+                        PauseRoboticActiveParam param = new PauseRoboticActiveParam();
+                        param.setTaskId(task.getId());
+                        param.setActiveType(2);
+                        companyVoiceRoboticService.pauseRoboticActive(param);
+                        companyVoiceRoboticMapper.update(null,
+                                new LambdaUpdateWrapper<CompanyVoiceRobotic>()
+                                        .eq(CompanyVoiceRobotic::getId, task.getId())
+                                        .set(CompanyVoiceRobotic::getPauseSource, null)
+                        );
+                        log.info("任务[{}]回到运行时间范围，已自动恢复", task.getId());
+                    } catch (Exception e) {
+                        log.error("自动恢复任务[{}]异常", task.getId(), e);
+                    }
+                }, cidWorkFlowExecutor));
+            }
+        }
+        if (!futures.isEmpty()) {
+            log.info("本次需自动恢复 {} 个任务，等待并行执行完成", futures.size());
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+        }
+    }
+
+    private boolean isOutOfRuntimeRange(CompanyVoiceRobotic task, LocalTime now) {
+        LocalTime start = task.getRuntimeRangeStart();
+        LocalTime end = task.getRuntimeRangeEnd();
+        if (start == null || end == null) {
+            return false;
+        }
+        if (start.isBefore(end)) {
+            return now.isBefore(start) || now.isAfter(end);
+        } else {
+            return now.isAfter(end) && now.isBefore(start);
+        }
+    }
+
+    private boolean isWithinRuntimeRange(CompanyVoiceRobotic task, LocalTime now) {
+        return !isOutOfRuntimeRange(task, now);
+    }
+
 }

fs-ai-call-task/src/main/java/com/fs/app/task/Task.java

@@ -63,4 +63,23 @@ public class Task {
         }
 
     }
+
+    @Scheduled(cron = "0 0/1 * * * ?")
+    public void taskStatusChange(){
+        // 防御性检查：确保依赖注入成功
+        if (tenantTaskRunner == null) {
+            log.error("[Task] tenantTaskRunner 未注入，请检查 Spring 配置");
+            return;
+        }
+        if (taskService == null) {
+            log.error("[Task] taskService 未注入，请检查 Spring 配置");
+            return;
+        }
+        if (saasTaskEnabled) {
+            tenantTaskRunner.runForResponsibleTenant("taskStatusChange", () -> taskService.taskStatusChange());
+        } else {
+            taskService.taskStatusChange();
+        }
+
+    }
 }

fs-service/src/main/java/com/fs/company/domain/CompanyVoiceRobotic.java

@@ -104,6 +104,7 @@ public class CompanyVoiceRobotic {
     private String runTaskFlow;
     // 任务状态0待执行1执行中2执行中断3执行完成
     private Integer taskStatus;
+    private String pauseSource;
     private Integer addWxTime;
     /** 创建人 */
     @Excel(name = "创建时间")

fs-service/src/main/java/com/fs/company/service/impl/CompanyVoiceRoboticServiceImpl.java

@@ -1431,6 +1431,8 @@ public class CompanyVoiceRoboticServiceImpl extends ServiceImpl<CompanyVoiceRobo
         final Long workflowId = robotic.getCompanyAiWorkflowId();
         final Long roboticId = robotic.getId();
 
+        companyWorkflowEngine.createSipTask(roboticId, workflowId);
+
         // 先初始化所有工作流实例
         List<ExecutionResult> initResults = new ArrayList<>();
         for (CompanyVoiceRoboticBusiness business : roboticBusinesseList) {
@@ -1973,27 +1975,22 @@ public class CompanyVoiceRoboticServiceImpl extends ServiceImpl<CompanyVoiceRobo
                 return R.error("任务不在执行中状态，无法暂停");
             }
             robotic.setTaskStatus(2);
+            robotic.setPauseSource("manual");
             updateCompanyVoiceRobotic(robotic);
-            // 更新Redis缓存
             redisCache2.setCacheObject("task:status:" + param.getTaskId(), 2);
-            // 调用EasyCall暂停
             pauseEasyCallTask(robotic);
             return R.ok("暂停成功");
         }
-        //恢复任务继续进入可运行
         else if (ACTIVE_TYPE_CONTINUE.equals(param.getActiveType())) {
             CompanyVoiceRobotic robotic = selectCompanyVoiceRoboticById(param.getTaskId());
             if (robotic == null || robotic.getTaskStatus() != 2) {
                 return R.error("任务不在中断状态，无法继续");
             }
             robotic.setTaskStatus(1);
+            robotic.setPauseSource(null);
             updateCompanyVoiceRobotic(robotic);
-            // 更新Redis缓存
             redisCache2.setCacheObject("task:status:" + param.getTaskId(), 1);
-            // 调用EasyCall恢复
             resumeEasyCallTask(robotic);
-            // 异步执行恢复扫描（分批处理，避免阻塞接口）
-            // 通过SpringUtils获取代理对象调用，确保@Async生效
             SpringUtils.getBean(CompanyVoiceRoboticServiceImpl.class).resumePausedInstances(param.getTaskId());
             return R.ok("继续成功");
         }

fs-service/src/main/java/com/fs/company/service/impl/CompanyWorkflowEngineImpl.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fs.common.core.redis.RedisCache;
 import com.fs.common.exception.CustomException;
 import com.fs.common.utils.StringUtils;
 import com.fs.company.domain.*;
@@ -29,6 +30,7 @@ import org.springframework.stereotype.Service;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author MixLiu
@@ -75,6 +77,9 @@ public class CompanyWorkflowEngineImpl implements CompanyWorkflowEngine {
     @Autowired
     CompanyWxClientMapper companyWxClientMapper;
 
+    @Autowired
+    private RedisCache redisCache;
+
     @Autowired
     private CloudHostProper cloudHostProper;
 
@@ -102,8 +107,6 @@ public class CompanyWorkflowEngineImpl implements CompanyWorkflowEngine {
                     definition.getStartNodeKey(), context, definition);
 
             log.info("工作流初始化成功: {} -> {}", workflowInstanceId, workflowDefinitionId);
-            //为任务创建sip任务并存入表数据
-            createSipTask(Long.parseLong(inputVariables.get("roboticId").toString()),workflowDefinitionId);
             return ExecutionResult.success()
                     .nextNodeKey(definition.getStartNodeKey())
                     .workflowInstanceId(workflowInstanceId).build();
@@ -572,35 +575,44 @@ public class CompanyWorkflowEngineImpl implements CompanyWorkflowEngine {
      * @param workFlowId
      */
     public Long createSipTask(Long roboticId, Long workFlowId) {
-        try {
-            List<String> nodeTypes = Arrays.asList(NodeTypeEnum.AI_CALL_TASK.getCode());
-            CompanyVoiceRobotic robotic = companyVoiceRoboticMapper.selectCompanyVoiceRoboticById(roboticId);
-            List<CompanyWorkflowNode> companyWorkflowNodes = companyWorkflowNodeMapper.selectNodesByWorkflowIdAndTypes(workFlowId, nodeTypes);
-            //为所有外呼节点创建任务的对应sip外呼任务
-            for (CompanyWorkflowNode callNode : companyWorkflowNodes) {
+        List<String> nodeTypes = Arrays.asList(NodeTypeEnum.AI_CALL_TASK.getCode());
+        CompanyVoiceRobotic robotic = companyVoiceRoboticMapper.selectCompanyVoiceRoboticById(roboticId);
+        List<CompanyWorkflowNode> companyWorkflowNodes = companyWorkflowNodeMapper.selectNodesByWorkflowIdAndTypes(workFlowId, nodeTypes);
+        for (CompanyWorkflowNode callNode : companyWorkflowNodes) {
+            String lockKey = "sipTask:lock:" + roboticId + ":" + callNode.getNodeKey();
+            boolean locked = redisCache.setIfAbsent(lockKey, "1", 30, TimeUnit.SECONDS);
+            if (!locked) {
+                log.info("createSipTask: 其他线程正在创建SIP任务，等待后重试 - roboticId: {}, nodeKey: {}", roboticId, callNode.getNodeKey());
+                for (int i = 0; i < 20; i++) {
+                    try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+                    CompanySiptaskInfo existing = companySiptaskInfoMapper.selectSipTaskInfoByTaskIdAndNodeKey(roboticId, callNode.getNodeKey());
+                    if (existing != null && existing.getBatchId() != null) {
+                        return existing.getBatchId();
+                    }
+                }
+                log.warn("createSipTask: 等待超时，尝试直接创建 - roboticId: {}, nodeKey: {}", roboticId, callNode.getNodeKey());
+            }
+            try {
+                CompanySiptaskInfo existingSipTask = companySiptaskInfoMapper.selectSipTaskInfoByTaskIdAndNodeKey(roboticId, callNode.getNodeKey());
+                if (existingSipTask != null && existingSipTask.getBatchId() != null) {
+                    log.info("createSipTask: SIP任务已存在，跳过创建 - roboticId: {}, nodeKey: {}, batchId: {}", roboticId, callNode.getNodeKey(), existingSipTask.getBatchId());
+                    return existingSipTask.getBatchId();
+                }
                 String nodeConfig = callNode.getNodeConfig();
                 AiCallConfigVO callConfigVo = JSONObject.parseObject(nodeConfig, AiCallConfigVO.class);
                 EasyCallCreateTaskParam createParam = new EasyCallCreateTaskParam();
-                // 任务名称：使用任务名称_工作流id_节点key
                 createParam.setBatchName(cloudHostProper.getCompanyName()+"-"+robotic.getName() + "_" + workFlowId + "_" + callNode.getNodeKey());
                 if (null != callConfigVo.getMaxConcurrency()) {
                     createParam.setThreadNum(Long.valueOf(callConfigVo.getMaxConcurrency()));
                 } else {
                     createParam.setThreadNum(3L);
                 }
-                // AI 外呼模式
                 createParam.setTaskType(1);
-                // 外呼线路（网关）
                 createParam.setGatewayId(callConfigVo.getGatewayId());
-                // 大模型底座
                 createParam.setLlmAccountId(callConfigVo.getLlmAccountId());
-                // 音色编号
                 createParam.setVoiceCode(callConfigVo.getVoiceCode());
-                // 音色来源（如未配置默认留空，由 EasyCallCenter365 使用默认值）
                 createParam.setVoiceSource(callConfigVo.getVoiceSource());
-                // 技能组（转人工客服分组，可选）
                 createParam.setGroupId(callConfigVo.getBusiGroupId());
-                // 模型参数
                 createParam.setTtsModels(callConfigVo.getTtsModels());
 
                 if (callConfigVo.getExtensionList() != null && !callConfigVo.getExtensionList().isEmpty()) {
@@ -628,9 +640,12 @@ public class CompanyWorkflowEngineImpl implements CompanyWorkflowEngine {
                 sipTaskInfo.setTaskJson(JSONObject.toJSONString(task));
                 companySiptaskInfoMapper.insertCompanySiptaskInfo(sipTaskInfo);
                 return task.getBatchId();
+            } catch (Exception ex) {
+                log.error("创建SIP任务失败：{}", ex);
+                return null;
+            } finally {
+                redisCache.deleteObject(lockKey);
             }
-        } catch (Exception ex) {
-            log.error("创建SIP任务失败：{}", ex);
         }
         return null;
     }

fs-service/src/main/java/com/fs/company/service/impl/call/node/AiCallTaskNode.java

@@ -422,9 +422,29 @@ public class AiCallTaskNode extends AbstractWorkflowNode {
         if(null != sipTaskInfo && null != sipTaskInfo.getBatchId()){
             return  sipTaskInfo.getBatchId();
         }
-        //没有的情况下创建任务并返回
-        CompanyAiWorkflowExec companyAiWorkflowExec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(workflowInstanceId);
-        return companyWorkflowEngine.createSipTask(roboticId, companyAiWorkflowExec.getWorkflowId());
+        String lockKey = "sipTask:lock:" + roboticId + ":" + nodeKey;
+        boolean locked = super.redisCache.setIfAbsent(lockKey, "1", 30, TimeUnit.SECONDS);
+        if (!locked) {
+            log.info("getTaskBatchId: 其他线程正在创建SIP任务，等待后重试 - roboticId: {}, nodeKey: {}", roboticId, nodeKey);
+            for (int i = 0; i < 20; i++) {
+                try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+                sipTaskInfo = companySiptaskInfoMapper.selectSipTaskInfoByTaskIdAndNodeKey(roboticId, nodeKey);
+                if (null != sipTaskInfo && null != sipTaskInfo.getBatchId()) {
+                    return sipTaskInfo.getBatchId();
+                }
+            }
+            log.warn("getTaskBatchId: 等待超时，尝试直接创建 - roboticId: {}, nodeKey: {}", roboticId, nodeKey);
+        }
+        try {
+            sipTaskInfo = companySiptaskInfoMapper.selectSipTaskInfoByTaskIdAndNodeKey(roboticId, nodeKey);
+            if (null != sipTaskInfo && null != sipTaskInfo.getBatchId()) {
+                return sipTaskInfo.getBatchId();
+            }
+            CompanyAiWorkflowExec companyAiWorkflowExec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(workflowInstanceId);
+            return companyWorkflowEngine.createSipTask(roboticId, companyAiWorkflowExec.getWorkflowId());
+        } finally {
+            super.redisCache.deleteObject(lockKey);
+        }
     }
 //    @Override
 //    protected void postExecute(ExecutionContext context, ExecutionResult result) {

fs-service/src/main/resources/db/tenant-initTable.sql

@@ -2645,6 +2645,7 @@ CREATE TABLE `company_voice_robotic`
     `available_start_time`   time NULL DEFAULT NULL COMMENT '应用场景指定时间范围开始',
     `available_end_time`     time NULL DEFAULT NULL COMMENT '应用场景指定时间范围结束',
     `del_flag`               tinyint NULL DEFAULT 0 COMMENT '删除标志 0正常 1删除',
+    `pause_source` varchar(20) NULL DEFAULT NULL COMMENT '暂停来源：auto-自动暂停(超出运行时间范围) manual-手动暂停',
     PRIMARY KEY (`id`) USING BTREE,
     INDEX                    `company_scene_idx`(`company_id`, `scene_type`, `del_flag`) USING BTREE COMMENT '公司，场景索引'
 ) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci COMMENT = '机器人外呼任务' ROW_FORMAT = DYNAMIC;

fs-service/src/main/resources/mapper/company/CompanyVoiceRoboticMapper.xml

@@ -27,6 +27,7 @@
         <result property="companyAiWorkflowId"    column="company_ai_workflow_id"    />
         <result property="delFlag"    column="del_flag"    />
         <result property="isWeCom"    column="is_we_com"    />
+        <result property="pauseSource"    column="pause_source"    />
 
     </resultMap>
 
@@ -187,6 +188,7 @@
             <if test="createUser != null">create_user = #{createUser},</if>
             <if test="companyAiWorkflowId != null">company_ai_workflow_id = #{companyAiWorkflowId},</if>
             <if test="taskStatus != null">task_status = #{taskStatus},</if>
+            <if test="pauseSource != null">pause_source = #{pauseSource},</if>
         </trim>
         where id = #{id}
     </update>