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-19 17:30:06 +08:00
parent c9efc86c16
commit 86a444c14b
12 changed files with 312 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
ai-analytic-center-biz/pom.xml
View File

@@ -126,6 +126,20 @@
<artifactId>xxl-job-core</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.3.2</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>com.zaxxer</groupId>-->
<!-- <artifactId>HikariCP</artifactId>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>com.alibaba</groupId>-->
<!-- <artifactId>druid-spring-boot-starter</artifactId>-->
<!-- <version>1.1.13</version>-->
<!-- </dependency>-->
</dependencies>
<build>

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

@@ -0,0 +1,40 @@
package com.volvo.ai.analytic.center.config;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.core.JdbcTemplate;
import javax.sql.DataSource;
@Configuration
public class ClickHouseConfig {
@Primary
@Bean(name = "dataSource")
@ConfigurationProperties(prefix="spring.datasource")
public DataSource dataSource() {
DataSource dataSource = DataSourceBuilder.create().build();
return dataSource;
}
@Bean(name = "clickHouseDataSource")
@ConfigurationProperties(prefix="spring.clickhouse")
public DataSource clickHouseDataSource(){
DataSource dataSource = DataSourceBuilder.create().build();
return dataSource;
}
@Bean(name = "jdbcTemplate")
public JdbcTemplate jdbcTemplate(@Qualifier("dataSource") DataSource dataSource) {
return new JdbcTemplate(dataSource);
}
@Bean(name = "clickhouseJdbcTemplate")
public JdbcTemplate clickhouseJdbcTemplate(@Qualifier("clickHouseDataSource") DataSource clickhouseJdbcTemplate) {
return new JdbcTemplate(clickhouseJdbcTemplate);
}
}

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

