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

@@ -436,7 +436,7 @@ public class DataHandlerController extends BaseController {
     @TycloudOperation(ApiLevel = UserType.AGENCY)
     @PutMapping(value = "/sync/db2es/equipments")
     @ApiOperation(httpMethod = "PUT", value = "2.同步设备数据到es新旧索引", notes = "同步设备数据到es新旧索引")
-    public ResponseModel<Integer> synEquipFromDb2Es(@ApiParam(value = "设备种类code") @RequestParam String equListCode,
+    public ResponseModel<Integer> syncEquipFromDb2Es(@ApiParam(value = "设备种类code") @RequestParam String equListCode,
                                                 @ApiParam(value = "设备类别code") @RequestParam(required = false) String equCategoryCode,
                                                 @ApiParam(value = "属地code") @RequestParam String orgBranchCode) {
         return ResponseHelper.buildResponse(dataHandlerService.synEquipFromDb2Es(equListCode, equCategoryCode, orgBranchCode));

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/data/fix/patcher/FilterableBatchDataPatcher.java

@@ -24,28 +24,41 @@ public abstract class FilterableBatchDataPatcher implements HistoricalDataPatche
         IdxBizJgUseInfoServiceImpl useInfoService = applicationContext.getBean(IdxBizJgUseInfoServiceImpl.class);
         Integer maxVersion = useInfoService.getBaseMapper().selectMaxVersionWithParams(buildFilter(params));
         Integer nextVersion = maxVersion + 1;
-        List<String> refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionWithParams(nextVersion,buildFilter(params));
+        List<String> refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionWithParams(nextVersion, buildFilter(params));
         int patchSize = refreshRecords.size();
         while (!refreshRecords.isEmpty()) {
-            refreshRecords.parallelStream().forEach(record -> {
-                try {
-                    beforePatching(record);
-                    patchSingleRecord(record);
-                    afterPatching(record);
-                } catch (Exception e) {
-                    // 异常数据跳过
-                    log.error("数据修补失败,设备:{}", record, e);
-                }
-            });
-            useInfoService.getBaseMapper().updateVersionBatch(refreshRecords, nextVersion);
-            refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionWithParams(nextVersion,buildFilter(params));
-            patchSize = patchSize + refreshRecords.size();
+            try {
+                refreshRecords.parallelStream().forEach(record -> {
+                    try {
+                        beforePatching(record);
+                        patchSingleRecord(record);
+                        afterPatching(record);
+                    } catch (Exception e) {
+                        // 异常数据跳过
+                        log.error("单个方式数据修补失败,设备:{}", record, e);
+                    }
+                });
+                patchBatchRecord(refreshRecords);
+            } catch (Exception e) {
+                // 本批次异常数据跳过
+                log.error("数据修补失败,设备:{}", refreshRecords, e);
+            } finally {
+                StopWatch watch1 = new StopWatch();
+                watch1.start();
+                useInfoService.getBaseMapper().updateVersionBatch(refreshRecords, nextVersion);
+                watch1.stop();
+                log.info("版本号批量更新条数：「{}」, 耗时：「{}」", refreshRecords.size(), watch1.getTotalTimeSeconds());
+                refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionWithParams(nextVersion, buildFilter(params));
+                patchSize = patchSize + refreshRecords.size();
+            }
         }
         watch.stop();
-        log.info("数据修补完成，共处理{}条记录，耗时: {}秒", patchSize, watch.getTotalTimeSeconds());
+        log.warn("数据修补完成，共处理{}条记录，耗时: {}秒", patchSize, watch.getTotalTimeSeconds());
         return patchSize;
     }
 
+    protected abstract void patchBatchRecord(List<String> refreshRecords);
+
     protected abstract Map<String, Object> buildFilter(Map<String, Object> params);
 
     protected abstract void beforePatching(String record);

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/refresh/handler/EquipmentRefreshHandler.java

@@ -88,11 +88,11 @@ public class EquipmentRefreshHandler implements IDataRefreshHandler {
     }
 
 
-    private List<ESEquipmentInfo.TechParam> buildTechParamByEquList(String record, String equListCode) {
+    public List<ESEquipmentInfo.TechParam> buildTechParamByEquList(String record, String equListCode) {
         return StringUtils.isNotEmpty(equListCode) ? statisticsDataUpdateService.getTechParams(equListCode, record) : new ArrayList<>();
     }
 
-    private LocalDate getIssueDate(String useRegistrationCode) {
+    public LocalDate getIssueDate(String useRegistrationCode) {
         if (StringUtils.isEmpty(useRegistrationCode)) {
             return null;
         }

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

@@ -27,10 +27,7 @@ import com.yeejoin.amos.boot.module.common.api.dao.EsBaseEnterpriseInfoDao;
 import com.yeejoin.amos.boot.module.common.api.dao.EsEquipmentDao;
 import com.yeejoin.amos.boot.module.common.api.dao.EsUserInfoDao;
 import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
-import com.yeejoin.amos.boot.module.common.api.entity.ESEquipmentInfo;
-import com.yeejoin.amos.boot.module.common.api.entity.EsBaseEnterpriseInfo;
-import com.yeejoin.amos.boot.module.common.api.entity.EsUserInfo;
-import com.yeejoin.amos.boot.module.common.api.entity.TzsUserPermission;
+import com.yeejoin.amos.boot.module.common.api.entity.*;
 import com.yeejoin.amos.boot.module.common.api.enums.CylinderTypeEnum;
 import com.yeejoin.amos.boot.module.common.biz.refresh.cm.RefreshCmService;
 import com.yeejoin.amos.boot.module.common.biz.service.impl.EsSearchServiceImpl;
@@ -207,6 +204,8 @@ public class DataHandlerServiceImpl {
 
     private final IdxBizJgUseInfoMapper idxBizJgUseInfoMapper;
 
+    private final EsBulkService esBulkService;
+
     /**
      * 安装告知压力管道历史数据修复-详情中的设备列表修改为汇总表格式
      *
@@ -2783,6 +2782,7 @@ public class DataHandlerServiceImpl {
                 if (!esEquipmentCategoryDtos.isEmpty()) {
                     totalUpdate = esEquipmentCategoryDtos.size() + totalUpdate;
                     esEquipmentCategory.saveAll(esEquipmentCategoryDtos);
+                    esBulkService.bulkUpsert(IDX_BIZ_VIEW_JG_ALL, esEquipmentCategoryDtos.stream().map(e-> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
                     idxBizJgSupervisionInfoServiceImpl.getBaseMapper().updateOrgBranchCodeBatch(esEquipmentCategoryDtos);
                 }
             }

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/EsBulkService.java

 package com.yeejoin.amos.boot.module.jg.biz.service.impl;
 
-import com.alibaba.fastjson.JSON;
 import com.yeejoin.amos.boot.module.common.api.entity.EsEntity;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.elasticsearch.action.bulk.BulkRequest;
-import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.xcontent.XContentType;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
+import java.util.Map;
 
 
 @Component
@@ -22,15 +21,18 @@ public class EsBulkService {
 
     private final RestHighLevelClient restHighLevelClient;
 
+    private final ElasticsearchConverter converter; // 依赖注入
+
+
     public <T> void bulkUpsert(String index, List<EsEntity<T>> list) {
         BulkRequest request = new BulkRequest();
         list.forEach(item -> {
+            Map<String, Object> docData = converter.mapObject(item.getData());
             request.add(new UpdateRequest(index, item.getId())
-                    .doc(JSON.toJSONString(item.getData()), XContentType.JSON)
-                    .upsert(JSON.toJSONString(item.getData()), XContentType.JSON));
+                    .doc(docData)
+                    .upsert(docData));
         });
         try {
-//            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
             restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
         } catch (Exception e) {
             log.error("批量写入数据失败:{}", e.getMessage(), e);

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

@@ -79,7 +79,7 @@
             <if test="params.orgBranchCode != null and params.orgBranchCode != ''">
                 and s."ORG_BRANCH_CODE" like concat(#{params.orgBranchCode},'%')
             </if>
-            limit 10000
+            limit 1000
     </select>
     <select id="selectMaxVersion" resultType="java.lang.Integer">
         SELECT
@@ -120,9 +120,16 @@
         WHERE ibjui."RECORD" = #{record}
     </select>
     <update id="updateVersionBatch">
-        <foreach collection="records" separator=";" item="record" open="" close="">
-            UPDATE idx_biz_jg_use_info SET "VERSION"=#{version} WHERE record = #{record}
-        </foreach>
+        UPDATE
+            idx_biz_jg_use_info
+        SET
+            "VERSION"=#{version}
+        WHERE
+        record = ANY(ARRAY[
+            <foreach collection="records" separator="," item="record" open="" close="">
+                #{record}
+            </foreach>
+        ])
     </update>
 
      <sql id="equip-detail-es">