企微语料ai解析开发

ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/PageDto.java

@@ -0,0 +1,22 @@
+package com.volvo.ai.analytic.center.dto;
+/**
+ * @ClassName PageDto
+ * @Description
+ * @Author renzhen
+ * @Date 2025-03-11 10:37
+ * @Version 1.0
+ **/
+
+import lombok.Data;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
+
+@Data
+public class PageDto {
+
+	// 每页大小
+	private static int currentPage = 1; // 当前页码
+	public static int getTotalPages(int totalCount,int pageSize) {
+		return (int) Math.ceil((double) totalCount / pageSize);
+	}
+
+}

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/QiWeiCorpusJob.java

@@ -0,0 +1,37 @@
+package com.volvo.ai.analytic.center.job;
+
+import com.volvo.ai.analytic.center.service.TmOdsVdqwMessagearchivingService;
+import com.volvo.common.core.util.ResultMsg;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+
+@Slf4j
+@Component
+@RestController
+public class QiWeiCorpusJob {
+
+    @Autowired
+    private TmOdsVdqwMessagearchivingService tmOdsVdqwMessagearchivingService;
+
+    /**
+     * 企微语料处理
+     */
+    @XxlJob("QiWeiCorpusTask")
+    @PostMapping("qiWeiCorpusTask")
+    public ResultMsg qiWeiCorpusTask(@RequestBody String paramJson) {
+        try {
+            log.info("qiWeiCorpusTask 企微语料查询处理:{}",paramJson);
+            tmOdsVdqwMessagearchivingService.runQiWeiCorpusDify(paramJson);
+        } catch (Exception e) {
+            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
+            throw new RuntimeException(e);
+        }
+        return ResultMsg.ok();
+    }
+}

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TmOdsVdqwMessagearchivingMapper.java

@@ -5,6 +5,7 @@ import com.volvo.ai.analytic.center.dto.corpus.OdsVdqwMessageOTD;
 import com.volvo.ai.analytic.center.entity.TmOdsVdqwMessagearchiving;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
 
 import java.util.List;
 
