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

企微，dcc结构调整

Browse Source
This commit is contained in:
zren25
2025-05-09 19:20:51 +08:00
parent a587dd679a
commit 5d8828829b
9 changed files with 153 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/NameplateTableKafkaDTO.java Normal file
View File

@@ -0,0 +1,19 @@
package com.volvo.ai.analytic.center.dto.corpus;
import com.volvo.ai.analytic.center.entity.TmNameplateCorpus;
import lombok.Data;
import java.util.List;
@Data
public class NameplateTableKafkaDTO {
private String database;
private String table;
private String type;
private List<TmNameplateCorpus> data;
}

5
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java
View File

@@ -7,7 +7,6 @@ import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
@@ -18,8 +17,6 @@ import java.util.Map;
@Configuration
@RefreshScope
public class KafkaConfig {
@Value("${kafka.dcc.consumer.key-deserializer}")
String keySerializer;
// 第一个Kafka配置
@Bean(name = "dccKafkaTemplate")
public KafkaTemplate<String, String> dccProducerFactory(
@@ -62,6 +59,7 @@ public class KafkaConfig {
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
@@ -79,6 +77,7 @@ public class KafkaConfig {
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));

72
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java Normal file
View File

@@ -0,0 +1,72 @@
package com.volvo.ai.analytic.center.mq;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.dto.corpus.NameplateTableKafkaDTO;
import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* @ClassName NameplateKafkaConsumer
* @Description 铭牌数据处理
* @Author renzhen
* @Date 2025-03-04 10:18
* @Version 1.0
**/
@Slf4j
@Component
@RestController
@RefreshScope
public class NameplateKafkaConsumer {
@Autowired
private TmNameplateCorpusService tmNameplateCorpusService;
private final ObjectMapper objectMapper = new ObjectMapper();
@Value("${rocketmq.producer.corpus.dcctopic}")
private String dccMqTipic;
@Resource
private RocketMQTemplate rocketMqTemplate;
@KafkaListener(topics = "${kafka.analyticCenter.consumer.topic}",
groupId = "${kafka.analyticCenter.consumer.group}" ,
containerFactory = "analyticCenterConsumerFactory",
concurrency = "3")
public void listen(String recordMessages) {
long startTime = System.currentTimeMillis();
log.info("nameplateKafkaConsumer message: {}", recordMessages);
if(StringUtils.isNotEmpty(recordMessages)){
try {
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
tmNameplateCorpus.getData().forEach(nameplate -> {
log.info("tmNameplateCorpus customerFlowId: {}",nameplate.getCustomerFlowId());
tmNameplateCorpusService.processItem(nameplate);
});
} catch (Exception e) {
log.error("tmNameplateCorpus 铭牌解析处理出错: {}", e.getMessage());
}
}
log.info("tmNameplateCorpus Kafka 消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);
}
}

121
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java
View File

@@ -1,121 +0,0 @@
package com.volvo.ai.analytic.center.mq;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.constant.Constant;
import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
import com.volvo.ai.analytic.center.dto.corpus.DisplayDTO;
import com.volvo.ai.analytic.center.entity.TmTelephoneCorpus;
import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @ClassName CorpusProcessKafkaConsumer
* @Description
* @Author renzhen
* @Date 2025-03-04 10:18
* @Version 1.0
**/
@Slf4j
@Component
@RestController
@RefreshScope
public class NameplateKafkaProducer {
@Autowired
private TmTelephoneCorpusService tmTelephoneCorpusService;
private final ObjectMapper objectMapper = new ObjectMapper();
@Value("${rocketmq.producer.corpus.dcctopic}")
private String dccMqTipic;
@Resource
private RocketMQTemplate rocketMqTemplate;
@KafkaListener(topics = "${kafka.analyticCenter.consumer.topic}", groupId = "${kafka.analyticCenter.consumer.group}" ,containerFactory = "analyticCenterConsumerFactory")
public void listen(List<ConsumerRecord<String, Object>> recordMessages) {
long startTime = System.currentTimeMillis();
try {
log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
// 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
log.info("获取的线程数：{}", optimalThreadPoolSize);
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
if(CollectionUtils.isNotEmpty(recordMessages)){
log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size());
for (ConsumerRecord<String, Object> record : recordMessages) {
log.info("CorpusProcessKafkaProducer message: {}",record);
executor.submit(() -> {
try {
String message = (String) record.value();
AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay());
DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
log.info("aicorpusTelephone display getSegments: {}", display.getSegments());
TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus();
BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus);
tmTelephoneCorpus.setCreateBy("kafka");
tmTelephoneCorpus.setCreateTime(LocalDateTime.now());
// 条件：只处理dcc的 10s通话时间以上
log.info("CorpusProcessKafkaProducer getCategoryCode: {}，audioFileId：{}，sourceId：{}", tmTelephoneCorpus.getCategoryCode(), aicorpusTelephone.getAudioFileId(), aicorpusTelephone.getSourceId());
if (Constant.CHANNEL_DCC.equals(tmTelephoneCorpus.getCategoryCode())) {
log.info(" dcc 语料开始处理");
tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus);
long startTimeDify = System.currentTimeMillis();
rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(message).build(),
new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("dcc发送MQ成功 消息体:{}", message);
}
@Override
public void onException(Throwable e) {
log.error("dcc 送MQ异常 消息体:{}, 异常：", message, e);
}
}, 10000);
log.info(" dify处理耗时：{}", System.currentTimeMillis() - startTimeDify);
}
} catch (Exception e) {
log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}", e.getMessage());
}
});
}
executor.shutdown();
}
log.info("Kafka 消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);
// 在这里可以添加对解析后的对象的进一步处理逻辑
} catch (Exception e) {
log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}" , e.getMessage());
}
}
}

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java
View File

