feat(jg):优化数据质量评分处理逻辑

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/controller/DataHandlerController.java

@@ -380,6 +380,6 @@ public class DataHandlerController extends BaseController {
     @PostMapping(value = "/handleDataQualityScore")
     @ApiOperation(httpMethod = "POST", value = "修复es中未正常生成数据等级的数据", notes = "修复es中未正常生成数据等级的数据")
     public ResponseModel<String> handleDataQualityScore() {
-        return ResponseHelper.buildResponse(dataHandlerService.handleDataQualityScore());
+        return ResponseHelper.buildResponse(dataHandlerService.handleDataQualityScore2());
     }
 }
\ No newline at end of file

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/CommonServiceImpl.java

@@ -3230,7 +3230,7 @@ public class CommonServiceImpl implements ICommonService {
      */
     public String castDataQualityScore2Name(String dataQualityScore, Boolean isIntoManagement){
         String name = dataDictTypeHandler.handle(DATA_QUALITY_SCORE, dataQualityScore);
-        return name == null ?  (isIntoManagement == null || !isIntoManagement) ? "二类" : "一类" : name;
+        return name == null ? "三类" : name;
     }
 
     public void setDataQualityScore2Json(Map<String, Object> result, String bizId, BusinessTypeEnum businessTypeEnum) {

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/DataHandlerServiceImpl.java

@@ -38,13 +38,16 @@ import com.yeejoin.amos.boot.module.jg.api.dto.JgInstallationNoticeEqDto;
 import com.yeejoin.amos.boot.module.jg.api.dto.JgUseRegistrationDto;
 import com.yeejoin.amos.boot.module.jg.api.dto.PieLineEquipContraptionDto;
 import com.yeejoin.amos.boot.module.jg.api.entity.*;
+import com.yeejoin.amos.boot.module.jg.api.enums.BusinessTypeEnum;
 import com.yeejoin.amos.boot.module.jg.api.enums.PipelineEnum;
 import com.yeejoin.amos.boot.module.jg.api.enums.SafetyProblemTypeEnum;
 import com.yeejoin.amos.boot.module.jg.api.mapper.*;
+import com.yeejoin.amos.boot.module.jg.biz.event.publisher.EventPublisher;
 import com.yeejoin.amos.boot.module.jg.biz.feign.TzsServiceFeignClient;
 import com.yeejoin.amos.boot.module.jg.biz.handler.strategy.ProblemHandleStrategy;
 import com.yeejoin.amos.boot.module.jg.biz.listener.SafetyProblemTopicMessage;
 import com.yeejoin.amos.boot.module.jg.biz.refresh.StatisticsDataUpdateService;
+import com.yeejoin.amos.boot.module.jg.biz.reminder.core.event.EquipCreateOrEditEvent;
 import com.yeejoin.amos.boot.module.ymt.api.dto.EquipWaitRefreshDataQualityScore;
 import com.yeejoin.amos.boot.module.ymt.api.dto.ProjectWaitRefreshDataQualityScore;
 import com.yeejoin.amos.boot.module.ymt.api.dto.RefreshDataDto;
@@ -63,6 +66,7 @@ import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -174,6 +178,8 @@ public class DataHandlerServiceImpl {
 
     private final RestHighLevelClient restHighLevelClient;
 
+    @Autowired
+    private EventPublisher eventPublisher;
 
     /**
      * 安装告知压力管道历史数据修复-详情中的设备列表修改为汇总表格式
@@ -2312,18 +2318,7 @@ public class DataHandlerServiceImpl {
         try {
             BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
 
-            BoolQueryBuilder meBuilder = QueryBuilders.boolQuery();
-            meBuilder.must(QueryBuilders.existsQuery("USE_ORG_CODE"));
-            meBuilder.mustNot(QueryBuilders.existsQuery("DATA_QUALITY_SCORE"));
-            meBuilder.must(QueryBuilders.termQuery("IS_INTO_MANAGEMENT", true));
-            boolQuery.must(meBuilder);
-
-            BoolQueryBuilder meBuilder1 = QueryBuilders.boolQuery();
-            meBuilder1.should(QueryBuilders.matchQuery("STATUS", "已认领"));
-            meBuilder1.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("STATUS")));
-            meBuilder1.should(QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("STATUS", "")));
-            meBuilder1.minimumShouldMatch(1);
-            boolQuery.must(meBuilder1);
+            buildQuery(boolQuery);
 
             esSearchService.searchResponseInBatch(IDX_BIZ_VIEW_JG_ALL, boolQuery, 2000, searchHits -> {
                 List<ESEquipmentCategoryDto> esEquipmentCategoryList = searchHits.stream().map(searchHit -> JSONObject.parseObject(searchHit.getSourceAsString(), ESEquipmentCategoryDto.class)).collect(Collectors.toList());
@@ -2346,4 +2341,111 @@ public class DataHandlerServiceImpl {
         log.info("处理数据质量评分结束，更新{}条设备", total.get());
         return "success";
     }
+
+    @Transactional(rollbackFor = Exception.class)
+    public String handleDataQualityScore2() {
+        log.info("处理数据质量评分开始");
+        AtomicLong total = new AtomicLong(0L);
+        try {
+            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
+
+            log.info("开始处理设备es没有等级的数据");
+            buildQuery(boolQuery);
+            processEquipmentData(boolQuery, total);
+            log.info("设备es没有等级的数据处理完成，当前总计处理{}条设备", total.get());
+
+            boolQuery = QueryBuilders.boolQuery();
+            log.info("开始处理设备es等级不对的数据");
+            buildQuery1(boolQuery);
+            processEquipmentData(boolQuery, total);
+            log.info("设备es等级不对的数据处理完成，当前总计处理{}条设备", total.get());
+        } catch (Exception e) {
+            log.error("处理数据质量评分过程中发生异常", e);
+            throw new RuntimeException(e);
+        }
+        log.info("处理数据质量评分结束，更新{}条设备", total.get());
+        return "success";
+    }
+
+    /**
+     * 处理设备数据的通用方法
+     *
+     * @param boolQuery 查询条件
+     * @param total     计数器
+     */
+    private void processEquipmentData(BoolQueryBuilder boolQuery, AtomicLong total) throws Exception {
+        log.debug("开始执行ES批量查询，查询条件: {}", boolQuery);
+        esSearchService.searchResponseInBatch(IDX_BIZ_VIEW_JG_ALL, boolQuery, 2000, searchHits -> {
+            log.debug("处理批次数据，当前批次包含{}条记录", searchHits.size());
+            List<ESEquipmentCategoryDto> esEquipmentCategoryList = searchHits.stream()
+                    .map(searchHit -> JSONObject.parseObject(searchHit.getSourceAsString(), ESEquipmentCategoryDto.class))
+                    .collect(Collectors.toList());
+
+            Set<String> projectContraptionIds = esEquipmentCategoryList.stream()
+                    .filter(i -> "8000".equals(i.getEQU_LIST_CODE()))
+                    .map(ESEquipmentCategoryDto::getProjectContraptionId)
+                    .collect(Collectors.toSet());
+
+            Set<String> equipRecord = esEquipmentCategoryList.stream()
+                    .filter(i -> !"8000".equals(i.getEQU_LIST_CODE()))
+                    .map(ESEquipmentCategoryDto::getSEQUENCE_NBR)
+                    .collect(Collectors.toSet());
+
+            log.debug("当前批次解析结果: 总设备数={}, 装置设备数={}, 设备数={}",
+                    esEquipmentCategoryList.size(), projectContraptionIds.size(), equipRecord.size());
+
+            if (!ValidationUtil.isEmpty(equipRecord)) {
+                log.info("发布设备事件，设备数量: {}", equipRecord.size());
+                eventPublisher.publish(new EquipCreateOrEditEvent(
+                        this,
+                        BusinessTypeEnum.JG_NEW_EQUIP.name(),
+                        equipRecord,
+                        EquipCreateOrEditEvent.EquipType.equip
+                ));
+            }
+
+            if (!ValidationUtil.isEmpty(projectContraptionIds)) {
+                log.info("发布装置设备事件，设备数量: {}", projectContraptionIds.size());
+                eventPublisher.publish(new EquipCreateOrEditEvent(
+                        this,
+                        BusinessTypeEnum.JG_NEW_PROJECT.name(),
+                        projectContraptionIds,
+                        EquipCreateOrEditEvent.EquipType.project
+                ));
+            }
+
+            total.addAndGet(esEquipmentCategoryList.size());
+            log.debug("当前批次处理完成，累计处理{}条设备", total.get());
+        });
+        log.debug("ES批量查询执行完成");
+    }
+
+    private void buildQuery(BoolQueryBuilder boolQuery) {
+        BoolQueryBuilder meBuilder = QueryBuilders.boolQuery();
+        meBuilder.must(QueryBuilders.existsQuery("USE_ORG_CODE"));
+        meBuilder.mustNot(QueryBuilders.existsQuery("DATA_QUALITY_SCORE"));
+        meBuilder.must(QueryBuilders.termQuery("IS_INTO_MANAGEMENT", true));
+        boolQuery.must(meBuilder);
+
+        BoolQueryBuilder meBuilder1 = QueryBuilders.boolQuery();
+        meBuilder1.should(QueryBuilders.matchQuery("STATUS", "已认领"));
+        meBuilder1.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("STATUS")));
+        meBuilder1.should(QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("STATUS", "")));
+        meBuilder1.minimumShouldMatch(1);
+        boolQuery.must(meBuilder1);
+    }
+
+    private void buildQuery1(BoolQueryBuilder boolQuery) {
+        BoolQueryBuilder meBuilder = QueryBuilders.boolQuery();
+        meBuilder.mustNot(QueryBuilders.existsQuery("USE_ORG_CODE"));
+        meBuilder.must(QueryBuilders.termQuery("DATA_QUALITY_SCORE", 1));
+        boolQuery.must(meBuilder);
+
+        BoolQueryBuilder meBuilder1 = QueryBuilders.boolQuery();
+        meBuilder1.should(QueryBuilders.matchQuery("STATUS", "已认领"));
+        meBuilder1.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("STATUS")));
+        meBuilder1.should(QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("STATUS", "")));
+        meBuilder1.minimumShouldMatch(1);
+        boolQuery.must(meBuilder1);
+    }
 }