开发画像企微场景，代码优化，多线程处理

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqConsumer.java

@@ -2,6 +2,8 @@
 package com.volvo.ai.analytic.center.mq;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.volvo.ai.analytic.center.constant.Constant;
@@ -52,6 +54,9 @@ public class CorpusDccMqConsumer implements RocketMQListener<MessageExt>  {
 			AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
 			//DCC场景
 			if(Objects.equals(aicorpusTelephone.getCategoryCode(), Constant.CHANNEL_DCC)){
+				//TODO  这里需要确认客户画像下这个字段值是什么？
+				JSONObject jsonObject = JSONObject.parseObject(	aicorpusTelephone.getDisplay());
+				log.info(" message: " + JSON.toJSONString(aicorpusTelephone));
 				tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone);
 			}
 			//channel_4in1场景

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java

@@ -1,6 +1,8 @@
 
 package com.volvo.ai.analytic.center.mq;
 
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.volvo.ai.analytic.center.constant.Constant;
 import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
@@ -92,6 +94,15 @@ public class CorpusProcessKafkaProducer {
 						try {
 							String message = (String) record.value();
 							AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
+							JSONObject jsonObject = JSONObject.parseObject(	aicorpusTelephone.getDisplay());
+
+							Long audioDuration = jsonObject.getLong("audio_duration"); // 毫秒
+							if(audioDuration/1000 <=10){
+								log.info("电话语料时长小于10秒，不进行dify处理");
+								return;
+							}
+
+
 							log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay());
 							DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
 							log.info("aicorpusTelephone display getSegments: {}", display.getSegments());

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java

@@ -259,11 +259,11 @@ public class CorpusPortraitServiceImpl  implements CorpusPortraitService {
 				corpusReportDTO.setUserId(userId);
 				// 获取配置
 				JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.CORPUS_PORTRAIT_QIWEI.getCode(), JSONObject.toJSONString(corpusReportDTO), null);
-				log.info("runDify execDifyFlow {}", execDifyFlow);
+				log.info("runDify execDifyFlow 企微语料，画像场景 Dify的返回 : {}", execDifyFlow);
 				String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId");
 				if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) {
 					JSONObject text = execDifyFlow.getJSONObject("outputs");
-
+					// TODO  检查是否发了 unionid ，  画像场景  "businessId": "业务ID",  // 根据分析不同传不同的值 1、unionId 2、通话ID 3、客流ID 4、试驾ID
 //					huaWeiService.sendEvent(setCloudEvents(aiAnalysisRequestId,  text.toJSONString(),"QIWEI"),ltoChannelId);
 					log.info("send mq ，企微语料，画像场景 {}", text);
 					tmTelephoneCorpusService.sendMq( BusinessTypeEnum.CORPUS_PORTRAIT_QIWEI.getCode(), text.toJSONString());

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

@@ -155,6 +155,8 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
 			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());

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

@@ -130,6 +130,9 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
                 // 使用 CompletableFuture 并行处理
                 CompletableFuture<?>[] futures = messageList.stream()
                         .map(item -> CompletableFuture.runAsync(() -> {
+							long startTime2 = System.currentTimeMillis();
+							log.info("处理企微语料开始 FromUserId={}, AcceptUserId={}",
+									item.getFromUserId(), item.getAcceptUserId());
                             try {
                                 processItem(item, finalStatTime, finalEndTime);
                             } catch (Exception e) {
@@ -141,8 +144,10 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 							} catch (Exception e) {
 								log.error("corpusPortrait企微画像处理异步任务执行失败", e);
 							}
+							log.info("处理企微语料 调用完成：  结束时间：{}, 总用时： {}", System.currentTimeMillis(),System.currentTimeMillis()-startTime2);
                         }, executor))
                         .toArray(CompletableFuture[]::new);
+
                 // 等待所有任务完成
                 CompletableFuture.allOf(futures).join();
 
@@ -218,7 +223,7 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 			corpusReportDTO.setAnalysisScene(1l);
 			// 获取配置
 			JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_QIWEI.getCode(), JSONObject.toJSONString(corpusReportDTO),null);
