@@ -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 < TmOdsImRealtimeMessageCurMapper , TmOdsImRealtimeMessageCur > 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 < DataMaskingRule > 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 < ImCurMessageDTO > 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语料内容处理： ： " , item . getSessionId ( ) ) ;
// 获取语料信息
List < ImCurMessageDTO > 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 < String , Object > 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 ) ;
}
}
jie.lu