diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/NameplateTableKafkaDTO.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/NameplateTableKafkaDTO.java new file mode 100644 index 0000000..b5ef245 --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/NameplateTableKafkaDTO.java @@ -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 data; + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java index da9a2be..3938ec7 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java @@ -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 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 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 factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java new file mode 100644 index 0000000..f27ac4d --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java @@ -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); + } + + + +} + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java deleted file mode 100644 index 0405aa3..0000000 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java +++ /dev/null @@ -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> 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 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()); - } - } - - - -} - diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java index 513446a..3c4cd2c 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java @@ -17,6 +17,8 @@ public interface TmNameplateCorpusService extends IService { void runNameplateCorpusDifyRetry(String paramJson); + void processItem(TmNameplateCorpus item); + void sendNameplateLto(JSONObject execDifyFlow, TmNameplateCorpus tmNameplateCorpus); ResultMsg updateNameplate(String message); 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 48356fd..8c012d1 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 @@ -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 inputMap = JSON.parseObject(JSONObject.toJSONString(diFyReq.getInputs()),Map.class); + inputMap.put("aiAnalysisRequestId",aiAnalysisRequestId); Map 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() diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java index 31a8a1b..8b11c3f 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java @@ -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 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 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 maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode()); + RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput(); + runMaskingRuleInput.setDataMaskingRules(maskingRuleItems); - runMaskingRuleInput.setOldStr(nameplateContent); - String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput); - Map 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 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 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); } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java index 16f85da..71f0cc4 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java @@ -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 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); } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmTelephoneCorpusServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmTelephoneCorpusServiceImpl.java index 117a212..199016c 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmTelephoneCorpusServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmTelephoneCorpusServiceImpl.java @@ -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 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