代码合并

ai-analytic-center-biz/pom.xml

@@ -220,6 +220,12 @@
 			<version>2.3.0</version>
 		</dependency>
 
+		<dependency>
+			<groupId>com.xuxueli</groupId>
+			<artifactId>xxl-job-core</artifactId>
+			<version>2.3.0</version>
+		</dependency>
+
 	</dependencies>
 
 	<build>

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java

@@ -6,6 +6,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.volvo.ai.analytic.center.dto.req.CallItem;
 import com.volvo.ai.analytic.center.dto.req.DiFyReq;
 import com.volvo.ai.analytic.center.dto.req.DiffDefeatanAlysis;
+import com.volvo.ai.analytic.center.dto.req.DifyImageWorkFlow;
 import com.volvo.ai.analytic.center.dto.resp.DiffDefeatAnalyseOutputResult;
 import com.volvo.ai.analytic.center.feign.DiFyFeign;
 import com.volvo.ai.analytic.center.service.DiFyService;
@@ -14,6 +15,7 @@ import com.volvo.common.core.util.ResultMsg;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -28,71 +30,20 @@ import java.util.Map;
 
 @RestController
 @Api(tags = "测试类API")
-@RequestMapping("/test")
+@RequestMapping("/ai-analytic-center")
 @Slf4j
 public class TestController {
 
     @Autowired
-    private DiFyFeign diFyFeign;
+    private RocketMQTemplate rocketMQTemplate;
-
-    @Autowired
-    private DiFyService diFyService;
-
-    @Autowired
-    private MqMessageRecordService mqMessageRecordService;
-
-    @Value("${dify.user}")
-    private String user;
-
-    @Value("${dify.flowId}")
-    private String flowId;
 
 
-    @PostMapping("/runWorkflows2")
-    @ApiOperation(value = "测试Dify")
-    public ResultMsg<Object> runWorkflows2(@RequestBody String contentStr) {
-        //"（1）语音电话（2024-11-09 12:10:57）客服：哎，哎，姚先生，你好打扰到您了，蹦蹦沃沃赛练呢。客户：喂你好。客服：哎，对对，小明那个昨天看了车不要着急走，今天可有时间到店来接待您试驾感受一下呢。客户：我现在已经开工了，我现在没时间过去啊。客服：好的，那等你忙完这段时间好吧。客户：好的，好的，好的好，再见啊。客服：好，那这边先不打扰您，哎，好，再见。";
-        Map<String, Object> record = new HashMap<>();
-        record.put("record",contentStr);
-        log.info("contentStr {}", contentStr);
-        DiFyReq diFyReq = new DiFyReq();
-        diFyReq.setUser(user);
-        diFyReq.setFlowId(flowId);
-        diFyReq.setInputs(record);
-        Object difyResult = diFyService.getDiFyObject(diFyReq);
-        JSONObject dify = JSONObject.parseObject(difyResult.toString());
-        String resultStr = dify.getString("result").replace("\n", "").replaceAll(" ", "");
-        return ResultMsg.ok(resultStr);
-    }
 
-    @PostMapping("/runWorkflows3")
+    @PostMapping("/mockMq")
-    @ApiOperation(value = "测试Dify")
-    public ResultMsg<Object> runWorkflows3(@RequestBody String contentStr) {
-        //[{"sessionId":"1","sourceId":"1","audioTime":"2024-11-09 12:10:57","audioBillsec":"10"},{"sessionId":"2","sourceId":"2","audioTime":"2024-11-09 12:11:57","audioBillsec":"20"}]
-        log.info("contentStr {}", contentStr);
-        List<CallItem> callList = JSON.parseArray(contentStr, CallItem.class);
-        DiffDefeatanAlysis input = new DiffDefeatanAlysis();
-        input.setCallList(callList);
-        DiffDefeatAnalyseOutputResult diffDefeatAnalyseOutputResult = mqMessageRecordService.processChatRecord(input);
-        return ResultMsg.ok(diffDefeatAnalyseOutputResult);
-    }
-
-    @PostMapping("/runWorkflows4")
-    @ApiOperation(value = "测试Dify")
-    public ResultMsg<Object> runWorkflows4(@RequestBody String contentStr) {
-
-        //{"formId": "123456789","sinceType": 2,"subSinceType": 51,"data": {"businessId": "123456789","vdqwUserId": "123","vdqwCustomerId": "uuid","defeatTime": "2024-10-27 08:09:09","callList": [{"sessionId": "123456789","sourceId": "123456789","audioTime": "2024-10-27 08:09:09","audioBillsec": 1800}, {"sessionId": "123456789","sourceId": "123456789","audioTime": "2024-10-27 08:09:09","audioBillsec": 1800}]} }
-        //{"formId": "123456789","sinceType": 2,"subSinceType": 52,"data": {"businessId": "123456789", "approveCode": "10001", "approveResult": "审批通过", "approveOpinion": "" } }
-        log.info("contentStr {}", contentStr);
-        boolean result = mqMessageRecordService.processMessageByMQ(contentStr);
-        return ResultMsg.ok(result);
-    }
-
-    @PostMapping("/runWorkflows5")
     @ApiOperation(value = "补偿处理消息")
-    public ResultMsg<Object> runWorkflows5() {
+    public ResultMsg<Object> mockMq(@RequestBody String message) {
-        mqMessageRecordService.processMessageByTask();
+        rocketMQTemplate.syncSend("COMMUNITY_POST_TO_AI_TOPIC", message);
-        return ResultMsg.ok("");
+        return ResultMsg.ok("ok");
     }
 }
 

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/MessageProcessJob.java

@@ -17,13 +17,13 @@ public class MessageProcessJob {
     /**
      * 邀约消息推送
      */
-    @XxlJob("processMessageByTask")
+    @XxlJob("communityMessageByTask")
     public ResultMsg processMessageByTask() {
         try {
-            log.info("processMessageByTask 开始执行定时任务补偿处理消息");
+            log.info("communityMessageByTask 开始执行定时任务补偿处理消息");
-            mqMessageRecordService.processMessageByTask();
+            mqMessageRecordService.communityMessageByTask();
         } catch (Exception e) {
-            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
+            log.error("communityMessageByTask 定时任务补偿处理消息异常",e.getMessage());
             throw new RuntimeException(e);
         }
         return ResultMsg.ok();

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CommunityMessageConsumer.java

@@ -20,7 +20,7 @@ public class CommunityMessageConsumer implements RocketMQListener<MessageExt>{
 
     @Override
     public void onMessage(MessageExt messageExt) {
-        log.info("Received message: " + messageExt);
+        log.info("Received 社区舆情分析 message: " + messageExt);
         mqMessageRecordService.processMessageByMQ(new String(messageExt.getBody()));
     }
 }

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java

@@ -1,3 +1,4 @@
+/*
 package com.volvo.ai.analytic.center.mq;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -16,13 +17,15 @@ import org.springframework.web.bind.annotation.RestController;
 
 import java.time.LocalDateTime;
 
+*/
 /**
  * @ClassName CorpusProcessKafkaConsumer
  * @Description
  * @Author renzhen
  * @Date 2025-03-04 10:18
  * @Version 1.0
- **/
+ **//*
+
 @Slf4j
 @Component
 @RestController
@@ -63,3 +66,4 @@ public class CorpusProcessKafkaProducer {
 		}
 	}
 }
+*/

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

@@ -16,4 +16,9 @@ public interface MqMessageRecordService extends IService<MqMessageRecord> {
 
     //定时任务补偿处理消息
     void processMessageByTask();
+
+    //定时任务补偿处理消息
+    void communityMessageByTask();
 }
+
+

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

@@ -36,8 +36,9 @@ public class DiFyServiceImpl implements DiFyService{
         map.put("inputs",diFyReq.getInputs());
         map.put("response_mode","blocking");
         map.put("user",diFyReq.getUser());
-
+        log.info("请求DiFy入参:{}",JSON.toJSONString(map));
         JSONObject difyResult = diFyFeign.runWorkflows("Bearer "+diFyReq.getFlowId(),map);
+        log.info("请求DiFy响应结果:{}",difyResult.toJSONString());
         JSONObject data = difyResult.getJSONObject("data");
         if (data != null && "succeeded".equals(data.get("status"))){
             JSONObject outputs = data.getJSONObject("outputs");

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

@@ -4,6 +4,9 @@ import cn.hutool.core.date.DateUtil;
 import com.alibaba.cloud.commons.lang.StringUtils;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.volvo.ai.analytic.center.constant.Constant;
 import com.volvo.ai.analytic.center.dto.req.*;
@@ -24,6 +27,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.jdbc.core.BeanPropertyRowMapper;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
@@ -61,10 +65,10 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
     @Value("${dify.user}")
     private String user;
 
-    @Value("${dify.flowId}")
+    @Value("${dify.community.targetToken}")
     private String flowId;
 
-    @Value("${dify.imageFlowId}")
+    @Value("${dify.community.imageToken}")
     private String imageFlowId;
 
     @Value("${rocketmq.producer.topic}")
@@ -87,15 +91,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         }
         // 生成ai分析请求id
         String aiAnalysisRequestId = AiAnalysisUtils.getAiAnalysisRequestId(BusinessTypeEnum.COMMUNITYTARGET.getCode());
-
+        // 异步保存请求日志
-        // 保存请求日志
+        syncSaveRequestLogs(message, aiAnalysisRequestId);
-        aiAnalysisRequestLogsMapper.insert(AiAnalysisRequestLogs.builder()
-                .aiAnalysisRequestId(aiAnalysisRequestId)
-                .businessRequest(message)
-                .difyAgentKey(flowId)
-                .difyRequest(JSON.toJSONString(communityTargetDTO))
-                .aiAnalysisRequestType(BusinessTypeEnum.COMMUNITYTARGET.getCode())
-                .build());
 
         JSONObject difResult = new JSONObject();
         try {
@@ -103,36 +100,23 @@ 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);
-                        diFyImageReq.setInputs(diFyImageWorkFlow);
-
-                        //调用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()))
                     .map(CommunityTargetDTO.ContentNode::getNodeContent)
-                    .collect(Collectors.joining(""));
+                    .collect(Collectors.joining(" "));
 
             //构建舆情分析请求对象
             DifyCommunityTargetDTO difyCommunityTargetDTO = new DifyCommunityTargetDTO();
             difyCommunityTargetDTO.setTargetId(communityTargetDTO.getTargetId());
             difyCommunityTargetDTO.setTargetType(communityTargetDTO.getTargetType());
-            difyCommunityTargetDTO.setTargetContent(textContent + sb.toString());
+            //脱敏处理
+            dataMasking(textContent, sb, difyCommunityTargetDTO);
+
+            //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
+            syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisRequestId);
 
             DiFyReq diFyReq = new DiFyReq();
             diFyReq.setUser(user);
@@ -140,32 +124,115 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             diFyReq.setInputs(difyCommunityTargetDTO);
             //调用舆情文本分析dify工作流
             difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
-            JSONObject difResultJSONObject = difResult.getJSONObject("text");
+            //处理结果
-            if (difResultJSONObject != null) {
+            processingCommunityDifyResponse(difResult);
-                DifyCommunityTargetResult difyCommunityTargetResult = new DifyCommunityTargetResult();
+            //异步更新请求日志表的difyResponse字段
-                String targetId = difResultJSONObject.getString("targetId");
+            syncUpdateDiFyResponse(difResult, aiAnalysisRequestId);
-                String targetType = difResultJSONObject.getString("targetType");
-                String commentAnswer = difResultJSONObject.getString("commentAnswer");
-
-                difyCommunityTargetResult.setTargetType(targetType);
-                difyCommunityTargetResult.setTargetId(targetId);
-                difyCommunityTargetResult.setCommentAnswer(commentAnswer);
-                //返回结果推送到社区的MQ
-                rocketMQTemplate.syncSend("", JSON.toJSONString(difyCommunityTargetResult));
-                log.info("舆情分析发送回调MQ完成: {}", JSON.toJSONString(difyCommunityTargetResult));
-            }
         } catch (Exception e) {
-            log.error("请求DiFy异常:{}", e);
+            log.error("舆情自动化异常:{}", e.getMessage());
             //保存错误日志
             aiAnalysisErrorsMapper.insert(AiAnalysisErrors.builder()
                     .aiAnalysisRequestId(aiAnalysisRequestId)
                     .difyResponse(difResult.toJSONString())
                     .aiAnalysisErrorMessage(e.getMessage())
+                    .aiAnalysisRequestType(BusinessTypeEnum.COMMUNITYTARGET.getCode())
                     .build());
         }
         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 processingCommunityDifyResponse(JSONObject difResult) {
+        if (difResult != null) {
+            DifyCommunityTargetResult difyCommunityTargetResult = new DifyCommunityTargetResult();
+            String targetId = difResult.getString("targetId");
+            String targetType = difResult.getString("targetType");
+            String commentAnswer = difResult.getString("commentAnswer");
+            String keyWords = difResult.getString("keyWords");
+            String targetTag = difResult.getString("targetTag");
+
+            difyCommunityTargetResult.setTargetType(targetType);
+            difyCommunityTargetResult.setTargetId(targetId);
+            difyCommunityTargetResult.setCommentAnswer(commentAnswer);
+            difyCommunityTargetResult.setTargetTag(targetTag);
+            difyCommunityTargetResult.setKeyWords(keyWords);
+            //返回结果推送到社区的MQ
+            rocketMQTemplate.syncSend(topic, JSON.toJSONString(difyCommunityTargetResult));
+            log.info("舆情分析发送回调MQ完成: {}", JSON.toJSONString(difyCommunityTargetResult));
+        }
+    }
+
+    @Async
+    protected void syncUpdateDiFyRequest(DifyCommunityTargetDTO difyCommunityTargetDTO, String aiAnalysisRequestId) {
+        aiAnalysisRequestLogsMapper.update(new AiAnalysisRequestLogs(),
+                new UpdateWrapper<AiAnalysisRequestLogs>().set("dify_request", JSON.toJSONString(difyCommunityTargetDTO))
+                        .eq("ai_analysis_request_id", aiAnalysisRequestId));
+    }
+
+    @Async
+    protected void syncUpdateDiFyResponse(JSONObject difResult, String aiAnalysisRequestId) {
+        aiAnalysisRequestLogsMapper.update(new AiAnalysisRequestLogs(),
+                new UpdateWrapper<AiAnalysisRequestLogs>().set("dify_response", difResult.toJSONString())
+                        .eq("ai_analysis_request_id", aiAnalysisRequestId));
+    }
+
+    @Async
+    protected void syncSaveRequestLogs(String message, String aiAnalysisRequestId) {
+        aiAnalysisRequestLogsMapper.insert(AiAnalysisRequestLogs.builder()
+                .aiAnalysisRequestId(aiAnalysisRequestId)
+                .businessRequest(message)
+                .difyAgentKey(flowId)
+                .aiAnalysisRequestType(BusinessTypeEnum.COMMUNITYTARGET.getCode())
+                .build());
+    }
+
     private void processMqSinceType51(MqFormData oldItem, MqMessageRecord curMQMessageRecord) {
         log.info("processMqSinceType51 辨别数据为分析请求 {}", SubSinceTypeEnum.SINCETYPE51.getCode());
         DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class);
