From 3b83ff86ba616dc6964ea9ec73fe602c1d6c3931 Mon Sep 17 00:00:00 2001 From: "jie.lu" Date: Fri, 18 Jul 2025 17:49:36 +0800 Subject: [PATCH] =?UTF-8?q?IM=E6=95=B0=E6=8D=AE=E5=A4=84=E7=90=86=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../center/dto/corpus/ImCurMessageDTO.java | 34 +++ .../center/enums/BusinessTypeEnum.java | 2 + .../center/enums/ImCustomTypeEnums.java | 27 ++ .../center/job/ImMessageAnalysisJob.java | 44 ++++ .../TmOdsImRealtimeMessageCurMapper.java | 12 + .../TmOdsImRealtimeMessageCurService.java | 15 ++ .../TmOdsImRealtimeMessageCurServiceImpl.java | 234 ++++++++++++++++++ .../TmOdsImRealtimeMessageCurMapper.xml | 53 ++++ .../TmOdsImRealtimeMessageCurMapper.xml | 53 ++++ 9 files changed, 474 insertions(+) create mode 100644 ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/ImCurMessageDTO.java create mode 100644 ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/ImCustomTypeEnums.java create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/ImMessageAnalysisJob.java create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmOdsImRealtimeMessageCurService.java create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsImRealtimeMessageCurServiceImpl.java create mode 100644 ai-analytic-center-biz/src/main/resources/mapper/TmOdsImRealtimeMessageCurMapper.xml create mode 100644 ai-analytic-center-biz/target/classes/mapper/TmOdsImRealtimeMessageCurMapper.xml diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/ImCurMessageDTO.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/ImCurMessageDTO.java new file mode 100644 index 0000000..98e921d --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/corpus/ImCurMessageDTO.java @@ -0,0 +1,34 @@ +package com.volvo.ai.analytic.center.dto.corpus; + +import lombok.Data; + +import java.util.Date; + +/** + * @description IM会话存档消息记录表-湖仓同步表 + * @author jlu + * @date 2025-07-17 + */ +@Data +public class ImCurMessageDTO { + + private static final long serialVersionUID = 1L; + + private String sessionId; + + /** + * 消息内容文本 + */ + private String msgContent; + + /** + * 客户类型0:用户 1：百度 2：GPT 3：人工 4：唯都Agent + */ + private Integer customerType; + + private Long msgTimestamp; + + public ImCurMessageDTO() { + } + +} \ No newline at end of file diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/BusinessTypeEnum.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/BusinessTypeEnum.java index c735c33..706a83d 100644 --- a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/BusinessTypeEnum.java +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/BusinessTypeEnum.java @@ -28,6 +28,8 @@ public enum BusinessTypeEnum { SMART_ASSISTANT_NAMEPLATE("SMART_ASSISTANT_NAMEPLATE", "智能助手-铭牌"), + IM_CORPUS_ANALYSIS("IM_CORPUS_ANALYSIS", "IM语料库"), + //特邀发言官 SPOKESMAN("SPOKESMAN", "特邀发言官"), diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/ImCustomTypeEnums.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/ImCustomTypeEnums.java new file mode 100644 index 0000000..5185164 --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/enums/ImCustomTypeEnums.java @@ -0,0 +1,27 @@ +package com.volvo.ai.analytic.center.enums; + +public enum ImCustomTypeEnums { + + CUSTOMER(0, "用户"), + BAIDU(1, "百度"), + GPT(2, "GPT"), + MANUAL(3, "人工"), + VOLVO_AGENT(4, "唯都Agent"), + ; + + private Integer code; + private String message; + + ImCustomTypeEnums(Integer code, String message) { + this.code = code; + this.message = message; + } + + public Integer getCode() { + return this.code; + } + + public String getMessage() { + return this.message; + } +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/ImMessageAnalysisJob.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/ImMessageAnalysisJob.java new file mode 100644 index 0000000..bb88f38 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/ImMessageAnalysisJob.java @@ -0,0 +1,44 @@ +package com.volvo.ai.analytic.center.job; + +import com.volvo.ai.analytic.center.service.TmOdsImRealtimeMessageCurService; +import com.volvo.common.core.util.ResultMsg; +import com.xxl.job.core.context.XxlJobHelper; +import com.xxl.job.core.handler.annotation.XxlJob; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +@Slf4j +@Component +@RestController +public class ImMessageAnalysisJob { + + @Autowired + private TmOdsImRealtimeMessageCurService tmOdsImRealtimeMessageCurService; + + /** + * 企微语料处理 + */ + @XxlJob("imMessageAnalysisTask") + public ResultMsg imMessageAnalysisTask(@RequestBody String paramJson) { + try { + // 获取任务参数 + String param = XxlJobHelper.getJobParam(); + if(StringUtils.isEmpty(param)){ + param = paramJson; + } + // 执行业务逻辑 + XxlJobHelper.log("任务参数: {}", param); + log.info("imMessageAnalysisTask IM语料分析处理:{}",param); + tmOdsImRealtimeMessageCurService.runImCorpusDify(param); + } catch (Exception e) { + log.error("imMessageAnalysisTask 定时任务处理消息异常",e.getMessage()); + throw new RuntimeException(e); + } + return ResultMsg.ok(); + } + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TmOdsImRealtimeMessageCurMapper.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TmOdsImRealtimeMessageCurMapper.java index ed5689f..ce70cb4 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TmOdsImRealtimeMessageCurMapper.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TmOdsImRealtimeMessageCurMapper.java @@ -2,8 +2,13 @@ package com.volvo.ai.analytic.center.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.volvo.ai.analytic.center.dto.corpus.ImCurMessageDTO; import com.volvo.ai.analytic.center.entity.TmOdsImRealtimeMessageCur; + +import java.util.List; + import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; /** * @description 湖仓同步实时IM消息表-Mapper (tm_ods_im_realtime_message_cur) @@ -11,4 +16,11 @@ import org.apache.ibatis.annotations.Mapper; */ @Mapper public interface TmOdsImRealtimeMessageCurMapper extends BaseMapper { + + // 统计IM消息数据 + int countImRealtimeMessageCurByDate(@Param("statTime") Long statTime, @Param("endTime") Long endTime); + // 分页获取处理数据 + List queryImRealtimeMessageCurByDate(@Param("statTime") Long statTime, @Param("endTime") Long endTime, @Param("offset") int offset, @Param("pageSize") int pageSize); + + List queryImRealtimeMessageCurBySessionId(@Param("statTime") Long statTime, @Param("endTime") Long endTime, @Param("sessionId") String sessionId); } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmOdsImRealtimeMessageCurService.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmOdsImRealtimeMessageCurService.java new file mode 100644 index 0000000..ce2b71f --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmOdsImRealtimeMessageCurService.java @@ -0,0 +1,15 @@ +package com.volvo.ai.analytic.center.service; + +import com.baomidou.mybatisplus.extension.service.IService; +import com.volvo.ai.analytic.center.entity.TmOdsImRealtimeMessageCur; +/** + * @description 电话语料表-同步表 + * @author BEJSON + * @date 2025-03-04 + */ +public interface TmOdsImRealtimeMessageCurService extends IService { + + void runImCorpusDify(String paramJson); + + +} \ No newline at end of file diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsImRealtimeMessageCurServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsImRealtimeMessageCurServiceImpl.java new file mode 100644 index 0000000..6f769d5 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsImRealtimeMessageCurServiceImpl.java @@ -0,0 +1,234 @@ +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.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.volvo.ai.analytic.center.dto.PageDto; +import com.volvo.ai.analytic.center.dto.corpus.CorpusReportDTO; +import com.volvo.ai.analytic.center.dto.corpus.ImCurMessageDTO; +import com.volvo.ai.analytic.center.dto.corpus.OdsVdqwMessageOTD; +import com.volvo.ai.analytic.center.dto.req.ConsultingRequestDTO; +import com.volvo.ai.analytic.center.dto.req.DiFyReq; +import com.volvo.ai.analytic.center.dto.req.RunMaskingRuleInput; +import com.volvo.ai.analytic.center.entity.*; +import com.volvo.ai.analytic.center.enums.BusinessTypeEnum; +import com.volvo.ai.analytic.center.enums.CategoryEnum; +import com.volvo.ai.analytic.center.enums.ImCustomTypeEnums; +import com.volvo.ai.analytic.center.feign.RemoteCarModelClient; +import com.volvo.ai.analytic.center.mapper.TmOdsImRealtimeMessageCurMapper; +import com.volvo.ai.analytic.center.mapper.TmOdsVdqwExternalcontactMapper; +import com.volvo.ai.analytic.center.mapper.TmOdsVdqwMessagearchivingMapper; +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; +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.stereotype.Service; + +import javax.annotation.Resource; + +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.Arrays; +import java.util.Date; +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.concurrent.atomic.AtomicInteger; + + +/** + * @description 企微语料表-同步表 + * @author jlu + * @date 2025-07-17 + */ +@RefreshScope +@Slf4j +@Service +public class TmOdsImRealtimeMessageCurServiceImpl extends ServiceImpl implements TmOdsImRealtimeMessageCurService { + + @Autowired + private TmOdsImRealtimeMessageCurMapper tmOdsImRealtimeMessageCurMapper; + + @Autowired + private DiFyService diFyService; + + @Value("${dify.corpus.imCurToken}") + private String imCurToken; + + @Value("${batch.size}") + public int pageSize = 100; + + @Autowired + private DataMaskingRuleService dataMaskingRuleService; + + @Autowired + private CorpusQuestionService corpusQuestionService; + + @Override + public void runImCorpusDify(String paramJson) { + long startTime = System.currentTimeMillis(); + log.info("runImCorpusDify paramJson {}", paramJson); + // 获取当前日期 + LocalDate today = LocalDate.now(); + // 获取前一天日期 + LocalDate yesterday = today.minusDays(1); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + // 格式化输出 + String formattedDate = yesterday.toString(); // 默认格式为 yyyy-MM-dd + String statTimeString = formattedDate.concat(" 00:00:00"); + String endTimeString = formattedDate.concat(" 23:59:59"); + if (StringUtils.isNotBlank(paramJson)) { + JSONObject paramJsonObj = JSONObject.parseObject(paramJson); + if (null != paramJsonObj && paramJsonObj.containsKey("statTime") && paramJsonObj.containsKey("endTime")) { + statTimeString = paramJsonObj.getString("statTime"); + endTimeString = paramJsonObj.getString("endTime"); + } + } + Long finalStatTime = stringTimeToTimeStamp(statTimeString, formatter); + Long finalEndTime = stringTimeToTimeStamp(endTimeString, formatter); + + Long statTime = stringTimeToTimeStamp(statTimeString, formatter); + Long endTime = stringTimeToTimeStamp(endTimeString, formatter); + + Integer total = tmOdsImRealtimeMessageCurMapper.countImRealtimeMessageCurByDate(statTime, endTime); + int totalPages = PageDto.getTotalPages(total, pageSize); + + // 设置脱敏规则 + List maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(BusinessTypeEnum.SMART_ASSISTANT_QIWEI.getCode()); + RunMaskingRuleInput runMaskingRuleInput = new RunMaskingRuleInput(); + runMaskingRuleInput.setDataMaskingRules(maskingRuleItems); + + // 创建线程池 + int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1; + log.info("IM处理总数据量：{},总页数：{}，获取的线程数：{}",total,totalPages,optimalThreadPoolSize); + ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小 + + try { + for (int i = 1; i <= totalPages; i++) { + int offset = (i - 1) * pageSize; + List messageList = tmOdsImRealtimeMessageCurMapper.queryImRealtimeMessageCurByDate(statTime, endTime, offset, pageSize); + // 处理查询到的数据 + // 使用 CompletableFuture 并行处理 + CompletableFuture[] futures = messageList.stream() + .map(item -> CompletableFuture.runAsync(() -> { + try { + processItem(item, finalStatTime, finalEndTime, runMaskingRuleInput, formatter); + } catch (Exception e) { + log.error("处理IM语料失败: sessionId={}, 异常: {}", + item.getSessionId(), e.getMessage(), e); + } + }, executor)) + .toArray(CompletableFuture[]::new); + // 等待所有任务完成 + CompletableFuture.allOf(futures).join(); + } + // 关闭线程池 + executor.shutdown(); + } catch (Exception e) { + log.error("企微数据跑批异常",e); + } finally { + // 关闭线程池 + executor.shutdown(); + } + log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime); + } + + private void processItem(ImCurMessageDTO item, Long statTime, Long endTime, RunMaskingRuleInput runMaskingRuleInput, DateTimeFormatter formatter) { + log.info("IM语料内容处理：sessionId：{}", item.getSessionId()); + + // 获取语料信息 + List contetnList = tmOdsImRealtimeMessageCurMapper.queryImRealtimeMessageCurBySessionId(statTime, endTime, item.getSessionId()); + + // 组装语料信息 + StringBuffer chatList = new StringBuffer(); + contetnList.forEach(contentItem -> { + String title = ""; + if(ImCustomTypeEnums.CUSTOMER.getCode().equals(contentItem.getCustomerType())){ + title="客户:"; + }else{ + title="顾问:"; + } + JSONObject contentJson = JSONObject.parseObject(contentItem.getMsgContent()); + String content = contentJson.getString("text"); + // 拼接 role 和 text + String chat = title.concat(content); + runMaskingRuleInput.setOldStr(chat); + String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput); + chatList.append(corpusChat).append("\n"); + }); + Map inputMap = new HashMap<>(); + inputMap.put("chat", chatList.toString()); + inputMap.put("businessType", BusinessTypeEnum.IM_CORPUS_ANALYSIS.getCode()); + inputMap.put("sourceCorpusId", item.getSessionId()); + inputMap.put("sourceCorpusTime", timeStampToLocalDateTime(item.getMsgTimestamp(), formatter)); + + DiFyReq diFyImageReq = new DiFyReq(); + diFyImageReq.setUser(ConstantStr.corpus_user); + diFyImageReq.setFlowId(imCurToken); + diFyImageReq.setInputs(inputMap); + // 调用dify + JSONObject resultJsonObject = (JSONObject) diFyService.getDiFyObject(diFyImageReq); + log.info("runDify difyflow {}", resultJsonObject); + + // 保存结果 + if (resultJsonObject != null) { + ConsultingRequestDTO corpusQuestion = new ConsultingRequestDTO(); + corpusQuestion.setText(resultJsonObject.getString("text")); + corpusQuestion.setBusinessType(resultJsonObject.getString("businessType")); + corpusQuestion.setSourceCorpusId(resultJsonObject.getString("sourceCorpusId")); + String sourceCorpusTimeStr = resultJsonObject.getString("sourceCorpusTime"); + if (StringUtils.isNotBlank(sourceCorpusTimeStr)) { + try { + corpusQuestion.setSourceCorpusTime(DateUtil.parse(sourceCorpusTimeStr)); + } catch (Exception e) { + log.error("转换 sourceCorpusTime 为 Date 类型失败，时间字符串: {}", sourceCorpusTimeStr, e); + } + } + corpusQuestionService.runCorpusQuestionDify(corpusQuestion); + } + + } + + /** + * 时间字符串转时间戳秒级 + * @param stringTime + * @param formatter + * @return + */ + private Long stringTimeToTimeStamp(String stringTime, DateTimeFormatter formatter) { + LocalDateTime localDateTime = LocalDateTime.parse(stringTime, formatter); + + // 转换为秒级时间戳 + long timestampSeconds = localDateTime.atZone(ZoneId.systemDefault()).toEpochSecond(); + return timestampSeconds; + } + + /** + * 秒级时间戳转时间字符串 + * @param timeStamp + * @param formatter + * @return + */ + private String timeStampToLocalDateTime(Long timeStamp, DateTimeFormatter formatter) { + Instant instant = Instant.ofEpochSecond(timeStamp); + LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault()); + return localDateTime.format(formatter); + } + +} \ No newline at end of file diff --git a/ai-analytic-center-biz/src/main/resources/mapper/TmOdsImRealtimeMessageCurMapper.xml b/ai-analytic-center-biz/src/main/resources/mapper/TmOdsImRealtimeMessageCurMapper.xml new file mode 100644 index 0000000..3fdd28b --- /dev/null +++ b/ai-analytic-center-biz/src/main/resources/mapper/TmOdsImRealtimeMessageCurMapper.xml @@ -0,0 +1,53 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/ai-analytic-center-biz/target/classes/mapper/TmOdsImRealtimeMessageCurMapper.xml b/ai-analytic-center-biz/target/classes/mapper/TmOdsImRealtimeMessageCurMapper.xml new file mode 100644 index 0000000..3fdd28b --- /dev/null +++ b/ai-analytic-center-biz/target/classes/mapper/TmOdsImRealtimeMessageCurMapper.xml @@ -0,0 +1,53 @@ + + + + + + + + + + + \ No newline at end of file