refactor(amos-boot-module-tcm): 重构企业信息更新逻辑- 新增更新企业信息和设备信息的方法

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

@@ -19,8 +19,12 @@ import com.yeejoin.amos.boot.biz.common.utils.DateUtils;
 import com.yeejoin.amos.boot.biz.common.utils.RedisKey;
 import com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant;
 import com.yeejoin.amos.boot.module.common.api.dao.ESEquipmentCategory;
+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.dto.ESEquipmentCategoryDto;
 import com.yeejoin.amos.boot.module.common.api.dto.UserPermissionDto;
+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.enums.UnitDataSourceEnum;
 import com.yeejoin.amos.boot.module.common.biz.event.CommonPublisher;
 import com.yeejoin.amos.boot.module.common.biz.refresh.DataRefreshEvent;
@@ -74,6 +78,8 @@ import java.nio.charset.StandardCharsets;
 import java.text.ParseException;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -137,6 +143,12 @@ public class TzBaseEnterpriseInfoServiceImpl
 	@Autowired
 	private ESEquipmentCategory esEquipmentCategory;
 
+	@Autowired
+	private EsEquipmentDao esEquipmentDao;
+
+	@Autowired
+	private EsBaseEnterpriseInfoDao esBaseEnterpriseInfoDao;
+
     private static final Map<String ,String> JYJC_CERT_MAP = new HashMap<>();
 
 	@Autowired
@@ -1471,11 +1483,15 @@ public class TzBaseEnterpriseInfoServiceImpl
 				log.info("end-查询旧属地监管部门orgCode对应的所有设备equipmentRecordList：{}，companySeqList：{}", equipmentRecordList.size(), companySeqList.size());
 				// 4. 更新idx_biz_view_jg_all es设备信息
 				updateEquipmentJgAllEs(newOrgCode, oldOrgCode, equipmentRecordList, newModel.getString("companyName"));
-				// 5. 发送数据刷新事件 - 内存分页处理
-				// 处理企业数据
-				publishDataRefreshEvents(companySeqList, DataRefreshEvent.DataType.enterprise, 1000);
-				// 处理设备数据
-				publishDataRefreshEvents(equipmentRecordList, DataRefreshEvent.DataType.equipment, 1000);
+				// 5. 更新idx_biz_enterprise_info es企业信息
+				updateEnterpriseInfoEs(newOrgCode, oldOrgCode, companySeqList);
+				// 6. 更新idx_biz_equipment_info es设备信息
+				updateEquipmentInfoEs(newOrgCode, oldOrgCode, equipmentRecordList);
+//				// 5. 发送数据刷新事件 - 内存分页处理
+//				// 处理企业数据
+//				publishDataRefreshEvents(companySeqList, DataRefreshEvent.DataType.enterprise, 1000);
+//				// 处理设备数据
+//				publishDataRefreshEvents(equipmentRecordList, DataRefreshEvent.DataType.equipment, 1000);
 			}
 		} catch (Exception e) {
 			willInfo("refreshCompanyOrgCode", dataResult, "orgCodeUpdate", e);
@@ -1484,33 +1500,59 @@ public class TzBaseEnterpriseInfoServiceImpl
 		log.info("刷新单位orgCode结束");
 	}
 
