diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqConsumer.java index 4c0d351..fa6cddf 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqConsumer.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqConsumer.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO; import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService; +import com.volvo.ai.analytic.center.service.CorpusPortraitService; import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.common.message.MessageExt; @@ -12,9 +13,13 @@ import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.web.bind.annotation.RestController; +import javax.annotation.Resource; +import java.util.concurrent.CompletableFuture; + /** * @ClassName CorpusProcessKafkaConsumer * @Description @@ -41,6 +46,12 @@ public class CorpusDccMqConsumer implements RocketMQListener { private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; private final ObjectMapper objectMapper = new ObjectMapper(); + @Autowired + @Resource(name = "threadPoolTaskExecutor") + private ThreadPoolTaskExecutor executor; + + @Autowired + private CorpusPortraitService corpusPortraitService; @Override public void onMessage(MessageExt messageExt) { @@ -51,6 +62,14 @@ public class CorpusDccMqConsumer implements RocketMQListener { log.info("dcc_mq message: " + message); AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class); tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone); + // 提交异步任务 + CompletableFuture.runAsync(() -> { + try { + corpusPortraitService.portraitDcc(aicorpusTelephone); + } catch (Exception e) { + log.error("ccorpusPortrait画像Dcc异步任务执行失败", e); + } + }, executor); log.info("corpusDccMqProducer mq 处理完成: {}", aicorpusTelephone.getSourceId()); log.info("dcc_mq 处理完成，耗时：{}", System.currentTimeMillis() - startTime); } catch (JsonProcessingException e) { 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 index 4276007..06ec577 100644 --- 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 @@ -3,6 +3,7 @@ package com.volvo.ai.analytic.center.mq; import com.alibaba.fastjson.JSON; import com.volvo.ai.analytic.center.dto.corpus.NameplateTableKafkaDTO; +import com.volvo.ai.analytic.center.service.CorpusPortraitService; import com.volvo.ai.analytic.center.service.TmNameplateCorpusService; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; @@ -14,10 +15,12 @@ import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.handler.annotation.Header; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.web.bind.annotation.RestController; import javax.annotation.Resource; +import java.util.concurrent.CompletableFuture; /** * @ClassName NameplateKafkaConsumer @@ -40,6 +43,13 @@ public class NameplateKafkaConsumer { @Resource private RocketMQTemplate rocketMqTemplate; + @Autowired + @Resource(name = "threadPoolTaskExecutor") + private ThreadPoolTaskExecutor executor; + + @Autowired + private CorpusPortraitService corpusPortraitService; + @KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", groupId = "${analyticCenterKafka.consumer.group}" , @@ -63,7 +73,19 @@ public class NameplateKafkaConsumer { return; } - tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpusService.processItem(nameplate)); + tmNameplateCorpus.getData().forEach(nameplate -> { + tmNameplateCorpusService.processItem(nameplate); + + CompletableFuture.runAsync(() -> { + try { + corpusPortraitService.portraitNameplate(nameplate); + } catch (Exception e) { + log.error("corpusPortrait画像铭牌异步任务执行失败", e); + } + }, executor); + } + + ); log.info("nameplateKafkaConsumeracknowledge：{}",partitionId, offset); // 手动提交 offset } catch (Exception e) { diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java index c557345..e9de950 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java @@ -3,27 +3,35 @@ package com.volvo.ai.analytic.center.service.impl; import cn.hutool.core.date.DatePattern; import cn.hutool.core.date.DateUtil; import com.alibaba.fastjson.JSONObject; +import com.huaweicloud.sdk.eg.v1.model.CloudEvents; import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO; import com.volvo.ai.analytic.center.dto.corpus.CorpusReportDTO; import com.volvo.ai.analytic.center.dto.corpus.OdsVdqwMessageOTD; import com.volvo.ai.analytic.center.dto.req.DiFyReq; import com.volvo.ai.analytic.center.dto.req.RunMaskingRuleInput; +import com.volvo.ai.analytic.center.dto.resp.CarModelRespDTO; +import com.volvo.ai.analytic.center.dto.resp.CompetingBrrandsDTO; +import com.volvo.ai.analytic.center.dto.resp.ResultDTO; import com.volvo.ai.analytic.center.entity.*; +import com.volvo.ai.analytic.center.enums.BizEnum; import com.volvo.ai.analytic.center.enums.BusinessTypeEnum; +import com.volvo.ai.analytic.center.feign.CompetingBrandsClient; import com.volvo.ai.analytic.center.mapper.TmOdsVdqwMessagearchivingMapper; import com.volvo.ai.analytic.center.service.*; import com.volvo.ai.analytic.center.utils.ConstantStr; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.StringUtils; 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.stereotype.Service; +import javax.annotation.Resource; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; /** @@ -47,6 +55,9 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { // @Value("${dify.corpus.portrait.nameplateToken}") private String nameplateAppKey; + + // @Value("${huawei.cloud.EG.channel.ltoSourceId}") + private String sourceId; @Autowired private DataMaskingRuleService dataMaskingRuleService; @Autowired @@ -63,6 +74,13 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { @Autowired private TmOdsVdqwMessagearchivingMapper tmOdsVdqwMessagearchivingMapper; + @Resource + HuaWeiEGService huaWeiService; + + @Autowired + CompetingBrandsClient competingBrandsClient; + + @Override public void portraitDcc(AicorpusTelephoneDTO aicorpusTelephone) { @@ -101,8 +119,9 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { log.info("runDify execDifyFlow {}", execDifyFlow); String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId"); if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) { + JSONObject text = execDifyFlow.getJSONObject("outputs"); - // 发送消息 + huaWeiService.sendEvent(setCloudEvents(aiAnalysisRequestId, text.toJSONString())); } try { @@ -121,6 +140,18 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { } } + + public CloudEvents setCloudEvents(String aiAnalysisRequestId, String text){ + // 发送消息 + CloudEvents cloudEvents = new CloudEvents(); + cloudEvents.setId(aiAnalysisRequestId); + cloudEvents.setType("APPLICATION"); + cloudEvents.setSource(sourceId); + cloudEvents.setSpecversion("1.0"); + cloudEvents.setDatacontenttype("application/json"); + cloudEvents.setData(text); + return cloudEvents; + } @Override public void portraitQiWei(OdsVdqwMessageOTD item, String statTime, String endTime) { log.info("画像-企微语料内容：FromUserId：{}, AcceptUserId：{}", item.getFromUserId(), item.getAcceptUserId()); @@ -186,7 +217,11 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { // 获取配置 JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_QIWEI.getCode(), JSONObject.toJSONString(corpusReportDTO), null); log.info("runDify execDifyFlow {}", execDifyFlow); + String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId"); if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) { + JSONObject text = execDifyFlow.getJSONObject("outputs"); + + huaWeiService.sendEvent(setCloudEvents(aiAnalysisRequestId, text.toJSONString())); } try { @@ -232,11 +267,31 @@ public class CorpusPortraitServiceImpl implements CorpusPortraitService { // 获取配置 JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode(), JSONObject.toJSONString(corpusReportDTO),null); log.info("runDify execDifyFlow {}", execDifyFlow); + String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId"); + if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) { + JSONObject text = execDifyFlow.getJSONObject("outputs"); + + huaWeiService.sendEvent(setCloudEvents(aiAnalysisRequestId, text.toJSONString())); + } } catch (Exception e) { log.error("nameplate processItem error {}", e.getMessage()); } + } + public String getBrand(){ + ResultDTO> result = competingBrandsClient.getBrand(); + if (BizEnum.SUCCESS.getCode().toString().equals(result.getReturnCode()) && CollectionUtils.isNotEmpty(result.getData())) { + List brandDtoList = result.getData(); + // 拼接 modelName + String brandName = brandDtoList.stream() + .map(CompetingBrrandsDTO::getBrandName) // 提取 modelName + .collect(Collectors.joining(", ")); // 用逗号和空格拼接 + log.info("拼接后的车型名称: {}", brandName); + + return brandName; + } + return null; } } \ No newline at end of file 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 a899bcb..6f012e3 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 @@ -91,6 +91,9 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl