refactor(ai-analytic-center): 重构代码并移除未使用的功能
- 删除了多个类中的冗余代码和未使用的属性- 移除了未使用的 AI 解析相关代码和接口 - 优化了部分代码结构，提高了代码可读性和维护性 - 删除了未使用的 constant 字符串 - 添加一些注释
This commit is contained in:
刘亚洲（A）
@@ -23,7 +23,9 @@ public class KafkaConfig {
|
||||
|
||||
|
||||
|
||||
// 第一个Kafka配置
|
||||
/**
|
||||
* 湖仓Kafka配置，用途：mock数据
|
||||
*/
|
||||
@Bean(name = "dccKafkaTemplate")
|
||||
public KafkaTemplate<String, String> dccKafkaTemplate(
|
||||
@Value("${spring.kafka.bootstrap-servers}") String bootstrapServers,
|
||||
@@ -39,7 +41,9 @@ public class KafkaConfig {
|
||||
}
|
||||
|
||||
|
||||
// 第二个Kafka配置
|
||||
/**
|
||||
* 分支中心kafka配置，湖仓Kafka配置
|
||||
*/
|
||||
@Bean(name = "analyticCenterKafkaTemplate")
|
||||
public KafkaTemplate<String, String> analyticCenterKafkaTemplate(
|
||||
@Value("${analyticCenterKafka.bootstrap-servers}") String bootstrapServers,
|
||||
@@ -54,6 +58,10 @@ public class KafkaConfig {
|
||||
return new KafkaTemplate<>(factory);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 分支中心kafka消费者配置
|
||||
*/
|
||||
@Bean(name = "analyticCenterConsumerFactory")
|
||||
public ConcurrentKafkaListenerContainerFactory<String, String> analyticCenterConsumerFactory(
|
||||
@Value("${analyticCenterKafka.bootstrap-servers}") String bootstrapServers,
|
||||
@@ -75,7 +83,6 @@ public class KafkaConfig {
|
||||
new ConcurrentKafkaListenerContainerFactory<>();
|
||||
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
|
||||
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
|
||||
// factory.setBatchListener(true);
|
||||
return factory;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package com.volvo.ai.analytic.center.controller;
|
||||
|
||||
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
|
||||
import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
|
||||
import com.volvo.common.core.util.ResultMsg;
|
||||
import io.swagger.annotations.Api;
|
||||
@@ -25,10 +24,8 @@ public class NameplateCorpusController {
|
||||
|
||||
@Autowired
|
||||
private TmNameplateCorpusService tmNameplateCorpusService;
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
@PostMapping("/update")
|
||||
@ApiOperation(value = "更新dify结果")
|
||||
@ApiOperation(value = "dify发起http 东西网关调用，异步更新dify结果")
|
||||
public ResultMsg<Object> updateNameplate(@RequestBody String message) {
|
||||
log.info("updateNameplate message: {}", message);
|
||||
return tmNameplateCorpusService.updateNameplate(message);
|
||||
|
||||
@@ -3,7 +3,6 @@ package com.volvo.ai.analytic.center.controller;
|
||||
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
|
||||
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
|
||||
import com.volvo.common.core.util.ResultMsg;
|
||||
import io.swagger.annotations.Api;
|
||||
import io.swagger.annotations.ApiOperation;
|
||||
@@ -30,9 +29,6 @@ public class TestController {
|
||||
@Autowired
|
||||
private RocketMQTemplate rocketMQTemplate;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("analyticCenterKafkaTemplate")
|
||||
private KafkaTemplate<String, String> kafkaProducer;
|
||||
@@ -65,7 +61,7 @@ public class TestController {
|
||||
return ResultMsg.ok("ok");
|
||||
}
|
||||
@PostMapping("/mockDccKafka")
|
||||
@ApiOperation(value = "补偿处理消息")
|
||||
@ApiOperation(value = "手动发送kafka消息，用于开发测试")
|
||||
public ResultMsg<Object> mockDccKafka(@RequestBody AicorpusTelephoneDTO message) {
|
||||
|
||||
for(int i=0;i<20;i++){
|
||||
|
||||
@@ -1,15 +1,12 @@
|
||||
package com.volvo.ai.analytic.center.job;
|
||||
|
||||
import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
|
||||
import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
|
||||
import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
|
||||
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.beans.factory.annotation.Value;
|
||||
import org.springframework.cloud.context.config.annotation.RefreshScope;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
@@ -29,39 +26,10 @@ public class NameplateCorpusJob {
|
||||
@Autowired
|
||||
private TmNameplateCorpusService tmNameplateCorpusService;
|
||||
|
||||
@Autowired
|
||||
private TmTelephoneCorpusService tmTelephoneCorpusService;
|
||||
|
||||
@Autowired
|
||||
private TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
|
||||
@Value("${dify.corpus.nameplate.isUse}")
|
||||
private boolean isUse;
|
||||
/**
|
||||
* 铭牌语料处理
|
||||
*/
|
||||
@XxlJob("nameplateCorpusTask")
|
||||
@PostMapping("nameplateCorpusTask")
|
||||
public ResultMsg nameplateCorpusTask(@RequestBody String paramJson) {
|
||||
try {
|
||||
// 获取任务参数
|
||||
String param = XxlJobHelper.getJobParam();
|
||||
if(StringUtils.isEmpty(param)){
|
||||
param = paramJson;
|
||||
}
|
||||
// 执行业务逻辑
|
||||
XxlJobHelper.log("任务参数: {}", param);
|
||||
log.info("nameplateCorpusTask 铭牌语料查询处理:{}",param);
|
||||
tmNameplateCorpusService.runNameplateCorpusDify(param);
|
||||
} catch (Exception e) {
|
||||
log.error("nameplateCorpusTask 定时任务补偿处理消息异常",e.getMessage());
|
||||
}
|
||||
return ResultMsg.ok();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 铭牌语料处理失败重试
|
||||
* @param paramJson
|
||||
* 手动铭牌语料处理失败重试
|
||||
* @param paramJson （statTime，endTime，customerFlowIds）
|
||||
* @return
|
||||
*/
|
||||
@XxlJob("nameplateCorpusTaskFailRetry")
|
||||
|
||||
@@ -24,7 +24,7 @@ import org.springframework.web.client.RestTemplate;
|
||||
|
||||
/**
|
||||
* @ClassName AnalysisDifyMqConsumer
|
||||
* @Description AI解析MQ-Callback处理
|
||||
* @Description AI解析MQ-Callback处理 （暂时未用）
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
* @Version 1.0
|
||||
@@ -44,17 +44,9 @@ public class AnalysisDifyCallbackMqConsumer implements RocketMQListener<MessageE
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
|
||||
@Value("${dify.corpus.checkDccRepeat}")
|
||||
private String checkDccRepeat;
|
||||
|
||||
@Autowired
|
||||
private DiFyService diFyService;
|
||||
|
||||
@Autowired
|
||||
private RestTemplate restTemplate;
|
||||
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@Override
|
||||
public void onMessage(MessageExt messageExt) {
|
||||
long startTime = System.currentTimeMillis();
|
||||
|
||||
@@ -5,10 +5,8 @@ import com.alibaba.fastjson.JSONObject;
|
||||
import com.volvo.ai.analytic.center.dto.req.DiFyReq;
|
||||
import com.volvo.ai.analytic.center.entity.AiAnalysisErrors;
|
||||
import com.volvo.ai.analytic.center.service.AiAnalysisErrorsService;
|
||||
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
|
||||
import com.volvo.ai.analytic.center.service.DiFyService;
|
||||
import com.volvo.ai.analytic.center.utils.ConstantStr;
|
||||
import com.volvo.ai.analytic.center.utils.RedisCounterRateLimiter;
|
||||
import com.volvo.ai.analytic.center.utils.RedisLockService;
|
||||
import com.volvo.ai.analytic.center.utils.RedisZSetUtil;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -18,7 +16,6 @@ import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
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.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
@@ -27,7 +24,7 @@ import java.util.concurrent.CompletableFuture;
|
||||
|
||||
/**
|
||||
* @ClassName AnalysisDifyMqConsumer
|
||||
* @Description AI解析 MQ处理
|
||||
* @Description AI解析 MQ处理 （暂时未用）
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
* @Version 1.0
|
||||
@@ -44,12 +41,6 @@ import java.util.concurrent.CompletableFuture;
|
||||
public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
|
||||
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
|
||||
@Value("${dify.corpus.checkDccRepeat}")
|
||||
private String checkDccRepeat;
|
||||
|
||||
@Value("${rocketmq.consumer.analysisDify.difyLimit}")
|
||||
private int difyLimit;
|
||||
|
||||
@@ -58,14 +49,9 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
|
||||
@Autowired
|
||||
private DiFyService diFyService;
|
||||
|
||||
@Autowired
|
||||
private RedisCounterRateLimiter redisCounterRateLimiter;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisErrorsService aiAnalysisErrorsService;
|
||||
|
||||
@Autowired
|
||||
private RedisTemplate<String, String> redisTemplate;
|
||||
|
||||
@Autowired
|
||||
private RedisZSetUtil redisZSetUtil;
|
||||
|
||||
@@ -4,7 +4,6 @@ package com.volvo.ai.analytic.center.mq;
|
||||
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.TmTelephoneCorpusService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
@@ -16,7 +15,7 @@ import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* @ClassName CorpusProcessKafkaConsumer
|
||||
* @ClassName 湖仓发送过来的kafka消息自己发MQ自己消费 DCC通话记录 （目前接入的需求：4IN1的智能工单-电话场景，LTO跟进记录AI总结）
|
||||
* @Description
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
@@ -37,9 +36,6 @@ public class CorpusDccMqConsumer implements RocketMQListener<MessageExt> {
|
||||
@Autowired
|
||||
private TmTelephoneCorpusService tmTelephoneCorpusService;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@Override
|
||||
|
||||
@@ -30,7 +30,7 @@ import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* @ClassName CorpusProcessKafkaConsumer
|
||||
* @Description
|
||||
* @Description 消费湖仓同步的DCC通话记录kafka消息
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
* @Version 1.0
|
||||
|
||||
@@ -4,7 +4,6 @@ import com.alibaba.fastjson.JSONObject;
|
||||
import com.volvo.ai.analytic.center.entity.TmAnalysisResult;
|
||||
import com.volvo.ai.analytic.center.entity.TtAnalysisResultInfo;
|
||||
import com.volvo.ai.analytic.center.enums.BusinessTypeEnum;
|
||||
import com.volvo.ai.analytic.center.mapper.AiAnalysisRequestLogsMapper;
|
||||
import com.volvo.ai.analytic.center.mapper.TtAnalysisResultInfoMapper;
|
||||
import com.volvo.ai.analytic.center.service.TmAnalysisResultService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -19,6 +18,9 @@ import java.util.Arrays;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* LTO跟进记录AI总结的点赞点踩
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.corpus.isLikeTpoicGroup}",
|
||||
@@ -29,9 +31,6 @@ public class CorpushIsLikeConsumer implements RocketMQListener<MessageExt>{
|
||||
@Autowired
|
||||
private TmAnalysisResultService tmAnalysisResultService;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsMapper aiAnalysisRequestLogsMapper;
|
||||
|
||||
@Autowired
|
||||
private TtAnalysisResultInfoMapper ttAnalysisResultInfoMapper;
|
||||
@Override
|
||||
|
||||
@@ -4,7 +4,6 @@ package com.volvo.ai.analytic.center.mq;
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.volvo.ai.analytic.center.dto.req.FourInOneRequestDTO;
|
||||
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
|
||||
import com.volvo.ai.analytic.center.service.IntelligentCustomerService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
@@ -17,7 +16,7 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* @ClassName CorpusProcessKafkaConsumer
|
||||
* @Description 智能客服 语料解析 四合一
|
||||
* @Description 消费4In1智能工单-在线客服的语料消息
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
* @Version 1.0
|
||||
@@ -37,8 +36,6 @@ public class IntelligentCustomerMqConsumer implements RocketMQListener<MessageEx
|
||||
@Autowired
|
||||
private IntelligentCustomerService intelligentCustomerService;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@Override
|
||||
|
||||
@@ -7,7 +7,6 @@ import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
|
||||
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.cloud.context.config.annotation.RefreshScope;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
@@ -17,11 +16,9 @@ import org.springframework.messaging.handler.annotation.Header;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
/**
|
||||
* @ClassName NameplateKafkaConsumer
|
||||
* @Description 铭牌数据处理
|
||||
* @Description 消费tm_nameplate_corpus表binlog的Kafka消息
|
||||
* @Author renzhen
|
||||
* @Date 2025-03-04 10:18
|
||||
* @Version 1.0
|
||||
@@ -37,17 +34,13 @@ public class NameplateKafkaConsumer {
|
||||
private TmNameplateCorpusService tmNameplateCorpusService;
|
||||
|
||||
|
||||
@Resource
|
||||
private RocketMQTemplate rocketMqTemplate;
|
||||
|
||||
|
||||
@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}",
|
||||
groupId = "${analyticCenterKafka.consumer.group}" ,
|
||||
containerFactory = "analyticCenterConsumerFactory",
|
||||
concurrency = "3")
|
||||
public void listen(String recordMessages, Acknowledgment ack,
|
||||
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partitionId,
|
||||
@Header(KafkaHeaders.OFFSET) Long offset) throws InterruptedException {
|
||||
@Header(KafkaHeaders.OFFSET) Long offset) {
|
||||
long startTime = System.currentTimeMillis();
|
||||
log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId());
|
||||
log.info("nameplateKafkaConsumerMessage: {}", recordMessages);
|
||||
|
||||
@@ -13,8 +13,6 @@ import java.util.List;
|
||||
*/
|
||||
public interface TmNameplateCorpusService extends IService<TmNameplateCorpus> {
|
||||
|
||||
void runNameplateCorpusDify(String paramJson);
|
||||
|
||||
void runNameplateCorpusDifyRetry(String paramJson);
|
||||
|
||||
void processItem(TmNameplateCorpus item);
|
||||
|
||||
@@ -54,8 +54,6 @@ public class IntelligentCustomerServiceImpl implements IntelligentCustomerServic
|
||||
private String topic;
|
||||
@Value("${dify.intelligentCustomer.fourInOneToken}")
|
||||
private String fourInOneToken;
|
||||
@Autowired
|
||||
private RemoteCarModelClient remoteCarModelClient;
|
||||
|
||||
@Autowired
|
||||
private DataMaskingRuleService dataMaskingRuleService;
|
||||
@@ -66,13 +64,6 @@ public class IntelligentCustomerServiceImpl implements IntelligentCustomerServic
|
||||
@Autowired
|
||||
private AiAnalysisErrorsService aiAnalysisErrorsService;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisErrorsMapper aiAnalysisErrorsMapper;
|
||||
|
||||
@Autowired
|
||||
private TcIntelligentCustomerMapper tcIntelligentCustomerMapper;
|
||||
|
||||
private Map<String, List<String>> tcIntelligentCustomer = Collections.EMPTY_MAP;
|
||||
|
||||
@Override
|
||||
public void fourInOneMqService(FourInOneRequestDTO fourInOneRequestDTO) {
|
||||
@@ -148,39 +139,4 @@ public class IntelligentCustomerServiceImpl implements IntelligentCustomerServic
|
||||
}, 10000);
|
||||
}
|
||||
|
||||
// 咨询类【201】/投诉类【202】/投诉类型【203】 过滤
|
||||
/** private String fourInOneMqServiceFilter(String customerType,String customerValue){
|
||||
if(tcIntelligentCustomer.isEmpty()){
|
||||
queryTcIntelligentCustomer();
|
||||
}
|
||||
List<String> typeValues = tcIntelligentCustomer.get(customerType);
|
||||
if(!typeValues.contains(customerValue)){
|
||||
return "";
|
||||
}
|
||||
return customerValue;
|
||||
}
|
||||
|
||||
|
||||
public Map<String, List<String>> queryTcIntelligentCustomer() {
|
||||
LambdaQueryWrapper<TcIntelligentCustomer> queryWrapper = new LambdaQueryWrapper<>();
|
||||
queryWrapper.eq(TcIntelligentCustomer::getIsDeleted, "0");
|
||||
List<TcIntelligentCustomer> tcIntelligentCustomerList =tcIntelligentCustomerMapper.selectList(queryWrapper);
|
||||
tcIntelligentCustomer = tcIntelligentCustomerList.stream()
|
||||
.collect(Collectors.groupingBy(
|
||||
TcIntelligentCustomer::getIntelligentCustomerType, // 按照 IntelligentCustomerType 分组
|
||||
Collectors.mapping(
|
||||
customer -> {
|
||||
if (customer.getIntelligentCustomerType().equals(IntelligentCustomerTypeEnum.CONSULT_TYPE.getCode())) {
|
||||
return customer.getTypeOne(); // 如果是咨询类型，返回 TypeOne
|
||||
} else {
|
||||
return customer.getTypeOne().concat("#").concat(customer.getTypeTwo()).concat("#").concat(customer.getTypeThree()); // 其他类型，拼接 TypeOne、TypeTwo、TypeThree
|
||||
}
|
||||
},
|
||||
Collectors.toList()
|
||||
)
|
||||
));
|
||||
return tcIntelligentCustomer;
|
||||
}
|
||||
**/
|
||||
|
||||
}
|
||||
@@ -10,10 +10,8 @@ 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.feign.RemoteCarModelClient;
|
||||
import com.volvo.ai.analytic.center.mapper.AiAnalysisErrorsMapper;
|
||||
import com.volvo.ai.analytic.center.mapper.TmNameplateCorpusMapper;
|
||||
import com.volvo.ai.analytic.center.mapper.TmOdsVdqwMessagearchivingMapper;
|
||||
import com.volvo.ai.analytic.center.mapper.TtNameplateRecordMapper;
|
||||
import com.volvo.ai.analytic.center.service.*;
|
||||
import com.volvo.ai.analytic.center.utils.ConstantStr;
|
||||
@@ -21,13 +19,11 @@ import com.volvo.common.core.util.ResultMsg;
|
||||
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.LocalDate;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
@@ -45,8 +41,6 @@ import java.util.concurrent.Executors;
|
||||
@Service
|
||||
public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusMapper, TmNameplateCorpus> implements TmNameplateCorpusService {
|
||||
|
||||
@Autowired
|
||||
private TmOdsVdqwMessagearchivingMapper tmOdsVdqwMessagearchivingMapper;
|
||||
|
||||
@Autowired
|
||||
private TmNameplateCorpusMapper tmNameplateCorpusMapper;
|
||||
@@ -63,72 +57,14 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
|
||||
private AiAnalysisErrorsMapper aiAnalysisErrorsMapper;
|
||||
@Autowired
|
||||
private DiFyService diFyService;
|
||||
|
||||
@Resource
|
||||
private RocketMQTemplate rocketMqTemplate;
|
||||
@Value("${dify.corpus.nameplate.appkey}")
|
||||
private String nameplateAppKey;
|
||||
|
||||
@Value("${batch.size}")
|
||||
public int pageSize = 100;
|
||||
@Autowired
|
||||
private RemoteCarModelClient remoteCarModelClient;
|
||||
|
||||
@Autowired
|
||||
private DataMaskingRuleService dataMaskingRuleService;
|
||||
|
||||
|
||||
@Override
|
||||
public void runNameplateCorpusDify(String paramJson) {
|
||||
long startTime = System.currentTimeMillis();
|
||||
log.info("runNameplateCorpusDify 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(ConstantStr.statTime) && paramJsonObj.containsKey(ConstantStr.endTime)) {
|
||||
statTime = paramJsonObj.getString(ConstantStr.statTime);
|
||||
endTime = paramJsonObj.getString(ConstantStr.endTime);
|
||||
}
|
||||
}
|
||||
|
||||
Integer total = tmNameplateCorpusMapper.countTmNameplateCorpusByData(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<TmNameplateCorpus> messageList = tmNameplateCorpusMapper.queryTmNameplateCorpusByData(statTime, endTime, offset, pageSize);
|
||||
// 处理查询到的数据
|
||||
// 使用 CompletableFuture 并行处理
|
||||
CompletableFuture<?>[] futures = messageList.stream()
|
||||
.map(item -> CompletableFuture.runAsync(() -> {
|
||||
try {
|
||||
processItem(item);
|
||||
} catch (Exception e) {
|
||||
log.error("铭牌语料失败: customerFlowId={}, AcceptUserId={}, 异常: {}",
|
||||
item.getCustomerFlowId(), e.getMessage(), e);
|
||||
}
|
||||
}, executor))
|
||||
.toArray(CompletableFuture[]::new);
|
||||
// 等待所有任务完成
|
||||
CompletableFuture.allOf(futures).join();
|
||||
}
|
||||
// 关闭线程池
|
||||
executor.shutdown();
|
||||
log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
|
||||
}
|
||||
@Override
|
||||
public void runNameplateCorpusDifyRetry(String paramJson) {
|
||||
long startTime = System.currentTimeMillis();
|
||||
|
||||
@@ -18,7 +18,6 @@ import com.volvo.ai.analytic.center.enums.BizEnum;
|
||||
import com.volvo.ai.analytic.center.enums.BusinessTypeEnum;
|
||||
import com.volvo.ai.analytic.center.enums.CategoryEnum;
|
||||
import com.volvo.ai.analytic.center.feign.RemoteCarModelClient;
|
||||
import com.volvo.ai.analytic.center.mapper.AiAnalysisErrorsMapper;
|
||||
import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
|
||||
import com.volvo.ai.analytic.center.service.*;
|
||||
import com.volvo.ai.analytic.center.utils.ConstantStr;
|
||||
@@ -77,9 +76,6 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
|
||||
@Autowired
|
||||
private AiAnalysisErrorsService aiAnalysisErrorsService;
|
||||
|
||||
@Autowired
|
||||
private AiAnalysisErrorsMapper aiAnalysisErrorsMapper;
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void saveTelephoneCorpus(TmTelephoneCorpus tmTelephoneCorpus) {
|
||||
@@ -217,10 +213,4 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
|
||||
}
|
||||
}, 10000);
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -23,4 +23,4 @@ public class ConstantStr {
|
||||
public static final String aiAnalysisRequestId = "aiAnalysisRequestId";
|
||||
|
||||
public static final String DIFY_COUNTERRATELIMIT = "DIFY_COUNTERRATELIMIT";
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user