舆情二期代码编写

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java

@@ -35,6 +35,7 @@ import org.springframework.util.CollectionUtils;
 import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -117,17 +118,32 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
             //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
             syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisRequestId);
-
+            
             DiFyReq diFyReq = new DiFyReq();
             diFyReq.setUser(user);
-            diFyReq.setFlowId(flowId);
             diFyReq.setInputs(difyCommunityTargetDTO);
-            //调用舆情文本分析dify工作流
+            // 并行调用三个workflow
-            difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
+            //舆情案件分析
+            CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq,"app-cB5rd2VGIZn95u5GWgAnysj0"));
+            //内容主题关键词打标
+            CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq,"app-zJCgRBAedbOAQQUqx1IOLXNv"));
+            //litecrm线索分析
+            CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq,"app-1NZjZVMl3WEel6zhSGZa2rv9"));
+            CompletableFuture<Void> allFutures = CompletableFuture.allOf(caseWorkFlow, keywordWorkFlow, clueAnalysisWorkFlow);
+            // 等待所有API调用完成
+            allFutures.get();
+            // 获取各个API的结果
+            JSONObject caseResult = caseWorkFlow.get();
+            JSONObject keywordResult = keywordWorkFlow.get();
+            JSONObject clueAnalysisResult = clueAnalysisWorkFlow.get();
             //处理结果
-            processingCommunityDifyResponse(difResult);
+            processingCommunityDifyResponse(difResult,caseResult,keywordResult,clueAnalysisResult);
             //异步更新请求日志表的difyResponse字段
             syncUpdateDiFyResponse(difResult, aiAnalysisRequestId);
+
+
+            //调用舆情文本分析dify工作流
+//            difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
         } catch (Exception e) {
             log.error("舆情自动化异常:{}", e.getMessage());
             //保存错误日志
@@ -141,6 +157,18 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         return true;
     }
 
+    /**
+     * 调用案件，关键词，线索工作流
+     * @param diFyReq
+     * @param token
+     * @return
+     */
+    private JSONObject callCommunityWorkFlow(DiFyReq diFyReq, String token) {
+        diFyReq.setFlowId(token);
+        JSONObject difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
+        return difResult;
+    }
+
     /**
      * 舆情数据脱敏
      * @param textContent
@@ -189,7 +217,10 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
      * 处理dify返回结果
      * @param difResult
      */
-    private void processingCommunityDifyResponse(JSONObject difResult) {
+    private void processingCommunityDifyResponse(JSONObject difResult,
+                                                 JSONObject caseResult,
+                                                 JSONObject keywordResult,
+                                                 JSONObject clueAnalysisResult) {
         if (difResult != null) {
             DifyCommunityTargetResult difyCommunityTargetResult = new DifyCommunityTargetResult();
             String targetId = difResult.getString("targetId");
@@ -601,7 +632,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                                 //调用舆情文本分析dify工作流
                                 JSONObject  difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
                                 //处理结果
-                                processingCommunityDifyResponse(difResult);
+                                processingCommunityDifyResponse(difResult,null,null,null);
 
                                 //根据ai_analysis_request_id更新ai_analysis_errors表中的retry_count字段+1,更新status字段为1
                                 aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper<AiAnalysisErrors>()