-			log.info("runDify execDifyFlow {}", execDifyFlow);
+			log.info("runDify execDifyFlow , 企微场景，总结需求的dify返回 ：{}", execDifyFlow);
 			if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) {
 				JSONObject text = execDifyFlow.getJSONObject("outputs");
 				// 发送MQ

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

@@ -43,6 +43,9 @@ import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 
@@ -65,11 +68,12 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
 
 	@Value("${rocketmq.producer.corpus.topic}") //实际字符； ai_corpus_report_topic
 	private String topic;
+
 	@Value("${dify.corpus.telephoneToken}")
 	private String telephoneToken;
 
 	@Value("${dify.corpus.portrait.oneToken}")
-	private String telephoneTokenPortrait;
+	private String oneTokenPortrait;
 
 	@Value("${dify.intelligentCustomer.channel4In1Token}")
 	private String channel4In1Token;
@@ -115,15 +119,7 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
 
 			JSONObject jsonObject = JSONObject.parseObject(	aicorpusTelephone.getDisplay());
 			JSONArray segments = jsonObject.getJSONArray("segments");
-			Long audioDuration = jsonObject.getLong("audio_duration"); // 毫秒
+
-			if(audioDuration/1000 <=10){
-				log.info("电话语料时长小于10秒，不进行dify处理");
-				return;
-			}
-			if (segments == null || segments.isEmpty()) {
-				log.info("电话语料内容segments为空，不进行dify处理");
-				return;
-			}
 			// 遍历 segments
 			StringBuffer chatList = new StringBuffer();
 			segments.stream()
@@ -158,18 +154,54 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
 			corpusReportDTO.setCorpusTime(formattedDateStartTime);
 			corpusReportDTO.setRecordId(aicorpusTelephone.getSourceId());
 			corpusReportDTO.setAnalysisScene(2l);
-			// 获取配置
+
+			// 使用多线程并行执行两个业务场景
+			ExecutorService executorService = Executors.newFixedThreadPool(2);
+			//第一个业务场景  开始：    总结和分类业务场景
+//			JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode(),
+//					JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
+//			log.info("runDify execDifyFlow 返回 ,总结场景 ： {}",execDifyFlow);
+//			parseDfiyResult(execDifyFlow, aicorpusTelephone.getSourceId(), formattedDateStartTime, aicorpusTelephone.getAiAnalysisRequestId(),
+//					BusinessTypeEnum.SMART_ASSISTANT.getCode());
+			long startTime = System.currentTimeMillis();
+			CompletableFuture<Void> summaryFuture = CompletableFuture.runAsync(() -> {
 				JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode(),
 						JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
-			log.info("runDify execDifyFlow 返回 ,总结 {}",execDifyFlow);
+				log.info("runDify execDifyFlow 返回 ,总结场景 ： {}", execDifyFlow);
-			parseDfiyResult(execDifyFlow, aicorpusTelephone.getSourceId(), formattedDateStartTime, aicorpusTelephone.getAiAnalysisRequestId(),BusinessTypeEnum.SMART_ASSISTANT.getCode());
+				parseDfiyResult(execDifyFlow, aicorpusTelephone.getSourceId(), formattedDateStartTime, aicorpusTelephone.getAiAnalysisRequestId(),
+						BusinessTypeEnum.SMART_ASSISTANT.getCode());
+			}, executorService);
+			long endTime = System.currentTimeMillis();
+			log.info("第一个业务场景（总结和分类）执行时间: {} ms", (endTime - startTime));
+             //第一个业务场景，    结束
 
-			// 获取配置
+			//TODO 是否传给LOT  穿行改成并行
-			diFyImageReq.setFlowId(telephoneTokenPortrait);
+			//第二个业务场景：  用户画像业务场景。  开始
+//			diFyImageReq.setFlowId(telephoneTokenPortrait);
+//			JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode(),
+//					JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());  //调用dify平台。
+//			log.info("runDify execDifyFlow 返回 ,客户画像场景： {}",execDifyFlowForPortrait);
+//			parseDfiyResult(execDifyFlowForPortrait, aicorpusTelephone.getSourceId(), formattedDateStartTime,
+//					aicorpusTelephone.getAiAnalysisRequestId(), BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode()); // 发送mq消息。
+
+
+			long startTime2 = System.currentTimeMillis();
+			CompletableFuture<Void> portraitFuture = CompletableFuture.runAsync(() -> {
+
+				// 创建新的DiFyReq对象以避免线程安全问题
+				diFyImageReq.setFlowId(oneTokenPortrait);
 				JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode(),
 						JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
-			log.info("runDify execDifyFlow 返回 ,客户画像 {}",execDifyFlow);
+
-			parseDfiyResult(execDifyFlowForPortrait, aicorpusTelephone.getSourceId(), formattedDateStartTime, aicorpusTelephone.getAiAnalysisRequestId(), BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode());
+				log.info("runDify execDifyFlow 返回 ,客户画像场景： {}", execDifyFlowForPortrait);
+
+				parseDfiyResult(execDifyFlowForPortrait, aicorpusTelephone.getSourceId(), formattedDateStartTime,
+						aicorpusTelephone.getAiAnalysisRequestId(), BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode());
+			}, executorService);
+			long endTime2 = System.currentTimeMillis();
+			log.info("第二个业务场景（用户画像）执行时间: {} ms", (endTime2 - startTime2));
+			//第二个业务场景，    结束
+
 		}
 	}