定时任务处理舆情异常全部流程补偿

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

@@ -101,26 +101,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                     .anyMatch(contentNode -> NodeTypeEnum.IMAGE.getCode().equals(contentNode.getNodeType()));
             //如果存在图片节点，则先处理图片节点
             StringBuilder sb = new StringBuilder();
-            if (hasImageNodeType) {
+            processingImageInfo(hasImageNodeType, communityTargetDTO, sb);
-                communityTargetDTO.getContent().forEach(contentNode -> {
-                    if (contentNode.getNodeType().equals(NodeTypeEnum.IMAGE.getCode())) {
-                        //构建图片分析请求对象
-                        DifyImageWorkFlow diFyImageWorkFlow = new DifyImageWorkFlow();
-                        diFyImageWorkFlow.setUrl(contentNode.getNodeContent());
 
-                        DiFyReq diFyImageReq = new DiFyReq();
-                        diFyImageReq.setUser(user);
-                        diFyImageReq.setFlowId(imageFlowId);
-                        JSONObject imageInputs = new JSONObject();
-                        imageInputs.put("image", diFyImageWorkFlow);
-                        diFyImageReq.setInputs(imageInputs);
-
-                        //调用dify图片分析workflow
-                        JSONObject difImageResult = (JSONObject) diFyService.getDiFyObject(diFyImageReq);
-                        sb.append(difImageResult.get("text"));
-                    }
-                });
-            }
             String textContent = Optional.ofNullable(communityTargetDTO.getContent())
                     .orElse(Collections.emptyList()).stream()
                     .filter(contentNode -> NodeTypeEnum.TEXT.getCode().equals(contentNode.getNodeType()))
@@ -132,12 +114,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             difyCommunityTargetDTO.setTargetId(communityTargetDTO.getTargetId());
             difyCommunityTargetDTO.setTargetType(communityTargetDTO.getTargetType());
             //脱敏处理
-            List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC);
+            dataMasking(textContent, sb, difyCommunityTargetDTO);
-            RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
-            runMaskingRuleInput.setDataMaskingRules(maskingRuleItems);
-            runMaskingRuleInput.setOldStr(textContent + sb);
-            String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
-            difyCommunityTargetDTO.setTargetContent(corpusChat);
 
             //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
             syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisRequestId);
@@ -149,7 +126,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             //调用舆情文本分析dify工作流
             difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
             //处理结果
-            dueCommunityDifyResponse(difResult);
+            processingCommunityDifyResponse(difResult);
             //异步更新请求日志表的difyResponse字段
             syncUpdateDiFyResponse(difResult, aiAnalysisRequestId);
         } catch (Exception e) {
@@ -165,11 +142,55 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         return true;
     }
 
+    /**
+     * 舆情数据脱敏
+     * @param textContent
+     * @param sb
+     * @param difyCommunityTargetDTO
+     */
+    private void dataMasking(String textContent, StringBuilder sb, DifyCommunityTargetDTO difyCommunityTargetDTO) {
+        List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC);
+        RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
+        runMaskingRuleInput.setDataMaskingRules(maskingRuleItems);
+        runMaskingRuleInput.setOldStr(textContent + sb);
+        String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
+        difyCommunityTargetDTO.setTargetContent(corpusChat);
+    }
+
+    /**
+     * 处理图片信息
+     * @param hasImageNodeType
+     * @param communityTargetDTO
+     * @param sb
+     */
+    private void processingImageInfo(boolean hasImageNodeType, CommunityTargetDTO communityTargetDTO, StringBuilder sb) {
+        if (hasImageNodeType) {
+            communityTargetDTO.getContent().forEach(contentNode -> {
+                if (contentNode.getNodeType().equals(NodeTypeEnum.IMAGE.getCode())) {
+                    //构建图片分析请求对象
+                    DifyImageWorkFlow diFyImageWorkFlow = new DifyImageWorkFlow();
+                    diFyImageWorkFlow.setUrl(contentNode.getNodeContent());
+
+                    DiFyReq diFyImageReq = new DiFyReq();
+                    diFyImageReq.setUser(user);
+                    diFyImageReq.setFlowId(imageFlowId);
+                    JSONObject imageInputs = new JSONObject();
+                    imageInputs.put("image", diFyImageWorkFlow);
+                    diFyImageReq.setInputs(imageInputs);
+
+                    //调用dify图片分析workflow
+                    JSONObject difImageResult = (JSONObject) diFyService.getDiFyObject(diFyImageReq);
+                    sb.append(difImageResult.get("text"));
+                }
+            });
+        }
+    }
+
     /**
      * 处理dify返回结果
      * @param difResult
      */
-    private void dueCommunityDifyResponse(JSONObject difResult) {
+    private void processingCommunityDifyResponse(JSONObject difResult) {
         if (difResult != null) {
             DifyCommunityTargetResult difyCommunityTargetResult = new DifyCommunityTargetResult();
             String targetId = difResult.getString("targetId");
@@ -549,14 +570,41 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                         if (aiAnalysisRequestLogs != null) {
                             //获取dify_request字段
                             String difyRequest = aiAnalysisRequestLogs.getDifyRequest();
+                            String message = aiAnalysisRequestLogs.getBusinessRequest();
                             if (StringUtils.isNotEmpty(difyRequest)) {
-                                //调用dify接口
+                                CommunityTargetDTO communityTargetDTO = JSON.parseObject(message, CommunityTargetDTO.class);
+
+                                boolean hasImageNodeType = communityTargetDTO.getContent().stream()
+                                        .anyMatch(contentNode -> NodeTypeEnum.IMAGE.getCode().equals(contentNode.getNodeType()));
+                                //如果存在图片节点，则先处理图片节点
+                                StringBuilder sb = new StringBuilder();
+                                processingImageInfo(hasImageNodeType, communityTargetDTO, sb);
+
+                                String textContent = Optional.ofNullable(communityTargetDTO.getContent())
+                                        .orElse(Collections.emptyList()).stream()
+                                        .filter(contentNode -> NodeTypeEnum.TEXT.getCode().equals(contentNode.getNodeType()))
+                                        .map(CommunityTargetDTO.ContentNode::getNodeContent)
+                                        .collect(Collectors.joining(" "));
+
+                                //构建舆情分析请求对象
+                                DifyCommunityTargetDTO difyCommunityTargetDTO = new DifyCommunityTargetDTO();
+                                difyCommunityTargetDTO.setTargetId(communityTargetDTO.getTargetId());
+                                difyCommunityTargetDTO.setTargetType(communityTargetDTO.getTargetType());
+                                //脱敏处理
+                                dataMasking(textContent, sb, difyCommunityTargetDTO);
+
+                                //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
+                                syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisError.getAiAnalysisRequestId());
+
                                 DiFyReq diFyReq = new DiFyReq();
                                 diFyReq.setUser(user);
                                 diFyReq.setFlowId(flowId);
-                                diFyReq.setInputs(JSONObject.parseObject(difyRequest));
+                                diFyReq.setInputs(difyCommunityTargetDTO);
-                                JSONObject difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
+                                //调用舆情文本分析dify工作流
-                                this.dueCommunityDifyResponse(difResult);
+                                JSONObject  difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
+                                //处理结果
+                                processingCommunityDifyResponse(difResult);
+
                                 //根据ai_analysis_request_id更新ai_analysis_errors表中的retry_count字段+1,更新status字段为1
                                 aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper<AiAnalysisErrors>()
                                         .eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId())