From 35c3bb05b55d5a58f2e88e7e17f1b5100854402d Mon Sep 17 00:00:00 2001
From: zhangfan <fan.zhang.9@volvocars.com>
Date: Thu, 26 Dec 2024 16:53:02 +0800
Subject: [PATCH] =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=89=93=E5=8D=B0=E4=BC=98?=
 =?UTF-8?q?=E5=8C=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../center/controller/TestController.java     |  1 +
 .../center/job/MessageProcessJob.java         |  4 +-
 .../impl/MqMessageRecordServiceImpl.java      | 70 +++++++++----------
 3 files changed, 38 insertions(+), 37 deletions(-)

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 06de818..6c25497 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
@@ -54,6 +54,7 @@ public class TestController {
         //"（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);
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 79dd629..a079214 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
@@ -20,10 +20,10 @@ public class MessageProcessJob {
     @XxlJob("processMessageByTask")
     public ResultMsg processMessageByTask() {
         try {
-            log.info("开始执行定时任务补偿处理消息");
+            log.info("processMessageByTask 开始执行定时任务补偿处理消息");
             mqMessageRecordService.processMessageByTask();
         } catch (Exception e) {
-            log.error("定时任务补偿处理消息异常",e.getMessage());
+            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
             throw new RuntimeException(e);
         }
         return ResultMsg.ok();
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 e94d297..3db85cd 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
@@ -27,7 +27,6 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 
-import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.util.*;
@@ -72,10 +71,10 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
     @Override
     @Transactional
     public boolean processMessageByMQ(String message) {
-        log.info("message: {}", message);
+        log.info("processMessageByMQ message: {}", message);
         LocalDateTime currTime = LocalDateTime.now();
         RabbitMqFormData oldItem = JSONObject.parseObject(message, RabbitMqFormData.class);
-        log.info("真假战败数据已获取 {}",oldItem.getFormId());
+        log.info("processMessageByMQ 真假战败数据已获取 {}",oldItem.getFormId());
         MqMessageRecord curMQMessageRecord = new MqMessageRecord();
         try {
             curMQMessageRecord.setSinceType(oldItem.getSinceType());
@@ -87,9 +86,9 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             curMQMessageRecord.setLastRetryTime(currTime);
             curMQMessageRecord.setTaskStatus(0);
             this.save(curMQMessageRecord);
-            log.info("数据入库成功(MQMessageRecord)");
+            log.info("processMessageByMQ 数据入库成功");
         } catch (Exception ex) {
-            log.error("数据入库失败(MQMessageRecord)", ex);
+            log.error("processMessageByMQ 数据入库失败", ex);
             return false;
         }
         if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),oldItem.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE51.getCode(),oldItem.getSubSinceType())) {
@@ -97,17 +96,17 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         } else if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),oldItem.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE52.getCode(),oldItem.getSubSinceType())) {
             this.processMqSinceType52(oldItem, curMQMessageRecord);
         } else {
-            log.info("辨别数据为未知类型sinceType {},subSinceType {}", oldItem.getSinceType(), oldItem.getSubSinceType());
+            log.info("processMessageByMQ 辨别数据为未知类型sinceType {},subSinceType {}", oldItem.getSinceType(), oldItem.getSubSinceType());
         }
         return true;
     }
 
     private void processMqSinceType51(RabbitMqFormData oldItem,MqMessageRecord curMQMessageRecord){
-        log.info("辨别数据为分析请求 {}" ,SubSinceTypeEnum.SINCETYPE51.getCode());
+        log.info("processMqSinceType51 辨别数据为分析请求 {}" ,SubSinceTypeEnum.SINCETYPE51.getCode());
         DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class);
         DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis);
         DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput();
-        log.info("数据分析完成,分析返回对象output数据为 {}", output);
+        log.info("processMqSinceType51 数据分析完成,分析返回对象output数据为 {}", output);
         DiffDefeatResult diffDefeatResult;
         if(Objects.equals(HandleStatusEnum.ANALYSIS_NORMAL.getCode(),output.getHandleStatus())){
             diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class);
@@ -124,10 +123,10 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         curMQMessageRecord.setRespCode(output.getHandleStatus() == null ? "" : output.getHandleStatus().toString());
         curMQMessageRecord.setRespContent(output.getResultStr());
         this.updateById(curMQMessageRecord);
