admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化结构

Browse Source
This commit is contained in:
zhangfan
2024-12-23 14:42:55 +08:00
parent a362e4e610
commit 597d7b6f3b
3 changed files with 167 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ClickHouseConfig.java
View File

@@ -22,7 +22,7 @@ public class ClickHouseConfig {
} }
@Bean(name = "clickHouseDataSource") @Bean(name = "clickHouseDataSource")
@ConfigurationProperties(prefix="spring.clickhouse") @ConfigurationProperties(prefix="spring.clickhouse-ads-ai")
public DataSource clickHouseDataSource(){ public DataSource clickHouseDataSource(){
DataSource dataSource = DataSourceBuilder.create().build(); DataSource dataSource = DataSourceBuilder.create().build();
return dataSource; return dataSource;

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

@@ -58,8 +58,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
/** /**
* 处理Mq消息 * 处理Mq消息
* @param message * @param message 消息具体内容
* @return * @return 返回处理情况
*/ */
@Override @Override
public boolean processMessageByMQ(String message) { public boolean processMessageByMQ(String message) {
@@ -83,90 +83,85 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
log.error("数据入库失败(MQMessageRecord)", ex); log.error("数据入库失败(MQMessageRecord)", ex);
return false; return false;
} }
if (SinceTypeEnum.SINCETYPE2.getCode().equals(oldItem.getSinceType()) && SubSinceTypeEnum.SINCETYPE51.getCode().equals(oldItem.getSubSinceType()) ) { if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),oldItem.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE51.getCode(),oldItem.getSubSinceType())) {
log.info("辨别数据为分析请求"); this.processMqSinceType51(oldItem, curMQMessageRecord);
DiffDefeatResult diffDefeatResult = null; } else if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),oldItem.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE52.getCode(),oldItem.getSubSinceType())) {
DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class); this.processMqSinceType52(oldItem, curMQMessageRecord);
DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis); } else {
DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput(); log.info("辨别数据为未知类型sinceType {},subSinceType {}", oldItem.getSinceType(), oldItem.getSubSinceType());
log.info("数据分析完成");
if(HandleStatusEnum.ANALYSIS_NORMAL.getCode().equals(output.getHandleStatus())){
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class);
} else if (HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getCode().equals(output.getHandleStatus())) {
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
diffDefeatResult = new DiffDefeatResult();
diffDefeatResult.setUserStatus("无效通话");
diffDefeatResult.setAppointmentResult("因为通话时间太短没有ASR转义文本，因此判断为无效通话");
} else {
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
}
curMQMessageRecord.setRespCode(output.getHandleStatus() == null ? "" : output.getHandleStatus().toString());
curMQMessageRecord.setRespContent(output.getResultStr());
this.updateById(curMQMessageRecord);
log.info("数据更改分析状态完成");
if (MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode().equals(curMQMessageRecord.getTaskStatus())){
log.info("数据分析结果正常");
DiffdefeatApprove curDiffdefeatApprove = new DiffdefeatApprove();
String userStatus = getUserStatus(diffDefeatResult.getUserStatus());
curDiffdefeatApprove.setFormId(oldItem.getFormId());
curDiffdefeatApprove.setBusinessId(diffDefeatanAlysis.getBusinessId());
curDiffdefeatApprove.setUserStatus(userStatus);
curDiffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult());
curDiffdefeatApprove.setInputStr(response.getContentStr());
curDiffdefeatApprove.setDefeatTime(diffDefeatanAlysis.getDefeatTime());
diffdefeatApproveService.save(curDiffdefeatApprove);
log.info("插入数据记录完成");
DiffDefeatanCallbak curDiffDefeatanCallbak = new DiffDefeatanCallbak();
curDiffDefeatanCallbak.setBusinessId(diffDefeatanAlysis.getBusinessId());
curDiffDefeatanCallbak.setResponseCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
curDiffDefeatanCallbak.setLabel(userStatus);
curDiffDefeatanCallbak.setDescribe(diffDefeatResult.getAppointmentResult());
RabbitMqToData rabbitMqToData = new RabbitMqToData();
rabbitMqToData.setFormId(oldItem.getFormId());
rabbitMqToData.setSinceType(SinceTypeEnum.SINCETYPE2.getCode());
rabbitMqToData.setSubSinceType(SubSinceTypeEnum.SINCETYPE53.getCode());
rabbitMqToData.setData(curDiffDefeatanCallbak);
String callbakInput = JSONObject.toJSONString(rabbitMqToData);
rabbitTemplate.convertAndSend(Constant.rabbitToFormQueue, message);
log.info("发送回调MQ完成: {}", callbakInput);
} else {
log.info("数据分析结果不正常,等待重试");
return false;
}
}
if (SinceTypeEnum.SINCETYPE2.getCode().equals(oldItem.getSinceType()) && SubSinceTypeEnum.SINCETYPE52.getCode().equals(oldItem.getSubSinceType())) {
log.info("辨别数据为审批结果");
com.volvo.ai.analytic.center.dto.req.DiffDefeatanApprove curDiffDefeatanApprove = 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) {
try {
approveEntity.setApproveCode(curDiffDefeatanApprove.getApproveCode());
approveEntity.setApproveResult(curDiffDefeatanApprove.getApproveResult());
approveEntity.setApproveOpinion(curDiffDefeatanApprove.getApproveOpinion());
diffdefeatApproveService.updateById(approveEntity);
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
this.updateById(curMQMessageRecord);
log.info("审批结果更新完成");
} catch (Exception e) {
log.error("审批结果更新失败",e.getMessage());
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
curMQMessageRecord.setRespContent(e.getMessage());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
this.updateById(curMQMessageRecord);
}
}
} }
return true; return true;
} }
private void processMqSinceType51(RabbitMqFormData oldItem,MqMessageRecord curMQMessageRecord){
log.info("辨别数据为分析请求 {}" ,SubSinceTypeEnum.SINCETYPE51.getCode());
DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class);
DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis);
DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput();
log.info("数据分析完成");
DiffDefeatResult diffDefeatResult;
if(Objects.equals(HandleStatusEnum.ANALYSIS_NORMAL.getCode(),output.getHandleStatus())){
diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class);
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
} else if (Objects.equals(HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getCode(),output.getHandleStatus())) {
diffDefeatResult = new DiffDefeatResult();
diffDefeatResult.setUserStatus("无效通话");
diffDefeatResult.setAppointmentResult("因为通话时间太短没有ASR转义文本，因此判断为无效通话");
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
} else {
diffDefeatResult = new DiffDefeatResult();
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
}
curMQMessageRecord.setRespCode(output.getHandleStatus() == null ? "" : output.getHandleStatus().toString());
curMQMessageRecord.setRespContent(output.getResultStr());
this.updateById(curMQMessageRecord);
log.info("数据更改分析状态完成");
if (Objects.equals(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode(),curMQMessageRecord.getTaskStatus())){
log.info("数据分析结果正常");
DiffdefeatApprove diffdefeatApprove = new DiffdefeatApprove();
String userStatus = getUserStatus(diffDefeatResult.getUserStatus());
diffdefeatApprove.setFormId(oldItem.getFormId());
diffdefeatApprove.setBusinessId(diffDefeatanAlysis.getBusinessId());
diffdefeatApprove.setUserStatus(userStatus);
diffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult());
diffdefeatApprove.setInputStr(response.getContentStr());
diffdefeatApprove.setDefeatTime(diffDefeatanAlysis.getDefeatTime());
diffdefeatApproveService.save(diffdefeatApprove);
log.info("插入数据记录完成，开始发送Mq消息");
this.toMqSinceType53(diffDefeatanAlysis,diffDefeatResult,oldItem);
} else {
log.info("数据分析结果不正常,等待重试");
}
}
private void processMqSinceType52(RabbitMqFormData oldItem,MqMessageRecord curMQMessageRecord){
log.info("辨别数据为分析请求 {}" ,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) {
try {
approveEntity.setApproveCode(curDiffDefeatApprove.getApproveCode());
approveEntity.setApproveResult(curDiffDefeatApprove.getApproveResult());
approveEntity.setApproveOpinion(curDiffDefeatApprove.getApproveOpinion());
diffdefeatApproveService.updateById(approveEntity);
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
this.updateById(curMQMessageRecord);
log.info("审批结果更新完成");
} catch (Exception e) {
log.error("审批结果更新失败",e);
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
curMQMessageRecord.setRespContent(e.getMessage());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
this.updateById(curMQMessageRecord);
}
}
}
/** /**
* 处理聊天和语音数据 * 处理聊天和语音数据
* @return
*/ */
@Override @Override
public DiffDefeatAnalyseOutputResult processChatRecord(DiffDefeatanAlysis input) { public DiffDefeatAnalyseOutputResult processChatRecord(DiffDefeatanAlysis input) {
@@ -175,7 +170,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
//获取脱敏配置信息 //获取脱敏配置信息
List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC); List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC);
List<DiffDefeatCorpuItem> curDiffDefeatCorpuItems = new ArrayList<>(); List<DiffDefeatCorpuItem> curDiffDefeatCorpItems = new ArrayList<>();
// 根据SourceId查询通话信息 // 根据SourceId查询通话信息
List<String> sourceIds = Optional.ofNullable(input.getCallList()).map(list -> list.stream().map(CallItem::getSourceId).collect(Collectors.toList())).orElse(Collections.emptyList()); List<String> sourceIds = Optional.ofNullable(input.getCallList()).map(list -> list.stream().map(CallItem::getSourceId).collect(Collectors.toList())).orElse(Collections.emptyList());
if (!CollectionUtils.isEmpty(sourceIds)) { if (!CollectionUtils.isEmpty(sourceIds)) {
@@ -183,28 +178,28 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
String sourceId = sourceIds.stream().map(code -> "'"+code+"'").collect(Collectors.joining(",")); String sourceId = sourceIds.stream().map(code -> "'"+code+"'").collect(Collectors.joining(","));
int asrCount = clickhouseJdbcTemplate.queryForObject("select count(1) from asr_speechdetail where source_id in ("+ sourceId+") and file_status='InProgress'",Integer.class); int asrCount = clickhouseJdbcTemplate.queryForObject("select count(1) from asr_speechdetail where source_id in ("+ sourceId+") and file_status='InProgress'",Integer.class);
if (asrCount > 0) { if (asrCount > 0) {
log.info("存在未完成解析的通话信息");
output.setHandleStatus(HandleStatusEnum.ANALYSIS_CALLING.getCode()); output.setHandleStatus(HandleStatusEnum.ANALYSIS_CALLING.getCode());
output.setResultStr(HandleStatusEnum.ANALYSIS_CALLING.getMessage()); output.setResultStr(HandleStatusEnum.ANALYSIS_CALLING.getMessage());
return new DiffDefeatAnalyseOutputResult(output, contentStr.toString()); return new DiffDefeatAnalyseOutputResult(output, contentStr.toString());
} }
//查询到的通话数据 //查询到的通话数据
List<Map<String, Object>> hishistoryList = clickhouseJdbcTemplate.queryForList("select id,msg_json,source_id from asr_hishistory where dialect_text=0 and source_id in ("+ sourceId+") order by id"); List<Map<String, Object>> historyList = clickhouseJdbcTemplate.queryForList("select id,msg_json,source_id from asr_hishistory where dialect_text=0 and source_id in ("+ sourceId+") order by id");
for (CallItem item : input.getCallList()) { for (CallItem item : input.getCallList()) {
Optional<String> result = CollectionUtils.isEmpty(hishistoryList) ? Optional.empty(): Optional<String> result = CollectionUtils.isEmpty(historyList) ? Optional.empty():
hishistoryList.stream() historyList.stream().filter(entity -> entity.get("source_id").equals(item.getSourceId()))
.filter(entity -> entity.get("source_id").equals(item.getSourceId()))
.findFirst().map(entity -> entity.get("msg_json") == null ? "" : entity.get("msg_json").toString()); .findFirst().map(entity -> entity.get("msg_json") == null ? "" : entity.get("msg_json").toString());
DiffDefeatCorpuItem curDiffDefeatCorpuItem = new DiffDefeatCorpuItem(); DiffDefeatCorpuItem curDiffDefeatCorpItem = new DiffDefeatCorpuItem();
curDiffDefeatCorpuItem.setSourceId(item.getSourceId()); curDiffDefeatCorpItem.setSourceId(item.getSourceId());
curDiffDefeatCorpuItem.setCategory(CategoryEnum.PHONE_VOICE.getCode()); curDiffDefeatCorpItem.setCategory(CategoryEnum.PHONE_VOICE.getCode());
curDiffDefeatCorpuItem.setCorpuText(result.isPresent() ? result.get() : ""); curDiffDefeatCorpItem.setCorpuText(result.orElse(""));
curDiffDefeatCorpuItem.setHappenTime(item.getAudioTime()); curDiffDefeatCorpItem.setHappenTime(item.getAudioTime());
curDiffDefeatCorpuItems.add(curDiffDefeatCorpuItem); curDiffDefeatCorpItems.add(curDiffDefeatCorpItem);
} }
} }
// 查询企微的数据 // 查询企微的数据
if (StringUtils.isBlank(input.getVdqwUserId())|| StringUtils.isBlank(input.getVdqwCustomerId())) { if (StringUtils.isBlank(input.getVdqwUserId()) || StringUtils.isBlank(input.getVdqwCustomerId())) {
log.info("企微用户信息为空,不附加企微数据"); log.info("企微用户信息为空,不附加企微数据");
}else { }else {
try { try {
@@ -213,7 +208,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
List<Map<String, Object>> userList = clickhouseJdbcTemplate.queryForList(userIdSql); List<Map<String, Object>> userList = clickhouseJdbcTemplate.queryForList(userIdSql);
if (!CollectionUtils.isEmpty(userList) && userList.size() == 2) { if (!CollectionUtils.isEmpty(userList) && userList.size() == 2) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String startTime = sdf.format(DateUtil.offsetDay(input.getDefeatTime(), -3)); int offsetDay = 3;
String startTime = sdf.format(DateUtil.offsetDay(input.getDefeatTime(), -offsetDay));
String endTime = sdf.format(input.getDefeatTime() == null ? new Date() : input.getDefeatTime()); String endTime = sdf.format(input.getDefeatTime() == null ? new Date() : input.getDefeatTime());
String user1 = userList.get(0).get("userid").toString(); String user1 = userList.get(0).get("userid").toString();
String user2 = userList.get(1).get("userid").toString(); String user2 = userList.get(1).get("userid").toString();
@@ -232,40 +228,39 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
"order by seq;"; "order by seq;";
List<SessionItem> userInfoList = clickhouseJdbcTemplate.queryForList(sql, SessionItem.class); List<SessionItem> userInfoList = clickhouseJdbcTemplate.queryForList(sql, SessionItem.class);
Date sliceSatrtTime = sdf.parse(startTime); for (int i=0;i<offsetDay;i++){
Date slice1StopTime = DateUtil.offsetDay(sliceSatrtTime,1); Date sliceStartTime = sdf.parse(startTime);
Date slice2StopTime = DateUtil.offsetDay(slice1StopTime,1); sliceStartTime = DateUtil.offsetDay(sliceStartTime,i);
Date slice3StopTime = sdf.parse(endTime); Date sliceStopTime = DateUtil.offsetDay(sliceStartTime,i+1);
this.addSliceData(curDiffDefeatCorpuItems, sliceSatrtTime, slice1StopTime, userInfoList); this.addSliceData(curDiffDefeatCorpItems, sliceStartTime, sliceStopTime, userInfoList);
this.addSliceData(curDiffDefeatCorpuItems, slice1StopTime, slice2StopTime, userInfoList); }
this.addSliceData(curDiffDefeatCorpuItems, slice2StopTime, slice3StopTime, userInfoList);
} }
} catch (ParseException e) { } catch (ParseException e) {
log.error("企微数据处理异常",e); log.error("企微数据处理异常",e);
} }
} }
// 对集合进行排序 // 对集合进行排序
curDiffDefeatCorpuItems.sort(Comparator.comparing(DiffDefeatCorpuItem::getHappenTime)); curDiffDefeatCorpItems.sort(Comparator.comparing(DiffDefeatCorpuItem::getHappenTime));
if (!CollectionUtils.isEmpty(curDiffDefeatCorpuItems)){ if (!CollectionUtils.isEmpty(curDiffDefeatCorpItems)){
for (DiffDefeatCorpuItem item : curDiffDefeatCorpuItems) { for (DiffDefeatCorpuItem item : curDiffDefeatCorpItems) {
if (CategoryEnum.ENTERPRISE_WECHAT.getCode().equals(item.getCategory())) { if (Objects.equals(CategoryEnum.ENTERPRISE_WECHAT.getCode(),item.getCategory())) {
contentStr.append(CategoryEnum.ENTERPRISE_WECHAT.getMessage()).append("\n"); contentStr.append(CategoryEnum.ENTERPRISE_WECHAT.getMessage()).append("\n");
contentStr.append(item.getCorpuText()); contentStr.append(item.getCorpuText());
} else if (CategoryEnum.PHONE_VOICE.getCode().equals(item.getCategory()) && StringUtils.isNotEmpty(item.getCorpuText())) { } else if (Objects.equals(CategoryEnum.PHONE_VOICE.getCode(),item.getCategory()) && StringUtils.isNotEmpty(item.getCorpuText())) {
// 反序列化CorpuText为KafkaJson对象 // 反序列化CorpuText为KafkaJson对象
KafkaJson oldItem = JSON.parseObject(item.getCorpuText(), KafkaJson.class); KafkaJson oldItem = JSON.parseObject(item.getCorpuText(), KafkaJson.class);
// 反序列化display为CollectTranscriberJobResponse对象 // 反序列化display为CollectTranscriberJobResponse对象
CollectTranscriberJobResponse resp = JSON.parseObject(oldItem.getDisplay(), CollectTranscriberJobResponse.class); CollectTranscriberJobResponse resp = JSON.parseObject(oldItem.getDisplay(), CollectTranscriberJobResponse.class);
// 检查状态并处理Segments // 检查状态并处理Segments
if ("FINISHED".equals(resp.getStatus()) && !CollectionUtils.isEmpty(resp.getSegments())) { if (Objects.equals("FINISHED",resp.getStatus()) && !CollectionUtils.isEmpty(resp.getSegments())) {
StringBuffer vocStr = new StringBuffer(); StringBuilder vocStr = new StringBuilder();
for (Segment segment : resp.getSegments()) { for (Segment segment : resp.getSegments()) {
if (RoleEnum.AGENT.getCode().equals(segment.getResult().getAnalysisInfo().getRole())) { if (Objects.equals(RoleEnum.AGENT.getCode(),segment.getResult().getAnalysisInfo().getRole())) {
vocStr.append(RoleEnum.AGENT.getMessage()+"：" + segment.getResult().getText() + "\n"); vocStr.append(RoleEnum.AGENT.getMessage()).append("：").append(segment.getResult().getText()).append("\n");
} else if (RoleEnum.USER.getCode().equals(segment.getResult().getAnalysisInfo().getRole())) { } else if (Objects.equals(RoleEnum.USER.getCode(),segment.getResult().getAnalysisInfo().getRole())) {
vocStr.append(RoleEnum.USER.getMessage()+"：" + segment.getResult().getText() + "\n"); vocStr.append(RoleEnum.USER.getMessage()).append("：").append(segment.getResult().getText()).append("\n");
} else { } else {
vocStr.append(RoleEnum.OTHER.getMessage()+"：" + segment.getResult().getText() + "\n"); vocStr.append(RoleEnum.OTHER.getMessage()).append("：").append(segment.getResult().getText()).append("\n");
} }
} }
contentStr.append(vocStr); contentStr.append(vocStr);
@@ -293,9 +288,9 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
diFyReq.setUser(user); diFyReq.setUser(user);
diFyReq.setFlowId(flowId); diFyReq.setFlowId(flowId);
diFyReq.setInputs(record); diFyReq.setInputs(record);
JSONObject difyResult = (JSONObject) diFyService.getDiFyObject(diFyReq); JSONObject difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
output.setHandleStatus(HandleStatusEnum.ANALYSIS_NORMAL.getCode()); output.setHandleStatus(HandleStatusEnum.ANALYSIS_NORMAL.getCode());
output.setResultStr(difyResult.getString("result")); output.setResultStr(difResult.getString("result"));
log.info("DiFy平台处理结果:{}", output.getResultStr()); log.info("DiFy平台处理结果:{}", output.getResultStr());
return new DiffDefeatAnalyseOutputResult(output, contentStr.toString()); return new DiffDefeatAnalyseOutputResult(output, contentStr.toString());
} }
@@ -319,26 +314,26 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
for (MqMessageRecord curMQMessageRecord : mqMessageRecords){ for (MqMessageRecord curMQMessageRecord : mqMessageRecords){
log.info("辨别数据为分析请求"); log.info("辨别数据为分析请求");
RabbitMqFormData oldItem = JSONObject.parseObject(curMQMessageRecord.getMessageContent(), RabbitMqFormData.class); RabbitMqFormData oldItem = JSONObject.parseObject(curMQMessageRecord.getMessageContent(), RabbitMqFormData.class);
if (curMQMessageRecord.getSinceType() == 2 && curMQMessageRecord.getSubSinceType() == SubSinceTypeEnum.SINCETYPE51.getCode() ) { try {
try { if (Objects.equals(SinceTypeEnum.SINCETYPE2.getCode(),curMQMessageRecord.getSinceType()) && Objects.equals(SubSinceTypeEnum.SINCETYPE51.getCode(),curMQMessageRecord.getSubSinceType())) {
DiffDefeatResult diffDefeatResult = null; DiffDefeatResult diffDefeatResult;
DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class); DiffDefeatanAlysis diffDefeatanAlysis = JSON.parseObject(JSON.toJSONString(oldItem.getData()), DiffDefeatanAlysis.class);
DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis); DiffDefeatAnalyseOutputResult response = this.processChatRecord(diffDefeatanAlysis);
DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput(); DiffDefeatAnalyseOutput output = response.getDiffDefeatAnalyseOutput();
log.info("数据分析完成"); log.info("数据分析完成");
if(HandleStatusEnum.ANALYSIS_NORMAL.getCode().equals(output.getHandleStatus())){ if(Objects.equals(HandleStatusEnum.ANALYSIS_NORMAL.getCode(),output.getHandleStatus())){
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode()); curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class); diffDefeatResult = JSONObject.parseObject(output.getResultStr(), DiffDefeatResult.class);
} else if (HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getCode().equals(output.getHandleStatus())) { } else if (Objects.equals(HandleStatusEnum.ANALYSIS_CONTENT_EMPTY.getCode(),output.getHandleStatus())) {
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode()); curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode());
diffDefeatResult = new DiffDefeatResult(); diffDefeatResult = new DiffDefeatResult();
diffDefeatResult.setUserStatus("无效通话"); diffDefeatResult.setUserStatus("无效通话");
diffDefeatResult.setAppointmentResult("因为通话时间太短没有ASR转义文本，因此判断为无效通话"); diffDefeatResult.setAppointmentResult("因为通话时间太短没有ASR转义文本，因此判断为无效通话");
} else { } else {
diffDefeatResult = new DiffDefeatResult();
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode()); curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
} }
if (curMQMessageRecord.getTaskStatus() == MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode()){ if (Objects.equals(MqTaskStatusEnum.ANALYSIS_COMPLETION.getCode(),curMQMessageRecord.getTaskStatus())){
String userStatus = getUserStatus(diffDefeatResult.getUserStatus());
DiffdefeatApprove curDiffdefeatApprove = diffdefeatApproveService.lambdaQuery() DiffdefeatApprove curDiffdefeatApprove = diffdefeatApproveService.lambdaQuery()
.eq(DiffdefeatApprove::getFormId, oldItem.getFormId()) .eq(DiffdefeatApprove::getFormId, oldItem.getFormId())
.eq(DiffdefeatApprove::getBusinessId, curMQMessageRecord.getSubBizNo()) .eq(DiffdefeatApprove::getBusinessId, curMQMessageRecord.getSubBizNo())
@@ -346,34 +341,22 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
.last("limit 1").one(); .last("limit 1").one();
if(curDiffdefeatApprove!=null){ if(curDiffdefeatApprove!=null){
curDiffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult()); curDiffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult());
curDiffdefeatApprove.setUserStatus(userStatus); curDiffdefeatApprove.setUserStatus(getUserStatus(diffDefeatResult.getUserStatus()));
curDiffdefeatApprove.setInputStr(response.getContentStr()); curDiffdefeatApprove.setInputStr(response.getContentStr());
diffdefeatApproveService.updateById(curDiffdefeatApprove);
} else { } else {
log.info("数据分析结果正常"); log.info("数据分析结果正常");
curDiffdefeatApprove = new DiffdefeatApprove(); curDiffdefeatApprove = new DiffdefeatApprove();
curDiffdefeatApprove.setFormId(oldItem.getFormId()); curDiffdefeatApprove.setFormId(oldItem.getFormId());
curDiffdefeatApprove.setBusinessId(diffDefeatanAlysis.getBusinessId()); curDiffdefeatApprove.setBusinessId(diffDefeatanAlysis.getBusinessId());
curDiffdefeatApprove.setUserStatus(userStatus); curDiffdefeatApprove.setUserStatus(getUserStatus(diffDefeatResult.getUserStatus()));
curDiffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult()); curDiffdefeatApprove.setAppointmentResult(diffDefeatResult.getAppointmentResult());
curDiffdefeatApprove.setInputStr(response.getContentStr()); curDiffdefeatApprove.setInputStr(response.getContentStr());
curDiffdefeatApprove.setDefeatTime(diffDefeatanAlysis.getDefeatTime()); curDiffdefeatApprove.setDefeatTime(diffDefeatanAlysis.getDefeatTime());
diffdefeatApproveService.save(curDiffdefeatApprove); diffdefeatApproveService.save(curDiffdefeatApprove);
log.info("插入数据记录完成"); log.info("插入数据记录完成");
} }
this.toMqSinceType53(diffDefeatanAlysis,diffDefeatResult,oldItem);
DiffDefeatanCallbak curDiffDefeatanCallbak = new DiffDefeatanCallbak();
curDiffDefeatanCallbak.setBusinessId(diffDefeatanAlysis.getBusinessId());
curDiffDefeatanCallbak.setResponseCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
curDiffDefeatanCallbak.setLabel(userStatus);
curDiffDefeatanCallbak.setDescribe(diffDefeatResult.getAppointmentResult());
RabbitMqToData rabbitMqToData = new RabbitMqToData();
rabbitMqToData.setFormId(oldItem.getFormId());
rabbitMqToData.setSinceType(2);
rabbitMqToData.setSubSinceType(SubSinceTypeEnum.SINCETYPE53.getCode());
rabbitMqToData.setData(curDiffDefeatanCallbak);
String callbakInput = JSONObject.toJSONString(rabbitMqToData);
rabbitTemplate.convertAndSend(Constant.rabbitToFormQueue, callbakInput);
log.info("发送回调MQ完成: {}", callbakInput);
} }
curMQMessageRecord.setRespCode(output.getHandleStatus()+""); curMQMessageRecord.setRespCode(output.getHandleStatus()+"");
curMQMessageRecord.setRespContent(output.getResultStr()); curMQMessageRecord.setRespContent(output.getResultStr());
@@ -381,19 +364,9 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1); curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
this.updateById(curMQMessageRecord); this.updateById(curMQMessageRecord);
log.info("数据更改分析状态完成"); log.info("数据更改分析状态完成");
} catch (Exception e) {
log.error("数据解析异常: {}", e.getMessage());
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
curMQMessageRecord.setRespContent(e.getMessage());
curMQMessageRecord.setLastRetryTime(currTime);
curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
this.updateById(curMQMessageRecord);
} }
}
if (curMQMessageRecord.getSinceType() == 2 && curMQMessageRecord.getSubSinceType() == SubSinceTypeEnum.SINCETYPE52.getCode()) { if (Objects.equals(curMQMessageRecord.getSinceType(),SinceTypeEnum.SINCETYPE2.getCode()) && Objects.equals(curMQMessageRecord.getSubSinceType(), SubSinceTypeEnum.SINCETYPE52.getCode())) {
try {
DiffdefeatApprove approveEntity = diffdefeatApproveService.lambdaQuery() DiffdefeatApprove approveEntity = diffdefeatApproveService.lambdaQuery()
.eq(DiffdefeatApprove::getFormId, oldItem.getFormId()) .eq(DiffdefeatApprove::getFormId, oldItem.getFormId())
.eq(DiffdefeatApprove::getBusinessId, curMQMessageRecord.getSubBizNo()) .eq(DiffdefeatApprove::getBusinessId, curMQMessageRecord.getSubBizNo())
@@ -413,29 +386,26 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1); curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
this.updateById(curMQMessageRecord); this.updateById(curMQMessageRecord);
} }
} catch (Exception e) {
log.error("数据解析异常: {}", e.getMessage());
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
curMQMessageRecord.setRespContent(e.getMessage());
curMQMessageRecord.setLastRetryTime(currTime);
curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
this.updateById(curMQMessageRecord);
} }
} catch (Exception e) {
log.error("数据处理异常", e);
curMQMessageRecord.setTaskStatus(MqTaskStatusEnum.ANALYSIS_FAILURE.getCode());
curMQMessageRecord.setRespCode(HandleStatusEnum.ANALYSIS_CALLING.getCode().toString());
curMQMessageRecord.setRespContent(e.getMessage());
curMQMessageRecord.setLastRetryTime(currTime);
curMQMessageRecord.setRetryCount(curMQMessageRecord.getRetryCount()+1);
this.updateById(curMQMessageRecord);
} }
} }
} }
private String getUserStatus(String oldStr){
if (Objects.equals(MessageConvertEnum.CONFIRMED.getCode(),oldStr)){
private String getUserStatus(String oldStr)
{
if (MessageConvertEnum.CONFIRMED.getCode().equals(oldStr)){
return MessageConvertEnum.CONFIRMED.getMessage(); return MessageConvertEnum.CONFIRMED.getMessage();
}else if (MessageConvertEnum.LOOK_ON.getCode().equals(oldStr)){ }else if (Objects.equals(MessageConvertEnum.LOOK_ON.getCode(),oldStr)){
return MessageConvertEnum.LOOK_ON.getMessage(); return MessageConvertEnum.LOOK_ON.getMessage();
}else if (MessageConvertEnum.GIVE_UP.getCode().equals(oldStr)){ }else if (Objects.equals(MessageConvertEnum.GIVE_UP.getCode(),oldStr)){
return MessageConvertEnum.GIVE_UP.getMessage(); return MessageConvertEnum.GIVE_UP.getMessage();
}else if (MessageConvertEnum.INVALID.getCode().equals(oldStr)){ }else if (Objects.equals(MessageConvertEnum.INVALID.getCode(),oldStr)){
return MessageConvertEnum.INVALID.getMessage(); return MessageConvertEnum.INVALID.getMessage();
}else { }else {
return oldStr; return oldStr;
@@ -443,7 +413,6 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
} }
public void addSliceData(List<DiffDefeatCorpuItem> curDiffDefeatCorpuItems, Date sliceStartTime, Date sliceStopTime, List<SessionItem> sessionItems) { public void addSliceData(List<DiffDefeatCorpuItem> curDiffDefeatCorpuItems, Date sliceStartTime, Date sliceStopTime, List<SessionItem> sessionItems) {
// 使用Java 8的Stream API来过滤sessionItems
List<SessionItem> slices = sessionItems.stream() List<SessionItem> slices = sessionItems.stream()
.filter(a -> !a.getMsgtime().before(sliceStartTime) && a.getMsgtime().before(sliceStopTime)) .filter(a -> !a.getMsgtime().before(sliceStartTime) && a.getMsgtime().before(sliceStopTime))
.collect(Collectors.toList()); .collect(Collectors.toList());
@@ -452,7 +421,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
StringBuilder str = new StringBuilder(); StringBuilder str = new StringBuilder();
// 遍历sessionItems，构建字符串 // 遍历sessionItems，构建字符串
for (SessionItem sessionItem : sessionItems) { for (SessionItem sessionItem : sessionItems) {
if ("text".equals(sessionItem.getMsgtype()) && "send".equals(sessionItem.getActiontype())) { if (Objects.equals("text",sessionItem.getMsgtype()) && Objects.equals("send",sessionItem.getActiontype())) {
try { try {
// 反序列化content为ContentText对象 // 反序列化content为ContentText对象
ContentText contentText = JSON.parseObject(sessionItem.getContent(), ContentText.class); ContentText contentText = JSON.parseObject(sessionItem.getContent(), ContentText.class);
@@ -467,17 +436,33 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
} }
} }
} }
// 检查字符串是否不为空 // 检查字符串是否不为空
if (str.length() > 0) { if (str.length() > 0) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
DiffDefeatCorpuItem curDiffDefeatCorpuItem = new DiffDefeatCorpuItem(); DiffDefeatCorpuItem curItem = new DiffDefeatCorpuItem();
curDiffDefeatCorpuItem.setSourceId(slices.get(0).getMsgid()); curItem.setSourceId(slices.get(0).getMsgid());
curDiffDefeatCorpuItem.setCategory(CategoryEnum.ENTERPRISE_WECHAT.getCode()); curItem.setCategory(CategoryEnum.ENTERPRISE_WECHAT.getCode());
curDiffDefeatCorpuItem.setHappenTime(sdf.format(slices.get(0).getMsgtime())); curItem.setHappenTime(sdf.format(slices.get(0).getMsgtime()));
curDiffDefeatCorpuItem.setCorpuText(str.toString()); curItem.setCorpuText(str.toString());
curDiffDefeatCorpuItems.add(curDiffDefeatCorpuItem); curDiffDefeatCorpuItems.add(curItem);
} }
} }
} }
public void toMqSinceType53(DiffDefeatanAlysis diffDefeatanAlysis, DiffDefeatResult diffDefeatResult, RabbitMqFormData oldItem){
String userStatus = getUserStatus(diffDefeatResult.getUserStatus());
DiffDefeatanCallbak curDiffDefeatCallback = new DiffDefeatanCallbak();
curDiffDefeatCallback.setBusinessId(diffDefeatanAlysis.getBusinessId());
curDiffDefeatCallback.setResponseCode(HandleStatusEnum.ANALYSIS_NORMAL.getCode().toString());
curDiffDefeatCallback.setLabel(userStatus);
curDiffDefeatCallback.setDescribe(diffDefeatResult.getAppointmentResult());
RabbitMqToData rabbitMqToData = new RabbitMqToData();
rabbitMqToData.setFormId(oldItem.getFormId());
rabbitMqToData.setSinceType(SinceTypeEnum.SINCETYPE2.getCode());
rabbitMqToData.setSubSinceType(SubSinceTypeEnum.SINCETYPE53.getCode());
rabbitMqToData.setData(curDiffDefeatCallback);
String callbackInput = JSONObject.toJSONString(rabbitMqToData);
rabbitTemplate.convertAndSend(Constant.rabbitToFormQueue, callbackInput);
log.info("发送回调MQ完成: {}", callbackInput);
}
} }

8
ai-analytic-center-biz/src/main/resources/application-uat.yml
View File

@@ -64,3 +64,11 @@ mybatis-plus:
type-aliases-package: com.volvo.reportbi.mapper type-aliases-package: com.volvo.reportbi.mapper
configuration: configuration:
call-setters-on-nulls: true call-setters-on-nulls: true
service:
dify:
url: https://artera-uat.digitalvolvo.com
dify:
user: voc
flowId: app-MVGxogM08CDCg7jMo7GNF6eS