@@ -16,7 +17,8 @@ import java.util.List;
 @Mapper
 public interface TmOdsVdqwMessagearchivingMapper extends BaseMapper<TmOdsVdqwMessagearchiving> {
 
-	List<OdsVdqwMessageOTD>  queryOdsVdqwMessageByData(@Param("statTime") String statTime, @Param("endTime") String endTime);
+	int countOdsVdqwMessageByData(@Param("statTime") String statTime, @Param("endTime") String endTime);
+	List<OdsVdqwMessageOTD>  queryOdsVdqwMessageByData(@Param("statTime") String statTime, @Param("endTime") String endTime, @Param("offset") int offset, @Param("pageSize") int pageSize);
 
 	List<OdsVdqwMessageOTD> queryOdsVdqwMessageByFromUserIdAndAcceptUserId(@Param("statTime") String statTime, @Param("endTime") String endTime, @Param("userIdList") List userIdList);
 

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

@@ -11,6 +11,6 @@ import java.util.*;
  */
 public interface TmOdsVdqwMessagearchivingService extends IService<TmOdsVdqwMessagearchiving> {
 
-	void runQiWeiCorpusDify();
+	void runQiWeiCorpusDify(String paramJson);
 
 }

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

@@ -7,6 +7,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.volvo.ai.analytic.center.constant.Constant;
+import com.volvo.ai.analytic.center.dto.PageDto;
 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;
@@ -20,6 +21,7 @@ import com.volvo.ai.analytic.center.service.DataMaskingRuleService;
 import com.volvo.ai.analytic.center.service.DiFyService;
 import com.volvo.ai.analytic.center.service.TmOdsVdqwMessagearchivingService;
 import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
+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;
@@ -29,14 +31,19 @@ import org.apache.rocketmq.client.producer.SendResult;
 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.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.time.LocalDate;
 import java.util.Arrays;
 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;
 
 
 /**
@@ -44,6 +51,7 @@ import java.util.Map;
  * @author rz
  * @date 2025-03-04
  */
+@RefreshScope
 @Slf4j
 @Service
 public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwMessagearchivingMapper, TmOdsVdqwMessagearchiving> implements TmOdsVdqwMessagearchivingService {
@@ -62,9 +70,13 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 	@Resource
 	private RocketMQTemplate rocketMqTemplate;
 
-	@Value("${rocketmq.corpusTelephone.topic}")
+	@Value("${rocketmq.producer.corpus.topic}")
 	private String topic;
+	@Value("${dify.corpus.qiweiToken}")
+	private String qiweiToken;
 
+	@Value("${batch.size}")
+	public  int pageSize = 100;
 	@Autowired
 	private RemoteCarModelClient remoteCarModelClient;
 
@@ -74,37 +86,83 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 	@Autowired
 	private TmTelephoneCorpusService tmTelephoneCorpusService;
 
-	@Override
-	public void runQiWeiCorpusDify() {
-		String statTime = "2025-03-01";
-		String endTime = "2025-03-02";
 
-		List<OdsVdqwMessageOTD>  messageList =  tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime,endTime);
-		messageList.stream().forEach(item->{
+	@Override
+	public void runQiWeiCorpusDify(String paramJson) {
+		log.info("runQiWeiCorpusDify paramJson {}", paramJson);
+		// 获取当前日期
+		LocalDate today = LocalDate.now();
+		// 获取前一天日期
+		LocalDate yesterday = today.minusDays(1);
+		// 格式化输出
+		String formattedDate = yesterday.toString(); // 默认格式为 yyyy-MM-dd
+		String statTime = formattedDate.concat(" 00:00:00");
+		String endTime = formattedDate.concat(" 23:59:59");
+		if (StringUtils.isNotBlank(paramJson)) {
+			JSONObject paramJsonObj = JSONObject.parseObject(paramJson);
+			if (null != paramJsonObj && paramJsonObj.containsKey("statTime") && paramJsonObj.containsKey("endTime")) {
+				statTime = paramJsonObj.getString("statTime");
+				endTime = paramJsonObj.getString("endTime");
+			}
+		}
+		String finalStatTime = statTime;
+		String finalEndTime = endTime;
+
+		Integer total = tmOdsVdqwMessagearchivingMapper.countOdsVdqwMessageByData(statTime, endTime);
+		int totalPages = PageDto.getTotalPages(total, pageSize);
+
+		// 获取消息列表
+		int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1;
+		log.info("获取的线程数：{}",optimalThreadPoolSize);
+		// 创建线程池
+		ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
+
+
+		for (int i = 1; i <= totalPages; i++) {
+			int offset = (i - 1) * pageSize;
+			List<OdsVdqwMessageOTD> messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize);
+			// 处理查询到的数据
+			// 使用 CompletableFuture 并行处理
+			CompletableFuture<?>[] futures = messageList.stream()
+					.map(item -> CompletableFuture.runAsync(() -> {
+						try {
+							processItem(item, finalStatTime, finalEndTime);
+						} catch (Exception e) {
+							log.error("处理消息失败: FromUserId={}, AcceptUserId={}, 异常: {}",
+									item.getFromUserId(), item.getAcceptUserId(), e.getMessage(), e);
+						}
+					}, executor))
+					.toArray(CompletableFuture[]::new);
+			// 等待所有任务完成
+			CompletableFuture.allOf(futures).join();
+		}
+		// 关闭线程池
+		executor.shutdown();
+
+	}
+
+	private void processItem(OdsVdqwMessageOTD item, String statTime, String endTime) {
 		log.info("企微语料内容：FromUserId：{}, AcceptUserId：{}", item.getFromUserId(), item.getAcceptUserId());
 		//  1，vdqw_workuserinfo 这个表对应是 B端认证中心userId
 		//  2，vdqw_externalcontact 这个表对应是 	企微客户unionId
 		String unionId = getUnionId(Arrays.asList(item.getFromUserId(), item.getAcceptUserId()));
 		String userId = getUserId(Arrays.asList(item.getFromUserId(), item.getAcceptUserId()));
-			log.info("企微查询信息为空：unionId：{}, userId：{}", unionId,userId);
+		log.info("企微查询信息：unionId：{}, userId：{}", unionId, userId);
 		if (StringUtils.isEmpty(unionId) || StringUtils.isNotBlank(userId)) {
 			log.info("企微查询信息为空 ");
 			return;
 		}
 
 		if (StringUtils.isNotBlank(item.getFromUserId()) && StringUtils.isNotBlank(item.getAcceptUserId())) {
-
-
 			List<OdsVdqwMessageOTD> contetnList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByFromUserIdAndAcceptUserId(statTime, endTime, Arrays.asList(item.getFromUserId(), item.getAcceptUserId()));
 			OdsVdqwMessageOTD maxMsgTimeItem = contetnList.stream()
 					.max((o1, o2) -> o1.getMsgTime().compareTo(o2.getMsgTime()))
 					.orElse(null);
-				contetnList.stream().forEach(contentItem->{
-
-					Map<String, Object> inputMap = new HashMap();
+			contetnList.forEach(contentItem -> {
+				Map<String, Object> inputMap = new HashMap<>();
 				DiFyReq diFyImageReq = new DiFyReq();
-					diFyImageReq.setUser("11111");
-					diFyImageReq.setFlowId("app-peJXSjHjVKdkYxjdOUuPnZ5b");
+				diFyImageReq.setUser(ConstantStr.corpus_user);
+				diFyImageReq.setFlowId(qiweiToken);
 
 				JSONObject contentJson = JSONObject.parseObject(contentItem.getContent());
 				String content = contentJson.getString("content");
@@ -126,11 +184,10 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 				JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode());
 				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);
-						Map<String, String> ltoMap = new HashMap();
+					Map<String, String> ltoMap = new HashMap<>();
 					ltoMap.put("analysisRecordId", execDifyFlow.getString("aiAnalysisRequestId"));
 					ltoMap.put("analysisScene", "1");
 					ltoMap.put("unionId", unionId);
@@ -139,7 +196,6 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 					ltoMap.put("analysisResult", resultStrOne);
 					ltoMap.put("analysisDetail", resultStrTwo);
 					// 发送MQ
-
 					log.info("send mq  {}", ltoMap);
 					rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(JSON.toJSONString(ltoMap)).build(),
 							new SendCallback() {
@@ -147,21 +203,15 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
 								public void onSuccess(SendResult sendResult) {
 									log.info("企微语料发送MQ成功  消息体:{}", JSON.toJSONString(ltoMap));
 								}
+
 								@Override
 								public void onException(Throwable e) {
 									log.error("企微语料发送MQ异常  消息体:{}, 异常：", JSON.toJSONString(ltoMap), e);
 								}
 							}, 10000);
-
 				}
-
 			});
 		}
-
-
-
-		});
-
 	}
 
 	private String getUnionId(List<String> userIds){

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java

@@ -0,0 +1,14 @@
+package com.volvo.ai.analytic.center.utils;
+/**
+ * @ClassName ConstantStr
+ * @Description
+ * @Author renzhen
+ * @Date 2025-03-11 11:02
+ * @Version 1.0
+ **/
+
+public class ConstantStr {
+
+	public static final String CHANNEL_DCC = "Channel_Dcc";
+	public static final String corpus_user = "corpush_user";
+}

ai-analytic-center-biz/src/main/resources/mapper/TmOdsVdqwMessagearchivingMapper.xml

@@ -1,6 +1,24 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
-<mapper namespace="com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper">
+<mapper namespace="com.volvo.ai.analytic.center.mapper.TmOdsVdqwMessagearchivingMapper">
+
+    <select id="countOdsVdqwMessageByData" resultType="java.lang.Integer">
+        select count(1) from (
+        SELECT
+        tovm.from_user_id as fromUserUnId,
+        tovm.accept_user_id as acceptUserId,
+        tovm.msg_time as msgTime,
+        tovm.chat_type as chatType
+        FROM
+        `tm_ods_vdqw_messagearchiving` tovm
+        WHERE
+        tovm.msg_time between #{statTime} and #{endTime}
+        AND tovm.chat_type = 0
+        GROUP BY
+        tovm.from_user_id,
+        tovm.accept_user_id
+        ) tab
+    </select>
 
     <select id="queryOdsVdqwMessageByData" resultType="com.volvo.ai.analytic.center.dto.corpus.OdsVdqwMessageOTD" >
 
@@ -17,6 +35,7 @@
         GROUP BY
         tovm.from_user_id,
         tovm.accept_user_id
+        LIMIT #{offset}, #{pageSize}
     </select>
 
     <select id="queryOdsVdqwMessageByFromUserIdAndAcceptUserId" resultType="com.volvo.ai.analytic.center.dto.corpus.OdsVdqwMessageOTD" >

ai-analytic-center-biz/src/main/resources/mapper/TmTelephoneCorpusMapper.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper">
+
+</mapper>