@@ -17,6 +17,8 @@ public interface TmNameplateCorpusService extends IService<TmNameplateCorpus> {
void runNameplateCorpusDifyRetry(String paramJson);
void processItem(TmNameplateCorpus item);
void sendNameplateLto(JSONObject execDifyFlow, TmNameplateCorpus tmNameplateCorpus);
ResultMsg<Object> updateNameplate(String message);

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

@@ -52,10 +52,13 @@ public class DiFyServiceImpl implements DiFyService{
public JSONObject executeDifyFlow(DiFyReq diFyReq, String businessType, String businessData, String oldAiAnalysisRequestId) {
String aiAnalysisRequestId = StringUtils.isEmpty(oldAiAnalysisRequestId)? AiAnalysisUtils.getAiAnalysisRequestId(businessType):oldAiAnalysisRequestId;
try {
Map<String, String> inputMap = JSON.parseObject(JSONObject.toJSONString(diFyReq.getInputs()),Map.class);
inputMap.put("aiAnalysisRequestId",aiAnalysisRequestId);
Map<String, Object> map = new HashMap<>();
map.put("inputs",diFyReq.getInputs());
map.put("inputs", inputMap);
map.put("response_mode","blocking");
map.put("user",diFyReq.getUser());
diFyReq.setInputs(inputMap);
// 保存请求日志
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder()

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

@@ -19,7 +19,6 @@ import com.volvo.ai.analytic.center.mapper.TmOdsVdqwMessagearchivingMapper;
import com.volvo.ai.analytic.center.mapper.TtNameplateRecordMapper;
import com.volvo.ai.analytic.center.service.*;
import com.volvo.ai.analytic.center.utils.ConstantStr;
import com.volvo.ai.analytic.center.utils.FlowResultSplitUtil;
import com.volvo.common.core.util.ResultMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
@@ -186,66 +185,59 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
log.info("重跑铭牌语料铭牌数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
}
private void processItem(TmNameplateCorpus item) {
public void processItem(TmNameplateCorpus item) {
Optional.ofNullable(item).filter(tmNameplateCorpus -> tmNameplateCorpus.getCustomerFlowId()!=null && tmNameplateCorpus.getNameplateContent()!=null).orElseThrow(()->new RuntimeException("铭牌语料为空"));
try {
Optional.ofNullable(item).filter(tmNameplateCorpus -> tmNameplateCorpus.getCustomerFlowId()!=null && tmNameplateCorpus.getNameplateContent()!=null).orElseThrow(()->new RuntimeException("铭牌语料为空"));
String customerFlowId = item.getCustomerFlowId();
String nameplateContent = item.getNameplateContent();
log.info("铭牌数据处理：customerFlowId：{}", customerFlowId);
String customerFlowId = item.getCustomerFlowId();
String nameplateContent = item.getNameplateContent();
log.info("铭牌数据处理：customerFlowId：{}", customerFlowId);
DiFyReq diFyImageReq = new DiFyReq();
diFyImageReq.setUser(ConstantStr.corpus_user);
diFyImageReq.setFlowId(nameplateAppKey);
List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode());
RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
runMaskingRuleInput.setDataMaskingRules(maskingRuleItems);
DiFyReq diFyImageReq = new DiFyReq();
diFyImageReq.setUser(ConstantStr.corpus_user);
diFyImageReq.setFlowId(nameplateAppKey);
List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode());
RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput();
runMaskingRuleInput.setDataMaskingRules(maskingRuleItems);
runMaskingRuleInput.setOldStr(nameplateContent);
String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
Map<String, Object> inputMap = new HashMap<>();
String carModel = tmTelephoneCorpusService.getCarModelList();
inputMap.put("dialogue", corpusChat);
inputMap.put("model", carModel);
inputMap.put("customerFlowId", item.getCustomerFlowId());
diFyImageReq.setInputs(inputMap);
CorpusReportDTO corpusReportDTO = new CorpusReportDTO();
corpusReportDTO.setCustomerFlowId(customerFlowId);
corpusReportDTO.setAnalysisScene(3l);
corpusReportDTO.setCarModel(carModel);
// 获取配置
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode(), JSONObject.toJSONString(corpusReportDTO),null);
log.info("runDify execDifyFlow {}", execDifyFlow);
runMaskingRuleInput.setOldStr(nameplateContent);
String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
Map<String, Object> inputMap = new HashMap<>();
String carModel = tmTelephoneCorpusService.getCarModelList();
inputMap.put("dialogue", corpusChat);
inputMap.put("model", carModel);
inputMap.put("customerFlowId", item.getCustomerFlowId());
inputMap.put("analysisScene", "3");
diFyImageReq.setInputs(inputMap);
CorpusReportDTO corpusReportDTO = new CorpusReportDTO();
corpusReportDTO.setCustomerFlowId(customerFlowId);
corpusReportDTO.setAnalysisScene(3l);
corpusReportDTO.setCarModel(carModel);
// 获取配置
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode(), JSONObject.toJSONString(corpusReportDTO),null);
log.info("runDify execDifyFlow {}", execDifyFlow);
} catch (Exception e) {
log.error("nameplate processItem error {}", e.getMessage());
}
}
@Override
public void sendNameplateLto(JSONObject execDifyFlow, TmNameplateCorpus tmNameplateCorpus) {
if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) {
String text = execDifyFlow.getJSONObject("outputs").getString("text");
String resultStrOne = FlowResultSplitUtil.flowOutputTextSplit(text, "任务1", "任务2");
String resultStrTwo = FlowResultSplitUtil.flowOutputTextSplit(text, "任务2", null);
if (StringUtils.isBlank(resultStrOne) || StringUtils.isBlank(resultStrTwo)){
log.info("铭牌语料解析为空，text:{}", text);
return;
}
Map<String, String> ltoMap = new HashMap<>();
ltoMap.put("analysisRecordId", execDifyFlow.getString(ConstantStr.aiAnalysisRequestId));
ltoMap.put("analysisScene", "3");
ltoMap.put("customerFlowId", tmNameplateCorpus.getCustomerFlowId());
ltoMap.put("analysisResult", resultStrOne);
ltoMap.put("analysisDetail", resultStrTwo);
String text = execDifyFlow.getString("outputs");
// 发送MQ
log.info("send mq {}", ltoMap);
tmTelephoneCorpusService.sendMq( CategoryEnum.NAMEPLATE_VOICE.getCode(), JSONObject.toJSONString(ltoMap));
log.info("send mq {}", text);
tmTelephoneCorpusService.sendMq( CategoryEnum.NAMEPLATE_VOICE.getCode(), text);
try {
ttNameplateRecordMapper.insert(TtNameplateRecord.builder().nameplateCorpusId(tmNameplateCorpus.getId()).customerFlowId(tmNameplateCorpus.getCustomerFlowId()).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(JSONObject.toJSONString(ltoMap)).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text).build());
} catch (Exception e) {
log.info(" 铭牌语料处理保存报告异常processItem：{} ", e);
}

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

@@ -20,7 +20,6 @@ import com.volvo.ai.analytic.center.mapper.TmOdsVdqwWorkuserinfoMapper;
import com.volvo.ai.analytic.center.mapper.TtVdqwRecordMapper;
import com.volvo.ai.analytic.center.service.*;
import com.volvo.ai.analytic.center.utils.ConstantStr;
import com.volvo.ai.analytic.center.utils.FlowResultSplitUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
@@ -197,6 +196,10 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
}
inputMap.put("chat", chatList.toString());
inputMap.put("analysisScene", "1");
inputMap.put("unionId", unionId);
inputMap.put("consultantId", userId);
inputMap.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN));
diFyImageReq.setInputs(inputMap);
CorpusReportDTO corpusReportDTO = new CorpusReportDTO();
@@ -208,28 +211,14 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_QIWEI.getCode(), JSONObject.toJSONString(corpusReportDTO),null);
log.info("runDify execDifyFlow {}", execDifyFlow);
if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) {
String text = execDifyFlow.getJSONObject("outputs").getString("text");
String resultStrOne = FlowResultSplitUtil.flowOutputTextSplit(text, "任务1", "任务2");
String resultStrTwo = FlowResultSplitUtil.flowOutputTextSplit(text, "任务2", null);
if (StringUtils.isBlank(resultStrOne) || StringUtils.isBlank(resultStrTwo)){
log.info("企微语料解析为空，text:{}", text);
return;
}
Map<String, String> ltoMap = new HashMap<>();
ltoMap.put("analysisRecordId", execDifyFlow.getString("aiAnalysisRequestId"));
ltoMap.put("analysisScene", "1");
ltoMap.put("unionId", unionId);
ltoMap.put("consultantId", userId);
ltoMap.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN));
ltoMap.put("analysisResult", resultStrOne);
ltoMap.put("analysisDetail", resultStrTwo);
JSONObject text = execDifyFlow.getJSONObject("outputs");
// 发送MQ
log.info("send mq {}", ltoMap);
tmTelephoneCorpusService.sendMq( CategoryEnum.ENTERPRISE_WECHAT.getCode(), JSONObject.toJSONString(ltoMap));
log.info("send mq {}", text);
tmTelephoneCorpusService.sendMq( CategoryEnum.ENTERPRISE_WECHAT.getCode(), text.toJSONString());
try {
ttVdqwRecordMapper.insert(TtVdqwRecord.builder().acceptUserId(item.getFromUserId()).fromUserId(item.getAcceptUserId()).msgTime(maxMsgTimeItem.getMsgTime()).from_accept_user_id(item.getFromAcceptUserId()).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(JSONObject.toJSONString(ltoMap)).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build());
} catch (Exception e) {
log.info(" 企业语料处理保存报告异常processItem：{} ", e);
}

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