-        log.info("数据更改分析状态完成");
+        log.info("processMqSinceType51 数据更改分析状态完成");
 
         if (Objects.equals(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode(),curMQMessageRecord.getTaskStatus())){
-            log.info("数据分析结果正常");
+            log.info("processMqSinceType51 数据分析结果正常");
             DiffdefeatApprove diffdefeatApprove = new DiffdefeatApprove();
             String userStatus = getUserStatus(diffDefeatResult.getUserStatus());
             diffdefeatApprove.setFormId(oldItem.getFormId());
@@ -137,15 +136,15 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             diffdefeatApprove.setInputStr(response.getContentStr());
             diffdefeatApprove.setDefeatTime(diffDefeatanAlysis.getDefeatTime());
             diffdefeatApproveService.save(diffdefeatApprove);
-            log.info("插入数据记录完成，开始发送Mq消息");
+            log.info("processMqSinceType51 插入数据记录完成，开始发送Mq消息");
             this.toMqSinceType53(diffDefeatanAlysis,diffDefeatResult,oldItem);
         } else {
-            log.info("数据分析结果不正常,等待重试");
+            log.info("processMqSinceType51 数据分析结果不正常,等待重试");
         }
     }
 
     private void processMqSinceType52(RabbitMqFormData oldItem,MqMessageRecord curMQMessageRecord){
-        log.info("辨别数据为分析请求 {}" ,SubSinceTypeEnum.SINCETYPE52.getCode());
+        log.info("processMqSinceType52 辨别数据为分析请求 {}" ,SubSinceTypeEnum.SINCETYPE52.getCode());
         com.volvo.ai.analytic.center.dto.req.DiffDefeatanApprove curDiffDefeatApprove = JSON.parseObject(JSON.toJSONString(oldItem.getData()), com.volvo.ai.analytic.center.dto.req.DiffDefeatanApprove.class);
         DiffdefeatApprove approveEntity = diffdefeatApproveService.lambdaQuery().eq(DiffdefeatApprove::getFormId, oldItem.getFormId()).last("limit 1").one();
         if (approveEntity != null) {
@@ -158,9 +157,9 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                 curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
                 curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
                 this.updateById(curMQMessageRecord);
-                log.info("审批结果更新完成");
+                log.info("processMqSinceType52 审批结果更新完成");
             } catch (Exception e) {
-                log.error("审批结果更新失败",e);
+                log.error("processMqSinceType52 审批结果更新失败",e);
                 curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
                 curMQMessageRecord.setRespContent(e.getMessage());
                 curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
@@ -191,7 +190,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 //            }
             //查询到的通话数据
             String sql = "select display,sourceId from ods_asrdetail_d where sourceId in ("+ sourceId+") order by transcribeTime";
-            log.info("查询通话数据sql {}",sql);
+            log.info("processChatRecord 查询通话数据sql {}",sql);
             List<Map<String, Object>> historyList = clickhouseJdbcTemplate.queryForList(sql);
             for (CallItem item : input.getCallList()) {
                 Optional<String> result = CollectionUtils.isEmpty(historyList) ? Optional.empty():
@@ -208,12 +207,12 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
         // 查询企微的数据
         if (StringUtils.isBlank(input.getVdqwUserId()) || StringUtils.isBlank(input.getVdqwCustomerId())) {
-            log.info("企微用户信息为空,不附加企微数据");
+            log.info("processChatRecord 企微用户信息为空,不附加企微数据");
         }else {
             try {
                 String userIdSql = "select distinct userid as userid from ods_workuserinfo_d where middleuserid ='" + input.getVdqwUserId()
                         + "' union ALL select externaluserid as userid from ods_externalcontact_d where unionid ='" + input.getVdqwCustomerId() + "'";
-                log.info("查询企微用户数据sql {}",userIdSql);
+                log.info("processChatRecord 查询企微用户数据sql {}",userIdSql);
                 List<Map<String, Object>> userList = clickhouseJdbcTemplate.queryForList(userIdSql);
                 if (!CollectionUtils.isEmpty(userList) && userList.size() == 2) {
                     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -235,7 +234,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                             "fromuserid ='" + user1 + "' and acceptuserid ='" + user2 + "') or (" +
                             "fromuserid ='" + user2 + "' and acceptuserid ='" + user1 + "'))" +
                             "order by seq;";
-                    log.info("企微数据查询sql {}",sql);
+                    log.info("processChatRecord 企微数据查询sql {}",sql);
                     List<SessionItem> userInfoList = clickhouseJdbcTemplate.queryForList(sql, SessionItem.class);
 
                     for (int i=0;i<offsetDay;i++){
@@ -246,7 +245,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                     }
                 }
             } catch (Exception e) {
-                log.error("企微数据处理异常",e);
+                log.error("processChatRecord 企微数据处理异常",e);
             }
         }
         // 对集合进行排序
@@ -281,20 +280,21 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             }
         }
         if (StringUtils.isEmpty(contentStr.toString())) {
+            log.info("processChatRecord 未获取到任何语音或企微语料信息");
             output.setHandleStatus(HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getCode());
             output.setResultStr(HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getMessage());
-            return new DiffDefeatAnalyseOutputResult(output, "{\"Content\": \""+contentStr.toString()+"\"}");
+            return new DiffDefeatAnalyseOutputResult(output, "{\"Content\": \""+contentStr+"\"}");
         }
         //获取脱敏配置信息
         List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC);
 
-        log.info("开始脱敏 {}", contentStr);
+        log.info("processChatRecord 开始脱敏 {}", contentStr);
         RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
         runMaskingRuleInput.setOpinionId(input.getBusinessId());
         runMaskingRuleInput.setOldStr(contentStr.toString());
         runMaskingRuleInput.setDataMaskingRules(maskingRuleItems);
         String summaryText = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
