package com.yeejoin.amos.boot.module.jg.biz.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.yeejoin.amos.boot.biz.common.dao.mapper.DataDictionaryMapper;
import com.yeejoin.amos.boot.biz.common.dto.KeyValueDto;
import com.yeejoin.amos.boot.biz.common.entity.BaseEntity;
import com.yeejoin.amos.boot.biz.common.entity.DataDictionary;
import com.yeejoin.amos.boot.biz.common.entity.TzsBaseEntity;
import com.yeejoin.amos.boot.biz.common.utils.SnowflakeIdUtil;
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.dao.EsUserInfoDao;
import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
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.ESEquipmentInfoService;
import com.yeejoin.amos.boot.module.common.biz.service.impl.EquipmentCategoryService;
import com.yeejoin.amos.boot.module.common.biz.service.impl.EsSearchServiceImpl;
import com.yeejoin.amos.boot.module.common.biz.utils.RefreshDataUtils;
import com.yeejoin.amos.boot.module.jg.api.common.BizCommonConstant;
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.data.fix.service.FilterableEquipInsert2EsPatcher;
import com.yeejoin.amos.boot.module.jg.biz.data.fix.service.ReceiveOrgFixService;
import com.yeejoin.amos.boot.module.jg.biz.data.fix.service.WeatherTankFieldPatcherFull;
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.refresh.handler.EquipmentRefreshHandler;
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;
import com.yeejoin.amos.boot.module.ymt.api.entity.*;
import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentClassifityEnum;
import com.yeejoin.amos.boot.module.ymt.api.enums.EquipmentEnum;
import com.yeejoin.amos.boot.module.ymt.api.enums.FlowStatusEnum;
import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StopWatch;
import org.typroject.tyboot.core.foundation.context.RequestContext;
import org.typroject.tyboot.core.foundation.utils.ValidationUtil;

import java.io.IOException;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static com.alibaba.fastjson.JSON.toJSONString;
import static com.yeejoin.amos.boot.module.jg.biz.service.impl.JgInstallationNoticeServiceImpl.CONSTRUCTION_TYPE;
import static com.yeejoin.amos.boot.module.jg.biz.service.impl.JgInstallationNoticeServiceImpl.CONSTRUCTION_TYPE_NAME;

