diff --git a/ai-analytic-center-biz/pom.xml b/ai-analytic-center-biz/pom.xml index 5dd5c3d..2ab5bba 100644 --- a/ai-analytic-center-biz/pom.xml +++ b/ai-analytic-center-biz/pom.xml @@ -220,6 +220,12 @@ 2.3.0 + + com.xuxueli + xxl-job-core + 2.3.0 + + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java index 1294dfd..902fdf7 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java +++ b/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; - - @Autowired - private DiFyService diFyService; - - @Autowired - private MqMessageRecordService mqMessageRecordService; - - @Value("${dify.user}") - private String user; - - @Value("${dify.flowId}") - private String flowId; + private RocketMQTemplate rocketMQTemplate; - @PostMapping("/runWorkflows2") - @ApiOperation(value = "测试Dify") - public ResultMsg runWorkflows2(@RequestBody String contentStr) { - //"（1）语音电话（2024-11-09 12:10:57）客服：哎，哎，姚先生，你好打扰到您了，蹦蹦沃沃赛练呢。客户：喂你好。客服：哎，对对，小明那个昨天看了车不要着急走，今天可有时间到店来接待您试驾感受一下呢。客户：我现在已经开工了，我现在没时间过去啊。客服：好的，那等你忙完这段时间好吧。客户：好的，好的，好的好，再见啊。客服：好，那这边先不打扰您，哎，好，再见。"; - Map 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") - @ApiOperation(value = "测试Dify") - public ResultMsg 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 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 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") + @PostMapping("/mockMq") @ApiOperation(value = "补偿处理消息") - public ResultMsg runWorkflows5() { - mqMessageRecordService.processMessageByTask(); - return ResultMsg.ok(""); + public ResultMsg mockMq(@RequestBody String message) { + rocketMQTemplate.syncSend("COMMUNITY_POST_TO_AI_TOPIC", message); + return ResultMsg.ok("ok"); } } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/MessageProcessJob.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/MessageProcessJob.java index a079214..270dd6f 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/MessageProcessJob.java +++ b/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 开始执行定时任务补偿处理消息"); - mqMessageRecordService.processMessageByTask(); + log.info("communityMessageByTask 开始执行定时任务补偿处理消息"); + mqMessageRecordService.communityMessageByTask(); } catch (Exception e) { - log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage()); + log.error("communityMessageByTask 定时任务补偿处理消息异常",e.getMessage()); throw new RuntimeException(e); } return ResultMsg.ok(); diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CommunityMessageConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CommunityMessageConsumer.java index e590206..d60bd52 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CommunityMessageConsumer.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CommunityMessageConsumer.java @@ -20,7 +20,7 @@ public class CommunityMessageConsumer implements RocketMQListener{ @Override public void onMessage(MessageExt messageExt) { - log.info("Received message: " + messageExt); + log.info("Received 社区舆情分析 message: " + messageExt); mqMessageRecordService.processMessageByMQ(new String(messageExt.getBody())); } } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java index cde79b9..5236ce0 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java +++ b/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 { } } } +*/ diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/MqMessageRecordService.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/MqMessageRecordService.java index 7613500..ac3900a 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/MqMessageRecordService.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/MqMessageRecordService.java @@ -16,4 +16,9 @@ public interface MqMessageRecordService extends IService { //定时任务补偿处理消息 void processMessageByTask(); + + //定时任务补偿处理消息 + void communityMessageByTask(); } + + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java index a347547..db5f84e 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java +++ b/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"); diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java index fcde389..d0331b7 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java +++ b/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 NodeTypeEnum.IMAGE.getCode().equals(contentNode.getNodeType())); //如果存在图片节点，则先处理图片节点 StringBuilder sb = new StringBuilder(); - if (hasImageNodeType) { - communityTargetDTO.getContent().forEach(contentNode -> { - if (contentNode.getNodeType().equals(NodeTypeEnum.IMAGE.getCode())) { - //构建图片分析请求对象 + processingImageInfo(hasImageNodeType, communityTargetDTO, sb); - 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 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().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().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 aiAnalysisErrors = aiAnalysisErrorsMapper.selectList(new LambdaQueryWrapper() + .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() + .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() + .eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId()) + .set(AiAnalysisErrors::getRetryCount, aiAnalysisError.getRetryCount() + 1) + .set(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "1")); + }else{ + aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper() + .eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisError.getAiAnalysisRequestId()) + .set(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "2")); + } + } + } catch (Exception e) { + log.error("补偿社区消息,AIID:{},异常:{}",aiAnalysisError.getAiAnalysisRequestId(), e); + aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper() + .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();