AI外呼调整企微加个微节点流程

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

@@ -190,14 +190,14 @@ public class AiQwAddWxTaskNode extends AbstractWorkflowNode {
      * getRedisCacheKey
      *
      */
-    public static String getDelayAddWxKeyPrefix(Long time) {
+    public static String getDelayAddWxKeyPrefix(Integer cidGroupNo,Long time) {
         Date nowDay;
         if (null != time) {
             nowDay = new Date(time);
         }else{
             nowDay = new Date();
         }
-        return String.format(DELAY_QW_ADD_WX_KEY, nowDay.getHours(), nowDay.getMinutes());
+        return String.format(DELAY_QW_ADD_WX_KEY,cidGroupNo, nowDay.getHours(), nowDay.getMinutes());
     }
 
     /**
@@ -227,7 +227,7 @@ public class AiQwAddWxTaskNode extends AbstractWorkflowNode {
                 //节点包含延时条件
                 if (null != condition.getAddTime() && !condition.isAdd()) {
                     long l = System.currentTimeMillis() + condition.getAddTime() * 60 * 1000;
-                    String redisKey = getDelayAddWxKeyPrefix(l) + workflowInstanceId;
+                    String redisKey = getDelayAddWxKeyPrefix(exec.getCidGroupNo(),l) + workflowInstanceId;
                     ExecutionContext nextContext = context.clone();
                     nextContext.setCurrentNodeKey(edge.getTargetNodeKey());
                     super.redisCache.setCacheObject(redisKey, nextContext);

fs-wx-task/src/main/java/com/fs/app/service/WxTaskService.java

@@ -925,7 +925,38 @@ public class WxTaskService {
         log.info("==========执行企微申请加个微结果查询任务结束==========");
     }
 
-    
+
+    /**
+     * 扫描企微加微工作流延时任务
+     */
+    public void cidWorkflowQwAddWxRun() {
+        log.info("===========企微加微工作流延时任务开始扫描===========");
+        String delayAddWxKeyPrefix = AiQwAddWxTaskNode.getDelayAddWxKeyPrefix(cidGroupNo,null) + "*";
+        Set<String> keys = redisKeyScanner.scanMatchKey(delayAddWxKeyPrefix);
+        log.info("企微加微共扫描到 {} 个待处理键", keys.size());
+        keys.parallelStream().forEach(key -> {
+            try {
+                //doExec
+                CompletableFuture.runAsync(()->{
+                    try {
+                        ExecutionContext context = redisCache2.getCacheObject(key);
+                        context.setVariable("callRedisKey",key);
+                        context.setVariable("callSource","qwAddWxTimer");
+                        companyWorkflowEngine.timeDoExecute(context.getWorkflowInstanceId(),context.getCurrentNodeKey(),context.getVariables());
+                    } catch (Exception e) {
+                        log.error("处理工作流延时任务异常 - key: {}", key, e);
+                    }
+                }, cidExcutor).thenRun(()->{
+                    redisCache2.deleteObject(key);
+                });
+
+            } catch (Exception ex) {
+                log.error("处理工作流延时任务异常 - key: {}", key, ex);
+            }
+        });
+        log.info("===========工作流延时任务扫描结束===========");
+    }
+
     /**
      * 获取过滤后的企微客户列表
      */
@@ -933,7 +964,7 @@ public class WxTaskService {
         List<CompanyWxClient> list = companyWxClientService.getAddWxList(accountIdList, 2);
         
         // 排除掉没到达加微步骤的人
-        List<CompanyVoiceRoboticCallees> excludeList = companyVoiceRoboticCalleesMapper.selectExcludeList(list);
+        List<CompanyVoiceRoboticCallees> excludeList = companyVoiceRoboticCalleesMapper.selectExcludeList(list,2);
         Set<String> excludeKeys = excludeList.stream()
                 .filter(e -> !Constants.QW_ADD_WX.equals(getNextTaskOptimized(e.getTaskFlow(), e.getRunTaskFlow())))
                 .map(callee -> callee.getRoboticId() + "_" + callee.getUserId())