/**
 * 用于业务变更过程中的历史数据处理的service层
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataHandlerServiceImpl {
    public static final String IDX_BIZ_VIEW_JG_ALL = "idx_biz_view_jg_all";
    public static final String IDX_BIZ_EQUIPMENT_INFO = "idx_biz_equipment_info";
    public static final String STATUS = "STATUS";
    public static final String record = "record";
    public static final String SEQUENCE_NBR = "SEQUENCE_NBR";

    private static final int BATCH_SIZE = 1000;
    private final ObjectMapper objectMapper;
    private final JgRegistrationHistoryServiceImpl registrationHistoryService;
    private final JgInstallationNoticeServiceImpl installationNoticeService;
    private final JgUseRegistrationServiceImpl useRegistrationService;
    private final JgVehicleInformationServiceImpl jgVehicleInformationService;
    private final JgUseRegistrationEqServiceImpl useRegistrationEqService;
    private final JgInstallationNoticeEqServiceImpl installationNoticeEqService;
    private final IdxBizJgUseInfoServiceImpl useInfoService;
    private final SnowflakeIdUtil sequence;
    private final IdxBizJgProjectContraptionServiceImplService projectContraptionService;
    private final IdxBizJgProjectConstructionMapper projectConstructionMapper;
    private final IdxBizJgTechParamsPipelineServiceImpl techParamsPipelineService;
    private final EquipmentCategoryMapper equipmentCategoryMapper;
    private final IdxBizJgRegisterInfoServiceImpl registerInfoService;
    private final IdxBizJgSupervisionInfoServiceImpl supervisionInfoService;
    private final IdxBizJgFactoryInfoServiceImpl factoryInfoService;
    private final ESEquipmentCategory esEquipmentCategory;
    private final TzBaseEnterpriseInfoMapper tzBaseEnterpriseInfoMapper;
    private final JgInstallationNoticeEqMapper jgInstallationNoticeEqMapper;
    private final DataDictionaryMapper dataDictionaryMapper;
    private final IdxBizJgProjectInspectionMapper projectInspectionMapper;
    private final IdxBizJgInspectionDetectionInfoServiceImpl idxBizJgInspectionDetectionInfoService;
    private final IdxBizJgProjectContraptionMapper idxBizJgProjectContraptionMapper;
    private final JgInstallationNoticeMapper jgInstallationNoticeMapper;
    private final DataHandlerMapper dataHandlerMapper;
    private final CategoryOtherInfoMapper categoryOtherInfoMapper;
    private final CommonMapper commonMapper;
    private final ProblemHandleStrategy equipmentStrategy;
    private final SafetyProblemTracingServiceImpl safetyProblemTracingService;
    private final JgChangeRegistrationUnitServiceImpl jgChangeRegistrationUnitService;
    private final JgChangeRegistrationUnitEqServiceImpl jgChangeRegistrationUnitEqService;
    private final UseInfoMapper useInfoMapper;
    private final TzsServiceFeignClient tzsServiceFeignClient;
    private final JgChangeVehicleRegistrationUnitMapper jgChangeVehicleRegistrationUnitMapper;
    private final JgChangeRegistrationTransferMapper jgChangeRegistrationTransferMapper;
    private final JgUseRegistrationManageServiceImpl jgUseRegistrationManageServiceImpl;
    private final JgChangeRegistrationReformMapper jgChangeRegistrationReformMapper;

    private final JgReformNoticeMapper jgReformNoticeMapper;

    private final EsBaseEnterpriseInfoDao enterpriseInfoDao;

    private final TzBaseUnitLicenceMapper licenceMapper;

    private final TzsUserInfoMapper userInfoMapper;

    private final TzsUserPermissionMapper userPermissionMapper;

    private final EsUserInfoDao esUserInfoDao;

    private final EsEquipmentDao esEquipmentDao;

    private final EsSearchServiceImpl esSearchService;

    private final IdxBizJgMaintenanceRecordInfoServiceImpl maintenanceRecordInfoService;

    private static final String ROOT_ORG_CODE = "50";

    private final StatisticsDataUpdateService statisticsDataUpdateService;

    private final IdxBizJgRegisterInfoMapper registerInfoMapper;

    private final IdxBizJgSupervisionInfoServiceImpl idxBizJgSupervisionInfoServiceImpl;
    private final JgMaintainNoticeMapper jgMaintainNoticeMapper;
    private final JgMaintainNoticeEqMapper jgMaintainNoticeEqMapper;
    private final JgMaintainNoticeServiceImpl jgMaintainNoticeServiceImpl;

    @Value("${jyjc.open.online: true}")
    private Boolean onlineJyjc;

    private final JgUseRegistrationManageServiceImpl registrationManageService;

    private final SuperviseInfoMapper superviseInfoMapper;

    private final IdxBizJgOtherInfoMapper otherInfoMapper;

    private final RestHighLevelClient restHighLevelClient;

    private final WeatherTankFieldPatcherFull weatherTankFieldPatcher;

    private final FilterableEquipInsert2EsPatcher equipInsert2EsPatcher;

    private final EventPublisher eventPublisher;

    private final JgUseRegistrationManageMapper jgUseRegistrationManageMapper;

    private final CommonServiceImpl commonServiceImpl;

    private final ReceiveOrgFixService receiveOrgFixService;

    private final IdxBizJgUseInfoMapper idxBizJgUseInfoMapper;

    private final EsBulkService esBulkService;

    private final EquipmentRefreshHandler refreshHandler;

    private final IdxBizJgSupervisionInfoMapper idxBizJgSupervisionInfoMapper;

    private final IdxBizJgMaintenanceRecordInfoMapper maintenanceRecordInfoMapper;

    private final IdxBizJgInspectionDetectionInfoMapper inspectionDetectionInfoMapper;

    private final ESEquipmentInfoService esEquipmentInfoService;

    private final EquipmentCategoryService equipmentCategoryService;



    /**
     * 安装告知压力管道历史数据修复-详情中的设备列表修改为汇总表格式
     *
     * @return result
     */
    @Transactional(rollbackFor = Exception.class)
    public Boolean deviceListInFormWithInstallNotice() {
        List<JgInstallationNotice> noticeList = installationNoticeService.list(new LambdaQueryWrapper<JgInstallationNotice>().eq(JgInstallationNotice::getEquCategoryCode, "8300"));

        List<JgRegistrationHistory> jgRegistrationHistories = noticeList.stream()
                .map(notice -> {
                    JgRegistrationHistory history = getRegistrationHistory(String.valueOf(notice.getSequenceNbr()));
                    if (history == null)
                        return null;
                    try {
                        Map<String, Object> hisData = objectMapper.readValue(
                                history.getChangeData(),
                                new TypeReference<Map<String, Object>>() {
                                }
                        );
                        // 获取设备列表并处理每个设备信息
                        List<Map<String, Object>> deviceList = objectMapper.readValue(
                                objectMapper.writeValueAsString(hisData.get("deviceList")),
                                new TypeReference<ArrayList<Map<String, Object>>>() {
                                }
                        );
                        if (!ValidationUtil.isEmpty(deviceList)) {
                            List<Map<String, Object>> pipelineListInfo = installationNoticeService.getBaseMapper().getPipelineEquInfoByRecords(deviceList.stream().map(device -> String.valueOf(device.get("record"))).collect(Collectors.toList()));
                            // 更新处理后的数据回到 history 对象
                            hisData.put("deviceList", pipelineListInfo);
                        }
                        history.setChangeData(objectMapper.writeValueAsString(hisData)); // 更新 changeData 字段
                    } catch (Exception e) {
                        log.error("JSON 数据处理失败!", e);
                        throw new RuntimeException(e.getMessage());
                    }
                    return history;
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        // 批量更新历史表数据
        registrationHistoryService.updateBatchById(jgRegistrationHistories);
        return Boolean.TRUE;
    }

    // 辅助方法：根据 DocumentId 获取历史记录
    private JgRegistrationHistory getRegistrationHistory(String documentId) {
        return registrationHistoryService.getBaseMapper().selectOne(
                new QueryWrapper<JgRegistrationHistory>().lambda()
                        .eq(JgRegistrationHistory::getCurrentDocumentId, documentId)
                        .eq(JgRegistrationHistory::getIsDelete, false)
        );
    }

    /**
     * 向新增的工程装置表刷入数据
     *
     * @return result
     */
    @Transactional(rollbackFor = Exception.class)
    public Boolean writeData2ProjectContraption() {

        // projectContraptionService.getBaseMapper().delete(null);

        List<IdxBizJgUseInfo> collect = useInfoService.lambdaQuery()
                .select(IdxBizJgUseInfo::getRecord, IdxBizJgUseInfo::getProjectContraption, IdxBizJgUseInfo::getUseUnitCreditCode, IdxBizJgUseInfo::getUseUnitName, IdxBizJgUseInfo::getIsIntoManagement)
                .isNotNull(IdxBizJgUseInfo::getProjectContraption)
                .ne(IdxBizJgUseInfo::getProjectContraption, "")
                .groupBy(IdxBizJgUseInfo::getProjectContraption, IdxBizJgUseInfo::getUseUnitCreditCode)
                .list().stream()
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        log.info("查询到的工程装置集合: {}", toJSONString(collect));


        List<IdxBizJgProjectContraption> projectContraptionList = new ArrayList<>();
        Iterator<IdxBizJgUseInfo> iterator = collect.iterator();
        while (iterator.hasNext()) {
            IdxBizJgUseInfo useInfo = iterator.next();
            boolean isLast = !iterator.hasNext();

            String record = useInfo.getRecord();
            Boolean isIntoManagement = ObjectUtils.isEmpty(useInfo.getIsIntoManagement()) ? Boolean.FALSE : useInfo.getIsIntoManagement();
            String projectContraption = useInfo.getProjectContraption();
            String useUnitCreditCode = useInfo.getUseUnitCreditCode();

            Long sequenceNbr = sequence.nextId();

            List<String> installNotSeqs = installationNoticeEqService.lambdaQuery()
                    .eq(JgInstallationNoticeEq::getEquId, record)
                    .list().stream()
                    .filter(Objects::nonNull)
                    .map(JgInstallationNoticeEq::getEquipTransferId)
                    .collect(Collectors.toList());

            JgInstallationNotice installationNotice = installNotSeqs.isEmpty() ? null : installationNoticeService.lambdaQuery()
                    .in(JgInstallationNotice::getSequenceNbr, installNotSeqs)
                    .eq(JgInstallationNotice::getUseUnitCreditCode, useUnitCreditCode)
                    .eq(JgInstallationNotice::getNoticeStatus, FlowStatusEnum.TO_BE_FINISHED.getCode())
                    .list().stream()
                    .findFirst().orElse(null);

            List<String> useRegSeqs = useRegistrationEqService.lambdaQuery()
                    .eq(JgUseRegistrationEq::getEquId, record)
                    .list().stream()
                    .filter(Objects::nonNull)
                    .map(JgUseRegistrationEq::getEquipTransferId)
                    .collect(Collectors.toList());

            JgUseRegistration useRegistration = useRegSeqs.isEmpty() ? null : useRegistrationService.lambdaQuery()
                    .in(JgUseRegistration::getSequenceNbr, useRegSeqs)
                    .eq(JgUseRegistration::getUseUnitCreditCode, useUnitCreditCode)
                    .eq(JgUseRegistration::getStatus, FlowStatusEnum.TO_BE_FINISHED.getName())
                    .list().stream()
                    .findFirst().orElse(null);

            List<IdxBizJgUseInfo> useInfos = useInfoService.lambdaQuery()
                    .eq(IdxBizJgUseInfo::getProjectContraption, projectContraption)
                    .eq(IdxBizJgUseInfo::getUseUnitCreditCode, useUnitCreditCode)
                    .list();

            List<IdxBizJgTechParamsPipeline> idxBizJgTechParamsPipelines = useInfos.isEmpty() ? new ArrayList<>() : techParamsPipelineService
                    .lambdaQuery()
                    .in(IdxBizJgTechParamsPipeline::getRecord, useInfos.stream()
                            .map(IdxBizJgUseInfo::getRecord)
                            .collect(Collectors.toList()))
                    .list();

            double pipeLengthSum = idxBizJgTechParamsPipelines.stream()
                    .filter(Objects::nonNull)
                    .filter(pip -> Objects.nonNull(pip.getPipeLength()))
                    .mapToDouble(pipeline -> Double.parseDouble(pipeline.getPipeLength()))
                    .sum();

            IdxBizJgRegisterInfo registerInfo = registerInfoService.lambdaQuery()
                    .eq(IdxBizJgRegisterInfo::getRecord, useInfos
                            .stream().findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgSupervisionInfo supervisionInfo = supervisionInfoService.lambdaQuery()
                    .eq(IdxBizJgSupervisionInfo::getRecord, useInfos
                            .stream()
                            .findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgFactoryInfo factoryInfo = factoryInfoService.lambdaQuery()
                    .eq(IdxBizJgFactoryInfo::getRecord, useInfos
                            .stream()
                            .findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgProjectContraption idxBizJgProjectContraption = IdxBizJgProjectContraption.builder()
                    .projectContraption(projectContraption)
                    .projectContraptionNo(projectContraption)
                    .useUnitCreditCode(isIntoManagement ? (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getUseUnitCreditCode()) : null)
                    .useUnitName(isIntoManagement ? (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getUseUnitName()) : null)
                    .uscUnitCreditCode(isIntoManagement ? (ObjectUtils.isEmpty(useRegistration) ? (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getInstallUnitCreditCode()) : null) : null)
                    .uscUnitName(isIntoManagement ? (ObjectUtils.isEmpty(useRegistration) ? (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getInstallUnitName()) : null) : null)
                    .equList(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquListCode())
                    .equListName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquList())
                    .equCategory(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode())
                    .equCategoryName(ObjectUtils.isEmpty(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode()))).getName())
                    .equDefine(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine())
                    .equDefineName(ObjectUtils.isEmpty(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, (ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine()))).getName())
                    .content(null)
                    .pipelineLength(pipeLengthSum)
                    .productPhoto(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getProductPhoto())
                    .otherAccessories(null)
                    .orgCode(ObjectUtils.isEmpty(supervisionInfo) ? null : supervisionInfo.getOrgBranchCode())
                    .orgName(ObjectUtils.isEmpty(supervisionInfo) ? null : supervisionInfo.getOrgBranchName())
                    .productQualificationCertificate(ObjectUtils.isEmpty(factoryInfo) ? null : factoryInfo.getProductQualityYieldProve())
                    .province(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getProvince())
                    .provinceName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getProvinceName())
                    .city(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCity())
                    .cityName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCityName())
                    .county(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCounty())
                    .countyName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCountyName())
                    .street(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getStreet())
                    .streetName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getStreetName())
                    .address(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getAddress())
                    .startLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getStartePosition())
                    .endLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getEndPosition())
                    .supervisoryCode(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getSupervisoryCode())
                    .useRegistrationCode(ObjectUtils.isEmpty(useRegistration) ? null : useRegistration.getUseRegistrationCode())
                    .isIntoManagement(isIntoManagement)
                    .dataSource("jg")
                    .build();
            idxBizJgProjectContraption.setSequenceNbr(sequenceNbr);
            idxBizJgProjectContraption.setRecDate(new Date());
            idxBizJgProjectContraption.setIsDelete(Boolean.FALSE);
            projectContraptionList.add(idxBizJgProjectContraption);
            if (projectContraptionList.size() % 1000 == 0 || isLast) {
                projectContraptionService.saveBatch(projectContraptionList);
                projectContraptionList.clear();
            }

            useInfoService.lambdaUpdate()
                    .set(IdxBizJgUseInfo::getProjectContraptionId, sequenceNbr)
                    .eq(IdxBizJgUseInfo::getProjectContraption, projectContraption)
                    .eq(IdxBizJgUseInfo::getUseUnitCreditCode, useUnitCreditCode)
                    .update();

            if (!installNotSeqs.isEmpty()) {
                installationNoticeService.lambdaUpdate()
                        .set(JgInstallationNotice::getProjectContraptionId, sequenceNbr)
                        .in(JgInstallationNotice::getSequenceNbr, installNotSeqs)
                        .eq(JgInstallationNotice::getProjectContraption, projectContraption)
                        .eq(JgInstallationNotice::getUseUnitCreditCode, useUnitCreditCode)
                        .update();
            }

            if (!useRegSeqs.isEmpty()) {
                useRegistrationService.lambdaUpdate()
                        .set(JgUseRegistration::getProjectContraptionId, sequenceNbr)
                        .in(JgUseRegistration::getSequenceNbr, useRegSeqs)
                        .eq(JgUseRegistration::getProjectContraption, projectContraption)
                        .eq(JgUseRegistration::getUseUnitCreditCode, useUnitCreditCode)
                        .update();
            }

        }
        return Boolean.TRUE;
    }

    /**
     * 刷入工程装置id到es的idx_biz_view_jg_all索引
     *
     * @return result
     */
    public String writeProjectContraptionId() {
        AtomicReference<Long> size = new AtomicReference<>(0L);
        StopWatch stopWatch = new StopWatch("writeProjectContraptionId");
        stopWatch.start();
        List<IdxBizJgUseInfo> useInfos = useInfoService.lambdaQuery()
                .select(IdxBizJgUseInfo::getRecord, IdxBizJgUseInfo::getProjectContraptionId)
                .isNotNull(IdxBizJgUseInfo::getProjectContraptionId)
                .list();
        for (IdxBizJgUseInfo useInfo : useInfos) {
            Optional<ESEquipmentCategoryDto> equInfosWithEs = esEquipmentCategory.findById(useInfo.getRecord());
            equInfosWithEs.ifPresent(equInfoEs -> {
                if (PipelineEnum.INDUSTRIAL_PIPELINE.getCode().equals(equInfoEs.getEQU_CATEGORY_CODE())) {
                    equInfoEs.setProjectContraptionId(useInfo.getProjectContraptionId());
                    esEquipmentCategory.save(equInfoEs);
                    size.getAndSet(size.get() + 1);
                }
            });
        }
        stopWatch.stop();
        return String.format("接口调用成功，更新数据：%s条，耗时：%s s", size.get(), stopWatch.getTotalTimeSeconds());
    }

    public Boolean writeDataUseCodeIsNull2ProjectContraption() {
        List<IdxBizJgUseInfo> collect = useInfoService.lambdaQuery()
                .select(IdxBizJgUseInfo::getRecord, IdxBizJgUseInfo::getProjectContraption, IdxBizJgUseInfo::getUseUnitCreditCode, IdxBizJgUseInfo::getUseUnitName, IdxBizJgUseInfo::getIsIntoManagement)
                .isNotNull(IdxBizJgUseInfo::getProjectContraption)
                .ne(IdxBizJgUseInfo::getProjectContraption, "")
                .isNull(IdxBizJgUseInfo::getUseUnitCreditCode)
                .groupBy(IdxBizJgUseInfo::getProjectContraption)
                .list().stream()
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        log.info("查询到的工程装置集合: {},总数：{}", toJSONString(collect), collect.size());

        List<IdxBizJgProjectContraption> projectContraptionList = new ArrayList<>();
        Iterator<IdxBizJgUseInfo> iterator = collect.iterator();
        while (iterator.hasNext()) {
            IdxBizJgUseInfo useInfo = iterator.next();
            boolean isLast = !iterator.hasNext();

            String record = useInfo.getRecord();
            Boolean isIntoManagement = ObjectUtils.isEmpty(useInfo.getIsIntoManagement()) ? Boolean.FALSE : useInfo.getIsIntoManagement();
            String projectContraption = useInfo.getProjectContraption();
            Long sequenceNbr = sequence.nextId();

            List<String> installNotSeqs = installationNoticeEqService.lambdaQuery()
                    .eq(JgInstallationNoticeEq::getEquId, record)
                    .list().stream()
                    .filter(Objects::nonNull)
                    .map(JgInstallationNoticeEq::getEquipTransferId)
                    .collect(Collectors.toList());

            JgInstallationNotice installationNotice = installNotSeqs.isEmpty() ? null : installationNoticeService.lambdaQuery()
                    .in(JgInstallationNotice::getSequenceNbr, installNotSeqs)
                    .list().stream()
                    .findFirst().orElse(null);

            List<IdxBizJgUseInfo> useInfos = useInfoService.lambdaQuery()
                    .eq(IdxBizJgUseInfo::getProjectContraption, projectContraption)
                    .list();

            // 技术参数表
            List<IdxBizJgTechParamsPipeline> idxBizJgTechParamsPipelines = useInfos.isEmpty() ? new ArrayList<>() : techParamsPipelineService
                    .lambdaQuery()
                    .in(IdxBizJgTechParamsPipeline::getRecord, useInfos.stream()
                            .map(IdxBizJgUseInfo::getRecord)
                            .collect(Collectors.toList()))
                    .list();

            double pipeLengthSum = idxBizJgTechParamsPipelines.stream()
                    .filter(Objects::nonNull)
                    .filter(pip -> Objects.nonNull(pip.getPipeLength()))
                    .mapToDouble(pipeline -> Double.parseDouble(pipeline.getPipeLength()))
                    .sum();

            IdxBizJgRegisterInfo registerInfo = registerInfoService.lambdaQuery()
                    .eq(IdxBizJgRegisterInfo::getRecord, useInfos
                            .stream().findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgSupervisionInfo supervisionInfo = supervisionInfoService.lambdaQuery()
                    .eq(IdxBizJgSupervisionInfo::getRecord, useInfos
                            .stream()
                            .findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgFactoryInfo factoryInfo = factoryInfoService.lambdaQuery()
                    .eq(IdxBizJgFactoryInfo::getRecord, useInfos
                            .stream()
                            .findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgProjectContraption idxBizJgProjectContraption = IdxBizJgProjectContraption.builder()
                    .projectContraption(projectContraption)
                    .projectContraptionNo(projectContraption)
                    .uscUnitCreditCode(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getInstallUnitCreditCode())
                    .uscUnitName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getInstallUnitName())
                    .equList(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquListCode())
                    .equListName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquList())
                    .equCategory(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode())
                    .equCategoryName(ObjectUtils.isEmpty(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, (ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getEquCategoryCode()))).getName())
                    .equDefine(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine())
                    .equDefineName(ObjectUtils.isEmpty(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, (ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine()))).getName())
                    .content(null)
                    .pipelineLength(pipeLengthSum)
                    .productPhoto(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getProductPhoto())
                    .otherAccessories(null)
                    .orgCode(ObjectUtils.isEmpty(supervisionInfo) ? null : supervisionInfo.getOrgBranchCode())
                    .orgName(ObjectUtils.isEmpty(supervisionInfo) ? null : supervisionInfo.getOrgBranchName())
                    .productQualificationCertificate(ObjectUtils.isEmpty(factoryInfo) ? null : factoryInfo.getProductQualityYieldProve())
                    .province(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getProvince())
                    .provinceName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getProvinceName())
                    .city(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCity())
                    .cityName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCityName())
                    .county(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCounty())
                    .countyName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getCountyName())
                    .street(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getStreet())
                    .streetName(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getStreetName())
                    .address(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getAddress())
                    .startLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getStartePosition())
                    .endLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getEndPosition())
                    .supervisoryCode(ObjectUtils.isEmpty(installationNotice) ? null : installationNotice.getSupervisoryCode())
                    .isIntoManagement(isIntoManagement)
                    .dataSource("jg")
                    .build();
            idxBizJgProjectContraption.setSequenceNbr(sequenceNbr);
            idxBizJgProjectContraption.setRecDate(new Date());
            idxBizJgProjectContraption.setIsDelete(Boolean.FALSE);
            projectContraptionList.add(idxBizJgProjectContraption);
            if (projectContraptionList.size() % 1000 == 0 || isLast) {
                projectContraptionService.saveBatch(projectContraptionList);
                projectContraptionList.clear();
            }

            useInfoService.lambdaUpdate()
                    .set(IdxBizJgUseInfo::getProjectContraptionId, sequenceNbr)
                    .eq(IdxBizJgUseInfo::getProjectContraption, projectContraption)
                    .update();

            if (!installNotSeqs.isEmpty()) {
                installationNoticeService.lambdaUpdate()
                        .set(JgInstallationNotice::getProjectContraptionId, sequenceNbr)
                        .in(JgInstallationNotice::getSequenceNbr, installNotSeqs)
                        .eq(JgInstallationNotice::getProjectContraption, projectContraption)
                        .update();
            }
        }
        return Boolean.TRUE;
    }

    public void writeSupervisionOrgCode2UseRegister() {
        // 使用历史表补充属地监管部门字段
        filledDataFromHistory();
        // 查询历史登记且单位办理的数据进行刷库（历史气瓶导入，manage_type=unit and reg_type=1）
        filledHisRegCylinderByUnit();
    }

    private void filledHisRegCylinderByUnit() {
        List<JgUseRegistration> useRegistrations = useRegistrationService.list(new LambdaQueryWrapper<JgUseRegistration>().isNull(JgUseRegistration::getSupervisionOrgCode).eq(JgUseRegistration::getManageType, "unit").eq(JgUseRegistration::getRegType, "1").select(BaseEntity::getSequenceNbr, JgUseRegistration::getUseUnitCreditCode));
        Map<String, String> unitSuperOrgCodeMap = new ConcurrentHashMap<>();
        useRegistrations.parallelStream().forEach(u -> {
            LambdaUpdateWrapper<JgUseRegistration> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(JgUseRegistration::getSequenceNbr, u.getSequenceNbr());
            if (unitSuperOrgCodeMap.get(u.getUseUnitCreditCode()) != null) {
                updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, unitSuperOrgCodeMap.get(u.getUseUnitCreditCode()));
            } else {
                TzBaseEnterpriseInfo baseEnterpriseInfo = tzBaseEnterpriseInfoMapper.selectByUseUnitCode(u.getUseUnitCreditCode());
                if (baseEnterpriseInfo != null) {
                    unitSuperOrgCodeMap.put(u.getUseUnitCreditCode(), baseEnterpriseInfo.getSuperviseOrgCode());
                    updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, baseEnterpriseInfo.getSuperviseOrgCode());
                } else {
                    updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, null);
                }
            }
            useRegistrationService.update(null, updateWrapper);
        });
    }

    private void filledDataFromHistory() {
        List<JgUseRegistration> useRegistrations = useRegistrationService.list(new LambdaQueryWrapper<JgUseRegistration>().isNull(JgUseRegistration::getSupervisionOrgCode).select(BaseEntity::getSequenceNbr));
        useRegistrations.parallelStream().forEach(u -> {
            JgRegistrationHistory history = getRegistrationHistory(u.getSequenceNbr() + "");
            if (history != null) {
                JSONObject jsonObject = JSON.parseObject(history.getChangeData());
                String orgBranchCodeWithName = jsonObject.getString("orgBranchCode");
                if (orgBranchCodeWithName != null && !orgBranchCodeWithName.isEmpty()) {
                    u.setSupervisionOrgCode(orgBranchCodeWithName.split("_")[0]);
                    LambdaUpdateWrapper<JgUseRegistration> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, u.getSupervisionOrgCode());
                    updateWrapper.eq(JgUseRegistration::getSequenceNbr, u.getSequenceNbr());
                    useRegistrationService.update(null, updateWrapper);
                }
            }
        });
    }

    @Transactional(rollbackFor = Exception.class)
    public void projectContraptionNoticeDataDeal() {
        // 1.安装告知完成但是设备状态未非已纳管状态：设备的idx_biz_jg_use_info表的IS_INTO_MANAGEMENT刷为已纳管, es的IS_INTO_MANAGEMENT刷为已纳管
        List<JgInstallationNoticeEq> eqs = jgInstallationNoticeEqMapper.selectErrorRecordOfInIntoManage();
        log.info("处理前统计：1.安装告知完成但是设备状态未非已纳管状态的设备数量：{}", eqs.size());
        for (JgInstallationNoticeEq eq : eqs) {
            // 使用信息表刷已纳管
            useInfoService.lambdaUpdate()
                    .set(IdxBizJgUseInfo::getIsIntoManagement, true)
                    .eq(IdxBizJgUseInfo::getRecord, eq.getEquId())
                    .update();
            // es刷已纳管
            Optional<ESEquipmentCategoryDto> equInfosWithEs = esEquipmentCategory.findById(eq.getEquId());
            equInfosWithEs.ifPresent(equInfoEs -> {
                equInfoEs.setIS_INTO_MANAGEMENT(true);
                esEquipmentCategory.save(equInfoEs);
            });
        }
        // 2.安装告知非已作废，但是缺少装置id的数据处理：创建装置、写入安装告知表project_contraption_id、写入设备es的装置id，设备使用信息表的projectContraptionId
        List<JgInstallationNotice> noticeList = getErrorInstallNoticeData();
        log.info("处理前统计：2.安装告知非已作废，但是缺少装置id安装告知单据数量：{}", noticeList.size());
        List<IdxBizJgProjectContraption> projectContraptionList = new ArrayList<>();
        Map<Long, List<String>> projectContraptionIdRecordsMap = new HashMap<>();
        for (JgInstallationNotice installationNotice : noticeList) {
            List<JgInstallationNoticeEq> noticeEqs = installationNoticeEqService.list(new LambdaQueryWrapper<JgInstallationNoticeEq>().eq(JgInstallationNoticeEq::getEquipTransferId, installationNotice.getSequenceNbr()).select(JgInstallationNoticeEq::getEquId, JgInstallationNoticeEq::getInformNumber));
            if (noticeEqs.isEmpty()) {
                continue;
            }
            List<String> records = noticeEqs.stream().map(JgInstallationNoticeEq::getEquId).collect(Collectors.toList());
            List<IdxBizJgUseInfo> useInfos = useInfoService.lambdaQuery().in(IdxBizJgUseInfo::getRecord, records).list();
            // 技术参数表
            List<IdxBizJgTechParamsPipeline> idxBizJgTechParamsPipelines = useInfos.isEmpty() ? new ArrayList<>() : techParamsPipelineService
                    .lambdaQuery()
                    .in(IdxBizJgTechParamsPipeline::getRecord, records)
                    .list();
            double pipeLengthSum = idxBizJgTechParamsPipelines.stream()
                    .filter(Objects::nonNull)
                    .filter(pip -> Objects.nonNull(pip.getPipeLength()))
                    .mapToDouble(pipeline -> Double.parseDouble(pipeline.getPipeLength()))
                    .sum();
            BigDecimal pipeLengthSumBig = new BigDecimal(String.valueOf(pipeLengthSum)).setScale(3, RoundingMode.HALF_UP);

            IdxBizJgRegisterInfo registerInfo = registerInfoService.lambdaQuery()
                    .eq(IdxBizJgRegisterInfo::getRecord, useInfos
                            .stream().findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();

            IdxBizJgFactoryInfo factoryInfo = factoryInfoService.lambdaQuery()
                    .eq(IdxBizJgFactoryInfo::getRecord, useInfos
                            .stream()
                            .findFirst().orElse(new IdxBizJgUseInfo())
                            .getRecord())
                    .one();
            Long sequenceNbr = sequence.nextId();
            installationNotice.setProjectContraptionId(sequenceNbr + "");
            // key为装置id,value为此装置下的管道数组
            projectContraptionIdRecordsMap.put(sequenceNbr, records);
            boolean isFinish = String.valueOf(FlowStatusEnum.TO_BE_FINISHED.getCode()).equals(installationNotice.getNoticeStatus());
            IdxBizJgProjectContraption idxBizJgProjectContraption = IdxBizJgProjectContraption.builder()
                    .projectContraption(installationNotice.getProjectContraption())
                    .projectContraptionNo(installationNotice.getProjectContraption())
                    .useUnitCreditCode(isFinish ? installationNotice.getUseUnitCreditCode() : null)
                    .useUnitName(isFinish ? installationNotice.getUseUnitName() : null)
                    .uscUnitCreditCode(installationNotice.getInstallUnitCreditCode())
                    .uscUnitName(installationNotice.getInstallUnitName())
                    .equList(installationNotice.getEquListCode())
                    .equListName(installationNotice.getEquList())
                    .equCategory(installationNotice.getEquCategoryCode())
                    .equCategoryName(ObjectUtils.isEmpty(installationNotice.getEquCategoryCode()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, installationNotice.getEquCategoryCode())).getName())
                    .equDefine(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine())
                    .equDefineName(ObjectUtils.isEmpty(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getEquDefine()) ? null : equipmentCategoryMapper.selectOne(new LambdaQueryWrapper<EquipmentCategory>().eq(EquipmentCategory::getCode, registerInfo.getEquDefine())).getName())
                    .content(null)
                    .pipelineLength(pipeLengthSumBig.doubleValue())
                    .productPhoto(ObjectUtils.isEmpty(registerInfo) ? null : registerInfo.getProductPhoto())
                    .productQualificationCertificate(ObjectUtils.isEmpty(factoryInfo) ? null : factoryInfo.getProductQualityYieldProve())
                    .otherAccessories(installationNotice.getOtherAccessories())
                    .orgCode(installationNotice.getOrgBranchCode())
                    .orgName(installationNotice.getOrgBranchName())
                    .province(installationNotice.getProvince())
                    .provinceName(installationNotice.getProvinceName())
                    .city(installationNotice.getCity())
                    .cityName(installationNotice.getCityName())
                    .county(installationNotice.getCounty())
                    .countyName(installationNotice.getCountyName())
                    .street(installationNotice.getFactoryUseSiteStreet())
                    .streetName(installationNotice.getStreetName())
                    .address(installationNotice.getAddress())
                    .startLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getStartePosition())
                    .endLatitudeLongitude(ObjectUtils.isEmpty(idxBizJgTechParamsPipelines) ? null : idxBizJgTechParamsPipelines.get(0).getEndPosition())
                    .supervisoryCode(installationNotice.getSupervisoryCode())
                    .isIntoManagement(isFinish)
                    .dataSource("jg")
                    .build();
            idxBizJgProjectContraption.setSequenceNbr(sequenceNbr);
            idxBizJgProjectContraption.setRecDate(new Date());
            idxBizJgProjectContraption.setIsDelete(Boolean.FALSE);
            // 安装告知完成后，创建装置的施工信息表
            saveProjectConstructionAfterFinish(installationNotice, isFinish, noticeEqs, sequenceNbr, records);
            // 安装告知表写入装置id
            updateProjectContraptionId2Notice(installationNotice, idxBizJgProjectContraption);
            projectContraptionList.add(idxBizJgProjectContraption);
        }
        // 创建装置
        if (!projectContraptionList.isEmpty()) {
            projectContraptionService.saveBatch(projectContraptionList);
        }
        // 更新设备es的装置id、使用信息的装置id
        projectContraptionIdRecordsMap.forEach((k, v) -> {
            IdxBizJgProjectContraption idxBizJgProjectContraption = projectContraptionService.getById(k);
            Iterable<ESEquipmentCategoryDto> equipmentCategories = esEquipmentCategory.findAllById(v);
            equipmentCategories.forEach(e -> e.setProjectContraptionId(k + ""));
            esEquipmentCategory.saveAll(equipmentCategories);
            Collection<IdxBizJgUseInfo> useInfos = useInfoService.getUseInfoListByEquIds(v);
            useInfos.forEach(u -> {
                LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.eq(IdxBizJgUseInfo::getRecord, u.getRecord());
                updateWrapper.set(IdxBizJgUseInfo::getProjectContraptionId, k);
                updateWrapper.set(IdxBizJgUseInfo::getProjectContraption, idxBizJgProjectContraption.getProjectContraption());
                useInfoService.update(updateWrapper);
            });
        });
        List<JgInstallationNoticeEq> eqsAfter = jgInstallationNoticeEqMapper.selectErrorRecordOfInIntoManage();
        log.info("处理后统计：3.安装告知完成但是设备状态未非已纳管状态的设备数量：{}", eqsAfter.size());
        List<JgInstallationNotice> noticeListAfter = getErrorInstallNoticeData();
        log.info("处理后统计：4.安装告知非已作废，但是缺少装置id安装告知单据数量：{}", noticeListAfter.size());
    }

    private void updateProjectContraptionId2Notice(JgInstallationNotice installationNotice, IdxBizJgProjectContraption idxBizJgProjectContraption) {
        LambdaUpdateWrapper<JgInstallationNotice> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.set(JgInstallationNotice::getProjectContraptionId, installationNotice.getProjectContraptionId());
        updateWrapper.eq(BaseEntity::getSequenceNbr, installationNotice.getSequenceNbr());
        installationNoticeService.update(null, updateWrapper);
    }

    private void saveProjectConstructionAfterFinish(JgInstallationNotice installationNotice, boolean isFinish, List<JgInstallationNoticeEq> eqs, Long sequenceNbr, List<String> records) {
        if (isFinish) {
            IdxBizJgProjectConstruction idxBizJgProjectConstruction = new IdxBizJgProjectConstruction();
            idxBizJgProjectConstruction.setUscUnitCreditCode(installationNotice.getInstallUnitCreditCode());
            idxBizJgProjectConstruction.setUscUnitName(installationNotice.getInstallUnitName());
            idxBizJgProjectConstruction.setUscDate(installationNotice.getInstallStartDate());
            idxBizJgProjectConstruction.setRecord(String.valueOf(installationNotice.getSequenceNbr()));
            idxBizJgProjectConstruction.setSequenceCode(String.valueOf(installationNotice.getSequenceNbr()));
            idxBizJgProjectConstruction.setNoticeId(String.valueOf(installationNotice.getSequenceNbr()));
            idxBizJgProjectConstruction.setInformCode(eqs.stream().findFirst().orElse(new JgInstallationNoticeEq()).getInformNumber());
            idxBizJgProjectConstruction.setConstructionLeaderId(installationNotice.getInstallLeaderId());
            idxBizJgProjectConstruction.setConstructionLeaderName(installationNotice.getInstallLeaderName());
            idxBizJgProjectConstruction.setConstructionLeaderPhone(installationNotice.getInstallLeaderPhone());
            idxBizJgProjectConstruction.setAddress(installationNotice.getAddress());
            idxBizJgProjectConstruction.setProvinceName(installationNotice.getProvinceName());
            idxBizJgProjectConstruction.setCityName(installationNotice.getCityName());
            idxBizJgProjectConstruction.setCountyName(installationNotice.getCountyName());
            idxBizJgProjectConstruction.setStreetName(installationNotice.getStreetName());
            idxBizJgProjectConstruction.setIsXixian(installationNotice.getIsXixian());
            idxBizJgProjectConstruction.setProjectContraptionId(sequenceNbr + "");
            idxBizJgProjectConstruction.setEquipRecords(StringUtils.join(records, ","));
            idxBizJgProjectConstruction.setConstructionContractAttachment(installationNotice.getInstallContractAttachment());
            idxBizJgProjectConstruction.setProxyStatementAttachment(installationNotice.getProxyStatementAttachment());
            idxBizJgProjectConstruction.setConstructionOtherAccessories(installationNotice.getOtherAccessories());
            idxBizJgProjectConstruction.setRecDate(new Date());
            idxBizJgProjectConstruction.setRecUserId(RequestContext.getExeUserId());
            LambdaQueryWrapper<DataDictionary> lambda = new LambdaQueryWrapper<>();
            // 获取施工类型id
            lambda.eq(DataDictionary::getType, CONSTRUCTION_TYPE);
            lambda.eq(DataDictionary::getName, CONSTRUCTION_TYPE_NAME);
            List<DataDictionary> dataDictionaries = dataDictionaryMapper.selectList(lambda);
            if (!CollectionUtils.isEmpty(dataDictionaries)) {
                idxBizJgProjectConstruction.setConstructionType(String.valueOf(dataDictionaries.get(0).getSequenceNbr()));
            }
            projectConstructionMapper.insert(idxBizJgProjectConstruction);
        }
    }

    private List<JgInstallationNotice> getErrorInstallNoticeData() {
        return installationNoticeService.list(new LambdaQueryWrapper<JgInstallationNotice>().ne(JgInstallationNotice::getNoticeStatus, FlowStatusEnum.TO_BE_DISCARD.getCode()).ne(JgInstallationNotice::getProjectContraption, "").isNull(JgInstallationNotice::getProjectContraptionId));
    }

    /**
     * 刷使用登记缺少装置id的管道的登记数据：1.填充装置id；2.已完成状状态清除装置表的安装单位信息字段,补充使用信息字段
     */
    @Transactional(rollbackFor = Exception.class)
    public void projectContraptionUseRegisterDataDeal() {
        // 查询使用登记：非作废状态 有装置名称但是缺少装置id的使用登记单据进行数据处理
        List<JgUseRegistration> useRegistrations = getUseRegisterOfError();
        log.info("处理前统计：1.使用登记装置id历史数据填充，存在缺少装置id的单据数量：{}", useRegistrations.size());
        // 循环按照使用登记的eq表关联安装告知eq表，在安装告知主表查询装置id
        for (JgUseRegistration useRegistration : useRegistrations) {
            // 登记的关联的告知单据【逻辑按照eq进行关联】
            JgInstallationNotice installationNotice = installationNoticeService.getBaseMapper().selectNoticeDataByUseRegisterId(useRegistration.getSequenceNbr() + "");
            if (installationNotice != null && StringUtils.isNotEmpty(installationNotice.getProjectContraptionId())) {
                // 0.填充装置id
                useRegistration.setProjectContraptionId(installationNotice.getProjectContraptionId());
                // 1.已完成状状态清除装置表的安装单位信息字段,填充使用信息数据，创建装置检验信息表
                if (FlowStatusEnum.TO_BE_FINISHED.getName().equals(useRegistration.getStatus())) {
                    updateProjectContraptionAfter(useRegistration, installationNotice);
                    // 创建装置检验流水信息
                    createProjectInspectInfoAfter(useRegistration);
                }
                // 2.使用登记写入装置id
                LambdaUpdateWrapper<JgUseRegistration> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.set(JgUseRegistration::getProjectContraptionId, useRegistration.getProjectContraptionId());
                updateWrapper.eq(BaseEntity::getSequenceNbr, useRegistration.getSequenceNbr());
                useRegistrationService.update(null, updateWrapper);
            }
        }
        List<JgUseRegistration> useRegistrationsAfter = getUseRegisterOfError();
        log.info("处理后统计：2.使用登记装置id历史数据填充，存在缺少装置id的单据数量：{}", useRegistrationsAfter.size());
    }

    private void updateProjectContraptionAfter(JgUseRegistration useRegistration, JgInstallationNotice installationNotice) {
        // 特殊处理 将任意一个的 record 放入到remark字段
        IdxBizJgUseInfo useInfo = getIdxBizJgUseInfo(installationNotice.getRemark());
        if (useInfo != null) {
            LambdaUpdateWrapper<IdxBizJgProjectContraption> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(BaseEntity::getSequenceNbr, installationNotice.getProjectContraptionId());
            updateWrapper.set(IdxBizJgProjectContraption::getUscUnitCreditCode, null);
            updateWrapper.set(IdxBizJgProjectContraption::getUscUnitName, null);
            updateWrapper.set(IdxBizJgProjectContraption::getUseRegistrationCode, useRegistration.getUseRegistrationCode());
            updateWrapper.set(IdxBizJgProjectContraption::getProvince, useInfo.getProvince());
            updateWrapper.set(IdxBizJgProjectContraption::getProvinceName, useInfo.getProvinceName());
            updateWrapper.set(IdxBizJgProjectContraption::getCity, useInfo.getCity());
            updateWrapper.set(IdxBizJgProjectContraption::getCityName, useInfo.getCityName());
            updateWrapper.set(IdxBizJgProjectContraption::getCounty, useInfo.getCounty());
            updateWrapper.set(IdxBizJgProjectContraption::getCountyName, useInfo.getCountyName());
            updateWrapper.set(IdxBizJgProjectContraption::getStreet, useInfo.getFactoryUseSiteStreet());
            updateWrapper.set(IdxBizJgProjectContraption::getStreetName, useInfo.getStreetName());
            updateWrapper.set(IdxBizJgProjectContraption::getAddress, useInfo.getAddress());
            updateWrapper.set(IdxBizJgProjectContraption::getUseDate, useRegistration.getAuditPassDate());
            projectContraptionService.update(null, updateWrapper);
        }
    }

    private void createProjectInspectInfoAfter(JgUseRegistration useRegistration) {
        JgRegistrationHistory jgRegistrationHistory = getRegistrationHistory(useRegistration.getSequenceNbr() + "");
        if (jgRegistrationHistory != null) {
            JSONObject mapData = JSONObject.parseObject(jgRegistrationHistory.getChangeData());
            IdxBizJgProjectInspection inspectionDetectionInfo = new IdxBizJgProjectInspection();
            BeanUtil.copyProperties(mapData, inspectionDetectionInfo);
            // 兼容最早的老数据，管道的检验信息单独维护
            if (inspectionDetectionInfo.getInspectConclusion() == null && mapData.containsKey("equipmentLists")) {
                List<Map<String, Object>> equipmentLists = null;
                try {
                    equipmentLists = objectMapper.readValue(
                            objectMapper.writeValueAsString(mapData.get("equipmentLists")),
                            new TypeReference<ArrayList<Map<String, Object>>>() {
                            }
                    );
                    if (!equipmentLists.isEmpty()) {
                        Object record = equipmentLists.get(0).get("record");
                        IdxBizJgInspectionDetectionInfo idxBizJgInspectionDetectionInfo = idxBizJgInspectionDetectionInfoService.queryNewestDetailByRecord(String.valueOf(record));
                        BeanUtil.copyProperties(idxBizJgInspectionDetectionInfo, inspectionDetectionInfo);
                        inspectionDetectionInfo.setRecord(null);
                    }
                } catch (JsonProcessingException e) {
                    throw new RuntimeException(e);
                }
            }
            inspectionDetectionInfo.setProjectContraptionId(useRegistration.getProjectContraptionId());
            inspectionDetectionInfo.setRecDate(new Date());

            // 使用登记创建新增的检验信息id设置为使用登记的主键id,在作废时级联作废
            inspectionDetectionInfo.setSequenceNbr(useRegistration.getSequenceNbr() + "");
            projectInspectionMapper.insert(inspectionDetectionInfo);
        }

    }

    private IdxBizJgUseInfo getIdxBizJgUseInfo(String record) {
        return useInfoService.getOneData(record);
    }

    /**
     * 查询非作废状态的装置名称不为空，装置id为空或者null的使用登记数据
     *
     * @return 符合条件列表
     */
    private List<JgUseRegistration> getUseRegisterOfError() {
        return useRegistrationService.getBaseMapper().selectList(new LambdaQueryWrapper<JgUseRegistration>().ne(JgUseRegistration::getStatus, FlowStatusEnum.TO_BE_DISCARD.getName()).ne(JgUseRegistration::getProjectContraption, "").isNull(JgUseRegistration::getProjectContraptionId).select(BaseEntity::getSequenceNbr, JgUseRegistration::getProjectContraption, JgUseRegistration::getUseRegistrationCode, JgUseRegistration::getStatus, JgUseRegistration::getAuditPassDate));
    }

    /**
     * 处理管道-安装告知审批完成但是装置是否纳管状态不正确的数据
     */
    @Transactional(rollbackFor = Exception.class)
    public void projectContraptionInfoUpdateV2() {
        List<IdxBizJgProjectContraption> projectConstructions = idxBizJgProjectContraptionMapper.selectErrorManagementProject();
        log.info("处理前：安装告知审批完成但是装置是否纳管状态不正确的数据量：{}", projectConstructions.size());
        projectConstructions.parallelStream().forEach(p -> {
            LambdaUpdateWrapper<IdxBizJgProjectContraption> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(BaseEntity::getSequenceNbr, p.getSequenceNbr());
            updateWrapper.set(IdxBizJgProjectContraption::getUscUnitName, p.getUscUnitName());
            updateWrapper.set(IdxBizJgProjectContraption::getUscUnitCreditCode, p.getUscUnitCreditCode());
            updateWrapper.set(IdxBizJgProjectContraption::getUseUnitName, p.getUseUnitName());
            updateWrapper.set(IdxBizJgProjectContraption::getUseUnitCreditCode, p.getUseUnitCreditCode());
            updateWrapper.set(BaseEntity::getRecDate, p.getRecDate());
            updateWrapper.set(IdxBizJgProjectContraption::getIsIntoManagement, true);
            projectContraptionService.update(null, updateWrapper);
        });
        List<IdxBizJgProjectContraption> projectConstructionsAfter = idxBizJgProjectContraptionMapper.selectErrorManagementProject();
        log.info("处理后：安装告知审批完成但是装置是否纳管状态不正确的数据量：{}", projectConstructionsAfter.size());
    }

    @Transactional(rollbackFor = Exception.class)
    public void projectUseInfoUpdateV3() {
        List<JgInstallationNoticeEq> eqs = jgInstallationNoticeEqMapper.selectPCIdNeErrorPieLineInUseInfo();
        log.info("处理前：查询安装告知（非删除、已作废）的装置id和设备的使用信息装置id不一样的设备：{}", eqs.size());
        eqs.forEach(eq -> {
            Optional<ESEquipmentCategoryDto> equInfosWithEs = esEquipmentCategory.findById(eq.getEquId());
            equInfosWithEs.ifPresent(equInfoEs -> {
                if (StringUtils.isNotEmpty(equInfoEs.getProjectContraptionId())) {
                    IdxBizJgProjectContraption idxBizJgProjectContraption = projectContraptionService.getById(equInfoEs.getProjectContraptionId());
                    LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
                    updateWrapper.eq(IdxBizJgUseInfo::getRecord, equInfoEs.getSEQUENCE_NBR());
                    updateWrapper.set(IdxBizJgUseInfo::getProjectContraptionId, idxBizJgProjectContraption.getSequenceNbr());
                    updateWrapper.set(IdxBizJgUseInfo::getProjectContraption, idxBizJgProjectContraption.getProjectContraption());
                    useInfoService.update(updateWrapper);
                }
            });
        });
        List<JgInstallationNoticeEq> eqsAfter = jgInstallationNoticeEqMapper.selectPCIdNeErrorPieLineInUseInfo();
        log.info("处理后：查询安装告知（非删除、已作废）的装置id和设备的使用信息装置id不一样的设备：{}", eqsAfter.size());
    }

    @Transactional(rollbackFor = Exception.class)
    public void projectContraptionInfoUpdateV3() {
        // 1220上线时, 已完成安装告知管道未写入装置的施工信息表的告知单据查询
        List<JgInstallationNotice> noticeList = getNotInsertProjectConstructionInstallNotice();
        log.info("处理前统计：1.管道已完成安装告知，但是未写入装置的施工信息表的告知单据数量：{}", noticeList.size());
        for (JgInstallationNotice installationNotice : noticeList) {
            List<JgInstallationNoticeEq> noticeEqs = installationNoticeEqService.list(new LambdaQueryWrapper<JgInstallationNoticeEq>().eq(JgInstallationNoticeEq::getEquipTransferId, installationNotice.getSequenceNbr()).select(JgInstallationNoticeEq::getEquId));
            if (noticeEqs.isEmpty()) {
                continue;
            }
            List<String> records = noticeEqs.stream().map(JgInstallationNoticeEq::getEquId).collect(Collectors.toList());
            // 安装告知完成后，创建装置的施工信息表
            saveProjectConstructionAfterFinish(installationNotice, true, noticeEqs, Long.parseLong(installationNotice.getProjectContraptionId()), records);
        }
        List<JgInstallationNotice> noticeListAfter = getNotInsertProjectConstructionInstallNotice();
        log.info("处理后统计：1.管道已完成安装告知，但是未写入装置的施工信息表的告知单据数量：{}", noticeListAfter.size());

    }

    private List<JgInstallationNotice> getNotInsertProjectConstructionInstallNotice() {
        return jgInstallationNoticeMapper.selectNotInsertProjectConstructionInstallNoticeList();
    }

    /**
     * 之前刷库问题导致的管道的装置id与安装告知的装置id的不一样的数据刷库
     * 已安装告知为准，刷1.使用信息表的装置id；2.es的装置id。进行比对装置的安装信息与安装告知的信息安装信息对上才刷【使用登记审批完成后，装置表的安装单位信息回清空，故不包括这部分数据】】
     */
    @Transactional(rollbackFor = Exception.class)
    public void projectUseInfoUpdateV4() {
        List<PieLineEquipContraptionDto> eqs = jgInstallationNoticeEqMapper.selectPCIdNeErrorPieLineInUseInfo2();
        log.info("处理前：查询管道的装置id与安装告知的装置id的不一样管道设备：{}", eqs.size());
        eqs.forEach(eq -> {
            // 1.已安装告知单据为准更新es
            Optional<ESEquipmentCategoryDto> equInfosWithEs = esEquipmentCategory.findById(eq.getRecord());
            equInfosWithEs.ifPresent(equInfoEs -> {
                equInfoEs.setProjectContraptionId(eq.getProjectContraptionId());
                equInfoEs.setPROJECT_CONTRAPTION(eq.getProjectContraption());
                esEquipmentCategory.save(equInfoEs);
            });
            // 2.已安装告知单据为准更新use_info
            LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(IdxBizJgUseInfo::getRecord, eq.getRecord());
            updateWrapper.set(IdxBizJgUseInfo::getProjectContraptionId, eq.getProjectContraptionId());
            updateWrapper.set(IdxBizJgUseInfo::getProjectContraption, eq.getProjectContraption());
            useInfoService.update(updateWrapper);
        });
        List<PieLineEquipContraptionDto> eqsAfter = jgInstallationNoticeEqMapper.selectPCIdNeErrorPieLineInUseInfo2();
        log.info("处理后：查询管道装置id与安装告知的装置id的不一样管道设备：{}", eqsAfter.size());
    }

    /**
     * 处理一个管道在多个安装告知里，刷管道的use_info、es的装置id,已最新安装告知为准
     */
    @Transactional(rollbackFor = Exception.class)
    public void projectUseInfoUpdateV5() {
        List<PieLineEquipContraptionDto> eqs = jgInstallationNoticeEqMapper.selectPCIdNeErrorPieLineInUseInfo3();
        log.info("处理前：一个管道在多个安装告知里，安装告知的装置id的不一样管道设备：{}", eqs.size());
        eqs.forEach(eq -> {
            JgInstallationNoticeEqDto jgInstallationNoticeEqDto = jgInstallationNoticeEqMapper.selectLatestNoticeByRecord(eq.getRecord());
            JgInstallationNotice jgInstallationNotice = jgInstallationNoticeMapper.selectOne(new LambdaQueryWrapper<JgInstallationNotice>().eq(BaseEntity::getSequenceNbr, jgInstallationNoticeEqDto.getEquipTransferId()).select(JgInstallationNotice::getProjectContraptionId, JgInstallationNotice::getProjectContraption));
            // 1.已安装告知单据为准更新es
            Optional<ESEquipmentCategoryDto> equInfosWithEs = esEquipmentCategory.findById(eq.getRecord());
            equInfosWithEs.ifPresent(equInfoEs -> {
                equInfoEs.setProjectContraptionId(jgInstallationNotice.getProjectContraptionId());
                equInfoEs.setPROJECT_CONTRAPTION(jgInstallationNotice.getProjectContraption());
                esEquipmentCategory.save(equInfoEs);
            });
            // 2.已安装告知单据为准更新use_info
            LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(IdxBizJgUseInfo::getRecord, eq.getRecord());
            updateWrapper.set(IdxBizJgUseInfo::getProjectContraptionId, jgInstallationNotice.getProjectContraptionId());
            updateWrapper.set(IdxBizJgUseInfo::getProjectContraption, jgInstallationNotice.getProjectContraption());
            useInfoService.update(updateWrapper);
        });
    }

    public Boolean writeProjectContraptionIdInJgUseRegistration() {
        List<JgUseRegistration> useRegistrations = useRegistrationService.list(
                new LambdaQueryWrapper<JgUseRegistration>()
                        .isNotNull(JgUseRegistration::getProjectContraptionId)
                        .isNull(JgUseRegistration::getOriginProjectContraptionIds)
                        .select(BaseEntity::getSequenceNbr, JgUseRegistration::getProjectContraptionId)
        );

        for (JgUseRegistration u : useRegistrations) {
            // 更新 JgUseRegistration 表
            useRegistrationService.lambdaUpdate()
                    .set(JgUseRegistration::getOriginProjectContraptionIds, u.getProjectContraptionId())
                    .eq(JgUseRegistration::getSequenceNbr, u.getSequenceNbr())
                    .update();

            // 更新 JgUseRegistrationEq 表
            LambdaQueryWrapper<JgUseRegistrationEq> lambda = new QueryWrapper<JgUseRegistrationEq>().lambda();
            lambda.eq(JgUseRegistrationEq::getEquipTransferId, u.getSequenceNbr());
            List<JgUseRegistrationEq> jgUseRegistrationEqList = useRegistrationEqService.getBaseMapper().selectList(lambda);

            if (jgUseRegistrationEqList != null) {
                for (JgUseRegistrationEq eq : jgUseRegistrationEqList) {
                    useRegistrationEqService.lambdaUpdate()
                            .set(JgUseRegistrationEq::getOriginProjectContraptionId, u.getProjectContraptionId())
                            .eq(JgUseRegistrationEq::getSequenceNbr, eq.getSequenceNbr())
                            .update();
                }
            }
        }
        return true;
    }

    /**
     * 将历史导入的普通气瓶和正常导入的普通气瓶刷入设备状态
     *
     * @return result
     */
    public String ordinaryGasCylinderWriteEquState() {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        // 所有的历史导入的普通气瓶 dataSource = 'jg_his' + 过滤出做了报废注销启用停用的业务设备 设备状态为 在用 （EquimentEnum.ZAIYONG.getCode()）
        List<String> allHisRecords = dataHandlerMapper.selectOrdinaryHisGasCylinder();
        useInfoService.lambdaUpdate().in(IdxBizJgUseInfo::getRecord, allHisRecords).set(IdxBizJgUseInfo::getEquState, EquipmentEnum.ZAIYONG.getCode()).update();
        allHisRecords.forEach(record -> {
            Optional<ESEquipmentCategoryDto> categoryDto = esEquipmentCategory.findById(record);
            categoryDto.ifPresent(categoryEs -> {
                categoryEs.setEQU_STATE(EquipmentEnum.ZAIYONG.getCode());
                esEquipmentCategory.save(categoryEs);
            });
        });
        // 所有的正常导入的普通气瓶 dataSource = 'jg' + 过滤出做了使用登记业务的设备 设备状态为 未登记 （EquimentEnum.WEIDENGJI.getCode()）
        List<String> allRecords = dataHandlerMapper.selectOrdinaryGasCylinder();
        useInfoService.lambdaUpdate().in(IdxBizJgUseInfo::getRecord, allRecords).set(IdxBizJgUseInfo::getEquState, EquipmentEnum.WEIDENGJI.getCode()).update();
        allRecords.forEach(record -> {
            Optional<ESEquipmentCategoryDto> categoryDto = esEquipmentCategory.findById(record);
            categoryDto.ifPresent(categoryEs -> {
                categoryEs.setEQU_STATE(EquipmentEnum.WEIDENGJI.getCode());
                esEquipmentCategory.save(categoryEs);
            });
        });

        stopWatch.stop();

        return String.format("刷入数据%s条，耗时：%s秒", allHisRecords.size() + allRecords.size(), stopWatch.getTotalTimeSeconds());
    }

    public void initFormUrl() {
        jgVehicleInformationService.initFormUrl();
        useRegistrationService.initFormUrl();
    }

    public void synchronizeEqData2ES(String records, Boolean isAgw, Boolean isIntoManagement) {
        if (records == null || records.isEmpty()) {
            return;
        }
        String[] recordArr = records.trim().split(",");
        for (String record : recordArr) {
            Map<String, Object> map = categoryOtherInfoMapper.selectDataById(record);
            categoryOtherInfoMapper.updateEsStatus(record);
            ESEquipmentCategoryDto dto = JSON.parseObject(toJSONString(map), ESEquipmentCategoryDto.class);
            Optional<ESEquipmentCategoryDto> data = esEquipmentCategory.findById(record);
            if (!ObjectUtils.isEmpty(data)) {
                esEquipmentCategory.deleteById(record);
            }
            if (!ObjectUtils.isEmpty(dto)) {
                long time = Timestamp.valueOf(map.get("REC_DATE").toString().substring(0, 19)).getTime();
                dto.setREC_DATE(time);
                // 需要安装的设备 安装告知审批通过 清除设备的USC_UNIT_CREDIT_CODE安装单位信息
                // 使用单位编辑 防止更新设备时将安改维单位信息更新到es中
                if (!isAgw) {
                    dto.setUSC_UNIT_CREDIT_CODE(null);
                    dto.setUSC_UNIT_NAME(null);
                }
                if (map.get("NEXT_INSPECT_DATE") != null) {
                    Date nextInspectDate = (Date) map.get("NEXT_INSPECT_DATE");
                    dto.setNEXT_INSPECT_DATE(DateUtil.parse(DateUtil.format(nextInspectDate, DatePattern.NORM_DATE_PATTERN)).getTime());
                }
                dto.setIS_INTO_MANAGEMENT(isIntoManagement);
                IdxBizJgUseInfo useInfo = useInfoService.lambdaQuery().eq(IdxBizJgUseInfo::getRecord, record).one();
                dto.setUSE_PLACE_CODE(useInfo.getProvince() + "#" + useInfo.getCity() + "#" + useInfo.getCounty() + "#" + useInfo.getFactoryUseSiteStreet());
                dto.setUSE_SITE_CODE(useInfo.getUsePlace());
                esEquipmentCategory.save(dto);
            }
        }
    }

    /**
     * 查询公司代码为空或者长度小于15位的西安设备,以tzs_jg_unit_import为准【按照单位名称】更新使用信息表的【使用单位编码】、es的【使用单位编码】
     */
    public void updateUseInfoOfCompanyV1() {
        List<IdxBizJgUseInfo> useInfos = useInfoService.getBaseMapper().selectXAList();
        log.info("1.西安导入数据需补充使用单位代码的设备数量：{}", useInfos.size());
        Map<String, String> companynameCodeMap = getCompanyNameCodeMap();
        AtomicInteger i = new AtomicInteger();
        useInfos.forEach(e -> {
            String companyCode = companynameCodeMap.get(e.getUseUnitName());
            if (companyCode != null) {
                i.getAndIncrement();
                // 1.数据库更新
                LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.eq(TzsBaseEntity::getSequenceNbr, e.getSequenceNbr());
                updateWrapper.set(IdxBizJgUseInfo::getUseUnitCreditCode, companyCode);
                updateWrapper.set(TzsBaseEntity::getRecDate, new Date());
                useInfoService.update(updateWrapper);
                // 2.es更新
                Optional<ESEquipmentCategoryDto> categoryDto = esEquipmentCategory.findById(e.getRecord());
                categoryDto.ifPresent(categoryEs -> {
                    categoryEs.setUSE_UNIT_CREDIT_CODE(companyCode);
                    // 捎带更新单位名称保持数据库和es一致
                    categoryEs.setUSE_UNIT_NAME(e.getUseUnitName());
                    esEquipmentCategory.save(categoryEs);
                });
            }
        });
        log.info("2.西安数据需补充使用单位设备数量：{}, 实际补充使用单位设备数量：{}", useInfos.size(), i.intValue());
    }

    private Map<String, String> getCompanyNameCodeMap() {
        List<KeyValueDto> companyList = commonMapper.queryAllCompanyDict();
        return companyList.stream().collect(Collectors.toMap(KeyValueDto::getStrKey, KeyValueDto::getStrValue, (k1, k2) -> k1));
    }

    public void restore() {
        LambdaQueryWrapper<JgChangeRegistrationUnit> jgChangeRegistrationUnitWrapper = new LambdaQueryWrapper<>();
        jgChangeRegistrationUnitWrapper.orderByAsc(JgChangeRegistrationUnit::getRecDate);
        List<JgChangeRegistrationUnit> changeRegistrationUnits = jgChangeRegistrationUnitService.list(jgChangeRegistrationUnitWrapper);
        for (JgChangeRegistrationUnit changeRegistrationUnit : changeRegistrationUnits) {
            String status = changeRegistrationUnit.getStatus();
            String useUnitName = null;
            String useUnitCreditCode = null;
            Date date = null;
            if ("已完成".equals(status)) {
                useUnitName = changeRegistrationUnit.getNewUseUnitName();
                useUnitCreditCode = changeRegistrationUnit.getNewUseUnitCreditCode();
                date = changeRegistrationUnit.getRecDate();
            } else {
                useUnitName = changeRegistrationUnit.getUseUnitName();
                useUnitCreditCode = changeRegistrationUnit.getUseUnitCreditCode();
                date = changeRegistrationUnit.getCreateDate();
            }
            LambdaQueryWrapper<JgChangeRegistrationUnitEq> unitEqWrapper = new LambdaQueryWrapper<>();
            unitEqWrapper.eq(JgChangeRegistrationUnitEq::getUnitChangeRegistrationId, changeRegistrationUnit.getSequenceNbr());
            List<JgChangeRegistrationUnitEq> changeRegistrationUnitEqs = jgChangeRegistrationUnitEqService.list(unitEqWrapper);
            Map<String, Map<String, Object>> resultMap = new HashMap<>();
            for (JgChangeRegistrationUnitEq changeRegistrationUnitEq : changeRegistrationUnitEqs) {
                String equipId = changeRegistrationUnitEq.getEquId();
                if (StringUtils.isNotEmpty(equipId)) {
                    LambdaUpdateWrapper<UseInfo> queryWrapper2 = new LambdaUpdateWrapper<>();
                    queryWrapper2.set(UseInfo::getUseUnitName, useUnitName);
                    queryWrapper2.set(UseInfo::getUseUnitCreditCode, useUnitCreditCode);
                    queryWrapper2.set(UseInfo::getRecDate, date);
                    queryWrapper2.eq(UseInfo::getRecord, equipId);
                    useInfoMapper.update(null, queryWrapper2);
                    Map<String, Object> map1 = new HashMap<>();
                    map1.put("USE_UNIT_CREDIT_CODE", useUnitCreditCode);
                    map1.put("USE_UNIT_NAME", useUnitName);
                    map1.put("REC_DATE", date);// 更新时间
                    resultMap.put(equipId, map1);
                    tzsServiceFeignClient.commonUpdateEsDataByIds(resultMap);
                }
            }
        }
    }

    /**
     * 存量数据刷属地监管部门到证管理表-无事务，每条可单独处理
     * 1.使用登记--tzs_jg_use_registration->supervision_org_code
     * 2.车用气瓶登记--tzs_jg_vehicle_information->org_branch_code|org_branch_name
     * 3.车用气瓶单位变更登记--tzs_jg_change_vehicle_registration_unit>org_branch_code|org_branch_name
     * 4.移装变更登记--tzs_jg_change_registration_transfer->tzs_jg_registration_history.change_data->>orgBranchCode
     * 5.历史有证气瓶导入--tzs_jg_use_registration->supervision_org_code
     * 6.历史有证压力管道-管道导入-工业管道--tzs_jg_use_registration->supervision_org_code
     *
     * @return 成功数量数量
     */
    public Long saveSupervise2UseRegManage() {

        log.info("存量数据刷属地监管部门到证管理表处理开始--->");
        // 1|5|6 预处理，存量数据导入时缺少记录到tzs_jg_use_registration->supervision_org_code，先补充supervision_org_code到tzs_jg_use_registration
        List<JgUseRegistrationDto> missingSupOrgCodeUses = useRegistrationService.getBaseMapper().selectMissingSupOrgCodeUseList();
        missingSupOrgCodeUses.parallelStream().forEach(u -> {
            // TODO 历史存量数据里有：610422_三原县市场监督管理局，需要转换为：50*73*123_三原县市场监督管理局, null则挂到根节点省上
            if (StringUtils.isNotEmpty(u.getSupervisoryCode()) && u.getSupervisoryCode().split("_").length > 1) {
                String[] orgCodeNameArray = u.getSupervisoryCode().split("_");
                String orgCodeOrCompanyCode = orgCodeNameArray[0];
                LambdaUpdateWrapper<JgUseRegistration> updateWrapper = new LambdaUpdateWrapper<>();
                if (orgCodeOrCompanyCode.contains("*") || ROOT_ORG_CODE.equals(orgCodeOrCompanyCode)) {
                    updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, orgCodeNameArray[0]);
                } else {
                    // 把companyCode换成orgCode
                    String orgCode = commonMapper.getOrgCodeByCompanyCode(orgCodeOrCompanyCode);
                    updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, orgCode == null ? ROOT_ORG_CODE : orgCode);
                }
                updateWrapper.eq(JgUseRegistration::getSequenceNbr, u.getSequenceNbr());
                useRegistrationService.update(updateWrapper);
            } else {
                // 没有的挂到根节点省
                LambdaUpdateWrapper<JgUseRegistration> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.eq(JgUseRegistration::getSequenceNbr, u.getSequenceNbr());
                updateWrapper.set(JgUseRegistration::getSupervisionOrgCode, ROOT_ORG_CODE);
                useRegistrationService.update(updateWrapper);
            }
        });
        // 1|5|6 处理
        AtomicReference<Long> dealNum = new AtomicReference<>(0L);
        List<JgUseRegistrationDto> useRegistrations = useRegistrationService.getBaseMapper().getListOfPass();
        useRegistrations.parallelStream().forEach(u -> {
            LambdaUpdateWrapper<JgUseRegistrationManage> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(JgUseRegistrationManage::getUseRegistrationCode, u.getUseRegistrationCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgCode, u.getSupervisionOrgCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgName, u.getSupervisionOrgName());
            dealNum.getAndSet(dealNum.get() + 1);
            jgUseRegistrationManageServiceImpl.update(updateWrapper);
        });
        // 2 处理
        List<JgVehicleInformation> vehicleUseRegs = jgVehicleInformationService.getBaseMapper().getListOfPass();
        vehicleUseRegs.parallelStream().forEach(u -> {
            LambdaUpdateWrapper<JgUseRegistrationManage> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(JgUseRegistrationManage::getUseRegistrationCode, u.getUseRegistrationCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgCode, u.getOrgBranchCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgName, u.getOrgBranchName());
            dealNum.getAndSet(dealNum.get() + 1);
            jgUseRegistrationManageServiceImpl.update(updateWrapper);
        });
        // 3 处理
        List<JgChangeVehicleRegistrationUnit> vehicleRegistrationUnits = jgChangeVehicleRegistrationUnitMapper.getListOfPass();
        vehicleRegistrationUnits.parallelStream().forEach(u -> {
            LambdaUpdateWrapper<JgUseRegistrationManage> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(JgUseRegistrationManage::getUseRegistrationCode, u.getUseRegistCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgCode, u.getOrgBranchCode());
            updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgName, u.getOrgBranchName());
            dealNum.getAndSet(dealNum.get() + 1);
            jgUseRegistrationManageServiceImpl.update(updateWrapper);
        });
        // 4 处理
        List<JgChangeRegistrationTransfer> transfers = jgChangeRegistrationTransferMapper.getListOfPass();
        transfers.parallelStream().forEach(u -> {
            if (StringUtils.isNotEmpty(u.getRemark()) && u.getRemark().split("_").length > 1) {
                String[] orgCodeNameArray = u.getRemark().split("_");
                LambdaUpdateWrapper<JgUseRegistrationManage> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.eq(JgUseRegistrationManage::getUseRegistrationCode, u.getUseRegistrationCode());
                updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgCode, orgCodeNameArray[0]);
                updateWrapper.set(JgUseRegistrationManage::getSuperviseOrgName, orgCodeNameArray[1]);
                dealNum.getAndSet(dealNum.get() + 1);
                jgUseRegistrationManageServiceImpl.update(updateWrapper);
            }
        });
        log.info("存量数据刷属地监管部门到证管理表处理结束，处理数据数量：{}", dealNum.get());
        return dealNum.get();
    }

    /**
     * 历史安全追溯问题闭环【维保备案】【检验超期】--问题源都是设备
     *
     * @return 闭环问题数
     */
    public String safeTraceabilityClosedLoop() {
        // 维保超期
        // 查询当天维保超期的设备
        Set<String> outOfMaintenanceRecords = commonMapper.queryOutOfMaintenanceRecord()
                .stream()
                .map(item -> Objects.toString(item.get("RECORD")))
                .collect(Collectors.toSet());
        // 查询未关闭的维保超期问题数
        Set<String> maintenanceProblemRecords = safetyProblemTracingService.lambdaQuery()
                .eq(SafetyProblemTracing::getProblemTypeCode, SafetyProblemTypeEnum.WBCQ.getProblemTypeCode())
                .eq(SafetyProblemTracing::getProblemStatusCode, 0)
                .select(SafetyProblemTracing::getSourceId)
                .list().stream()
                .map(SafetyProblemTracing::getSourceId)
                .collect(Collectors.toSet());
        JSONArray mainJsonArr = new JSONArray();
        maintenanceProblemRecords.stream()
                .filter(Objects::nonNull)
                .filter(item -> !outOfMaintenanceRecords.contains(item))
                .forEach(item -> mainJsonArr.add(new JSONObject().fluentPut("record", item)));
        // 处理问题数据
        equipmentStrategy.handleProblem(mainJsonArr, SafetyProblemTypeEnum.WBBA);

        // 检验超期
        // 查询当天检验超期的设备
        Set<String> outOfInspectionRecords = commonMapper.queryOutOfInspectionRecord()
                .stream().map(item -> Objects.toString(item.get("RECORD")))
                .collect(Collectors.toSet());
        // 查询未关闭的检验超期问题数
        Set<String> inspectProblemRecords = safetyProblemTracingService.lambdaQuery()
                .eq(SafetyProblemTracing::getProblemTypeCode, SafetyProblemTypeEnum.JYCQ.getProblemTypeCode())
                .eq(SafetyProblemTracing::getProblemStatusCode, 0)
                .select(SafetyProblemTracing::getSourceId)
                .list().stream()
                .map(SafetyProblemTracing::getSourceId)
                .collect(Collectors.toSet());
        JSONArray inspectJsonArr = new JSONArray();
        inspectProblemRecords.stream()
                .filter(Objects::nonNull)
                .filter(item -> !outOfInspectionRecords.contains(item))
                .forEach(item -> inspectJsonArr.add(new JSONObject().fluentPut("record", item)));
        // 处理问题数据
        equipmentStrategy.handleProblem(inspectJsonArr, SafetyProblemTypeEnum.JYBJ);
        return String.format("修正历史维保超期未闭环数:%s起,检验超期未闭环数：%s起。总数：%s起", mainJsonArr.size(), inspectJsonArr.size(), mainJsonArr.size() + inspectJsonArr.size());
    }

    public String writeEstateUnitInfoInUseInfo() {
        List<Map<String, Object>> vehicleUseRegs = jgVehicleInformationService.getBaseMapper().getVehicleInfoOfPassWithEquId();
        vehicleUseRegs.parallelStream().forEach(u -> {
            LambdaUpdateWrapper<IdxBizJgUseInfo> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.eq(IdxBizJgUseInfo::getRecord, u.get("equId"));
            updateWrapper.set(IdxBizJgUseInfo::getEstateUnitName, u.get("estateUnitName"));
            updateWrapper.set(IdxBizJgUseInfo::getEstateUnitCreditCode, u.get("estateUnitCreditCode"));
            useInfoService.update(updateWrapper);
        });
        return String.format("修正车用气瓶登记未更新产权信息数据:%s条。", vehicleUseRegs.size());
    }

    public String initDataQualityScore() {
        StopWatch watch = new StopWatch();
        watch.start();
        long dealEquipNum = 0L;
        Integer maxVersion = useInfoService.getBaseMapper().selectMaxVersion();
        // 1.非管道数据刷数据质量等级
        List<RefreshDataDto> refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersion(maxVersion);
        while (!refreshRecords.isEmpty()) {
            dealEquipNum = refreshRecords.size() + dealEquipNum;
            List<EquipWaitRefreshDataQualityScore> refreshDataQualityScores = new ArrayList<>();
            refreshRecords.forEach(equip -> {
                if (CylinderTypeEnum.CYLINDER.getCode().equals(equip.getEquCategory())) { // 气瓶更新逻辑
                    //  1.已纳管且有使用登记证的刷为：1----3级
                    if (equip.getIsIntoManagement() && StringUtils.isNotEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 1);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    //  2.已纳管且无使用登记证的刷为：3----1级
                    if (equip.getIsIntoManagement() && StringUtils.isEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    //  3.未纳管且有使用登记证为空
                    String whetherVehicleCylinder = StringUtils.isEmpty(equip.getWhetherVehicleCylinder()) ? "0" : equip.getWhetherVehicleCylinder();
                    // 3.1非车用气瓶逻辑 3 未纳管气瓶没有使用登记，直接都为3级
                    if ((!equip.getIsIntoManagement()) && StringUtils.isNotEmpty(equip.getUseOrgCode()) && "0".equals(whetherVehicleCylinder)) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    // 3.2车用气瓶逻辑, 单位内部编号、出场编号全有时刷为1；缺少任意一项时刷为2
                    if ((!equip.getIsIntoManagement()) && StringUtils.isNotEmpty(equip.getUseOrgCode()) && "1".equals(whetherVehicleCylinder)) {
                        if (StringUtils.isEmpty(equip.getUseInnerCode()) || StringUtils.isEmpty(equip.getFactoryNum())) {
                            EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                            refreshDataQualityScores.add(waitRefreshDataQualityScore);
                        }
                        if (StringUtils.isNotEmpty(equip.getUseInnerCode()) && StringUtils.isNotEmpty(equip.getFactoryNum())) {
                            EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 2);
                            refreshDataQualityScores.add(waitRefreshDataQualityScore);
                        }
                    }
                    //  4.未纳管且无使用登记证的刷为：3-1级
                    if ((!equip.getIsIntoManagement()) && StringUtils.isEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                } else { // 非气瓶外逻辑
                    //  1.已纳管且有使用登记证的刷为：1----3级
                    if (equip.getIsIntoManagement() && StringUtils.isNotEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 1);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    //  2.已纳管且无使用登记证的刷为：3----1级
                    if (equip.getIsIntoManagement() && StringUtils.isEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    //  3.未纳管且有使用登记证的刷为：1或者2，单位内部编号、出场编号全有时刷为1；缺少任意一项时刷为2
                    if ((!equip.getIsIntoManagement()) && StringUtils.isNotEmpty(equip.getUseOrgCode()) && (StringUtils.isEmpty(equip.getUseInnerCode()) || StringUtils.isEmpty(equip.getFactoryNum()))) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 2);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    if ((!equip.getIsIntoManagement()) && StringUtils.isNotEmpty(equip.getUseOrgCode()) && StringUtils.isNotEmpty(equip.getUseInnerCode()) && StringUtils.isNotEmpty(equip.getFactoryNum())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 1);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                    //  4.未纳管且无使用登记证的刷为：3-1级
                    if ((!equip.getIsIntoManagement()) && StringUtils.isEmpty(equip.getUseOrgCode())) {
                        EquipWaitRefreshDataQualityScore waitRefreshDataQualityScore = new EquipWaitRefreshDataQualityScore(equip.getRecord(), 3);
                        refreshDataQualityScores.add(waitRefreshDataQualityScore);
                    }
                }
            });
            // es 更新
            refreshDataQualityScores.parallelStream().forEach(d -> {
                this.updateEsDataQualityScore(d.getRecord(), d.getDataQualityScore());
            });
            // 数据库更新
            if (!refreshDataQualityScores.isEmpty()) {
                useInfoService.getBaseMapper().updateDataQualityScoreBatch(refreshDataQualityScores, maxVersion + 1);
            }
            refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersion(maxVersion);
        }
        // 2.管道逻辑
        List<IdxBizJgProjectContraption> projectContraptionList = projectContraptionService.getBaseMapper().selectList(new LambdaQueryWrapper<IdxBizJgProjectContraption>().isNull(IdxBizJgProjectContraption::getProjectContraptionParentId).select(BaseEntity::getSequenceNbr, IdxBizJgProjectContraption::getUseRegistrationCode));
        List<ProjectWaitRefreshDataQualityScore> projectWaitRefreshDataQualityScores = projectContraptionList.stream().map(p -> {
            ProjectWaitRefreshDataQualityScore projectWaitRefreshDataQualityScore = new ProjectWaitRefreshDataQualityScore();
            projectWaitRefreshDataQualityScore.setProjectContraptionId(p.getSequenceNbr());
            if (StringUtils.isNotEmpty(p.getUseRegistrationCode())) {
                projectWaitRefreshDataQualityScore.setDataQualityScore(1);
            } else {
                projectWaitRefreshDataQualityScore.setDataQualityScore(3);
            }
            return projectWaitRefreshDataQualityScore;
        }).collect(Collectors.toList());
        if (!projectWaitRefreshDataQualityScores.isEmpty()) {
            projectContraptionService.getBaseMapper().updateDataQualityScoreBatch(projectWaitRefreshDataQualityScores);
        }
        watch.stop();
        log.info("耗时秒：{}", watch.getTotalTimeSeconds());
        return "success";
    }

    private void updateEsDataQualityScore(String record, Integer dataQualityScore) {
        // 1.es更新
        Optional<ESEquipmentCategoryDto> categoryDto = esEquipmentCategory.findById(record);
        categoryDto.ifPresent(categoryEs -> {
            categoryEs.setDataQualityScore(dataQualityScore);
            esEquipmentCategory.save(categoryEs);
        });
    }

    public Integer initCompanyStatistData2Es() {
        log.info("综合统计企业信息入库开始");
        StopWatch watch = new StopWatch();
        watch.start();
        List<TzBaseEnterpriseInfo> units = tzBaseEnterpriseInfoMapper.selectList(null);
        List<EsBaseEnterpriseInfo> esBaseEnterpriseInfos = units.parallelStream().map(enterpriseInfo -> {
            EsBaseEnterpriseInfo esBaseEnterpriseInfo = new EsBaseEnterpriseInfo();
            BeanUtil.copyProperties(enterpriseInfo, esBaseEnterpriseInfo);
            esBaseEnterpriseInfo.setSequenceNbr(enterpriseInfo.getSequenceNbr() + "");
            esBaseEnterpriseInfo.setContactUser(enterpriseInfo.getUseContact());
            esBaseEnterpriseInfo.setEquipCategory(RefreshDataUtils.castStrList2String(enterpriseInfo.getEquipCategory()));
            esBaseEnterpriseInfo.setEquipCategoryName(RefreshDataUtils.castEquCategory2Name(esBaseEnterpriseInfo.getEquipCategory()));
            List<TzBaseUnitLicence> unitLicences = licenceMapper.selectList(new LambdaQueryWrapper<TzBaseUnitLicence>()
                    .eq(TzBaseUnitLicence::getUnitCode, enterpriseInfo.getUseUnitCode())
                    .eq(TzBaseUnitLicence::getIsDelete, false));
            List<EsBaseEnterpriseInfo.License> licenses = unitLicences.stream().map(lis -> {
                EsBaseEnterpriseInfo.License esLicense = new EsBaseEnterpriseInfo.License();
                BeanUtil.copyProperties(lis, esLicense);
                return esLicense;
            }).collect(Collectors.toList());
            if (!ObjectUtils.isEmpty(licenses)) {
                esBaseEnterpriseInfo.setLicenses(licenses);
                try {
                    esBaseEnterpriseInfo.setLicenseMinExpiryDate(RefreshCmService.getMinDateFromLicensesEp(licenses));
                } catch (Exception e) {
                    log.error("企业信息单位类型或者取资质最小日期失败:{}", enterpriseInfo, e);
                }
            }
            esBaseEnterpriseInfo.setUnitType(onlineJyjc ? enterpriseInfo.getUnitType() : RefreshCmService.updateJyjcUnitType(enterpriseInfo.getUnitType(), licenses));
            return esBaseEnterpriseInfo;
        }).collect(Collectors.toList());
        enterpriseInfoDao.saveAll(esBaseEnterpriseInfos);
        watch.stop();
        log.info("综合统计企业信息入库结束，耗时：{}秒", watch.getTotalTimeSeconds());
        return units.size();
    }

    public Integer initUserStatistData2Es() {
        log.info("综合统计人员信息入库开始");
        StopWatch watch = new StopWatch();
        watch.start();
        List<TzBaseEnterpriseInfo> units = tzBaseEnterpriseInfoMapper.selectList(new LambdaQueryWrapper<TzBaseEnterpriseInfo>()
                .select(BaseEntity::getSequenceNbr,
                        TzBaseEnterpriseInfo::getUseCode,
                        TzBaseEnterpriseInfo::getUseUnit,
                        TzBaseEnterpriseInfo::getUnitType,
                        TzBaseEnterpriseInfo::getSuperviseOrgCode,
                        TzBaseEnterpriseInfo::getSuperviseOrgName,
                        TzBaseEnterpriseInfo::getOfficeRegion
                ));
        Map<String, TzBaseEnterpriseInfo> useCodeEnterpriseMap = units.stream().collect(Collectors.toMap(TzBaseEnterpriseInfo::getUseCode, Function.identity(), (k1, k2) -> k2));
        Iterable<EsBaseEnterpriseInfo> esBaseEnterpriseInfoIterable = enterpriseInfoDao.findAll();
        List<EsBaseEnterpriseInfo> esBaseEnterpriseInfos = Lists.newArrayList(esBaseEnterpriseInfoIterable);
        Map<String, String> esUseCodeEnterpriseMap = esBaseEnterpriseInfos.stream().collect(Collectors.toMap(EsBaseEnterpriseInfo::getUseCode, EsBaseEnterpriseInfo::getUnitType, (k1, k2) -> k2));
        LambdaQueryWrapper<TzsUserInfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(BaseEntity::getIsDelete, false)
                .orderByDesc(BaseEntity::getSequenceNbr);
        Integer userNum = userInfoMapper.selectCount(wrapper);
        int pageSize = 1000;
        int totalPage = Math.max(1, (int) Math.ceil((double) userNum / pageSize));
        for (int i = 1; i <= totalPage; i++) {
            IPage<TzsUserInfo> tzsUserInfoPage = userInfoMapper.selectPage(new Page<>(i, pageSize), wrapper);
            List<EsUserInfo> esUserInfos = tzsUserInfoPage.getRecords().stream().map(u -> {
                EsUserInfo esUserInfo = new EsUserInfo();
                BeanUtil.copyProperties(u, esUserInfo);
                esUserInfo.setPost(RefreshDataUtils.castStrList2String(u.getPost()));
                esUserInfo.setNewPost(RefreshDataUtils.castStrList2String(u.getNewPost()));
                esUserInfo.setSubPost(RefreshDataUtils.castStrList2String(u.getSubPost()));
                esUserInfo.setEquipType(RefreshDataUtils.castStrList2String(u.getEquipType()));
                esUserInfo.setEquipTypeName(RefreshDataUtils.castEquCategory2Name(esUserInfo.getEquipType()));
                // 绑定设备状态
                esUserInfo.setBindEquStatus(this.queryUserBindEquStatus(u.getSequenceNbr()));
                // 人员类型暂时取esBaseEnterpriseInfo按照资质分开后的，等检验和检测分开后，按照实际情况取就即可：useCodeEnterpriseMap.getOrDefault(u.getUnitCode(), new TzBaseEnterpriseInfo()).getUnitType()
                String unitType = esUseCodeEnterpriseMap.getOrDefault(u.getUnitCode(), useCodeEnterpriseMap.getOrDefault(u.getUnitCode(), new TzBaseEnterpriseInfo()).getUnitType());
                esUserInfo.setUnitType(unitType);
                esUserInfo.setSuperviseOrgName(useCodeEnterpriseMap.getOrDefault(u.getUnitCode(), new TzBaseEnterpriseInfo()).getSuperviseOrgName());
                esUserInfo.setSuperviseOrgCode(useCodeEnterpriseMap.getOrDefault(u.getUnitCode(), new TzBaseEnterpriseInfo()).getSuperviseOrgCode());
                esUserInfo.setOfficeRegion(useCodeEnterpriseMap.getOrDefault(u.getUnitCode(), new TzBaseEnterpriseInfo()).getOfficeRegion());
                esUserInfo.setLicenses(this.buildUserLicenses(u.getSequenceNbr()));
                esUserInfo.setLicenseMinExpiryDate(RefreshCmService.getMinDateFromLicensesUser(esUserInfo.getLicenses()));
                return esUserInfo;
            }).collect(Collectors.toList());
            esUserInfoDao.saveAll(esUserInfos);
        }
        watch.stop();
        log.info("综合统计人员信息入库结束，耗时：{}秒", watch.getTotalTimeSeconds());
        return units.size();
    }

    private String queryUserBindEquStatus(Long sequenceNbr) {
        return commonMapper.queryUserBindEquNum(sequenceNbr.toString()) > 0 ? "1" : "0";
    }

    private List<EsUserInfo.License> buildUserLicenses(Long sequenceNbr) {
        List<TzsUserPermission> permissions = userPermissionMapper.selectList(new LambdaQueryWrapper<TzsUserPermission>().eq(TzsUserPermission::getUserSeq, sequenceNbr));
        return permissions.stream().map(p -> {
            EsUserInfo.License license = new EsUserInfo.License();
            BeanUtil.copyProperties(p, license);
            license.setPermissionItem(RefreshDataUtils.castStrList2String(p.getPermissionItem()));
            license.setJobItem(RefreshDataUtils.castStrList2String(p.getJobItem()));
            return license;
        }).collect(Collectors.toList());
    }

    public Integer initEquipStatistData2EsBatch() {
        log.info("综合统计设备信息入库开始");
        StopWatch watch = new StopWatch();
        watch.start();
        Integer maxVersion = useInfoService.getBaseMapper().selectMaxVersion();
        Integer nextVersion = maxVersion + 1;
        List<String> refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionAll(nextVersion);
        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.getNextInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getNextInspectDate))));
            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()));
                    if ("8000".equals(esEquipmentInfo.getEQU_LIST_CODE())) {
                        List<ESEquipmentInfo.TechParam> techParams = esEquipmentInfo.getTechParams();
                        List<ESEquipmentInfo.TechParam> pipeLength = techParams.stream().filter(e -> e.getParamKey().equals("pipeLength") && e.getDoubleValue() != null).collect(Collectors.toList());
                        if (!ObjectUtils.isEmpty(pipeLength)) {
                            esEquipmentInfo.setPipeLength(pipeLength.get(0).getDoubleValue());
                        }
                    }
                } 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());
            }
            StopWatch watch2 = new StopWatch();
            watch2.start();
            useInfoService.getBaseMapper().updateVersionBatch(refreshRecords, nextVersion);
            watch2.stop();
            log.warn("批量更新条数：{} 使用信息版本号，耗时：{}s", esEquipmentInfos.size(), watch2.getTotalTimeSeconds());
            StopWatch watch3 = new StopWatch();
            watch3.start();
            refreshRecords = useInfoService.getBaseMapper().selectUseInfoOfOneVersionAll(nextVersion);
            watch3.stop();
            log.warn("查询最多版本号耗时：{}s", watch3.getTotalTimeSeconds());
        }
        watch.stop();
        log.info("综合统计设备信息入库结束，耗时：{}秒", watch.getTotalTimeSeconds());
        return refreshRecords.size();
    }


    public void addStatistData2EsBatch() {
        log.info("增量添加ORG_BRANCH_CODE为50X综合搜索数据-设备信息入库开始");
        StopWatch watch = new StopWatch();
        watch.start();
        Integer count = useInfoService.getBaseMapper().selectAddDataRecordsCount();
        Integer times = 0;
        if (count != 0) {
            times = count / 5000;
            int last = count % 5000;
            if (last > 0) {
                times++;
            }
        } else {
            return;
        }
        Page<String> recordDtoPage = new Page<>();
        for (int j = 0; j <= times; j++) {
            recordDtoPage.setCurrent(j + 1);
            recordDtoPage.setSize(5000);

            Page<String> refreshRecords = useInfoService.getBaseMapper().selectAddDataRecords(recordDtoPage);

            if (!ObjectUtils.isEmpty(refreshRecords) && refreshRecords.getRecords().size() > 0) {
                List<String> records = refreshRecords.getRecords();
                StopWatch watch0 = new StopWatch();
                watch0.start();
                List<Map<String, Object>> details = useInfoService.getBaseMapper().queryDetailBatch(records);
                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(records);
                Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));

                // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
                List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = idxBizJgInspectionDetectionInfoService.getBaseMapper().selectLastedGroupByInspectTypeBatch(records);
                Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));

                // 设备最新的检验信息-维度：设备
                Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e -> e.getNextInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getNextInspectDate))));
                List<ESEquipmentInfo> esEquipmentInfos = records.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()));
                        if ("8000".equals(esEquipmentInfo.getEQU_LIST_CODE())) {
                            List<ESEquipmentInfo.TechParam> techParams = esEquipmentInfo.getTechParams();
                            List<ESEquipmentInfo.TechParam> pipeLength = techParams.stream().filter(e -> e.getParamKey().equals("pipeLength") && e.getDoubleValue() != null).collect(Collectors.toList());
                            if (!ObjectUtils.isEmpty(pipeLength)) {
                                esEquipmentInfo.setPipeLength(pipeLength.get(0).getDoubleValue());
                            }
                        }
                    } 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());
    }


    private List<ESEquipmentInfo.TechParam> buildTechParamByEquList(String record, String equListCode) {
        return StringUtils.isNotEmpty(equListCode) ? statisticsDataUpdateService.getTechParams(equListCode, record) : new ArrayList<>();
    }

    public Integer initFactoryCarNumber2Es() {
        List<IdxBizJgRegisterInfo> registerInfoList = registerInfoMapper.selectList(new QueryWrapper<IdxBizJgRegisterInfo>().lambda().isNotNull(IdxBizJgRegisterInfo::getCarNumber));
        Map<String, Map<String, Object>> resultMap = new HashMap<>();
        if (!CollectionUtils.isEmpty(registerInfoList)) {
            registerInfoList.parallelStream().forEach(v -> {
                Map<String, Object> param = new HashMap<>();
                param.put("CAR_NUMBER", v.getCarNumber());
                resultMap.put(v.getRecord(), param);
            });
            tzsServiceFeignClient.commonUpdateEsDataByIds(resultMap);
        }
        return registerInfoList.size();
    }

    private LocalDate getIssueDate(String useRegistrationCode) {
        if (StringUtils.isEmpty(useRegistrationCode)) {
            return null;
        }
        LambdaQueryWrapper<JgUseRegistrationManage> useRegistrationManageWrapper = new LambdaQueryWrapper<>();
        useRegistrationManageWrapper.eq(JgUseRegistrationManage::getUseRegistrationCode, useRegistrationCode)
                .isNotNull(JgUseRegistrationManage::getUseRegistrationCode)
                .eq(JgUseRegistrationManage::getCertificateStatus, "已登记")
                .eq(BaseEntity::getIsDelete, false).select(BaseEntity::getSequenceNbr, JgUseRegistrationManage::getRegDate);
        JgUseRegistrationManage manage = jgUseRegistrationManageServiceImpl.getBaseMapper().selectOne(useRegistrationManageWrapper);
        return Optional.ofNullable(manage).map(JgUseRegistrationManage::getRegDate).map(d -> d.toInstant().atZone(ZoneId.systemDefault()).toLocalDate()).orElse(null);
    }

    public void addPipeLength2Es() {
        log.info("压力管道长度刷入es-设备信息入库开始");
        StopWatch watch = new StopWatch();
        watch.start();
        Integer count = useInfoService.getBaseMapper().selectPiPeCount();
        Integer times = 0;
        if (count != 0) {
            times = count / 5000;
            int last = count % 5000;
            if (last > 0) {
                times++;
            }
        } else {
            return;
        }
        Page<String> recordDtoPage = new Page<>();
        for (int j = 0; j <= times; j++) {
            recordDtoPage.setCurrent(j + 1);
            recordDtoPage.setSize(5000);

            Page<String> refreshRecords = useInfoService.getBaseMapper().selectPiPeRecords(recordDtoPage);

            if (!ObjectUtils.isEmpty(refreshRecords) && refreshRecords.getRecords().size() > 0) {
                List<String> records = refreshRecords.getRecords();
                StopWatch watch0 = new StopWatch();
                watch0.start();
                List<Map<String, Object>> details = useInfoService.getBaseMapper().queryDetailBatch(records);
                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(records);
                Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));

                // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
                List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = idxBizJgInspectionDetectionInfoService.getBaseMapper().selectLastedGroupByInspectTypeBatch(records);
                Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));

                // 设备最新的检验信息-维度：设备
                Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e -> e.getNextInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getNextInspectDate))));
                List<ESEquipmentInfo> esEquipmentInfos = records.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()));
                        if ("8000".equals(esEquipmentInfo.getEQU_LIST_CODE())) {
                            List<ESEquipmentInfo.TechParam> techParams = esEquipmentInfo.getTechParams();
                            List<ESEquipmentInfo.TechParam> pipeLength = techParams.stream().filter(e -> e.getParamKey().equals("pipeLength") && e.getDoubleValue() != null).collect(Collectors.toList());
                            if (!ObjectUtils.isEmpty(pipeLength)) {
                                esEquipmentInfo.setPipeLength(pipeLength.get(0).getDoubleValue());
                            }
                        }
                    } 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());
    }

    public Integer deleteEquipWithStatusNotInClaimed(String indices, boolean isDelete) throws InterruptedException {
        if (!Arrays.asList(IDX_BIZ_VIEW_JG_ALL, IDX_BIZ_EQUIPMENT_INFO).contains(indices)) {
            throw new IllegalArgumentException("非法索引名，只允许使用 idx_biz_view_jg_all 或 idx_biz_equipment_info");
        }

        List<?> esList;
        List<String> records;

        if (IDX_BIZ_VIEW_JG_ALL.equals(indices)) {
            esList = queryUnclaimedEquipments(indices, ESEquipmentCategoryDto.class);
        } else {
            esList = queryUnclaimedEquipments(indices, ESEquipmentInfo.class);
        }

        if (CollectionUtils.isEmpty(esList)) {
            return 0;
        }

        records = extractSequenceNbr(esList);
        if (records.isEmpty()) {
            return 0;
        }

        if (isDelete) {
            List<String> successDeleteList = Collections.synchronizedList(new ArrayList<>());
            List<List<String>> allDataList = Lists.partition(records, BATCH_SIZE);
            ExecutorService executor = new ThreadPoolExecutor(
                    10,
                    20,
                    60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(1000),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.CallerRunsPolicy()
            );

            CountDownLatch latch = new CountDownLatch(allDataList.size());

            for (List<String> batch : allDataList) {
                executor.submit(() -> {
                    try {
                        superviseInfoMapper.deleteDataAll(batch);
                        successDeleteList.addAll(batch);
                    } catch (Exception e) {
                        log.error("批次删除失败: {}", batch, e);
                    } finally {
                        latch.countDown();
                    }
                });
            }

            latch.await();
            executor.shutdown();

            if (!successDeleteList.isEmpty()) {
                List<?> successEsList = esList.stream()
                        .filter(e -> successDeleteList.contains(getSequenceNbr(e)))
                        .collect(Collectors.toList());

                if (IDX_BIZ_VIEW_JG_ALL.equals(indices)) {
                    esEquipmentCategory.deleteAll((List<ESEquipmentCategoryDto>) successEsList);
                } else {
                    esEquipmentDao.deleteAll((List<ESEquipmentInfo>) successEsList);
                }
            }
            return successDeleteList.size();
        }

        return records.size();
    }

    private String getSequenceNbr(Object e) {
        try {
            if (e instanceof ESEquipmentCategoryDto) {
                return ((ESEquipmentCategoryDto) e).getSEQUENCE_NBR();
            } else if (e instanceof ESEquipmentInfo) {
                return ((ESEquipmentInfo) e).getSEQUENCE_NBR();
            }
        } catch (Exception ex) {
            log.warn("无法获取sequenceNbr: {}", e, ex);
        }
        return null;
    }


    private <T> List<T> queryUnclaimedEquipments(String index, Class<T> clazz) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        BoolQueryBuilder statusFilter = QueryBuilders.boolQuery()
                .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(STATUS, "已认领")))
                .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery(STATUS)))
                .should(QueryBuilders.termQuery(STATUS, ""))
                .minimumShouldMatch(1);
        boolQuery.must(statusFilter);

        try {
            return registrationManageService.searchResponse(
                    index,
                    boolQuery,
                    hit -> {
                        JSONObject json = JSONObject.parseObject(hit.getSourceAsString(), JSONObject.class);
                        json.put(SEQUENCE_NBR, Objects.toString(json.get(SEQUENCE_NBR), ""));
                        return json.toJavaObject(clazz);
                    }
            );
        } catch (Exception ex) {
            log.error("查询未认领设备异常：", ex);
            return Collections.emptyList();
        }
    }

    private <T> List<String> extractSequenceNbr(List<T> list) {
        return list.stream()
                .map(item -> {
                    try {
                        Method method = item.getClass().getMethod("getSEQUENCE_NBR");
                        Object value = method.invoke(item);
                        return Objects.toString(value, null);
                    } catch (Exception e) {
                        log.warn("反射获取 SEQUENCE_NBR 失败", e);
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    public Integer deleteEquipExistDB(boolean isDelete) {
        LambdaQueryWrapper<IdxBizJgOtherInfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.select(IdxBizJgOtherInfo::getRecord)
                .notIn(IdxBizJgOtherInfo::getClaimStatus, "已认领", "");
        List<IdxBizJgOtherInfo> jgOtherInfos = otherInfoMapper.selectList(wrapper);
        List<String> records = Optional.ofNullable(jgOtherInfos)
                .orElse(Collections.emptyList())
                .stream()
                .filter(Objects::nonNull)
                .map(IdxBizJgOtherInfo::getRecord)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (!records.isEmpty() && isDelete) {
            int batchSize = 1000;
            for (int i = 0; i < records.size(); i += batchSize) {
                int end = Math.min(i + batchSize, records.size());
                List<String> batch = records.subList(i, end);
                superviseInfoMapper.deleteDataAll(batch);
                deleteFromEs(batch);
            }
        }
        return records.size();
    }

    public Integer modifySupervisionOrgBranchCode(boolean isModify) {
        // 1. 查询 SuperviseInfo 中 orgBranchCode like "50*X" 的记录
        List<SuperviseInfo> supervisionInfos = superviseInfoMapper.selectList(
                new LambdaQueryWrapper<SuperviseInfo>()
                        .select(SuperviseInfo::getRecord)
                        .like(SuperviseInfo::getOrgBranchCode, "50*X")
        );

        Set<String> records = supervisionInfos.stream()
                .map(SuperviseInfo::getRecord)
                .filter(StringUtils::isNotEmpty)
                .collect(Collectors.toSet());

        if (records.isEmpty()) {
            return 0;
        }

        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.termsQuery("SEQUENCE_NBR.keyword", records));

        List<ESEquipmentCategoryDto> esResults;
        try {
            esResults = registrationManageService.searchResponse(
                    IDX_BIZ_VIEW_JG_ALL,
                    boolQuery,
                    hit -> JSONObject.parseObject(hit.getSourceAsString(), ESEquipmentCategoryDto.class)
            );
        } catch (Exception ex) {
            log.error("ES 查询异常: {}", ex.getMessage(), ex);
            throw new RuntimeException("ES 查询失败", ex);
        }

        List<Map<String, Object>> updateList = esResults.stream()
                .filter(dto -> StringUtils.isNotEmpty(dto.getOrgBranchCode())
                        && StringUtils.isNotEmpty(dto.getSEQUENCE_NBR())
                )
                .map(this::buildUpdateMap)
                .collect(Collectors.toList());

        if (updateList.isEmpty()) {
            return 0;
        }

        int updateCount = 0;
        List<List<Map<String, Object>>> partitions = Lists.partition(updateList, BATCH_SIZE);
        for (List<Map<String, Object>> batch : partitions) {
            if (isModify) {
                int count = superviseInfoMapper.batchUpdateOrgBranch(batch);
                updateCount += count;
                log.info("更新批次：{}, 成功记录数：{}", batch.size(), count);
            }
        }
        return updateList.size();
    }

    private Map<String, Object> buildUpdateMap(ESEquipmentCategoryDto dto) {
        Map<String, Object> map = new HashMap<>();
        map.put("record", dto.getSEQUENCE_NBR());
        map.put("orgBranchCode", dto.getOrgBranchCode());
        map.put("orgBranchName", dto.getORG_BRANCH_NAME());
        log.info("==========>record,{},orgBranchCode,{},orgBranchName,{}", dto.getSEQUENCE_NBR(), dto.getOrgBranchCode(), dto.getORG_BRANCH_NAME());
        return map;
    }

    public Integer deleteImportedPipelineData(String uscUnitCreditCode, String projectContraption,
                                              String dataSource, boolean isDelete) throws IOException {
        List<String> records = this.queryImportedPipelineRecords(uscUnitCreditCode, projectContraption, dataSource);
        if (records.isEmpty()) {
            return 0;
        }
        if (isDelete) {
            int batchSize = 1000;
            for (int i = 0; i < records.size(); i += batchSize) {
                int end = Math.min(i + batchSize, records.size());
                List<String> batch = records.subList(i, end);
                superviseInfoMapper.deleteDataAll(batch);
                deleteFromEs(batch);
            }
        }
        return records.size();
    }

    /**
     * 查询需要删除的设备
     */
    private List<String> queryImportedPipelineRecords(String uscUnitCreditCode,
                                                      String projectContraption,
                                                      String dataSource) throws IOException {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery("DATA_SOURCE.keyword", dataSource))
                .must(QueryBuilders.termQuery("IS_INTO_MANAGEMENT", false))
                .must(QueryBuilders.termQuery("PROJECT_CONTRAPTION.keyword", projectContraption))
                .must(QueryBuilders.wildcardQuery(
                        "USC_UNIT_CREDIT_CODE", "*" + QueryParser.escape(uscUnitCreditCode.toLowerCase()) + "*"))
                .must(QueryBuilders.termQuery("STATUS", "已认领"));
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(boolQuery)
                .fetchSource(new String[]{"SEQUENCE_NBR"}, null)
                .size(10000);

        SearchRequest request = new SearchRequest(IDX_BIZ_VIEW_JG_ALL).source(sourceBuilder);
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        return Arrays.stream(response.getHits().getHits())
                .map(hit -> (String) hit.getSourceAsMap().get("SEQUENCE_NBR"))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    /**
     * 删除ES中的数据（两个索引）
     */
    private void deleteFromEs(List<String> records) {
        List<ESEquipmentCategoryDto> esDtoList = buildEsCategoryDtos(records);
        List<ESEquipmentInfo> esNewDtoList = buildEsInfoDtos(records);

        if (!esDtoList.isEmpty()) {
            esEquipmentCategory.deleteAll(esDtoList);
        }
        if (!esNewDtoList.isEmpty()) {
            esEquipmentDao.deleteAll(esNewDtoList);
        }
    }

    private List<ESEquipmentCategoryDto> buildEsCategoryDtos(List<String> records) {
        return records.stream().map(v -> {
            ESEquipmentCategoryDto dto = new ESEquipmentCategoryDto();
            dto.setSEQUENCE_NBR(v);
            return dto;
        }).collect(Collectors.toList());
    }

    private List<ESEquipmentInfo> buildEsInfoDtos(List<String> records) {
        return records.stream().map(v -> {
            ESEquipmentInfo dto = new ESEquipmentInfo();
            dto.setSEQUENCE_NBR(v);
            return dto;
        }).collect(Collectors.toList());
    }

    public List<Long> generateSeq(String num) {
        List<Long> result = new ArrayList<>();
        for (long l = Long.parseLong(num); l > 0; l--) {
            result.add(sequence.nextId());
        }
        return result;
    }

    public String safetyProblemDataWriteRegionCodeAndName() {
        int maxNum = 5000;
        int pageNum = 1;
        int result;
        int count = result = safetyProblemTracingService.count();
        while (count > 0) {
            Page<SafetyProblemTracing> page = new Page<>(pageNum++, maxNum);
            IPage<SafetyProblemTracing> safetyProblemTracingIPage = safetyProblemTracingService.page(page);
            List<SafetyProblemTracing> safetyProblemTracingList = safetyProblemTracingIPage.getRecords();
            safetyProblemTracingList.forEach(x -> {
                JSONObject region = SafetyProblemTopicMessage.getRegionBySourceTypeAndId(x.getSourceType(), x.getSourceId());
                x.setRegionCode(region.getString("regionCode"));
                x.setRegionName(region.getString("regionName"));
            });
            safetyProblemTracingService.saveOrUpdateBatch(safetyProblemTracingList);
            count -= maxNum;
        }
        return String.format("刷新数据数量:%s", result);
    }

    /**
     * 刷新西安电梯数据，使其可以做后续业务
     * 使用登记证表的isDoBusiness 0 -> 1 ，与之对应的设备数据 isDoBusiness 0 -> 1
     *
     * @return 刷新使用登记证数量和设备数量
     */
    public String refreshXianData() {
        // 证管理表中sDoBusiness 0 数据
        List<JgUseRegistrationManage> manageList = jgUseRegistrationManageServiceImpl.lambdaQuery()
                .eq(JgUseRegistrationManage::getIsDoBusiness, Boolean.FALSE)
                .eq(JgUseRegistrationManage::getIsDelete, Boolean.FALSE)
                .list();
        // 对应的设备数据
        List<Long> manageSeqs = manageList.stream()
                .map(JgUseRegistrationManage::getSequenceNbr)
                .collect(Collectors.toList());

        List<List<Long>> managePartitions = Lists.partition(manageSeqs, BATCH_SIZE);
        for (List<Long> part : managePartitions) {
            jgUseRegistrationManageServiceImpl.lambdaUpdate()
                    .set(JgUseRegistrationManage::getIsDoBusiness, Boolean.TRUE)
                    .in(JgUseRegistrationManage::getSequenceNbr, part)
                    .update();
        }
        List<String> useRegistrantionCodeList = manageList.stream()
                .map(JgUseRegistrationManage::getUseRegistrationCode)
                .collect(Collectors.toList());
        List<List<String>> useRegistrationCodes = Lists.partition(useRegistrantionCodeList, BATCH_SIZE);
        for (List<String> useRegistrations : useRegistrationCodes) {
            List<String> recordList = registerInfoService.lambdaQuery()
                    .in(IdxBizJgRegisterInfo::getUseOrgCode, useRegistrations)
                    .list()
                    .stream()
                    .map(IdxBizJgRegisterInfo::getRecord)
                    .collect(Collectors.toList());

            // 批量查询 & 更新设备数据
            Iterable<ESEquipmentInfo> esEquipmentBatch = esEquipmentDao.findAllById(recordList);
            esEquipmentBatch.forEach(x -> {
                x.setUSE_PLACE("陕西省/西安市");
                x.setUSE_PLACE_CODE("610000#610100");
                x.setIS_DO_BUSINESS(Boolean.TRUE);
            });
            esEquipmentDao.saveAll(esEquipmentBatch);

            // 批量查询 & 更新设备分类数据
            Iterable<ESEquipmentCategoryDto> equipmentCategoryBatch = esEquipmentCategory.findAllById(recordList);
            equipmentCategoryBatch.forEach(x -> {
                x.setIS_DO_BUSINESS(Boolean.TRUE);
                x.setUSE_PLACE("陕西省/西安市");
                x.setUSE_PLACE_CODE("610000#610100");
            });
            esEquipmentCategory.saveAll(equipmentCategoryBatch);
        }
        return String.format("刷新证管理表数据：%s 条；对应设备ES数据:%s 条。", manageList.size(), useRegistrantionCodeList.size());
    }

    public String refreshXianData2ES(boolean isUpdate) throws IOException {
        List<String> records = this.queryIsDoBusinessRecords();
        if (records.isEmpty()) {
            return "没有需要刷新数据！";
        }
        if (isUpdate) {
            int batchSize = 1000;
            int total = records.size();
            for (int i = 0; i < total; i += batchSize) {
                int end = Math.min(i + batchSize, total);
                List<String> batch = new ArrayList<>(records.subList(i, end));
                Iterable<ESEquipmentCategoryDto> equipmentCategoryBatch = esEquipmentCategory.findAllById(batch);
                // 批量修改
                for (ESEquipmentCategoryDto x : equipmentCategoryBatch) {
                    x.setIS_DO_BUSINESS(Boolean.TRUE);
                    x.setUSE_PLACE("陕西省/西安市");
                    x.setUSE_PLACE_CODE("610000#610100");
                }
                esEquipmentCategory.saveAll(equipmentCategoryBatch);

                Iterable<ESEquipmentInfo> esEquipmentBatch = esEquipmentDao.findAllById(batch);
                esEquipmentBatch.forEach(x -> {
                    x.setUSE_PLACE("陕西省/西安市");
                    x.setUSE_PLACE_CODE("610000#610100");
                    x.setIS_DO_BUSINESS(Boolean.TRUE);
                });
                esEquipmentDao.saveAll(esEquipmentBatch);
            }
        }
        return String.format("刷新对应设备ES数据: %d 条。", records.size());
    }

    /**
     * 查询需要删除的设备
     */
    private List<String> queryIsDoBusinessRecords() throws IOException {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery("IS_DO_BUSINESS", false))
                .must(QueryBuilders.termQuery("IS_INTO_MANAGEMENT", false))
                .must(QueryBuilders.termQuery("STATUS", "已认领"));
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(boolQuery)
                .fetchSource(new String[]{"SEQUENCE_NBR"}, null)
                .size(10000);

        SearchRequest request = new SearchRequest(IDX_BIZ_VIEW_JG_ALL).source(sourceBuilder);
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        return Arrays.stream(response.getHits().getHits())
                .map(hit -> (String) hit.getSourceAsMap().get("SEQUENCE_NBR"))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    public Integer deleteEquipIsNotClaimed(String useUnitCreditCode, String equList,
                                           String dataSource, boolean isDelete) {
        List<IdxBizJgRegisterInfo> jgRegisterInfoList = registerInfoMapper.selectDeleteEquipIsNotClaimed(useUnitCreditCode, equList, dataSource);
        List<String> records = Optional.ofNullable(jgRegisterInfoList)
                .orElse(Collections.emptyList())
                .stream()
                .filter(Objects::nonNull)
                .map(IdxBizJgRegisterInfo::getRecord)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (!records.isEmpty() && isDelete) {
            int batchSize = 1000;
            for (int i = 0; i < records.size(); i += batchSize) {
                int end = Math.min(i + batchSize, records.size());
                List<String> batch = records.subList(i, end);
                superviseInfoMapper.deleteDataAll(batch);
                deleteFromEs(batch);
            }
        }
        return records.size();
    }


    public Integer initTank2Es() {
        return weatherTankFieldPatcher.patchBatchData(null);
    }

    @Transactional(rollbackFor = Exception.class)
    public String handleDataQualityScore() {
        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)) {
                equipRecord.removeIf(Objects::isNull);
                log.info("发布设备事件，设备数量: {}", equipRecord.size());
                if (!ValidationUtil.isEmpty(equipRecord)) {
                    eventPublisher.publish(new EquipCreateOrEditEvent(
                            this,
                            BusinessTypeEnum.JG_NEW_EQUIP.name(),
                            equipRecord,
                            EquipCreateOrEditEvent.EquipType.equip
                    ));
                }
            }

            if (!ValidationUtil.isEmpty(projectContraptionIds)) {
                projectContraptionIds.removeIf(Objects::isNull);
                log.info("发布装置设备事件，设备数量: {}", projectContraptionIds.size());
                if (!ValidationUtil.isEmpty(projectContraptionIds)) {
                    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);
    }

    public Boolean equDefineChangeFix() {
        String equCategory = "6E00";
        Map<String, String> oldFixEquDefineCodeMap = new HashMap<>();
        oldFixEquDefineCodeMap.put("6E20", "6E40");
        oldFixEquDefineCodeMap.put("6E30", "6E50");
        oldFixEquDefineCodeMap.put("6E40", "6E60");
        List<IdxBizJgRegisterInfo> regs = registerInfoService.list(new LambdaQueryWrapper<IdxBizJgRegisterInfo>()
                .eq(IdxBizJgRegisterInfo::getEquCategory, equCategory)
                .select(IdxBizJgRegisterInfo::getRecord, IdxBizJgRegisterInfo::getEquDefine));
        log.info("需要更新设备的品种的无动力游乐设施设备数量：{}", regs.size());
        regs.parallelStream().forEach(re -> {
            if (oldFixEquDefineCodeMap.containsKey(re.getEquDefine())) {
                // 1.idx_biz_jg_register_info 更新
                LambdaUpdateWrapper<IdxBizJgRegisterInfo> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
                lambdaUpdateWrapper.set(IdxBizJgRegisterInfo::getEquDefine, oldFixEquDefineCodeMap.get(re.getEquDefine()));
                lambdaUpdateWrapper.eq(IdxBizJgRegisterInfo::getRecord, re.getRecord());
                registerInfoService.update(lambdaUpdateWrapper);
                // 2.设备老索引修改
                Optional<ESEquipmentCategoryDto> op1 = esEquipmentCategory.findById(re.getRecord());
                if (op1.isPresent()) {
                    ESEquipmentCategoryDto esEquipmentCategoryDto = op1.get();
                    esEquipmentCategoryDto.setEQU_DEFINE_CODE(oldFixEquDefineCodeMap.get(re.getEquDefine()));
                    esEquipmentCategory.save(esEquipmentCategoryDto);
                }
                // 3.设备新索引修改
                Optional<ESEquipmentInfo> op2 = esEquipmentDao.findById(re.getRecord());
                if (op2.isPresent()) {
                    ESEquipmentInfo esEquipmentInfo = op2.get();
                    esEquipmentInfo.setEQU_DEFINE_CODE(oldFixEquDefineCodeMap.get(re.getEquDefine()));
                    esEquipmentDao.save(esEquipmentInfo);
                }
            }
        });
        // 4.证tzs_jg_use_registration_manage更新
        List<JgUseRegistrationManage> manages = jgUseRegistrationManageMapper.selectList(new LambdaQueryWrapper<JgUseRegistrationManage>()
                .eq(JgUseRegistrationManage::getEquCategory, equCategory)
                .select(JgUseRegistrationManage::getEquDefine, BaseEntity::getSequenceNbr));
        log.info("需要更新设备的品种的无动力游乐设施证数量：{}", regs.size());
        manages.parallelStream().forEach(m -> {
            if (oldFixEquDefineCodeMap.containsKey(m.getEquDefine())) {
                LambdaUpdateWrapper<JgUseRegistrationManage> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
                lambdaUpdateWrapper.set(JgUseRegistrationManage::getEquDefine, oldFixEquDefineCodeMap.get(m.getEquDefine()));
                lambdaUpdateWrapper.eq(JgUseRegistrationManage::getSequenceNbr, m.getSequenceNbr());
                jgUseRegistrationManageMapper.update(null, lambdaUpdateWrapper);
            }
        });
        return Boolean.TRUE;
    }

    /**
     * 压力管道-已完成及作废状态的单据的历史数据管道长度补充
     *
     * @return 是否成功
     */
    public Long pipeLenFix() {
        cn.hutool.core.date.StopWatch watch = new cn.hutool.core.date.StopWatch();
        // 1.已完成及作废状态的安装告知
        List<JgInstallationNotice> notices = installationNoticeService.list(new LambdaQueryWrapper<JgInstallationNotice>()
                .and(w -> w.eq(JgInstallationNotice::getNoticeStatus, FlowStatusEnum.TO_BE_FINISHED.getCode() + "")
                        .or().eq(JgInstallationNotice::getNoticeStatus, FlowStatusEnum.TO_BE_DISCARD.getCode() + ""))
                .eq(JgInstallationNotice::getEquListCode, EquipmentClassifityEnum.YLGD.getCode())
                .select(BaseEntity::getSequenceNbr)
        );
        watch.start("安装告知" + notices.size());
        notices.parallelStream().forEach(n -> {
            JSONObject jsonObject = commonServiceImpl.queryHistoryData(n.getSequenceNbr());
            if (jsonObject != null) {
                JSONArray jsonArray = jsonObject.getJSONArray("deviceList");
                List<JSONObject> pipelines = jsonArray.stream().map(e -> {
                    JSONObject pipeline = JSONObject.parseObject(e.toString());
                    if (!pipeline.containsKey(BizCommonConstant.PIPE_LENGTH)) {
                        pipeline.put(BizCommonConstant.PIPE_LENGTH, pipeline.get("pipeLength"));
                    }
                    return pipeline;
                }).collect(Collectors.toList());
                jsonObject.put("deviceList", pipelines);
                commonServiceImpl.saveOrUpdateHistory(null, jsonObject, null, n.getSequenceNbr() + "");
            }
        });
        watch.stop();
        // 2.已完成及作废状态的使用登记
        List<JgUseRegistration> useRegistrations = useRegistrationService.list(new LambdaQueryWrapper<JgUseRegistration>()
                .and(w -> w.eq(JgUseRegistration::getStatus, FlowStatusEnum.TO_BE_FINISHED.getName()).or()
                        .eq(JgUseRegistration::getStatus, FlowStatusEnum.TO_BE_DISCARD.getName()))
                .ne(JgUseRegistration::getProjectContraptionId, "")
                .select(BaseEntity::getSequenceNbr));
        watch.start("使用登记" + useRegistrations.size());
        useRegistrations.parallelStream().forEach(u -> {
            JSONObject jsonObject = commonServiceImpl.queryHistoryData(u.getSequenceNbr());
            if (jsonObject != null) {
                String pipelistKey;
                if (jsonObject.containsKey("equipmentLists")) {
                    pipelistKey = "equipmentLists";
                } else {
                    pipelistKey = "pipelineList";
                }
                JSONArray jsonArray = jsonObject.getJSONArray(pipelistKey);
                Optional.ofNullable(jsonArray).ifPresent(d -> {
                    List<JSONObject> pipelines = d.stream().map(e -> {
                        JSONObject pipeline = JSONObject.parseObject(e.toString());
                        if (!pipeline.containsKey(BizCommonConstant.PIPE_LENGTH)) {
                            pipeline.put(BizCommonConstant.PIPE_LENGTH, pipeline.get("pipeLength"));
                        }
                        return pipeline;
                    }).collect(Collectors.toList());
                    jsonObject.put(pipelistKey, pipelines);
                    commonServiceImpl.saveOrUpdateHistory(null, jsonObject, null, u.getSequenceNbr() + "");
                });
            }
        });
        watch.stop();
        // 3.已完成及作废状态的改造变更登记
        List<JgChangeRegistrationReform> changeRegistrationReforms = jgChangeRegistrationReformMapper.selectList(new LambdaQueryWrapper<JgChangeRegistrationReform>()
                .and(w -> w.eq(JgChangeRegistrationReform::getStatus, FlowStatusEnum.TO_BE_FINISHED.getName()).or()
                        .eq(JgChangeRegistrationReform::getStatus, FlowStatusEnum.TO_BE_DISCARD.getName()))
                .ne(JgChangeRegistrationReform::getProjectContraptionId, "")
                .select(JgChangeRegistrationReform::getApplyNo));
        watch.start("改造变更登记" + changeRegistrationReforms.size());
        changeRegistrationReforms.parallelStream().forEach(u -> {
            JSONObject jsonObject = commonServiceImpl.queryHistoryData(u.getApplyNo());
            if (jsonObject != null) {
                JSONArray jsonArray = jsonObject.getJSONArray("equipmentLists");
                Optional.ofNullable(jsonArray).ifPresent(d -> {
                    List<JSONObject> pipelines = d.stream().map(e -> {
                        JSONObject pipeline = JSONObject.parseObject(e.toString());
                        if (!pipeline.containsKey(BizCommonConstant.PIPE_LENGTH)) {
                            pipeline.put(BizCommonConstant.PIPE_LENGTH, pipeline.get("pipeLength"));
                        }
                        return pipeline;
                    }).collect(Collectors.toList());
                    jsonObject.put("equipmentLists", pipelines);
                    commonServiceImpl.saveOrUpdateHistory(null, jsonObject, null, u.getApplyNo());
                });
            }
        });
        watch.stop();
        // 4.已完成及作废状态的改造告知
        List<JgReformNotice> reformNotices = jgReformNoticeMapper.selectList(new LambdaQueryWrapper<JgReformNotice>()
                .and(w -> w.eq(JgReformNotice::getNoticeStatus, FlowStatusEnum.TO_BE_FINISHED.getCode() + "").or()
                        .eq(JgReformNotice::getNoticeStatus, FlowStatusEnum.TO_BE_DISCARD.getCode() + ""))
                .ne(JgReformNotice::getProjectContraptionId, "")
                .select(BaseEntity::getSequenceNbr));
        watch.start("改造告知" + reformNotices.size());
        reformNotices.parallelStream().forEach(u -> {
            JSONObject jsonObject = commonServiceImpl.queryHistoryData(u.getSequenceNbr());
            if (jsonObject != null) {
                JSONArray jsonArray = jsonObject.getJSONArray("deviceList");
                Optional.ofNullable(jsonArray).ifPresent(d -> {
                    List<JSONObject> pipelines = d.stream().map(e -> {
                        JSONObject pipeline = JSONObject.parseObject(e.toString());
                        if (!pipeline.containsKey(BizCommonConstant.PIPE_LENGTH)) {
                            pipeline.put(BizCommonConstant.PIPE_LENGTH, pipeline.get("pipeLength"));
                        }
                        return pipeline;
                    }).collect(Collectors.toList());
                    jsonObject.put("deviceList", pipelines);
                    commonServiceImpl.saveOrUpdateHistory(null, jsonObject, null, u.getSequenceNbr() + "");
                });
            }
        });
        watch.stop();
        int num = notices.size() + useRegistrations.size() + reformNotices.size() + changeRegistrationReforms.size();
        log.info("压力管道业务单据补充字段pipeLengthText，总处理数量：{}， 耗时信息：{}", num, watch.prettyPrint(TimeUnit.SECONDS));
        return (long) (num);
    }


    public Integer receiveOrgFix() {
        return receiveOrgFixService.doFix();
    }

    public void addDbData2EsBatch(Map<String, String> paramMap) {
        log.info("开始同步数据库设备数据到ES");
        StopWatch watch = new StopWatch();
        watch.start("同步设备数据到ES");

        String dataSource = MapUtil.getStr(paramMap, "dataSource");

        if (!ValidationUtil.isEmpty(MapUtil.getStr(paramMap, "records"))) {
            List<String> records = Stream.of(MapUtil.getStr(paramMap, "records").split(",")).map(String::trim).collect(Collectors.toList());
            queryAndSave2Es(records);
        } else {
            List<String> extraInEs;
            List<String> dbRecords;
            // 1. 查询数据库中所有设备 record
            LambdaQueryWrapper<IdxBizJgUseInfo> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.select(IdxBizJgUseInfo::getRecord).likeLeft(!ValidationUtil.isEmpty(dataSource), IdxBizJgUseInfo::getDataSource, dataSource);
            List<IdxBizJgUseInfo> list = idxBizJgUseInfoMapper.selectList(lambdaQueryWrapper);
            dbRecords = list.stream().map(IdxBizJgUseInfo::getRecord).collect(Collectors.toList());

            // 2. 从ES中查询对应记录
            List<String> esRecords = getEsData(dataSource);

            // 3. 对比差异（在DB没在ES的记录）
            extraInEs = findElementsInANotInB(dbRecords, esRecords);
            queryAndSave2Es(extraInEs);
        }

        watch.stop();
        log.info("同步数据库设备数据到ES结束，耗时：{}s", watch.getTotalTimeSeconds());
    }

    private void queryAndSave2Es(List<String> records) {
        // 4.extraInEs分批从数据库获取数据写入ES，每批次1000
        List<List<String>> extraInEsBatches = Lists.partition(records, 1000);
        AtomicInteger batchIndex = new AtomicInteger(0);
        extraInEsBatches.parallelStream().forEach(batch -> {
            StopWatch batchWatch = new StopWatch();
            batchWatch.start("批次" + batchIndex.get());
            int currentIndex = batchIndex.getAndIncrement();
            log.info("开始处理第 {} 个批次", currentIndex);
            List<Map<String, Object>> batchData = idxBizJgUseInfoMapper.queryDetailBatch(batch);
            List<ESEquipmentCategoryDto> esEquipmentInfos = batchData.parallelStream().map(data -> {
                ESEquipmentCategoryDto esEquipmentInfo = null;
                try {
                    esEquipmentInfo = new ESEquipmentCategoryDto();
                    StatisticsDataUpdateService.formatUseDate(data);
                    BeanUtil.copyProperties(data, esEquipmentInfo, true);
                } catch (Exception e) {
                    log.error("批次{}设备刷数据处理失败：{}", currentIndex, record, e);
                }
                return esEquipmentInfo;
            }).collect(Collectors.toList());
            esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
            if (!esEquipmentInfos.isEmpty()) {
                esEquipmentCategory.saveAll(esEquipmentInfos);
            }
            batchWatch.stop();
            log.info("批次{}处理完成，耗时：{}s", currentIndex, batchWatch.getTotalTimeSeconds());
        });
    }

    /**
     * 从 ES 中拉取索引中所有的 SEQUENCE_NBR
     */
    private List<String> getEsData(String dataSource) {
        // 分批拉取文档
        log.info("ES 查询开始, index={}, 时间={}", DataHandlerServiceImpl.IDX_BIZ_VIEW_JG_ALL, System.currentTimeMillis());
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        if (!ValidationUtil.isEmpty(dataSource)) {
            boolQuery = QueryBuilders.boolQuery()
                    .must(QueryBuilders.termQuery("DATA_SOURCE", dataSource));
        }
        // 线程安全的结果容器
        List<String> esIds = Collections.synchronizedList(new ArrayList<>());

        try {
            esSearchService.searchResponseInBatch(DataHandlerServiceImpl.IDX_BIZ_VIEW_JG_ALL, boolQuery, 2000, searchHits -> {
                // 每个批次只解析需要的字段
                List<String> batchIds = searchHits.stream()
                        .map(hit -> {
                            ESEquipmentCategoryDto dto = JSONObject.parseObject(hit.getSourceAsString(), ESEquipmentCategoryDto.class);
                            return dto == null ? null : dto.getSEQUENCE_NBR();
                        })
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());
                esIds.addAll(batchIds);
            });
        } catch (Exception ex) {
            log.error("ES 查询异常, index={}, 错误={}", DataHandlerServiceImpl.IDX_BIZ_VIEW_JG_ALL, ex.getMessage(), ex);
            throw new RuntimeException("ES 查询失败: " + ex.getMessage(), ex);
        }
        // 去重并返回
        return esIds.stream().distinct().collect(Collectors.toList());
    }

    public static List<String> findElementsInANotInB(List<String> listA, List<String> listB) {
        Set<String> bSet = Collections.synchronizedSet(new HashSet<>(listB));

        return listA.parallelStream()
                .filter(element -> !bSet.contains(element))
                .collect(Collectors.toList());
    }

    public Long orgBranchCode2Db(String equListCode, String equCategoryCode, String orgBranchCode, Integer seq) {
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.trackTotalHits(true);
        BoolQueryBuilder boolMust = QueryBuilders.boolQuery();
        boolMust.must(QueryBuilders.boolQuery().must(QueryBuilders.wildcardQuery("ORG_BRANCH_CODE.keyword", QueryParser.escape(orgBranchCode) + "*")));
        boolMust.must(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("EQU_LIST_CODE", equListCode)));
        if (StringUtils.isNotEmpty(equCategoryCode)) {
            boolMust.must(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("EQU_CATEGORY_CODE", equCategoryCode)));
        }
        // 数据库更新属地的设备总数
        long totalUpdate = 0L;
        try {
            String version = DateUtil.today() + seq;
            buildVersionFilter(version, boolMust);
            CountRequest countRequest = new CountRequest();
            countRequest.indices(IDX_BIZ_VIEW_JG_ALL);
            countRequest.query(boolMust);
            CountResponse countResponse = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
            long total = countResponse.getCount();
            int pageSize = 1000;
            int totalPage = (int) Math.ceil((double) total / pageSize);
            builder.size(pageSize);
            builder.query(boolMust);
            builder.sort("REC_DATE", SortOrder.DESC);
            SearchRequest request = new SearchRequest();
            request.indices(IDX_BIZ_VIEW_JG_ALL);
            for (int i = 1; i <= totalPage; i++) {
                request.source(builder);
                SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
                List<ESEquipmentCategoryDto> esEquipmentCategoryDtos = new ArrayList<>();
                for (SearchHit hit : response.getHits()) {
                    JSONObject jsonObject = (JSONObject) JSONObject.toJSON(hit);
                    ESEquipmentCategoryDto equipmentCategoryDto = JSONObject.toJavaObject(jsonObject.getJSONObject("sourceAsMap"), ESEquipmentCategoryDto.class);
                    Integer recordCount = useInfoService.lambdaQuery().eq(IdxBizJgUseInfo::getRecord, equipmentCategoryDto.getSEQUENCE_NBR()).count();
                    if (recordCount > 0) {
                        equipmentCategoryDto.setVersion(version);
                    } else {
                        // 数据库不存在的做标记
                        equipmentCategoryDto.setVersion("-1");
                    }
                    esEquipmentCategoryDtos.add(equipmentCategoryDto);
                }
                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);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return totalUpdate;
    }

    private static void buildVersionFilter(String version, BoolQueryBuilder boolMust) {
        BoolQueryBuilder meBuilder = QueryBuilders.boolQuery();
        meBuilder.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery("version", version, "-1")));
        meBuilder.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("version")));
        meBuilder.minimumShouldMatch(1);
        boolMust.must(meBuilder);
    }

    public Integer synEquipFromDb2Es(String equListCode, String equCategoryCode, String orgBranchCode) {
        Map<String, Object> p = new HashMap<>();
        p.put("equListCode", equListCode);
        p.put("equCategoryCode", equCategoryCode);
        p.put("orgBranchCode", orgBranchCode);
        return equipInsert2EsPatcher.patchBatchData(p);
    }

    public Integer updateMainNoticeJson() {
        log.info("处理维修告知的json历史数据开始");
        StopWatch watch = new StopWatch();
        watch.start();
        List<JgMaintainNotice> maintainNoticesNotices = jgMaintainNoticeServiceImpl.list();
        maintainNoticesNotices.parallelStream().forEach(r -> {
            List<String> records = jgMaintainNoticeEqMapper.selectList(
                            new LambdaQueryWrapper<JgMaintainNoticeEq>()
                                    .eq(JgMaintainNoticeEq::getEquipTransferId, r.getSequenceNbr()))
                    .stream()
                    .map(JgMaintainNoticeEq::getEquId)
                    .collect(Collectors.toList());
            if (!records.isEmpty()) {
                List<Map<String, Object>> deviceList = jgMaintainNoticeServiceImpl.getDeviceList(r, records);
                JSONObject hisData = new JSONObject();
                BeanUtil.copyProperties(r, hisData);
                hisData.put("deviceList", deviceList);
                commonServiceImpl.saveOrUpdateHistory(BusinessTypeEnum.JG_RENOVATION_REGISTRATION.getName(), hisData, null, r.getSequenceNbr() + "");
            }
        });
        watch.stop();
        log.info("处理维修告知的json历史数据结束：处理数量：「{}」，耗时：「{}」", maintainNoticesNotices.size(), watch.getTotalTimeSeconds());
        return maintainNoticesNotices.size();
    }

    public Integer insertEquipFromDb2NewEs(Map<String, String> paramMap) {
        List<String> refreshRecords = Collections.emptyList();
        if (!ValidationUtil.isEmpty(MapUtil.getStr(paramMap, "records"))) {
            refreshRecords = Stream.of(MapUtil.getStr(paramMap, "records").split(",")).map(String::trim).collect(Collectors.toList());
            this.patchBatchRecord(refreshRecords);
        }
        return refreshRecords.size();
    }

    protected void patchBatchRecord(List<String> refreshRecords) {
        log.info("批量处理设备信息到es开始");
        StopWatch watch = new StopWatch();
        watch.start("批量查询设备信息");
        List<Map<String, Object>> details = idxBizJgUseInfoMapper.queryDetailBatch(refreshRecords);
        watch.stop();
        Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
//        watch.start("组装es设备老索引更新及新增的对象数据");
//        // 组装es设备老索引更新及新增的对象数据
//        List<ESEquipmentCategoryDto> esEquipmentCategoryDtos = getEsEquipmentCategoryDtos(refreshRecords, recordDetailMap);
//        watch.stop();
        watch.start("组装es设备新索引更新及新增的对象数据");
        // 组装es设备新索引更新及新增的对象数据
        List<ESEquipmentInfo> esEquipmentInfos = getEsEquipmentInfos(refreshRecords, recordDetailMap);
        watch.stop();
        watch.start("es设备新旧索引保存");
        // 多线程保存
        List<CompletableFuture<Void>> futures = new ArrayList<>();
//        if (!esEquipmentCategoryDtos.isEmpty()) {
//            futures.add(CompletableFuture.runAsync(() -> {
//                StopWatch watch4 = new StopWatch();
//                watch4.start();
//                esBulkService.bulkUpsert(IDX_BIZ_VIEW_JG_ALL, esEquipmentCategoryDtos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
//                watch4.stop();
//                log.warn("[设备老索引] 批量入库 {} 条，耗时 {}s",
//                        esEquipmentCategoryDtos.size(), watch4.getTotalTimeSeconds());
//            }));
//        }
        if (!esEquipmentInfos.isEmpty()) {
            futures.add(CompletableFuture.runAsync(() -> {
                StopWatch watch4 = new StopWatch();
                watch4.start();
                esBulkService.bulkUpsert(IDX_BIZ_EQUIPMENT_INFO, esEquipmentInfos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
                watch4.stop();
                log.warn("[设备新索引] 批量入库 {} 条，耗时 {}s",
                        esEquipmentInfos.size(), watch4.getTotalTimeSeconds());
            }));
        }
        // 等待所有任务完成（阻塞当前线程）
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        watch.stop();
        log.warn("新索引数据补充匹配补充，总耗时情况：{}", watch.prettyPrint());
    }


    private List<ESEquipmentInfo> getEsEquipmentInfos(List<String> refreshRecords, Map<String, Map<String, Object>> recordDetailMap) {
        // 设备最新的维保信息-维度：设备
        List<IdxBizJgMaintenanceRecordInfo> lastMaintenanceRecordInfos = maintenanceRecordInfoMapper.selectLastedMainInfoBatch(refreshRecords);
        Map<String, List<IdxBizJgMaintenanceRecordInfo>> recordLastMaintMap = lastMaintenanceRecordInfos.stream().collect(Collectors.groupingBy(IdxBizJgMaintenanceRecordInfo::getRecord));

        // 设备、各检验类型下最新的检验信息-维度：设备、检验类型
        List<IdxBizJgInspectionDetectionInfo> lastedInspectInfosGroupByInspectType = inspectionDetectionInfoMapper.selectLastedGroupByInspectTypeBatch(refreshRecords);
        Map<String, List<IdxBizJgInspectionDetectionInfo>> recordInspectInfosGroupByInspectTypeMap = lastedInspectInfosGroupByInspectType.stream().collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord));

        // 设备最新的检验信息-维度：设备
        Map<String, Optional<IdxBizJgInspectionDetectionInfo>> recordLastInspectionMap = lastedInspectInfosGroupByInspectType.stream().filter(e -> e.getNextInspectDate() != null).collect(Collectors.groupingBy(IdxBizJgInspectionDetectionInfo::getRecord, Collectors.maxBy(Comparator.comparing(IdxBizJgInspectionDetectionInfo::getNextInspectDate))));
        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);
                // 最新检验信息-维度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(refreshHandler.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(refreshHandler.buildTechParamByEquList(record, esEquipmentInfo.getEQU_LIST_CODE()));
                if ("8000".equals(esEquipmentInfo.getEQU_LIST_CODE())) {
                    List<ESEquipmentInfo.TechParam> techParams = esEquipmentInfo.getTechParams();
                    List<ESEquipmentInfo.TechParam> pipeLength = techParams.stream().filter(e -> e.getParamKey().equals("pipeLength") && e.getDoubleValue() != null).collect(Collectors.toList());
                    if (!ObjectUtils.isEmpty(pipeLength)) {
                        esEquipmentInfo.setPipeLength(pipeLength.get(0).getDoubleValue());
                    }
                }
            } catch (Exception e) {
                // 异常数据跳过
                log.error("准备新设备索引数据失败：{}", record, e);
            }
            return esEquipmentInfo;
        }).collect(Collectors.toList());
        esEquipmentInfos.remove(null);
        esEquipmentInfos = esEquipmentInfos.stream().filter(e -> StringUtils.isNotEmpty(e.getSEQUENCE_NBR())).collect(Collectors.toList());
        return esEquipmentInfos;
    }



    @SneakyThrows
    public boolean syncEquipShard(Integer shard, Integer slots, String batchNo) {
        StopWatch watch = new StopWatch();
        watch.start();
        // 0.数据准备
        // 查询数据库 批量
        List<String> records = idxBizJgUseInfoMapper.selectRecordList(shard, slots);
        List<List<String>> partitions = Lists.partition(records, 2000);
        // 查询es-旧索引
        List<String> existsOldEsEquipRecords = partitions.parallelStream().map(equipmentCategoryService::findSequenceNbrByIds).flatMap(List::stream).collect(Collectors.toList());
        //  标记老索引es-缺少数据
        List<String> oldEsMissingRecords = records.parallelStream().filter(r -> !existsOldEsEquipRecords.contains(r)).collect(Collectors.toList());
        // 查询es-新索引
        List<String> existsNewEsEquipRecords =  partitions.parallelStream().map(esEquipmentInfoService::findSequenceNbrByIds).flatMap(List::stream).collect(Collectors.toList());
        // 标记新索引es-缺少数据
        List<String> newEsMissingRecords = records.parallelStream().filter(r -> !existsNewEsEquipRecords.contains(r)).collect(Collectors.toList());
        // 1.分批次保存,数据补充至es
        Set<String> allMissingRecords = Stream.concat(oldEsMissingRecords.stream(), newEsMissingRecords.stream()).collect(Collectors.toSet());
        List<List<String>> savePartitions = Lists.partition(new ArrayList<>(allMissingRecords), 2000);
        savePartitions.parallelStream().forEach(rs->{
            List<Map<String, Object>> details = idxBizJgUseInfoMapper.queryDetailBatch(rs);
            save2OldEquipEs(details, oldEsMissingRecords, batchNo);
            save2NewEquipEs(rs, details, newEsMissingRecords, batchNo);
        });
        // 2.更新本切片数据的es的版本号,用来清除es的数据标记用
        Map<String, Object> updateFields = MapUtil.of("version", batchNo);
        partitions.parallelStream().forEach(batch->{
            updateEsVersion(batch, updateFields, IDX_BIZ_VIEW_JG_ALL);
            updateEsVersion(batch, updateFields, IDX_BIZ_EQUIPMENT_INFO);
        });
        watch.stop();
        log.info("处理耗时:{}", watch.getTotalTimeSeconds());
        return true;
    }

    private void updateEsVersion(List<String> batch, Map<String, Object> updateFields, String index) {
        BulkRequest bulkRequest = new BulkRequest();
        // 构建批量请求
        batch.stream()
                .map(id -> new UpdateRequest(index, id)
                        .doc(updateFields, XContentType.JSON))
                .forEach(bulkRequest::add);
        try {
            BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            // 更详细的错误日志记录
            if (response.hasFailures()) {
                Arrays.stream(response.getItems())
                        .filter(BulkItemResponse::isFailed)
                        .forEach(item -> log.error("Failed to update {}: {}",
                                item.getId(),
                                item.getFailureMessage()));
            }
        } catch (IOException e) {
            log.error("批量更新版本号失败: batchSize={}, index={}", batch.size(), index, e);
            throw new RuntimeException("ES批量更新失败", e);
        }
    }

    private void save2NewEquipEs(List<String> rs, List<Map<String, Object>> details, List<String> waitSync2EquipmentInfoRecords, String batchNo) {
        Map<String, Map<String, Object>> recordDetailMap = details.stream().collect(Collectors.toMap(e -> (String) e.get("SEQUENCE_NBR"), Function.identity(), (k1, k2) -> k2));
        List<String> equipmentInfoSaveRecords = rs.stream().filter(waitSync2EquipmentInfoRecords::contains).collect(Collectors.toList());
        List<ESEquipmentInfo> esEquipmentInfos = getEsEquipmentInfos(equipmentInfoSaveRecords, recordDetailMap);
        esEquipmentInfos.forEach(e->e.setVersion(batchNo));
        esBulkService.bulkUpsert(IDX_BIZ_EQUIPMENT_INFO, esEquipmentInfos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
    }

    private void save2OldEquipEs(List<Map<String, Object>> details, List<String> waitSync2JgAllRecords, String batchNo) {
        List<Map<String, Object>> jgAllSaves = details.stream().filter(e-> waitSync2JgAllRecords.contains((String) e.get("SEQUENCE_NBR"))).collect(Collectors.toList());
        List<ESEquipmentCategoryDto> esEquipmentInfos = jgAllSaves.parallelStream().map(data -> {
            ESEquipmentCategoryDto esEquipmentInfo = null;
            try {
                esEquipmentInfo = new ESEquipmentCategoryDto();
                StatisticsDataUpdateService.formatUseDate(data);
                BeanUtil.copyProperties(data, esEquipmentInfo, true);
                esEquipmentInfo.setVersion(batchNo);
            } catch (Exception e) {
                log.error("批次{}设备刷数据处理失败", record, e);
            }
            return esEquipmentInfo;
        }).collect(Collectors.toList());
        if(!esEquipmentInfos.isEmpty()){
            esBulkService.bulkUpsert(IDX_BIZ_VIEW_JG_ALL, esEquipmentInfos.stream().map(e -> new EsEntity<>(e.getSEQUENCE_NBR(), e)).collect(Collectors.toList()));
        }
    }
}