-	private void updateEquipmentJgAllEs(String newOrgCode, String oldOrgCode, List<String> equipmentRecordList, String supervisionName) {
-		log.info("开始更新idx_biz_view_jg_all es设备属地监管部门信息，新orgCode{}, 旧orgCode:{}", newOrgCode, oldOrgCode);
+	private void updateEquipmentInfoEs(String newOrgCode, String oldOrgCode, List<String> equipmentRecordList) {
+		log.info("开始更新idx_biz_equipment_info es设备属地监管部门信息，新orgCode{}, 旧orgCode:{}", newOrgCode, oldOrgCode);
 		if (!ValidationUtil.isEmpty(equipmentRecordList)) {
-			log.info("待更新equipmentRecordList数量：{}", equipmentRecordList.size());
-			int batchSize = 1000;
-			for (int i = 0; i < equipmentRecordList.size(); i += batchSize) {
-				int endIndex = Math.min(i + batchSize, equipmentRecordList.size());
-				List<String> subList = equipmentRecordList.subList(i, endIndex);
-
-				Iterable<ESEquipmentCategoryDto> equipmentCategoryIterator = esEquipmentCategory.findAllById(subList);
-				List<ESEquipmentCategoryDto> updatedCategories = new ArrayList<>();
-
-				equipmentCategoryIterator.forEach(equipmentCategory -> {
-					if (ValidationUtil.isEmpty(equipmentCategory.getOrgBranchCode())) {
-						equipmentCategory.setOrgBranchCode(newOrgCode);// 补偿旧数据es属地监管部门为空，则用数据库中设备属地监管部门更新
-					} else {
-						equipmentCategory.setOrgBranchCode(equipmentCategory.getOrgBranchCode().replace(oldOrgCode, newOrgCode));
-					}
-					updatedCategories.add(equipmentCategory);
-				});
+			updateEsInfo(newOrgCode, oldOrgCode, equipmentRecordList, esEquipmentDao::findAllById, ESEquipmentInfo::setORG_BRANCH_CODE, ESEquipmentInfo::getORG_BRANCH_CODE, esEquipmentDao::saveAll);
+		}
+		log.info("更新idx_biz_equipment_info es设备属地监管部门信息结束");
+	}
 
-				if (!updatedCategories.isEmpty()) {
-					log.info("待更新updatedCategories数量：{}", updatedCategories.size());
-					esEquipmentCategory.saveAll(updatedCategories);
+	private void updateEnterpriseInfoEs(String newOrgCode, String oldOrgCode, List<String> companySeqList) {
+		log.info("开始更新idx_biz_enterprise_info es企业属地监管部门信息，新orgCode{}, 旧orgCode:{}", newOrgCode, oldOrgCode);
+		if (!ValidationUtil.isEmpty(companySeqList)) {
+			updateEsInfo(newOrgCode, oldOrgCode, companySeqList, esBaseEnterpriseInfoDao::findAllById, EsBaseEnterpriseInfo::setSuperviseOrgCode, EsBaseEnterpriseInfo::getSuperviseOrgCode, esBaseEnterpriseInfoDao::saveAll);
+		}
+		log.info("更新idx_biz_enterprise_info es设备属地监管部门信息结束");
+	}
+
+	private <T> void updateEsInfo(String newOrgCode, String oldOrgCode, List<String> recordList,
+								  Function<List<String>, Iterable<T>> findAllByIdFunc,
+								  BiConsumer<T, String> updateOrgCodeFunc,
+								  Function<T, String> getOrgCodeFunc,
+								  Consumer<List<T>> saveAllFunc) {
+		log.info("待更新es recordList数量：{}", recordList.size());
+		int batchSize = 1000;
+		for (int i = 0; i < recordList.size(); i += batchSize) {
+			int endIndex = Math.min(i + batchSize, recordList.size());
+			List<String> subList = recordList.subList(i, endIndex);
+
+			Iterable<T> recordIterator = findAllByIdFunc.apply(subList);
+			List<T> updatedRecord = new ArrayList<>();
+
+			recordIterator.forEach(record -> {
+				String equipOrgBranchCode = getOrgCodeFunc.apply(record);
+				if (ValidationUtil.isEmpty(equipOrgBranchCode)) {
+					updateOrgCodeFunc.accept(record, newOrgCode); // 补偿旧数据es属地监管部门为空，则用数据库中设备属地监管部门更新
+				} else {
+					String updatedOrgCode = equipOrgBranchCode.replace(oldOrgCode, newOrgCode);
+					updateOrgCodeFunc.accept(record, updatedOrgCode);
 				}
+				updatedRecord.add(record);
+			});
+
+			if (!updatedRecord.isEmpty()) {
+				log.info("待更新updatedRecord数量：{}", updatedRecord.size());
+				saveAllFunc.accept(updatedRecord);
 			}
 		}
+	}
+
+	private void updateEquipmentJgAllEs(String newOrgCode, String oldOrgCode, List<String> equipmentRecordList, String supervisionName) {
+		log.info("开始更新idx_biz_view_jg_all es设备属地监管部门信息，新orgCode{}, 旧orgCode:{}", newOrgCode, oldOrgCode);
+		if (!ValidationUtil.isEmpty(equipmentRecordList)) {
+			updateEsInfo(newOrgCode, oldOrgCode, equipmentRecordList, esEquipmentCategory::findAllById, ESEquipmentCategoryDto::setOrgBranchCode, ESEquipmentCategoryDto::getOrgBranchCode, esEquipmentCategory::saveAll);
+		}
 		// 根据旧属地监管部门orgCode查找es对应设备，更新数据库中设备属地字段
 		List<ESEquipmentCategoryDto> esEquipmentCategoryList = esEquipmentCategory.findByOrgBranchCodeKeywordStartingWith(oldOrgCode);
 		if (!ValidationUtil.isEmpty(esEquipmentCategoryList)) {