feat: 新增设备增量添加接口及实现

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

@@ -253,6 +253,13 @@ public class DataHandlerController extends BaseController {
         return ResponseHelper.buildResponse(dataHandlerService.initEquipStatistData2EsBatch());
     }
 
+    @TycloudOperation(ApiLevel = UserType.AGENCY)
+    @ApiOperation(httpMethod = "PUT", value = "增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备", notes = "增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备")
+    @PutMapping(value = "/equip/addStatistData2Es")
+    public ResponseModel<Integer> addStatistData2EsBatch() {
+        return ResponseHelper.buildResponse(dataHandlerService.addStatistData2EsBatch());
+    }
+
 
     /**
      * @apiNote 场车车牌号刷入ES

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

@@ -1597,6 +1597,77 @@ public class DataHandlerServiceImpl {
     }
 
 
+    public Integer addStatistData2EsBatch() {
+        log.info("增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备信息入库开始");
+        StopWatch watch = new StopWatch();
+        watch.start();
+        List<String> refreshRecords = useInfoService.getBaseMapper().selectAddDataRecords();
+        while (!refreshRecords.isEmpty()) {
+            StopWatch watch0 = new StopWatch();
+            watch0.start();
+            List<Map<String, Object>> details = useInfoService.getBaseMapper().queryDetailBatch(refreshRecords);
+            Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
+
+            // 设备最新的维保信息-维度：设备
+            List<IdxBizJgMaintenanceRecordInfo> lastMaintenanceRecordInfos = maintenanceRecordInfoService.getBaseMapper().selectLastedMainInfoBatch(refreshRecords);
+            Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));
+
+            // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
+            List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = idxBizJgInspectionDetectionInfoService.getBaseMapper().selectLastedGroupByInspectTypeBatch(refreshRecords);
+            Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));
+
+            // 设备最新的检验信息-维度：设备
+            Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e->e.getInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getInspectDate))));
+            List<ESEquipmentInfo> esEquipmentInfos = refreshRecords.parallelStream().map(record -> {
+                ESEquipmentInfo esEquipmentInfo = null;
+                try {
+                    esEquipmentInfo = new ESEquipmentInfo();
+                    Map<String, Object> detail = recordDetailMap.get(record);
+                    StatisticsDataUpdateService.formatUseDate(detail);
+                    BeanUtil.copyProperties(detail, esEquipmentInfo, true);
+                    Optional<ESEquipmentCategoryDto> esOptional = esEquipmentCategory.findById(record);
+                    if (esOptional.isPresent()) {
+                        ESEquipmentCategoryDto dto = esOptional.get();
+                        esEquipmentInfo.setUSC_UNIT_CREDIT_CODE(dto.getUSC_UNIT_CREDIT_CODE());
+                        esEquipmentInfo.setUSC_UNIT_NAME(dto.getUSC_UNIT_NAME());
+                        esEquipmentInfo.setIS_DO_BUSINESS(dto.getIS_DO_BUSINESS());
+                    }
+                    // 最新检验信息-维度record
+                    IdxBizJgInspectionDetectionInfo inspectionDetectionInfo = Optional.ofNullable(recordLastInspectionMap.get(record)).flatMap(i->i).orElse(new IdxBizJgInspectionDetectionInfo());
+                    // 最新维保信息-维度record
+                    IdxBizJgMaintenanceRecordInfo lastMaintenanceRecordInfo = Optional.ofNullable(recordLastMaintMap.get(record)).filter(l -> !l.isEmpty()).map(list -> list.get(0)).orElse(new IdxBizJgMaintenanceRecordInfo());
+                    // 最新检验信息-维度record、检验类型，存最新的一条
+                    List<IdxBizJgInspectionDetectionInfo> inspectionDetectionInfos = recordInspectInfosGroupByInspectTypeMap.getOrDefault(record, new ArrayList<>());
+                    StatisticsDataUpdateService.formatInspectDate(esEquipmentInfo, inspectionDetectionInfo, record);
+                    esEquipmentInfo.setIssueDate(getIssueDate(esEquipmentInfo.getUSE_ORG_CODE()));
+                    esEquipmentInfo.setMAINTAIN_UNIT(lastMaintenanceRecordInfo.getMeUnitCreditCode());
+                    esEquipmentInfo.setMAINTAIN_UNIT_NAME(lastMaintenanceRecordInfo.getMeUnitName());
+                    esEquipmentInfo.setInspections(BeanUtil.copyToList(inspectionDetectionInfos, ESEquipmentInfo.Inspection.class));
+                    esEquipmentInfo.setMaintenances(lastMaintenanceRecordInfo.getSequenceNbr() != null ? Collections.singletonList(BeanUtil.copyProperties(lastMaintenanceRecordInfo, ESEquipmentInfo.Maintenance.class)) : new ArrayList<>());
+                    esEquipmentInfo.setTechParams(this.buildTechParamByEquList(record, esEquipmentInfo.getEQU_LIST_CODE()));
+                } catch (Exception e) {
+                    // 异常数据跳过
+                    log.error("设备刷数据处理失败：{}", record, e);
+                }
+                return esEquipmentInfo;
+            }).collect(Collectors.toList());
+            watch0.stop();
+            log.warn("多线程处理查询设备详情耗时：{}", watch0.getTotalTimeSeconds());
+            esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
+            if (!esEquipmentInfos.isEmpty()) {
+                StopWatch watch1 = new StopWatch();
+                watch1.start();
+                esEquipmentDao.saveAll(esEquipmentInfos);
+                watch1.stop();
+                log.warn("es批量入库条数：{}，耗时：{}s", esEquipmentInfos.size(), watch1.getTotalTimeSeconds());
+            }
+        }
+        watch.stop();
+        log.info("增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备信息入库结束，耗时：{}秒", watch.getTotalTimeSeconds());
+        return refreshRecords.size();
+    }
+
+
     private List<ESEquipmentInfo.TechParam> buildTechParamByEquList(String record, String equListCode) {
         return StringUtils.isNotEmpty(equListCode) ? statisticsDataUpdateService.getTechParams(equListCode, record) : new ArrayList<>();
     }

amos-boot-system-tzs/amos-boot-module-ymt/amos-boot-module-ymt-api/src/main/java/com/yeejoin/amos/boot/module/ymt/api/mapper/IdxBizJgUseInfoMapper.java

@@ -33,4 +33,6 @@ public interface IdxBizJgUseInfoMapper extends BaseMapper<IdxBizJgUseInfo> {
     Map<String,Object> queryDetail(@Param("record") String record);
 
     List<Map<String,Object>> queryDetailBatch(@Param("records") List<String> records);
+
+    List<String> selectAddDataRecords();
 }

amos-boot-system-tzs/amos-boot-module-ymt/amos-boot-module-ymt-api/src/main/resources/mapper/IdxBizJgUseInfoMapper.xml

@@ -148,5 +148,8 @@
                 #{record}
             </foreach>
     </select>
+    <select id="selectAddDataRecords" resultType="java.lang.String">
+        select "RECORD" from amos_tzs_biz.idx_biz_jg_supervision_info where "ORG_BRANCH_CODE" = '50*X'
+    </select>
 
 </mapper>