@@ -5,6 +5,7 @@ import com.alibaba.cloud.commons.lang.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.constant.Constant;
import com.volvo.ai.analytic.center.dto.req.*;
import com.volvo.ai.analytic.center.dto.resp.*;
@@ -20,13 +21,16 @@ import com.volvo.ai.analytic.center.service.DiffdefeatApproveService;
import com.volvo.ai.analytic.center.service.MqMessageRecordService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
@Slf4j
@@ -42,6 +46,15 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
@Autowired
private DiFyService diFyService;
@Autowired
private JdbcTemplate clickhouseJdbcTemplate;
@Value("${dify.user}")
private String user;
@Value("${dify.flowId}")
private String flowId;
/**
* 处理Mq消息
* @param message
@@ -65,7 +78,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
this.save(curMQMessageRecord);
log.info("数据入库成功(MQMessageRecord)");
} catch (Exception ex) {
log.error("数据入库失败(MQMessageRecord)", ex.getMessage());
log.error("数据入库失败(MQMessageRecord)", ex);
return false;
}
if (oldItem.getSinceType() == 2 && oldItem.getSubSinceType() == SubSinceTypeEnum.SinceType51.getCode() ) {
@@ -160,33 +173,109 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
//获取脱敏配置信息
List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(Constant.CHANNEL_DCC);
List<DiffDefeatCorpuItem> curDiffDefeatCorpuItems = new ArrayList<>();
// 根据SourceId查询通话信息
List<String> sourceIds = input.getCallList().stream().map(CallItem::getSourceId).collect(Collectors.toList());
if (!sourceIds.isEmpty()) {
int asrCount = 0;
//todo 查询未完成解析的通话信息
List<String> sourceIds = Optional.ofNullable(input.getCallList()).map(list -> list.stream().map(CallItem::getSourceId).collect(Collectors.toList())).orElse(Collections.emptyList());
if (!CollectionUtils.isEmpty(sourceIds)) {
//查询未完成解析的通话信息
String sourceId = sourceIds.stream().map(code -> "'"+String.valueOf(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);
if (asrCount > 0) {
output.setHandleStatus(500);
output.setResultStr("存在未完成解析的通话信息");
return new DiffDefeatAnalyseOutputResult(output, contentStr);
}
//todo 查询到的通话数据
//查询到的通话数据
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");
for (CallItem item : input.getCallList()) {
Optional<String> result = CollectionUtils.isEmpty(hishistoryList) ? Optional.empty():
hishistoryList.stream()
.filter(entity -> entity.get("source_id").equals(item.getSourceId()))
.findFirst().map(entity -> entity.get("msg_json") == null ? "" : entity.get("msg_json").toString());
DiffDefeatCorpuItem curDiffDefeatCorpuItem = new DiffDefeatCorpuItem();
curDiffDefeatCorpuItem.setSourceId(item.getSourceId());
curDiffDefeatCorpuItem.setCategory("语音");
curDiffDefeatCorpuItem.setCorpuText(result.isPresent() ? result.get() : "");
curDiffDefeatCorpuItem.setHappenTime(item.getAudioTime());
curDiffDefeatCorpuItems.add(curDiffDefeatCorpuItem);
}
}
// 查询企微的数据
if (input.getVdqwUserId() == null || input.getVdqwCustomerId() == null) {
if (StringUtils.isBlank(input.getVdqwUserId())|| StringUtils.isBlank(input.getVdqwCustomerId())) {
log.info("企微用户信息为空,不附加企微数据");
}else {
//todo 查询到的企微数据
}
//todo 组装数据
try {
String userIdSql = "select distinct userid as userid from ods_vdqw.ods_vdqw_workuserinfo_d where middleuserid ='" + input.getVdqwUserId()
+ "' union ALL select externaluserid as userid from ods_vdqw.ods_vdqw_externalcontact_d where unionid ='" + input.getVdqwCustomerId() + "'";
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");
String startTime = sdf.format(DateUtil.offsetDay(input.getDefeatTime(), -3));
String endTime = sdf.format(input.getDefeatTime() == null ? new Date() : input.getDefeatTime());
String user1 = userList.get(0).get("userid").toString();
String user2 = userList.get(1).get("userid").toString();
String sql = "with tab As(" +
"select *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY msgtime DESC) as rn " +
"from ods_vdqw.ods_vdqw_messagearchiving_d " +
"where msgtime >= '" + startTime + "' and msgtime < '" + endTime + "' " +
"and chattype = '0' and msgtype='text')" +
"select id,fromuserid,acceptuserid,msgtime,content,seq,actiontype,msgid,msgtype," +
"(case when tab2.userid is null or tab2.userid='' then 0 else 1 end) as fromuserrole " +
"from tab LEFT JOIN (select distinct userid from ods_vdqw.ods_vdqw_workuserinfo_d) tab2 " +
"on tab.fromuserid = tab2.userid " +
"where rn = 1 and ((" +
"fromuserid ='" + user1 + "' and acceptuserid ='" + user2 + "') or (" +
"fromuserid ='" + user2 + "' and acceptuserid ='" + user1 + "'))" +
"order by seq;";
List<SessionItem> userInfoList = clickhouseJdbcTemplate.queryForList(sql, SessionItem.class);
Date sliceSatrtTime = sdf.parse(startTime);
Date slice1StopTime = DateUtil.offsetDay(sliceSatrtTime,1);
Date slice2StopTime = DateUtil.offsetDay(slice1StopTime,1);
Date slice3StopTime = sdf.parse(endTime);
this.addSliceData(curDiffDefeatCorpuItems, sliceSatrtTime, slice1StopTime, userInfoList);
this.addSliceData(curDiffDefeatCorpuItems, slice1StopTime, slice2StopTime, userInfoList);
this.addSliceData(curDiffDefeatCorpuItems, slice2StopTime, slice3StopTime, userInfoList);
}
} catch (ParseException e) {
e.printStackTrace();
}
}
// 对集合进行排序
curDiffDefeatCorpuItems.sort(Comparator.comparing(DiffDefeatCorpuItem::getHappenTime));
if (!CollectionUtils.isEmpty(curDiffDefeatCorpuItems)){
for (DiffDefeatCorpuItem item : curDiffDefeatCorpuItems) {
if ("企微".equals(item.getCategory())) {
contentStr += "企微记录\n";
contentStr += item.getCorpuText();
} else if ("语音".equals(item.getCategory()) && StringUtils.isNotEmpty(item.getCorpuText())) {
// 反序列化CorpuText为KafkaJson对象
KafkaJson oldItem = JSON.parseObject(item.getCorpuText(), KafkaJson.class);
// 反序列化display为CollectTranscriberJobResponse对象
CollectTranscriberJobResponse resp = JSON.parseObject(oldItem.getDisplay(), CollectTranscriberJobResponse.class);
// 检查状态并处理Segments
if ("FINISHED".equals(resp.getStatus()) && !CollectionUtils.isEmpty(resp.getSegments())) {
String vocStr = "";
for (Segment segment : resp.getSegments()) {
if ("AGENT".equals(segment.getResult().getAnalysisInfo().getRole())) {
vocStr += "客服：" + segment.getResult().getText() + "\n";
} else if ("USER".equals(segment.getResult().getAnalysisInfo().getRole())) {
vocStr += "客户：" + segment.getResult().getText() + "\n";
} else {
vocStr += "未知：" + segment.getResult().getText() + "\n";
}
}
contentStr += vocStr;
}
}
}
}
if (StringUtils.isEmpty(contentStr)) {
output.setHandleStatus(400);
output.setResultStr("语料信息为空");
return new DiffDefeatAnalyseOutputResult(output, contentStr);
}
log.info("开始脱敏 {}", contentStr);
RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
runMaskingRuleInput.setOpinionId(input.getBusinessId());
@@ -199,8 +288,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
record.put("record",summaryText);
DiFyReq diFyReq = new DiFyReq();
diFyReq.setUser("voc");
diFyReq.setFlowId("app-MVGxogM08CDCg7jMo7GNF6eS");
diFyReq.setUser(user);
diFyReq.setFlowId(flowId);
diFyReq.setInputs(record);
JSONObject difyResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
output.setHandleStatus(200);
@@ -349,4 +438,43 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
}
return oldStr;
}
public void addSliceData(List<DiffDefeatCorpuItem> curDiffDefeatCorpuItems, Date sliceStartTime, Date sliceStopTime, List<SessionItem> sessionItems) {
// 使用Java 8的Stream API来过滤sessionItems
List<SessionItem> slices = sessionItems.stream()
.filter(a -> !a.getMsgtime().before(sliceStartTime) && a.getMsgtime().before(sliceStopTime))
.collect(Collectors.toList());
// 检查是否有任何匹配的slices
if (!slices.isEmpty()) {
StringBuilder str = new StringBuilder();
// 遍历sessionItems，构建字符串
for (SessionItem sessionItem : sessionItems) {
if ("text".equals(sessionItem.getMsgtype()) && "send".equals(sessionItem.getActiontype())) {
try {
// 反序列化content为ContentText对象
ContentText contentText = JSON.parseObject(sessionItem.getContent(), ContentText.class);
// 根据fromuserrole添加内容到字符串
if (sessionItem.getFromuserrole() == 1) {
str.append("客服：").append(contentText.getContent()).append("\n");
} else {
str.append("客户：").append(contentText.getContent()).append("\n");
}
} catch (Exception e) {
log.error("反序列化异常: ", e);
}
}
}
// 检查字符串是否不为空
if (str.length() > 0) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
DiffDefeatCorpuItem curDiffDefeatCorpuItem = new DiffDefeatCorpuItem();
curDiffDefeatCorpuItem.setSourceId(slices.get(0).getMsgid());
curDiffDefeatCorpuItem.setCategory("企微");
curDiffDefeatCorpuItem.setHappenTime(sdf.format(slices.get(0).getMsgtime()));
curDiffDefeatCorpuItem.setCorpuText(str.toString());
curDiffDefeatCorpuItems.add(curDiffDefeatCorpuItem);
}
}
}
}