@@ -22,7 +22,6 @@ import com.volvo.ai.analytic.center.mapper.AiAnalysisErrorsMapper;
import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
import com.volvo.ai.analytic.center.service.*;
import com.volvo.ai.analytic.center.utils.ConstantStr;
import com.volvo.ai.analytic.center.utils.FlowResultSplitUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
@@ -140,38 +139,28 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
ZonedDateTime zonedDateTime = ZonedDateTime.parse(jsonObject.getString("start_time"));
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String formattedDateStartTime = zonedDateTime.format(formatter);
String carModel = getCarModelList();
inputMap.put("chat",chatList.toString());
inputMap.put("model",getCarModelList());
inputMap.put("model",carModel);
inputMap.put("analysisScene", "2");
inputMap.put("recordId", aicorpusTelephone.getSourceId());
inputMap.put("communicateDate", formattedDateStartTime);
diFyImageReq.setInputs(inputMap);
CorpusReportDTO corpusReportDTO = new CorpusReportDTO();
corpusReportDTO.setCorpusTime(formattedDateStartTime);
corpusReportDTO.setRecordId(aicorpusTelephone.getSourceId());
corpusReportDTO.setAnalysisScene(2l);
corpusReportDTO.setCarModel(carModel);
// 获取配置
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode(), JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
log.info("runDify execDifyFlow {}",execDifyFlow);
if(null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")){
String text = execDifyFlow.getJSONObject("outputs").getString("text");
String resultStrOne = FlowResultSplitUtil.flowOutputTextSplit(text, "任务1", "任务2");
String resultStrTwo =FlowResultSplitUtil.flowOutputTextSplit(text, "任务2", null);
if (StringUtils.isBlank(resultStrOne) || StringUtils.isBlank(resultStrTwo)){
log.info("电话语料解析为空，text:{}", text);
return;
}
JSONObject text = execDifyFlow.getJSONObject("outputs");
String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId");
Map<String, String> ltoMap = new HashMap();
ltoMap.put("analysisRecordId", aiAnalysisRequestId);
ltoMap.put("analysisScene", "2");
ltoMap.put("recordId", aicorpusTelephone.getSourceId());
ltoMap.put("communicateDate", formattedDateStartTime);
ltoMap.put("analysisResult", resultStrOne);
ltoMap.put("analysisDetail", resultStrTwo);
// 发送MQ
log.info("send mq {}",ltoMap);
sendMq( CategoryEnum.PHONE_VOICE.getCode(), JSONObject.toJSONString(ltoMap));
log.info("send mq {}",text);
sendMq( CategoryEnum.PHONE_VOICE.getCode(), text.toJSONString());
try {
if(StringUtils.isNotEmpty(aicorpusTelephone.getAiAnalysisRequestId())){
AiAnalysisErrors aiAnalysisErrors = new AiAnalysisErrors();
@@ -179,7 +168,7 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
aiAnalysisErrors.setAiAnalysisErrorHandlingStatus("1");
aiAnalysisErrorsService.updateAiAnalysisErrors(aiAnalysisErrors);
}
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(JSONObject.toJSONString(ltoMap)).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build());
} catch (Exception e) {
log.info(" 电话语料处理保存报告异常processItem：{} ", e);