@@ -481,6 +548,85 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         }
     }
 
+    /**
+     * 处理社区异常消息
+     */
+    @Override
+    public void communityMessageByTask() {
+
+        try {
+            //捞取异常表中属于社区的异常数据
+            List<AiAnalysisErrors> aiAnalysisErrors = aiAnalysisErrorsMapper.selectList(new LambdaQueryWrapper<AiAnalysisErrors>()
+                    .eq(AiAnalysisErrors::getAiAnalysisRequestType, BusinessTypeEnum.COMMUNITYTARGET.getCode())
+                    .eq(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "0")
+                    .lt(AiAnalysisErrors::getRetryCount, 4));
+            if (aiAnalysisErrors != null && aiAnalysisErrors.size() > 0) {
+                //根据ai_analysis_request_id获取AiAnalysisRequestLogs表中的对应的dify_request字段
+                for (AiAnalysisErrors aiAnalysisError : aiAnalysisErrors) {
+                    try {
+                        AiAnalysisRequestLogs aiAnalysisRequestLogs = aiAnalysisRequestLogsMapper.selectOne(new LambdaQueryWrapper<AiAnalysisRequestLogs>()
+                                .eq(AiAnalysisRequestLogs::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId()));
+                        if (aiAnalysisRequestLogs != null) {
+                            //获取dify_request字段
+                            String message = aiAnalysisRequestLogs.getBusinessRequest();
+                            if (StringUtils.isNotEmpty(message)) {
+                                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(difyCommunityTargetDTO);
+                                //调用舆情文本分析dify工作流
+                                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())
+                                        .set(AiAnalysisErrors::getRetryCount, aiAnalysisError.getRetryCount() + 1)
+                                        .set(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "1"));
+                            }else{
+                                aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper<AiAnalysisErrors>()
+                                        .eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId())
+                                        .set(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "2"));
+                            }
+                        }
+                    } catch (Exception e) {
+                        log.error("补偿社区消息,AIID:{},异常:{}",aiAnalysisError.getAiAnalysisRequestId(), e);
+                        aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper<AiAnalysisErrors>()
+                                .eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId())
+                                .set(AiAnalysisErrors::getRetryCount, aiAnalysisError.getRetryCount() + 1));
+                    }
+                }
+            }
+        } catch (Exception e) {
+           log.error("处理社区异常消息异常:{}", e);
+        }
+    }
+
     private String getUserStatus(String oldStr) {
         if (Objects.equals(MessageConvertEnum.CONFIRMED.getCode(), oldStr)) {
             return MessageConvertEnum.CONFIRMED.getMessage();