画像开发

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<MessageExt>  {
 	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<MessageExt>  {
 			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) {

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) {

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<List<CompetingBrrandsDTO>> result =  competingBrandsClient.getBrand();
+		if (BizEnum.SUCCESS.getCode().toString().equals(result.getReturnCode()) && CollectionUtils.isNotEmpty(result.getData())) {
+			List<CompetingBrrandsDTO> brandDtoList = result.getData();
+			// 拼接 modelName
+			String brandName = brandDtoList.stream()
+					.map(CompetingBrrandsDTO::getBrandName) // 提取 modelName
+					.collect(Collectors.joining(", ")); // 用逗号和空格拼接
 
+			log.info("拼接后的车型名称: {}", brandName);
+
+			return brandName;
+		}
+		return null;
 	}
 }

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<TmOdsVdqwM
 	@Resource(name = "threadPoolTaskExecutor")
 	private ThreadPoolTaskExecutor executor;
 
+	@Autowired
+	private CorpusPortraitService corpusPortraitService;
+
 
 	@Override
 	public void runQiWeiCorpusDify(String paramJson) {
@@ -133,10 +136,16 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
                                 log.error("处理企微语料失败: FromUserId={}, AcceptUserId={}, 异常: {}",
                                         item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e);
                             }
+							try {
+								corpusPortraitService.portraitQiWei(item, finalStatTime, finalEndTime);
+							} catch (Exception e) {
+								log.error("corpusPortrait企微画像处理异步任务执行失败", e);
+							}
                         }, executor))
                         .toArray(CompletableFuture[]::new);
                 // 等待所有任务完成
                 CompletableFuture.allOf(futures).join();
+
             }
         } catch (Exception e) {
 			log.error("企微数据跑批异常",e);