-        log.info("脱敏结果 {}", summaryText);
+        log.info("processChatRecord 脱敏结果 {}", summaryText);
 
         Map<String, Object> record = new HashMap<>();
         record.put("record",summaryText);
@@ -306,13 +306,13 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         JSONObject difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
         output.setHandleStatus(HandleStatusEnum.ANALYSIS_NORMAL.getCode());
         output.setResultStr(difResult.getString("result"));
-        log.info("DiFy平台处理结果:{}", output.getResultStr());
-        return new DiffDefeatAnalyseOutputResult(output, "{\"Content\": \""+contentStr.toString()+"\"}");
+        log.info("processChatRecord DiFy平台处理结果:{}", output.getResultStr());
+        return new DiffDefeatAnalyseOutputResult(output, "{\"Content\": \""+contentStr+"\"}");
     }
 
     @Override
     public void processMessageByTask() {
-        log.info("开始处理任务");
+        log.info("processMessageByTask 开始处理任务");
         LocalDateTime currTime = LocalDateTime.now();
         LocalDateTime startTime = currTime.plusDays(-3);
         LocalDateTime stopTime = currTime.plusMinutes(-5);
@@ -327,7 +327,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
             return ;
         }
         for (MqMessageRecord curMQMessageRecord : mqMessageRecords){
-            log.info("辨别数据为分析请求");
+            log.info("processMessageByTask 辨别数据为分析请求");
             RabbitMqFormData oldItem = JSONObject.parseObject(curMQMessageRecord.getMessageContent(), RabbitMqFormData.class);
             try {
                 if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),curMQMessageRecord.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE51.getCode(),curMQMessageRecord.getSubSinceType())) {
@@ -335,7 +335,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                     DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class);
                     DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis);
                     DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput();
-                    log.info("数据分析完成");
+                    log.info("processMessageByTask 数据分析完成");
                     if(Objects.equals(HandleStatusEnum.ANALYSIS_NORMAL.getCode(),output.getHandleStatus())){
                         curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
                         diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class);
@@ -360,7 +360,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                             approve.setInputStr(response.getContentStr());
                             diffdefeatApproveService.updateById(approve);
                         } else {
-                            log.info("数据分析结果正常");
+                            log.info("processMessageByTask 数据分析结果正常");
                             approve = new DiffdefeatApprove();
                             approve.setFormId(oldItem.getFormId());
                             approve.setBusinessId(diffDefeatanAlysis.getBusinessId());
@@ -369,7 +369,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                             approve.setInputStr(response.getContentStr());
                             approve.setDefeatTime(diffDefeatanAlysis.getDefeatTime());
                             diffdefeatApproveService.save(approve);
-                            log.info("插入数据记录完成");
+                            log.info("processMessageByTask 插入数据记录完成");
                         }
                         this.toMqSinceType53(diffDefeatanAlysis,diffDefeatResult,oldItem);
                     }
@@ -378,7 +378,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                     curMQMessageRecord.setLastRetryTime(currTime);
                     curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
                     this.updateById(curMQMessageRecord);
-                    log.info("数据更改分析状态完成");
+                    log.info("processMessageByTask 数据更改分析状态完成");
                 } else if (Objects.equals(curMQMessageRecord.getSinceType(),SinceTypeEnum.SINCETYPE2.getCode()) && Objects.equals(curMQMessageRecord.getSubSinceType(), SubSinceTypeEnum.SINCETYPE52.getCode())) {
                     DiffdefeatApprove approveEntity = diffdefeatApproveService.lambdaQuery()
                             .eq(DiffdefeatApprove::getFormId, oldItem.getFormId())
@@ -400,10 +400,10 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                         this.updateById(curMQMessageRecord);
                     }
                 } else {
-                    log.info("辨别数据为未知类型sinceType {},subSinceType {}", oldItem.getSinceType(), oldItem.getSubSinceType());
+                    log.info("processMessageByTask 辨别数据为未知类型sinceType {},subSinceType {}", oldItem.getSinceType(), oldItem.getSubSinceType());
                 }
             } catch (Exception e) {
-                log.error("数据处理异常", e);
+                log.error("processMessageByTask 数据处理异常", e);
                 curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
                 curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
                 curMQMessageRecord.setRespContent(e.getMessage());
@@ -447,7 +447,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                             str.append(RoleEnum.USER.getMessage()).append("：").append(contentText.getContent()).append("\n");
                         }
                     } catch (Exception e) {
-                        log.error("反序列化异常: ", e);
+                        log.error("addSliceData 反序列化异常: ", e);
                     }
                 }
             }
@@ -479,6 +479,6 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         String callbackInput = JSONObject.toJSONString(rabbitMqToData);
 //        rabbitTemplate.convertAndSend(Constant.rabbitToFormQueue, callbackInput);
         rocketMQTemplate.syncSend(topic, callbackInput);
-        log.info("发送回调MQ完成: {}", callbackInput);
+        log.info("toMqSinceType53 发送回调MQ完成: {}", callbackInput);
     }
 }