package com.celnet.datadump.service.impl;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.*;
import com.celnet.datadump.enums.FileType;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.DataDumpSpecialParam;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.*;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.soap.partner.*;
import com.sforce.soap.partner.sobject.SObject;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Response;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections4.Get;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;


@Service
@Slf4j
public class DataImportNewServiceImpl implements DataImportNewService {

    @Autowired
    private SalesforceTargetConnect salesforceTargetConnect;

    @Autowired
    private DataBatchHistoryService dataBatchHistoryService;

    @Autowired
    private SalesforceExecutor salesforceExecutor;

    @Autowired
    private SalesforceConnect salesforceConnect;

    @Autowired
    private DataObjectService dataObjectService;

    @Autowired
    private DataBatchService dataBatchService;

    @Autowired
    private DataFieldService dataFieldService;

    @Autowired
    private CustomMapper customMapper;

    @Autowired
    private CommonService commonService;

    @Autowired
    private LinkConfigService linkConfigService;

    @Autowired
    private DataLogService dataLogService;

    @Autowired
    private RichTextLogService richTextLogService;

    private static final String TEMP_FILE_PATH = "verify/";

    private static String FILE_UPLOAD_URL = "";

    private static String FILE_DOWNLOAD_URL = "";

    @PostConstruct
    public void init() {
        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
        for (Map<String, Object> map1 : poll) {
            if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
                FILE_UPLOAD_URL = (String) map1.get("value");
            }
            if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) {
                FILE_DOWNLOAD_URL = (String) map1.get("value");
            }
        }
    }

    static {
        // 检测路径是否存在 不存在则创建
        File excel = new File(TEMP_FILE_PATH);
        if (!excel.exists()) {
            boolean mkdir = excel.mkdir();
        }
    }

    /**
     * Get返写个人客户联系人入口
     */
    @Override
    public void getPersonContact(SalesforceParam param) throws Exception {
        List<Future<?>> futures = Lists.newArrayList();
        try {
            if (StringUtils.isNotBlank(param.getApi())) {
                // 手动任务
                ReturnT<String> result = manualGetPersonContact(param, futures);
                if (result != null) {
                }
            }
        } catch (Exception exception) {
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
            log.error("immigration error", exception);
            throw exception;
        }
    }

    /**
     * 组装Get执行参数
     */
    public ReturnT<String> manualGetPersonContact(SalesforceParam param, List<Future<?>> futures) throws Exception {
        String api = "Contact";

        TimeUnit.MILLISECONDS.sleep(1);
        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                return new ReturnT<>(500, "api:" + apiNames + " is locked");
            }
        }

        PartnerConnection connect = salesforceTargetConnect.createConnect();
        DataObject update = new DataObject();
        TimeUnit.MILLISECONDS.sleep(1);
        try {
            List<SalesforceParam> salesforceParams = null;

            update.setName(api);
            update.setDataLock(1);
            dataObjectService.updateById(update);

            QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
            dbQw.eq("name", api);
            List<DataBatch> list = dataBatchService.list(dbQw);
            AtomicInteger batch = new AtomicInteger(1);

            if (CollectionUtils.isNotEmpty(list)) {
                salesforceParams = list.stream().map(t -> {
                    SalesforceParam salesforceParam = param.clone();
                    salesforceParam.setApi(t.getName());
                    salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                    salesforceParam.setEndCreateDate(t.getSyncEndDate());
                    salesforceParam.setBatch(batch.getAndIncrement());
                    return salesforceParam;
                }).collect(Collectors.toList());
            }

            // 手动任务优先执行
            for (SalesforceParam salesforceParam : salesforceParams) {
                Future<?> future = salesforceExecutor.execute(() -> {
                    try {
                        manualGetPersonContactId(salesforceParam, connect);
                    } catch (Throwable throwable) {
                        log.error("salesforceExecutor error", throwable);
                        throw new RuntimeException(throwable);
                    }
                }, salesforceParam.getBatch(), 1);
                futures.add(future);
            }
            // 等待当前所有线程执行完成
            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            update.setDataWork(0);
        } catch (InterruptedException e) {
            throw e;
        } catch (Throwable e) {
            log.error("manualImmigration error", e);
            throw new RuntimeException(e);
        } finally {
            if (isFull) {
                update.setName(api);
                update.setDataLock(0);
                dataObjectService.updateById(update);
            }
        }
        return null;
    }

    /**
     * 获取newId,写入oldId
     */
    private void manualGetPersonContactId(SalesforceParam param, PartnerConnection partnerConnection) throws Exception {

        String api = param.getApi();
        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
        Field[] dsrFields = dsr.getFields();

        //表内数据总量
        Integer count = customMapper.countBySQL(api, "where new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");

        if (count == 0) {
            return;
        }

        //批量插入200一次
        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        log.info("总反写 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);

        for (int i = 0; i < page; i++) {
            List<Map<String, Object>> data = customMapper.list("*", api, "new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 200");
            int size = data.size();

            log.info("总反写 count:{}；当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
            SObject[] accounts = new SObject[size];

            // 新客户ID，旧联系人ID，用于更新本地数据
            Map<String,String> idMaps = new HashMap<>();
            // 旧客户ID，新联系人ID，用于更新SF数据
            Map<String,String> idMapa = new HashMap<>();

            for (Map<String, Object> map : data) {
                Map<String, Object> idMap = customMapper.getById("new_id", "Account", map.get("AccountId").toString());
                if(idMap.get("new_id") != null && StringUtils.isNotEmpty(idMap.get("new_id").toString())){
                    // 新客户ID，旧联系人ID
                    idMaps.put(idMap.get("new_id").toString(),map.get("Id").toString());
                }
            }

            String idStr = "(";
            for (String ids : idMaps.keySet()) {
                idStr += "'" + ids + "',"; // 拼接每个ID
            }
            if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束
                idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号
            }
            idStr += ")"; // 添加右括号

            try {
                String sql = "SELECT Id,AccountId,Account.old_sfdc_id__c FROM Contact where AccountId in " + idStr ;
                QueryResult queryResult = partnerConnection.queryAll(sql);
                if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
                    break;
                }
                SObject[] records = queryResult.getRecords();
                com.alibaba.fastjson2.JSONArray objects = DataUtil.toJsonArray(records, dsrFields);
                for (int z = 0; z < objects.size(); z++) {
                    JSONObject jsonObject = objects.getJSONObject(z);
                    String contactId = jsonObject.getString(Const.ID);
                    String accountId = jsonObject.getString("AccountId");
                    String oldAccountId = jsonObject.getString("Account_old_sfdc_id__c");
                    String id = idMaps.get(accountId);
                    List<Map<String, Object>> maps = Lists.newArrayList();
                    Map<String, Object> paramMap = Maps.newHashMap();
                    paramMap.put("key", "new_id");
                    paramMap.put("value", contactId);
                    maps.add(paramMap);
                    customMapper.updateById(api, maps, id);
                    idMapa.put(oldAccountId, contactId);
                }
                TimeUnit.MILLISECONDS.sleep(1);

                int index = 0;
                for (Map<String, Object> map : data) {
                    SObject account = new SObject();
                    account.setType(api);
                    account.setField("old_owner_id__c", map.get("OwnerId"));
                    account.setField("old_sfdc_id__c", map.get("Id"));
                    account.setId(idMapa.get(map.get("AccountId").toString()));
                    accounts[index] = account;
                    index++;
                }

                SaveResult[] saveResults = partnerConnection.update(accounts);
                for (SaveResult saveResult : saveResults) {
                    if (!saveResult.getSuccess()) {
                        log.info("-------------saveResults: {}", JSON.toJSONString(saveResult));
                        String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s, \n数据实体类:\n%s",  api, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), JSON.toJSONString(saveResult));
                        EmailUtil.send("DataDump ERROR", format);
                        return;
                    }
                }

            } catch (Exception e) {
                log.error("manualGetPersonContactId error api:{}", api, e);
                throw e;
            }
        }
        SalesforceParam countParam = new SalesforceParam();
        countParam.setApi(api);
        countParam.setBeginCreateDate(beginDate);
        countParam.setEndCreateDate(DateUtils.addSeconds(endDate, -1));
        // 存在isDeleted 只查询IsDeleted为false的
        if (dataFieldService.hasDeleted(countParam.getApi())) {
            countParam.setIsDeleted(false);
        } else {
            // 不存在 过滤
            countParam.setIsDeleted(null);
        }
        Integer sfNum = 0;
        // 不能 count Id
        if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
            sfNum = commonService.countSfNum(partnerConnection, countParam);
        }

        UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
        updateQw.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
                .set("target_sf_num", sfNum);
        dataBatchHistoryService.update(updateQw);

        UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
        updateQw2.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", endDate)
                .set("sf_add_num", sfNum);
        dataBatchService.update(updateQw2);
    }

    /**
     * 数据更新Update入口
     */
    @Override
    public ReturnT<String> immigrationUpdateNew(SalesforceParam param) throws Exception {
        List<Future<?>> futures = Lists.newArrayList();
        try {
            if (StringUtils.isNotBlank(param.getApi())) {
                if (1 == param.getType()){
                    return maualupdateSfDataNew(param, futures);
                }else {
                    return updateIncrementalSfDataNew(param, futures);
                }
            } else {
                if (1 == param.getType()){
                    autoUpdateSfDataNew(param, futures);
                }else {
                    autoUpdateIncrementalSfDataNew(param, futures);
                }
            }
            return ReturnT.SUCCESS;
        }  catch (InterruptedException e) {
            throw e;
        }  catch (Throwable throwable) {
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
            log.error("immigrationUpdate error", throwable);
            throw throwable;
        }
    }

    /**
     * 组装【单表】【存量】Update参数
     */
    public ReturnT<String> maualupdateSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
        List<String> apis = DataUtil.toIdList(param.getApi());

        String beginDateStr = null;
        String endDateStr = null;
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1).in("name", apis);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                String message = "api:" + apiNames + " is locked";
                String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s",  apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
                EmailUtil.send("DataDump ERROR", format);
                return new ReturnT<>(500, message);
            }
        }

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
        for (String api : apis) {
            DataObject update = dataObjectService.getById(api);
            try {
                List<SalesforceParam> salesforceParams = null;
                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                dbQw.eq("name", api);
                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
                    dbQw.eq("sync_start_date", beginDateStr);  // 等于开始时间
                    dbQw.eq("sync_end_date", endDateStr);    // 等于结束时间
                }
                List<DataBatch> list = dataBatchService.list(dbQw);
                AtomicInteger batch = new AtomicInteger(1);
                if (CollectionUtils.isNotEmpty(list)) {
                    salesforceParams = list.stream().map(t -> {
                        SalesforceParam salesforceParam = param.clone();
                        salesforceParam.setApi(t.getName());
                        salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                        salesforceParam.setEndCreateDate(t.getSyncEndDate());
                        salesforceParam.setBatch(batch.getAndIncrement());
                        return salesforceParam;
                    }).collect(Collectors.toList());
                }

                // 手动任务优先执行
                for (SalesforceParam salesforceParam : salesforceParams) {
                    if (param.getIsSingleThread()){
                        UpdateSfDataNew(salesforceParam, partnerConnection,update);
                    }else {
                        Future<?> future = salesforceExecutor.execute(() -> {
                            try {
                                UpdateSfDataNew(salesforceParam, partnerConnection,update);
                            } catch (Throwable throwable) {
                                log.error("salesforceExecutor error", throwable);
                                throw new RuntimeException(throwable);
                            }
                        }, salesforceParam.getBatch(), 1);
                        futures.add(future);
                    }
                }
                // 等待当前所有线程执行完成
                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));

            } catch (Exception e) {
                throw e;
            } finally {
                if (isFull) {
                    update.setNeedUpdate(false);
                    update.setName(api);
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 组装【单表】【增量】Update参数
     */
    public ReturnT<String> updateIncrementalSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
        List<String> apis = DataUtil.toIdList(param.getApi());

        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1).in("name", apis);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                return new ReturnT<>(500, "api:" + apiNames + " is locked");
            }
        }

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
        for (String api : apis) {
            DataObject update = dataObjectService.getById(api);
            try {
                //无创建时间对象
                if (!dataFieldService.hasCreatedDate(api)) {
                    UpdateSfShareDataNew(api, partnerConnection, update);
                    continue;
                }
                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
                DataBatch t = dataBatchService.getOne(dbQw);
                SalesforceParam salesforceParam = param.clone();
                salesforceParam.setApi(t.getName());
                salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                salesforceParam.setEndCreateDate(t.getSyncEndDate());
                salesforceParam.setBatch(new AtomicInteger(1).getAndIncrement());
                if (param.getIsSingleThread()){
                    UpdateSfDataNew(salesforceParam, partnerConnection,update);
                }else {
                    // 手动任务优先执行
                    Future<?> future = salesforceExecutor.execute(() -> {
                        try {
                            UpdateSfDataNew(salesforceParam, partnerConnection,update);
                        } catch (Throwable throwable) {
                            log.error("salesforceExecutor error", throwable);
                            throw new RuntimeException(throwable);
                        }
                    }, salesforceParam.getBatch(), 1);
                    futures.add(future);
                }
                update.setNeedUpdate(false);
            } catch (Exception e) {
                throw e;
            } finally {
                if (isFull) {
                    update.setName(api);
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
        }
        // 等待当前所有线程执行完成
        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
        futures.clear();

        return ReturnT.SUCCESS;
    }

    /**
     * 组装【多表】【存量】Update参数
     */
    private void autoUpdateSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {

        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                String message = "api:" + apiNames + " is locked";
                String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s",  apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
                EmailUtil.send("DataDump ERROR", format);
                return ;
            }
        }

        String beginDateStr = null;
        String endDateStr = null;
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        QueryWrapper<DataObject> qw = new QueryWrapper<>();
        qw.eq("need_update", 1)
                .orderByAsc("data_index")
                .last(" limit 10");

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();

        while (true) {
            List<DataObject> dataObjects = dataObjectService.list(qw);
            //判断dataObjects是否为空
            if (CollectionUtils.isEmpty(dataObjects)) {
                return;
            }

            for (DataObject dataObject : dataObjects) {
                TimeUnit.MILLISECONDS.sleep(1);

                DataObject update = new DataObject();
                update.setName(dataObject.getName());
                update.setDataLock(1);
                dataObjectService.updateById(update);
                TimeUnit.MILLISECONDS.sleep(1);
                try {
                    String api = dataObject.getName();
                    List<SalesforceParam> salesforceParams = null;
                    QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                    dbQw.eq("name", api);
                    if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
                        dbQw.eq("sync_start_date", beginDateStr);  // 等于开始时间
                        dbQw.eq("sync_end_date", endDateStr);    // 等于结束时间
                    }
                    List<DataBatch> list = dataBatchService.list(dbQw);
                    AtomicInteger batch = new AtomicInteger(1);
                    if (CollectionUtils.isNotEmpty(list)) {
                        salesforceParams = list.stream().map(t -> {
                            SalesforceParam salesforceParam = param.clone();
                            salesforceParam.setApi(t.getName());
                            salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                            salesforceParam.setEndCreateDate(t.getSyncEndDate());
                            salesforceParam.setBatch(batch.getAndIncrement());
                            return salesforceParam;
                        }).collect(Collectors.toList());
                    }

                    // 手动任务优先执行
                    for (SalesforceParam salesforceParam : salesforceParams) {
                        if (param.getIsSingleThread()){
                            UpdateSfDataNew(salesforceParam, partnerConnection,update);
                        }else {
                            Future<?> future = salesforceExecutor.execute(() -> {
                                try {
                                    UpdateSfDataNew(salesforceParam, partnerConnection, dataObject);
                                } catch (Throwable throwable) {
                                    log.error("salesforceExecutor error", throwable);
                                    throw new RuntimeException(throwable);
                                }
                            }, salesforceParam.getBatch(), 0);
                            futures.add(future);
                        }
                    }
                    // 等待当前所有线程执行完成
                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                } catch (Exception e) {
                    throw e;
                } finally {
                    update.setNeedUpdate(false);
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
            // 等待当前所有线程执行完成
            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            futures.clear();
        }
    }

    /**
     * 组装【多表】【增量】Update参数
     */
    private void autoUpdateIncrementalSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {

        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                String message = "api:" + apiNames + " is locked";
                String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s",  apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
                EmailUtil.send("DataDump ERROR", format);
                return ;
            }
        }

        QueryWrapper<DataObject> qw = new QueryWrapper<>();
        qw.eq("need_update", 1)
                .orderByAsc("data_index")
                .last(" limit 10");

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
        while (true) {
            List<DataObject> dataObjects = dataObjectService.list(qw);
            //判断dataObjects是否为空
            if (CollectionUtils.isEmpty(dataObjects)) {
                return;
            }

            for (DataObject dataObject : dataObjects) {
                TimeUnit.MILLISECONDS.sleep(1);
                DataObject update = new DataObject();
                update.setName(dataObject.getName());
                update.setDataLock(1);
                dataObjectService.updateById(update);
                TimeUnit.MILLISECONDS.sleep(1);
                try {
                    String api = dataObject.getName();
                    QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                    dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
                    DataBatch t = dataBatchService.getOne(dbQw);

                    AtomicInteger batch = new AtomicInteger(1);
                    SalesforceParam salesforceParam = param.clone();
                    salesforceParam.setApi(t.getName());
                    salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                    salesforceParam.setEndCreateDate(t.getSyncEndDate());
                    salesforceParam.setBatch(batch.getAndIncrement());

                    if (param.getIsSingleThread()){
                        UpdateSfDataNew(salesforceParam, partnerConnection,update);
                    }else {
                        Future<?> future = salesforceExecutor.execute(() -> {
                            try {
                                UpdateSfDataNew(salesforceParam, partnerConnection,dataObject);
                            } catch (Throwable throwable) {
                                log.error("salesforceExecutor error", throwable);
                                throw new RuntimeException(throwable);
                            }
                        }, salesforceParam.getBatch(), 0);
                        futures.add(future);
                    }

                } catch (Exception e) {
                    throw e;
                } finally {
                    update.setNeedUpdate(false);
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
            // 等待当前所有线程执行完成
            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            futures.clear();
        }
    }

    /**
     * 执行Update更新数据
     */
    private void UpdateSfDataNew(SalesforceParam param, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {

        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+ SystemConfigCode.INFO_FLAG+"'").get(0);
        String api = param.getApi();
        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);
        List<DataField> list = dataFieldService.list(dbQw);

        String sql = "";
        String sql2 = "";

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        if (1 == param.getType()) {
            if (api.endsWith("Share")){
                sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
                sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000 ";
            }else {
                sql = "where new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
                sql2 = "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000 ";
            }
        }else {
            String updateDateField = dataFieldService.returnUpdateDateField(api);
            if (api.endsWith("Share")){
                sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
                sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000 ";
            }else {
                sql = "where new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' ";
                sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000 ";
            }
        }
        //表内数据总量
        Integer count = customMapper.countBySQL(api, sql);

        if(count == 0){
            log.info("没有需要更新的数据，API: {}，开始时间: {}，结束时间: {}", api, beginDateStr, endDateStr);
            return;
        }

        //查询当前对象多态字段映射
        QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
        List<LinkConfig> configs = linkConfigService.list(queryWrapper);
        Map<String, String> fieldMap = new HashMap<>();
        if (!configs.isEmpty())  {
            fieldMap = configs.stream()
                    .collect(Collectors.toMap(
                            LinkConfig::getField,      // Key提取器
                            LinkConfig::getLinkField,  // Value提取器
                            (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
                    ));
        }


        int targetCount = 0;

        int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;

        log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);

        for (int z = 0; z < num; z++){

            List<Map<String, Object>> mapsList = customMapper.list("*", api,  sql2);
            int size = mapsList.size();
            log.info("总批次:{}；当前批次：{}；当前批次数据量:{}；-开始时间:{}；-结束时间:{}；-api:{}；", num,z,size, beginDateStr, endDateStr, api);

            //批量插入200一次
            int page = size%200 == 0 ? size/200 : (size/200) + 1;

            for (int i = 0; i < page; i++) {

                DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，查询并组装第" + (z*50 + i) + "批数据", "QueryDB");
                int startIndex = 200 * i;
                int endIndex = Math.min(startIndex + 200, size);
                if (startIndex >= endIndex) {
                    log.info("当前批次：{} 结束!! -开始时间:{}；-结束时间:{}；-api:{}；", z, beginDateStr, endDateStr, api);
                    break;
                }
                List<Map<String, Object>> mapList = mapsList.subList(startIndex, endIndex);

                SObject[] accounts = new SObject[mapList.size()];
                int j = 0;
                try {
                    for (Map<String, Object> map : mapList) {
                        //需要置空的参数
                        List<String> fieldsToNull = new ArrayList<>();
                        SObject account = new SObject();
                        account.setType(api);
                        String message = null;
                        //给对象赋值
                        for (DataField dataField : list) {
                            String field = dataField.getField();
                            String reference_to = dataField.getReferenceTo();

                            String value = String.valueOf(map.get(field));
                            //根据旧sfid查找引用对象新sfid
                            if (field.equals("Id")) {
                                account.setId(String.valueOf(map.get("new_id")));
                            } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
                                continue;
                            } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {

                                if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
                                    //判断reference_to内是否包含User字符串
                                    if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
                                        reference_to = "User";
                                    }
                                    //引用类型字段
                                    String linkfield = fieldMap.get(dataField.getField());
                                    if (StringUtils.isNotBlank(linkfield)){
                                        if (map.get(linkfield)!=null){
                                            reference_to = map.get(linkfield).toString();
                                        }else {
                                            message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！";
                                            List<Map<String, Object>> maps = new ArrayList<>();
                                            Map<String, Object> linkMap = new HashMap<>();
                                            linkMap.put("key", "is_update");
                                            linkMap.put("value", 2);
                                            maps.add(linkMap);
                                            Map<String, Object> linkMap1 = new HashMap<>();
                                            linkMap1.put("key", "error_message");
                                            linkMap1.put("value", message);
                                            maps.add(linkMap1);
                                            customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
                                            log.info(message);
                                        }
                                    }
                                    if (reference_to == null){
                                        continue;
                                    }
                                    Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
                                    if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
                                        account.setField(field, m.get("new_id"));
                                    }else {
                                        message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"异常！";
                                        List<Map<String, Object>> maps = new ArrayList<>();
                                        Map<String, Object> linkMap = new HashMap<>();
                                        linkMap.put("key", "is_update");
                                        linkMap.put("value", 2);
                                        maps.add(linkMap);
                                        Map<String, Object> linkMap1 = new HashMap<>();
                                        linkMap1.put("key", "error_message");
                                        linkMap1.put("value", message);
                                        maps.add(linkMap1);
                                        customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
                                        log.info(message);
                                        break;
                                    }
                                }
                            } else {
                                if(StringUtils.isNotBlank(dataField.getSfType())){
                                    if (map.get(field) != null){
                                        account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
                                    }else {
                                        fieldsToNull.add(field);
                                    }
                                }
                            }
                        }
                        if (message != null){
                            continue;
                        }

                        if (!fieldsToNull.isEmpty()){
                            String[] fieldsToNullArray = fieldsToNull.toArray(new String[0]);
                            account.setField("fieldsToNull", fieldsToNullArray);
                        }

                        if (dataObject.getIsEditable()){
                            account.setField("old_owner_id__c", map.get("OwnerId"));
                            account.setField("old_sfdc_id__c", map.get("Id"));
                        }
                        accounts[j++] = account;
                    }
                    dataLog.setEndTime(new Date());
                    dataLogService.save(dataLog);

                    if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                        printlnAccountsDetails(accounts,list);
                    }
                    DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，更新SF第" + (z*50 + i) + "批数据", "UpdateSF");
                    SaveResult[] saveResults = partnerConnection.update(accounts);
                    for (SaveResult saveResult : saveResults) {
                        if (!saveResult.getSuccess()) {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 2);
                            maps.add(linkMap);
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", JSON.toJSONString(saveResult.getErrors()));
                            maps.add(linkMap1);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                            String format = String.format("数据更新 error, api name: %s, \ncause:\n%s",  api, JSON.toJSONString(saveResult));
                            log.info(format);
                        }else {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 1);
                            maps.add(linkMap);
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", null);
                            maps.add(linkMap1);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                            targetCount ++;
                        }
                    }
                    dataLog1.setEndTime(new Date());
                    dataLogService.save(dataLog1);

                    TimeUnit.MILLISECONDS.sleep(1);

                }catch (InterruptedException interruptedException){
                    return;
                }catch (Throwable e) {
                    log.info(JSON.toJSONString(e));
                }
            }
        }

        UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
        updateQw.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
                .set("target_update_num", num);
        dataBatchHistoryService.update(updateQw);

        UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
        updateQw2.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", endDate)
                .set("sf_update_num", targetCount);
        dataBatchService.update(updateQw2);

        log.info("数据更新完成，API: {}，总更新数: {}，开始时间: {}，结束时间: {}", api, targetCount, beginDateStr, endDateStr);
    }

    /**
     * 执行Update更新数据
     */
    private void UpdateSfShareDataNew(String api, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {

        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);
        List<DataField> list = dataFieldService.list(dbQw);

        String sql = "";
        String sql2 = "";

        String beginDateStr = null;
        String endDateStr = null;

        String updateDateField = dataFieldService.returnUpdateDateField(api);
        if (api.endsWith("Share")){
            sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
            sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
        }else {
            sql = "where new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' ";
            sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
        }

        //表内数据总量
        Integer count = customMapper.countBySQL(api, sql);

        if(count == 0){
            return;
        }

        //查询当前对象多态字段映射
        QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
        List<LinkConfig> configs = linkConfigService.list(queryWrapper);
        Map<String, String> fieldMap = new HashMap<>();
        if (!configs.isEmpty())  {
            fieldMap = configs.stream()
                    .collect(Collectors.toMap(
                            LinkConfig::getField,      // Key提取器
                            LinkConfig::getLinkField,  // Value提取器
                            (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
                    ));
        }


        int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;

        log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);

        for (int z = 0; z < num; z++){

            List<Map<String, Object>> mapsList = customMapper.list("*", api,  sql2+ z * 10000 + ",10000");
            int size = mapsList.size();
            log.info("总Update数据 count:{}；当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,z, beginDateStr, endDateStr, api);

            //批量插入200一次
            int page = size%200 == 0 ? count/200 : (count/200) + 1;

            for (int i = 0; i < page; i++) {

                int startIndex = 200 * i;
                int endIndex = Math.min(startIndex + 200, size);
                if (startIndex >= endIndex) {
                    break;
                }
                List<Map<String, Object>> mapList = mapsList.subList(startIndex, endIndex);

                SObject[] accounts = new SObject[mapList.size()];
                int j = 0;
                try {
                    for (Map<String, Object> map : mapList) {

                        SObject account = new SObject();
                        account.setType(api);
                        String message = null;
                        //给对象赋值
                        for (DataField dataField : list) {
                            String field = dataField.getField();
                            String reference_to = dataField.getReferenceTo();

                            String value = String.valueOf(map.get(field));
                            //根据旧sfid查找引用对象新sfid
                            if (field.equals("Id")) {
                                account.setId(String.valueOf(map.get("new_id")));
                            } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
                                continue;
                            } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {

                                if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
                                    //判断reference_to内是否包含User字符串
                                    if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
                                        reference_to = "User";
                                    }
                                    //引用类型字段
                                    String linkfield = fieldMap.get(dataField.getField());
                                    if (StringUtils.isNotBlank(linkfield)){
                                        if (map.get(linkfield)!=null){
                                            reference_to = map.get(linkfield).toString();
                                        }else {
                                            log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
                                        }
                                    }
                                    if (reference_to == null){
                                        continue;
                                    }
                                    Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
                                    if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
                                        account.setField(field, m.get("new_id"));
                                    }else {
                                        message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"异常！";
                                        log.info(message);
                                        break;
                                    }
                                }
                            } else {
                                if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                    account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
                                }else {
                                    account.setField(field, map.get(field));
                                }
                            }
                        }
                        if (message!=null){
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", message);
                            maps.add(linkMap1);
                            customMapper.updateById(api, maps, map.get("Id").toString());
                            continue;
                        }
                        if (dataObject.getIsEditable()){
                            account.setField("old_owner_id__c", map.get("OwnerId"));
                            account.setField("old_sfdc_id__c", map.get("Id"));
                        }
                        accounts[j++] = account;
                    }

                    if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                        printlnAccountsDetails(accounts,list);
                    }

                    SaveResult[] saveResults = partnerConnection.update(accounts);
                    for (SaveResult saveResult : saveResults) {
                        if (!saveResult.getSuccess()) {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 2);
                            maps.add(linkMap);
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", JSON.toJSONString(saveResult.getErrors()));
                            maps.add(linkMap1);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                            String format = String.format("数据更新 error, api name: %s, \ncause:\n%s",  api, JSON.toJSONString(saveResult));
                            log.info(format);
                        }else {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 1);
                            maps.add(linkMap);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                        }
                    }

                } catch (Throwable e) {
                    log.info(JSON.toJSONString(e));
                }
            }
        }

    }

    /**
     * 打印SF交互数据明细
     */
    public void printlnAccountsDetails(SObject[] accounts,List<DataField> list) {
        for (int i = 0; i < accounts.length; i++) {
            SObject account = accounts[i];
            System.out.println("---  对象数据[" + i + "] ---");
            if (account == null){
                continue;
            }
            // 获取对象所有字段名
            for (DataField dataField : list) {
                try {
                    Object value = account.getField(dataField.getField());
                    System.out.println(dataField.getField() + ": " + (value != null ? value.toString() : "null"));
                } catch (Exception e) {
                    System.out.println(dataField.getField() + ": [权限不足或字段不存在]");
                }
            }
            System.out.println("old_owner_id__c: " + account.getField("old_owner_id__c"));
            System.out.println("old_sfdc_id__c: " + account.getField("old_sfdc_id__c"));

        }
    }

    /**
     * 返回SF交互数据错误明细
     */
    public Map<String,String> returnErrorAccountsDetails(SObject[] accounts,List<DataField> list,String errorId) {
        HashMap<String, String> map = new HashMap<>();
        for (int i = 0; i < accounts.length;  i++) {
            SObject account = accounts[i];
            if (account == null){
                continue;
            }
            if (errorId != null && (errorId.equals(account.getId()) || errorId.equals(account.getField("Id")))){
                for (DataField dataField : list) {
                    try {
                        Object value = account.getField(dataField.getField());
                        map.put(dataField.getField(),String.valueOf(value));
                        System.out.println(dataField.getField()  + ": " + value);
                    } catch (Exception e) {
                        System.out.println(dataField.getField() + ": [权限不足或字段不存在]");
                    }
                }
                map.put("old_owner_id__c",String.valueOf(account.getField("old_owner_id__c")));
                map.put("old_sfdc_id__c",String.valueOf(account.getField("old_sfdc_id__c")));
            }
        }
        return map;
    }

    /**
     * 获取DocumentLink
     */
    @Override
    public ReturnT<String> dumpDocumentLinkJob(String paramStr) throws Exception {
        String api = "ContentDocumentLink";
        PartnerConnection partnerConnection = salesforceConnect.createConnect();
        Integer count = customMapper.countBySQL("ContentDocument", "where new_id is not null");
        log.info("ContentDocument Total size：" + count);

        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        int totalSize = 0;
        for (int i = 0; i < page; i++) {
            List<Map<String, Object>> list = customMapper.list("Id", "ContentDocument", "new_id is not null order by Id limit " + i * 200 + ",200");
            if (list.isEmpty()){
                break;
            }
            totalSize = totalSize +list.size();
            log.info("dumpContentDocumentLink By ContentDocument Now size：" + totalSize);
            DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
            List<String> fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList());
            Field[] dsrFields = dsr.getFields();
            try {
                for (Map<String, Object> map : list) {
                    String contentDocumentId = (String) map.get("Id");
                    String sql = "SELECT Id, LinkedEntityId, LinkedEntity.Type, ContentDocumentId, Visibility, ShareType, SystemModstamp, IsDeleted FROM ContentDocumentLink where ContentDocumentId = '" + contentDocumentId + "'";
                    com.alibaba.fastjson2.JSONArray objects = null;
                    QueryResult queryResult = partnerConnection.queryAll(sql);
                    SObject[] records = queryResult.getRecords();
                    objects = DataUtil.toJsonArray(records, dsrFields);
                    commonService.saveOrUpdate(api, fields, records, objects, true);
                }
            } catch (Throwable e) {
                log.error("dumpDocumentLinkJob error message:{}", e.getMessage());
                return ReturnT.FAIL;
            }
        }
        log.info("dumpDocumentLink Success !!! ");
        return ReturnT.SUCCESS;
    }

    /**
     * 推送DocumentLink
     */
    @Override
    public ReturnT<String> uploadDocumentLinkJob(String paramStr) throws Exception {
        String api = "ContentDocumentLink";

        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);

        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);
        List<DataField> dataFields = dataFieldService.list(dbQw);

        PartnerConnection connection = salesforceTargetConnect.createConnect();
        List<Map<String, Object>> list = customMapper.list("Id", "ContentDocument", "new_id is not null");
        try {
            if (list != null && !list.isEmpty()) {
                //表内数据总量
                Integer count = customMapper.countBySQL(api, "where ShareType = 'V' ");
                //批量插入200一次
                int page = count % 200 == 0 ? count / 200 : (count / 200) + 1;
                for (int i = 0; i < page; i++) {
                    List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit 200");
                    SObject[] accounts = new SObject[linkList.size()];
                    String[] ids = new String[linkList.size()];
                    int index = 0;
                    for (Map<String, Object> map : linkList) {
                        String linkedEntityId = (String) map.get("LinkedEntityId");
                        String id = (String) map.get("Id");
                        String contentDocumentId = (String) map.get("ContentDocumentId");
                        String linkedEntityType = (String) map.get("LinkedEntity_Type");
                        String shareType = (String) map.get("ShareType");
                        String Visibility = (String) map.get("Visibility");

                        // dataObject查询
                        QueryWrapper<DataObject> qw = new QueryWrapper<>();
                        qw.eq("name", linkedEntityType);
                        List<DataObject> objects = dataObjectService.list(qw);
                        if (!objects.isEmpty()) {
                            Map<String, Object> dMap = customMapper.getById("new_id", "ContentDocument", contentDocumentId);
                            Map<String, Object> lMap = customMapper.getById("new_id", linkedEntityType, linkedEntityId);
                            SObject account = new SObject();
                            account.setType(api);
                            if (dMap != null){
                                account.setField("ContentDocumentId", dMap.get("new_id").toString());
                            }else {
                                String errorMessage = String.format("ContentDocumentLink Id: %s，对应的ContentDocumentId: %s 数据不存在！", id, contentDocumentId);
                                log.info(errorMessage);
                                List<Map<String, Object>> maps = new ArrayList<>();
                                Map<String, Object> linkMap1 = new HashMap<>();
                                linkMap1.put("key", "error_message");
                                linkMap1.put("value", errorMessage);
                                maps.add(linkMap1);
                                customMapper.updateById(api, maps, id);
                                continue;
                            }
                            if (lMap != null){
                                account.setField("LinkedEntityId", lMap.get("new_id").toString());
                            }else {
                                String errorMessage = String.format("ContentDocumentLink Id: %s，对应的LinkedEntityId: %s 数据不存在！", id, linkedEntityId);
                                log.info(errorMessage);
                                List<Map<String, Object>> maps = new ArrayList<>();
                                Map<String, Object> linkMap1 = new HashMap<>();
                                linkMap1.put("key", "error_message");
                                linkMap1.put("value", errorMessage);
                                maps.add(linkMap1);
                                customMapper.updateById(api, maps, id);
                                continue;
                            }
                            account.setField("ShareType", shareType);
                            account.setField("Visibility", Visibility);
                            ids[index] = id;
                            accounts[index] = account;
                            index++;
                        }
                    }
                    try {
                        if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                            printlnAccountsDetails(accounts,dataFields);
                        }
                        SaveResult[] saveResults = connection.create(accounts);
              ;
                        for (int j = 0; j < saveResults.length; j++) {
                            if (saveResults[j].getSuccess()) {
                                List<Map<String, Object>> maps = new ArrayList<>();
                                Map<String, Object> m = new HashMap<>();
                                m.put("key", "new_id");
                                m.put("value", saveResults[j].getId());
                                maps.add(m);
                                customMapper.updateById(api, maps, ids[j]);
                                log.info("ContentDocumentLink Id: {}，对应的new_id: {} 更新成功！ " , ids[j], saveResults[j].getId());
                            }else{
                                List<Map<String, Object>> maps = new ArrayList<>();
                                Map<String, Object> linkMap1 = new HashMap<>();
                                linkMap1.put("key", "error_message");
                                linkMap1.put("value", JSON.toJSONString(saveResults[j].getErrors()));
                                maps.add(linkMap1);
                                customMapper.updateById(api, maps, ids[j]);
                                log.error("Id:{},saveResults: {}",ids[j], JSON.toJSONString(saveResults[j]));
                            }
                        }
                    } catch (Exception e) {
                        log.error("uploadDocumentLinkJob error message:{}", e.getMessage());
                        throw new RuntimeException(e);
                    }
                }
            }
        } catch (Exception e) {
            log.error("uploadDocumentLink error api:{}, data:{}", api, com.alibaba.fastjson2.JSON.toJSONString(list), e);
            return ReturnT.FAIL;
        }
        log.info("uploadDocumentLink Success !!! ");
        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<String> dumpFileNew(SalesforceParam param) throws Exception {
        String downloadUrl = null;

        QueryWrapper<DataObject> wrapper = new QueryWrapper<>();
        if (StringUtils.isNotBlank(param.getApi())) {
            wrapper.in("name",DataUtil.toIdList(param.getApi()));
        }else {
            wrapper.isNotNull("blob_field");
        }
        List<DataObject> objectList = dataObjectService.list(wrapper);
        if (objectList.isEmpty()){
            log.info("没有对象存在文件二进制字段！不进行文件下载");
            return ReturnT.FAIL;
        }

        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);

        for (Map<String, Object> map1 : poll) {
            if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) {
                downloadUrl = (String) map1.get("value");
            }
        }
        if (StringUtils.isEmpty(downloadUrl)) {
            EmailUtil.send("DumpFile ERROR", "文件下载失败！下载地址未配置");
            return ReturnT.FAIL;
        }

        PartnerConnection connect = salesforceConnect.createConnect();

        List<Future<?>> futures = Lists.newArrayList();

        for (DataObject dataObject : objectList) {
            DataObject update = new DataObject();
            log.info("dump file api:{}, field:{}", dataObject.getName(), dataObject.getBlobField());

            try {
                String api = dataObject.getName();
                update.setName(dataObject.getName());
                List<SalesforceParam> salesforceParams = null;
                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                dbQw.eq("name", api);
                List<DataBatch> list = dataBatchService.list(dbQw);
                AtomicInteger batch = new AtomicInteger(1);
                if (CollectionUtils.isNotEmpty(list)) {
                    salesforceParams = list.stream().map(t -> {
                        SalesforceParam salesforceParam = param.clone();
                        salesforceParam.setApi(t.getName());
                        salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                        salesforceParam.setEndCreateDate(t.getSyncEndDate());
                        salesforceParam.setBatch(batch.getAndIncrement());
                        return salesforceParam;
                    }).collect(Collectors.toList());
                }

                if (Const.FILE_TYPE == FileType.SERVER) {
                    Date defaultBeginDate = DataUtil.DEFAULT_BEGIN_DATE;
                    Calendar calendar = Calendar.getInstance();
                    int currentYear = calendar.get(Calendar.YEAR);
                    calendar.setTime(defaultBeginDate);
                    int beginYear = calendar.get(Calendar.YEAR);

                    // 检测路径是否存在 不存在则创建
                    File baseDir = new File(Const.SERVER_FILE_PATH + "/" + api);
                    if (!baseDir.exists()) {
                        baseDir.mkdirs();
                    }

                    // 创建从开始年份到当前年份的子目录
                    for (int year = beginYear; year <= currentYear; year++) {
                        File yearDir = new File(Const.SERVER_FILE_PATH + "/" + api + "-" + year);
                        if (!yearDir.exists()) {
                            yearDir.mkdir();
                        }
                    }
                }

                // 手动任务优先执行
                for (SalesforceParam salesforceParam : salesforceParams) {
                    String finalDownloadUrl = downloadUrl;
                    Future<?> future = salesforceExecutor.execute(() -> {
                        try {
                            saveFile(salesforceParam, connect, finalDownloadUrl,dataObject.getName(),dataObject.getBlobField());
                        } catch (Throwable throwable) {
                            log.error("salesforceExecutor error", throwable);
                            throw new RuntimeException(throwable);
                        }
                    }, salesforceParam.getBatch(), 0);
                    futures.add(future);
                }
                // 等待当前所有线程执行完成
                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                update.setDataWork(0);
            } catch (InterruptedException e) {
                salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
                throw e;
            } catch (Throwable e) {
                throw new RuntimeException(e);
            } finally {
                update.setDataLock(0);
                dataObjectService.updateById(update);
            }
        }
        // 等待当前所有线程执行完成
        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
        futures.clear();
        return ReturnT.SUCCESS;
    }


    /**
     * 下载文件
     */
    private void saveFile(SalesforceParam param, PartnerConnection Connection ,String downloadUrl, String api, String field) throws Exception {

        String extraSql = "";
        if (dataFieldService.hasDeleted(api)) {
            extraSql += " AND IsDeleted = false ";
        }

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }
        String token = Connection.getSessionHeader().getSessionId();

        String name = "Name";
        if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
            name = "Title,FileExtension";
        }

        Map<String, String> headers = Maps.newHashMap();
        headers.put("Authorization", "Bearer " + token);
        headers.put("connection", "keep-alive");

        Integer count = customMapper.countBySQL(api, "where is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql );

        if (count == 0) {
            return;
        }

        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        log.info("api：{}；总文件数 count:{}；总批次 ：{}；-开始时间:{}；-结束时间:{}；",api, count, page,beginDateStr, endDateStr);

        //总插入数
        for (int i = 0; i < page; i++) {

            log.info("api：{} 批次:{}；-开始时间:{}；-结束时间:{}；",api, i, beginDateStr, endDateStr);

            // 获取未存储的附件id
            List<Map<String, Object>> list = customMapper.list("Id,CreatedDate, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit 200");

            for (Map<String, Object> map : list) {
                int failCount = 0;
                String id = null;
                // 上传完毕 更新附件信息
                List<Map<String, Object>> maps = Lists.newArrayList();
                try {
                    String fileName;
                    id = (String) map.get("Id");
                    if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
                        fileName = map.get("Title") + "." + map.get("FileExtension");
                    }else {
                        fileName = (String) map.get("Name");
                    }
                    String year = map.get("CreatedDate").toString().substring(0, 4);

                    // 判断路径是否为空
                    if (StringUtils.isNotBlank(fileName)) {
                        String filePath = api + "-" + year + "/" + id + "_" + fileName;
                        // 拼接url
                        String url = downloadUrl + String.format(Const.SF_FILE_URL, api, id, field);

                        log.info("文件下载请求地址：{}",url);
                        while (failCount < Const.MAX_FAIL_COUNT){
                            Response response = HttpUtil.doGet(url, null, headers);
                            if (response.body() != null && response.code() == 200) {
                                InputStream inputStream = response.body().byteStream();
                                dumpToServer(headers, id, filePath, url, response, inputStream);
                                Map<String, Object> paramMap = Maps.newHashMap();
                                if ("Document".equals(api)) {
                                    paramMap.put("key", "localUrl");
                                } else {
                                    paramMap.put("key", "url");
                                }
                                paramMap.put("value", filePath);
                                maps.add(paramMap);
                                inputStream.close();
                                break;
                            }else {
                                failCount++;
                                log.info("文件下载失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + response.message());
                            }
                        }

                        if (failCount == Const.MAX_FAIL_COUNT) {
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_dump");
                            paramMap.put("value", 2);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                        }else {
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_dump");
                            paramMap.put("value", 1);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                            TimeUnit.MILLISECONDS.sleep(1);
                        }
                    }

                } catch (InterruptedException interruptedException){
                    log.info("文件下载手动终止！");
                    return;
                } catch (Exception e) {
                    log.error("文件下载失败！, id: {}, 错误信息：{}", id ,e.getMessage());
                    Map<String, Object> paramMap = Maps.newHashMap();
                    paramMap.put("key", "is_dump");
                    paramMap.put("value", 2);
                    maps.add(paramMap);
                    customMapper.updateById(api, maps, id);
                    EmailUtil.send("File Dump ERROR", "文件下载失败！, id: "+ id + "，错误信息：" + e.getMessage());
                }
            }
        }
    }


    /**
     * 下载文件到服务器
     *
     * @param headers     请求头
     * @param id          id
     * @param filePath    文件路径
     * @param url         链接
     * @param response    响应
     * @param inputStream 流
     * @throws IOException exception
     */
    private void dumpToServer(Map<String, String> headers, String id, String filePath, String url, Response response, InputStream inputStream) throws IOException {
        String path = Const.SERVER_FILE_PATH + "/" + filePath;
        log.info("--------文件名称："+path);
        long offset = 0L;
        RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
        while (true) {
            try {
                TimeUnit.MILLISECONDS.sleep(1);

                // 保存到本地
                byte[] buf = new byte[8192];
                int len = 0;
                if (offset > 0) {
                    inputStream.skip(offset);
                    accessFile.seek(offset);
                }
                while ((len = inputStream.read(buf)) != -1) {
                    accessFile.write(buf, 0, len);
                    offset += len;
                }
                break;
            }catch (InterruptedException interruptedException){
                return;
            }catch (Exception e) {
                if (offset <= 0) {
                    throw e;
                }
                System.out.println("file dump to server EOF ERROR try to reconnect");
                response.close();
                response = HttpUtil.doGet(url, null, headers);
                assert response.body() != null;
                inputStream = response.body().byteStream();
            }
        }
        accessFile.close();
    }


    @Override
    public ReturnT<String> uploadFileNew(SalesforceParam param, DataObject dataObject) throws Exception {
        String uploadUrl = null;
        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
        for (Map<String, Object> map1 : poll) {
            if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
                uploadUrl = (String) map1.get("value");
            }
        }
        if (StringUtils.isBlank(uploadUrl)) {
            EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
            return ReturnT.FAIL;
        }

        if (StringUtils.isEmpty(dataObject.getBlobField())){
            log.info("没有对象存在文件二进制字段！不进行文件上传");
        }

        PartnerConnection connect = salesforceTargetConnect.createConnect();

        List<Future<?>> futures = Lists.newArrayList();

        DataObject update = new DataObject();

        log.info("dump file api:{}, field:{}", dataObject.getName(), dataObject.getBlobField());

        try {
            String api = dataObject.getName();
            update.setName(dataObject.getName());
            List<SalesforceParam> salesforceParams = null;
            QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
            dbQw.eq("name", api);
            List<DataBatch> list = dataBatchService.list(dbQw);
            AtomicInteger batch = new AtomicInteger(1);
            if (CollectionUtils.isNotEmpty(list)) {
                salesforceParams = list.stream().map(t -> {
                    SalesforceParam salesforceParam = param.clone();
                    salesforceParam.setApi(t.getName());
                    salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                    salesforceParam.setEndCreateDate(t.getSyncEndDate());
                    salesforceParam.setBatch(batch.getAndIncrement());
                    return salesforceParam;
                }).collect(Collectors.toList());
            }

            // 手动任务优先执行
            for (SalesforceParam salesforceParam : salesforceParams) {
                String finalDownloadUrl = uploadUrl;
                Future<?> future = salesforceExecutor.execute(() -> {
                    try {
                        uploadFile(salesforceParam, connect, finalDownloadUrl,dataObject.getName());
                    } catch (Throwable throwable) {
                        log.error("salesforceExecutor error", throwable);
                        throw new RuntimeException(throwable);
                    }
                }, salesforceParam.getBatch(), 0);
                futures.add(future);
            }

            // 等待当前所有线程执行完成
            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            update.setDataWork(0);
        } catch (InterruptedException e) {
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
            throw e;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        } finally {
            update.setDataLock(0);
            dataObjectService.updateById(update);
        }
        // 等待当前所有线程执行完成
        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
        futures.clear();
        return ReturnT.SUCCESS;
    }

    /**
     * 上传Attachment
     */
    private void uploadAttachment(SalesforceParam param, PartnerConnection connection ,String uploadUrl, String api) throws Exception {

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        String token = connection.getSessionHeader().getSessionId();

        Map<String, String> headers = Maps.newHashMap();
        headers.put("Authorization", "Bearer " + token);
        headers.put("connection", "keep-alive");

        Integer count = customMapper.countBySQL(api, "where is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'");

        if (count == 0) {
            return;
        }

        int page = count%1000 == 0 ? count/1000 : (count/1000) + 1;

        log.info("总文件数 count:{}；总批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);

        //总插入数
        for (int i = 0; i < page; i++) {

            log.info("总文件数 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);

            List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_Type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000");

            for (Map<String, Object> map : list) {
                int failCount = 0;
                String id = null;
                // 上传完毕 更新附件信息
                List<Map<String, Object>> maps = Lists.newArrayList();
                CloseableHttpResponse response = null;
                String respContent = null;
                try {
                    id = (String) map.get("Id");
                    String url_fileName = (String) map.get("url");
                    String parentId = (String) map.get("ParentId");
                    String parentType = (String) map.get("Parent_Type");
                    // 判断路径是否为空
                    if (StringUtils.isNotBlank(url_fileName)) {
                        String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
                        File file = new File(filePath);
                        boolean exists = file.exists();
                        if (!exists) {
                            log.info("文件不存在");
                        }
                        String fileName = (String) map.get("Name");
                        log.info("文件名称：" + fileName);
                        // dataObject查询
                        QueryWrapper<DataObject> qw = new QueryWrapper<>();
                        qw.eq("name", parentType);
                        List<DataObject> objects = dataObjectService.list(qw);
                        if (objects.isEmpty()) {
                            log.info("关联对象不存在： {}", parentType);
                        }

                        Map<String, Object> lMap = customMapper.getById("new_id",parentType, parentId);

                        // 拼接url
                        String url = uploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
                        HttpPost httpPost = new HttpPost(url);
                        httpPost.setHeader("Authorization", "Bearer " + token);
                        httpPost.setHeader("connection", "keep-alive");

                        com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
                        credentialsJsonParam.put("parentId", lMap.get("new_id"));
                        credentialsJsonParam.put("Name", fileName);

                        MultipartEntityBuilder builder = MultipartEntityBuilder.create();
                        builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
                        builder.addBinaryBody("Body", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
                        HttpEntity entity = builder.build();
                        httpPost.setEntity(entity);

                        CloseableHttpClient httpClient = HttpClients.createDefault();

                        while (failCount < Const.MAX_FAIL_COUNT) {
                            response = httpClient.execute(httpPost);
                            if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400) {
                                HttpEntity he = response.getEntity();
                                if (he != null) {
                                    respContent = EntityUtils.toString(he, "UTF-8");
                                    String newId = com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id").toString();
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "new_id");
                                    paramMap.put("value", newId);
                                    maps.add(paramMap);
                                }
                                break;
                            } else {
                                failCount++;
                                log.error("文件上传失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + JSON.toJSONString(response));
                            }
                        }

                        if (failCount == Const.MAX_FAIL_COUNT) {
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_upload");
                            paramMap.put("value", 2);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                        }else {
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_upload");
                            paramMap.put("value", 1);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                            TimeUnit.MILLISECONDS.sleep(1);
                        }
                    }
                }  catch (InterruptedException interruptedException){
                    return;
                } catch (Exception e) {
                    if (response != null) {
                        try {
                            response.close();
                        } catch (IOException re) {
                            log.error("exception message", re);
                        }
                    }
                    log.error("文件上传失败！, id: {}, 错误信息：{}", id ,JSON.toJSONString(e));
                    Map<String, Object> paramMap = Maps.newHashMap();
                    paramMap.put("key", "is_upload");
                    paramMap.put("value", 2);
                    maps.add(paramMap);
                    customMapper.updateById(api, maps, id);
                    EmailUtil.send("File Upload ERROR", "文件上传失败！, id: "+ id + "，错误信息：" + JSON.toJSONString(e));
                }
            }
        }
    }

    /**
     * 上传文件
     */
    private void uploadFile(SalesforceParam param, PartnerConnection connection ,String uploadUrl, String api) throws Exception {

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }
        String token = connection.getSessionHeader().getSessionId();

        Map<String, String> headers = Maps.newHashMap();
        headers.put("Authorization", "Bearer " + token);
        headers.put("connection", "keep-alive");

        Integer count = customMapper.countBySQL("ContentDocument", "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'");

        if (count == 0) {
            return;
        }

        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        log.info("总文件数 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);

        //总插入数
        for (int i = 0; i < page; i++) {

            // 获取未存储的附件id
            List<Map<String, Object>> documentList = customMapper.list("Id " , "ContentDocument", " new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 200 + ",200");

            for (Map<String, Object> documentMap : documentList) {

                String documentId = (String) documentMap.get("Id");
                // 获取未存储的附件id
                List<Map<String, Object>> list = customMapper.list("Id, url, PathOnClient, Title, ContentDocumentId, VersionNumber", api, "ContentDocumentId = '" + documentId + "' and new_id is null ORDER BY VersionNumber ASC");
                if (CollectionUtils.isEmpty(list)) {
                    continue;
                }
                String finalUploadUrl = uploadUrl;
                String newDocumentId = null;
                for (Map<String, Object> map : list) {
                    int failCount = 0;
                    String id = null;
                    // 上传完毕 更新附件信息
                    List<Map<String, Object>> maps = Lists.newArrayList();
                    CloseableHttpResponse response = null;
                    String respContent = null;
                    try {
                        id = (String) map.get("Id");
                        String url_fileName = (String) map.get("url");
                        String fileName = (String) map.get("PathOnClient");
                        String title = (String) map.get("Title");
                        String oldDocumentId = (String) map.get("ContentDocumentId");
                        Integer versionNumber = Integer.valueOf(map.get("VersionNumber").toString());
                        log.info("文件名称：" + fileName);
                        // 判断路径是否为空
                        if (StringUtils.isNotBlank(url_fileName)) {
                            String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
                            File file = new File(filePath);
                            boolean exists = file.exists();
                            if (!exists) {
                                log.info("文件不存在");
                                break;
                            }
                            // 拼接url
                            String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
                            HttpPost httpPost = new HttpPost(url);
                            httpPost.setHeader("Authorization", "Bearer " + token);
                            httpPost.setHeader("connection", "keep-alive");

                            com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
                            credentialsJsonParam.put("title", title);
                            credentialsJsonParam.put("pathOnClient", fileName);
                            if (newDocumentId != null) {
                                credentialsJsonParam.put("ContentDocumentId", newDocumentId);
                            }
                            MultipartEntityBuilder builder = MultipartEntityBuilder.create();
                            builder.addTextBody("entity_content", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
                            builder.addBinaryBody("VersionData", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
                            HttpEntity entity = builder.build();
                            httpPost.setEntity(entity);

                            CloseableHttpClient httpClient = HttpClients.createDefault();
                            while (failCount < Const.MAX_FAIL_COUNT) {
                                response = httpClient.execute(httpPost);
                                if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400 && response.getEntity() != null) {
                                    HttpEntity he = response.getEntity();
                                    respContent = EntityUtils.toString(he, "UTF-8");
                                    String newId = String.valueOf(com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id"));
                                    if (StringUtils.isBlank(newId)) {
                                        log.error("文件上传错误，返回实体信息：" + respContent);
                                        break;
                                    }
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "new_id");
                                    paramMap.put("value", newId);
                                    maps.add(paramMap);

                                    if (versionNumber == 1) {
                                        // 更新documentId
                                        String dId = commonService.getDocumentId(connection, newId);
                                        List<Map<String, Object>> dList = new ArrayList<>();
                                        Map<String, Object> dMap = Maps.newHashMap();
                                        dMap.put("key", "new_id");
                                        dMap.put("value", dId);
                                        dList.add(dMap);
                                        customMapper.updateById("ContentDocument", dList, oldDocumentId);
                                        newDocumentId = dId;
                                    }
                                    break;
                                } else {
                                    failCount++;
                                    log.error("文件上传失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + JSON.toJSONString(response));
                                }
                            }

                            if (failCount == Const.MAX_FAIL_COUNT) {
                                Map<String, Object> paramMap = Maps.newHashMap();
                                paramMap.put("key", "is_upload");
                                paramMap.put("value", 2);
                                maps.add(paramMap);
                                customMapper.updateById(api, maps, id);
                            }else {
                                Map<String, Object> paramMap = Maps.newHashMap();
                                paramMap.put("key", "is_upload");
                                paramMap.put("value", 1);
                                maps.add(paramMap);
                                customMapper.updateById(api, maps, id);
                                TimeUnit.MILLISECONDS.sleep(1);
                            }
                        }
                    } catch (Exception e) {
                        if (response != null) {
                            try {
                                response.close();
                            } catch (IOException re) {
                                log.error("exception message", re);
                            }
                        }
                        log.error("文件上传失败！, id: {}, 错误信息：{}", id ,JSON.toJSONString(e));
                        Map<String, Object> paramMap = Maps.newHashMap();
                        paramMap.put("key", "is_upload");
                        paramMap.put("value", 2);
                        maps.add(paramMap);
                        customMapper.updateById(api, maps, id);
                        EmailUtil.send("File Upload ERROR", "文件上传失败！, id: "+ id + "，错误信息：" + JSON.toJSONString(e));
                    }
                }
            }
        }
    }


    /**
     * updateLinkType入口
     */
    @Override
    public ReturnT<String> updateLinkTypeJob(SalesforceParam param) throws Exception {
        List<Future<?>> futures = Lists.newArrayList();
        try {
            if (param.getType() == 1){
                return updateLinkType(param, futures);
            }else {
                return updateLinkTypeBatch(param,futures);
            }
        }  catch (InterruptedException e) {
            throw e;
        }  catch (Throwable throwable) {
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
            log.error("updateLinkTypeJob error", throwable);
            throw throwable;
        }
    }


    /**
     * 【batch逻辑】组装updateLinkType执行参数
     */
    public ReturnT<String> updateLinkTypeBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
        List<String> apis;
        if (StringUtils.isBlank(param.getApi())) {
            QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("is_link","1").eq("is_create",1);
            apis = linkConfigService.list(queryWrapper).stream().map(LinkConfig::getApi).collect(Collectors.toList());
        } else {
            apis = DataUtil.toIdList(param.getApi());
        }
        String beginDateStr = null;
        String endDateStr = null;
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }
        QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
        wrapper.in("api", apis);
        Map<String, String> resultMap = dataObjectService.list().stream()
                .collect(Collectors.toMap(
                        DataObject::getKeyPrefix,
                        DataObject::getName,
                        (existing, replacement) -> replacement));

        for (String api : apis) {
            try {
                if (!dataFieldService.hasCreatedDate(api)){
                    log.info("当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
                    EmailUtil.send("DataDump ERROR", "当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
                    break;
                }
                QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
                dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
                List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
                if (linkConfigs.isEmpty()){
                    continue;
                }
                List<SalesforceParam> salesforceParams = null;
                QueryWrapper<DataBatch> dbQw1 = new QueryWrapper<>();
                dbQw1.eq("name", api);
                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
                    dbQw1.eq("sync_start_date", beginDateStr);  // 等于开始时间
                    dbQw1.eq("sync_end_date", endDateStr);    // 等于结束时间
                }
                List<DataBatch> list = dataBatchService.list(dbQw1);
                AtomicInteger batch = new AtomicInteger(1);
                if (CollectionUtils.isNotEmpty(list)) {
                    salesforceParams = list.stream().map(t -> {
                        SalesforceParam salesforceParam = param.clone();
                        salesforceParam.setApi(t.getName());
                        salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                        salesforceParam.setEndCreateDate(t.getSyncEndDate());
                        salesforceParam.setBatch(batch.getAndIncrement());
                        return salesforceParam;
                    }).collect(Collectors.toList());
                }
                Set<String> safeSet = new HashSet<>();

                // 手动任务优先执行
                for (SalesforceParam salesforceParam : salesforceParams) {
                    Future<?> future = salesforceExecutor.execute(() -> {
                        try {
                            updateLinkBatch(salesforceParam, resultMap,safeSet,linkConfigs);
                        } catch (Throwable throwable) {
                            log.error("salesforceExecutor error", throwable);
                            throw new RuntimeException(throwable);
                        }
                    }, salesforceParam.getBatch(), 1);
                    futures.add(future);
                }

                if (!safeSet.isEmpty()){
                    String format = String.format("更新关联类型 error, api name: %s,  \ncause:\n%s 前三位编码对象不存在！！",  api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
                    EmailUtil.send("DataDump ERROR", format);
                }

                // 等待当前所有线程执行完成
                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            } catch (Exception e) {
                throw e;
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 执行数据updateLinkType
     */
    private void updateLinkBatch(SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        String sql = "CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
        String allFiled = "Id";

        if (linkConfigs.size() == 1){
            for (LinkConfig linkConfig : linkConfigs) {
                sql = sql + " and " + linkConfig.getField() + " is not null";
                allFiled = allFiled + "," + linkConfig.getField();
            }
        }else {
            String info = null;
            for (LinkConfig linkConfig : linkConfigs) {
                if (StringUtils.isBlank(info)){
                    info = linkConfig.getField() + " is not null" ;
                }else {
                    info = info + " or " + linkConfig.getField() + " is not null";
                }
                allFiled = allFiled + "," + linkConfig.getField();
            }
            sql = sql +" and ("+ info +")";
        }

        // 表内数据总量
        Integer count = customMapper.countBySQL(param.getApi(), "where " + sql);

        log.info("表api：{} 存在" +count+ "条需更新数据！开始时间:{},结束时间：{}", param.getApi(), beginDateStr, endDateStr);

        if (count >0 ) {
            int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;

            for (int i = 0; i < page; i++) {

                log.info("表api：{}，批次：{}，单批次数：2000，开始时间：{}，结束时间：{}，执行数据更新!", param.getApi(), i,beginDateStr, endDateStr);

                List<Map<String, Object>> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000");

                for (int j = 1; j <= mapList.size(); j++) {
                    List<Map<String, Object>> updateMapList = new ArrayList<>();
                    Map<String, Object> map = mapList.get(j - 1);
                    for (LinkConfig config : linkConfigs) {
                        if (map.get(config.getField()) != null){
                            String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", config.getLinkField());
                            paramMap.put("value", type);
                            updateMapList.add(paramMap);
                            if (StringUtils.isBlank(type)){
                                safeSet.add(map.get(config.getField()).toString().substring(0, 3));
                            }else {

                            }
                        }
                    }
                    if (!updateMapList.isEmpty()) {
                        customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
                    }
                }
            }
        }

    }

    /**
     * 组装updateLinkType执行参数
     */
    public ReturnT<String> updateLinkType(SalesforceParam param, List<Future<?>> futures) throws Exception {
        Set<String> apis;
        if (StringUtils.isBlank(param.getApi())) {
            apis = linkConfigService.list().stream().map(LinkConfig::getApi).collect(Collectors.toSet());
        } else {
            apis = DataUtil.toIdSet(param.getApi());
        }

        //查询所有创建字段的关联类型
        QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
        wrapper.in("api", apis).eq("is_link",1).eq("is_create",1);
        Map<String, String> resultMap = dataObjectService.list().stream()
                .collect(Collectors.toMap(
                        DataObject::getKeyPrefix,
                        DataObject::getName,
                        (existing, replacement) -> replacement));



        for (String api : apis) {
            //邮件发送前三位编码不存在的对象
            Set<String> safeSet = new HashSet<>();
            QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
            dbQw.eq("api", api).eq("is_link",1).eq("is_create",1);
            List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
            if (linkConfigs.isEmpty()){
                continue;
            }
            Future<?> future = salesforceExecutor.execute(() -> {
                try {
                    updateLink(api,param, resultMap,safeSet,linkConfigs);
                } catch (Throwable throwable) {
                    log.error("salesforceExecutor error", throwable);
                    throw new RuntimeException(throwable);
                }
            }, 0, 1);
            futures.add(future);

            if (!safeSet.isEmpty()){
                String format = String.format("更新关联类型 error, api name: %s,  \ncause:\n%s 前三位编码对象不存在！！",  api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
                EmailUtil.send("DataDump ERROR", format);
            }
        }
        // 等待当前所有线程执行完成
        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));

        return ReturnT.SUCCESS;
    }


    /**
     * 执行数据updateLinkType
     */
    private void updateLink(String api,SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {

        String beginDateStr = null;
        Date beginDate = param.getBeginModifyDate();
        if (param.getBeginModifyDate() != null ){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
        }
        String sql1 = null;
        String sql2 = null;
        if (beginDateStr != null){
            String updateDateField = dataFieldService.returnUpdateDateField(api);
            sql1 = "where " + updateDateField + " >= '" + beginDateStr + "'";
            sql2 = updateDateField + " >= '"+ beginDateStr +"' " ;
        }else {
            sql1 = "where 1=1";
            sql2 = "1=1  " ;
        }

        String allFiled = "Id";

        if (linkConfigs.size() == 1){
            for (LinkConfig linkConfig : linkConfigs) {
                sql1 = sql1 + " and " + linkConfig.getField() + " is not null";
                sql2 = sql2 + " and " + linkConfig.getField() + " is not null";
                allFiled = allFiled + "," + linkConfig.getField();
            }
        }else {
            String info = null;
            for (LinkConfig linkConfig : linkConfigs) {
                if (StringUtils.isBlank(info)){
                    info = linkConfig.getField() + " is not null" ;
                }else {
                    info = info + " or " + linkConfig.getField() + " is not null";
                }
                allFiled = allFiled + "," + linkConfig.getField();
            }
            sql1 = sql1 +" and ("+ info +")";
            sql2 = sql2 +" and ("+ info +")";
        }


        // 表内数据总量
        Integer count = customMapper.countBySQL(api, sql1);

        log.info("表api：{} 存在" +count+ "条数据！", api);

        if (count >0 ) {
            int page = count % 10000 == 0 ? count / 10000 : (count / 10000) + 1;

            for (int i = 0; i < page; i++) {

                log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 10000);

                List<Map<String, Object>> mapList = customMapper.list(allFiled, api, sql2 +" order by Id limit "+ + i * 10000 + ",10000" );

                for (int j = 1; j <= mapList.size(); j++) {
                    List<Map<String, Object>> updateMapList = new ArrayList<>();
                    Map<String, Object> map = mapList.get(j - 1);
                    for (LinkConfig config : linkConfigs) {
                        if (map.get(config.getField()) != null){
                            String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", config.getLinkField());
                            paramMap.put("value", type);
                            updateMapList.add(paramMap);
                            if (StringUtils.isBlank(type)){
                                safeSet.add(map.get(config.getField()).toString().substring(0, 3));
                            }
                        }
                    }
                    if (!updateMapList.isEmpty()){
                        customMapper.updateById(api, updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
                    }
                }
            }
        }

    }


    /**
     * 校验删除数据入口
     */
    @Override
    public ReturnT<String> checkDeletedData(SalesforceParam param) throws Exception {

        if (param.getType() == 1){
            return manualCheckDeletedData(param);
        }else {
            return autoCheckDeletedData(param);
        }

    }

    /**
     * 删除目标ORG中的数据
     * @param param 参数
     * @return 执行结果
     * @throws Exception 异常
     */
    @Override
    public ReturnT<String> deleteTargetOrgData(SalesforceParam param) throws Exception {
        log.info("deleteTargetOrgData execute start ..................");

        QueryWrapper<DataObject> wrapper = new QueryWrapper<>();
        if (StringUtils.isNotBlank(param.getApi())) {
            wrapper.in("name", DataUtil.toIdList(param.getApi()));
        } else {
            wrapper.eq("data_work", 1);
        }
        List<DataObject> objectList = dataObjectService.list(wrapper);
        if (objectList.isEmpty()) {
            log.info("没有对象需要删除数据！!!");
            return ReturnT.FAIL;
        }

        PartnerConnection connect = salesforceTargetConnect.createConnect();

        try {
            for (DataObject dataObject : objectList) {

                String api = dataObject.getName();

                boolean moreData = true;

                DescribeSObjectResult dsr = connect.describeSObject(api);

                Field[] dsrFields = dsr.getFields();

                String sql = "select Id from " + api + " Limit 200";

                log.info("执行SQL：{}", sql);

                while (moreData){
                    int size = 0;

                    QueryResult queryResult = connect.queryAll(sql);

                    if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
                        break;
                    }
                    SObject[] records = queryResult.getRecords();

                    com.alibaba.fastjson2.JSONArray objects = DataUtil.toJsonArray(records, dsrFields);
                    String[] ids = new String[objects.size()];
                    for (int i = 0; i < objects.size(); i++) {
                        JSONObject jsonObject = objects.getJSONObject(i);
                        String id = jsonObject.getString("Id");
                        ids[i] = id;
                        size ++;
                    }
                    DeleteResult[] deleteResults = connect.delete(ids);
                    for (DeleteResult deleteResult : deleteResults) {
                        if (deleteResult.isSuccess()) {
                            log.info("成功删除目标ORG中的数据，ID: {}，类型: {}", deleteResult.getId(), dataObject.getLabel());
                        } else {
                            log.error("ID:{}删除目标ORG数据失败：{}", deleteResult.getId(), deleteResult.getErrors());
                        }
                    }
                    if (size < 200){
                        moreData = false;
                    }
                }
            }
            TimeUnit.MILLISECONDS.sleep(1);

        }catch (InterruptedException interruptedException){
            return ReturnT.FAIL;
        }catch (Exception e){
            log.info("目标ORG数据删除失败！", e);
        }

        log.info("目标ORG数据删除完成！");

        return ReturnT.SUCCESS;
    }

    private ReturnT<String> manualCheckDeletedData(SalesforceParam param) throws Exception {

        String api = "DeleteEvent";

        PartnerConnection connect = salesforceTargetConnect.createConnect();

        if ((param.getEndCreateDate() == null || param.getBeginCreateDate() == null) && param.getIds().isEmpty()){
            log.info("手动删除数据必须指定参数：开始与结束时间/Ids !!!!");
            return ReturnT.FAIL;
        }
        String sql = "";

        if (!param.getIds().isEmpty()){
            sql = "where Record in (" + param.getIds().stream().map(id -> "'" + id + "'").collect(Collectors.joining(",")) + ")";
        }else {
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            String beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            String endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
            sql = "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
        }

        Integer count = customMapper.countBySQL(api, sql);

        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        log.info("删除数据，总批次：{}",page);
        try{
            for (int i = 0; i < page; i++) {

                List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");

                log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());

                Map<String, String> idNameMap = new HashMap<>();
                String[] ids = new String[list.size()];

                for (int z = 0; z < list.size(); z++) {
                    Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
                    if (objectMap != null && objectMap.get("new_id") != null) {
                        ids[z] = objectMap.get("new_id").toString();
                        idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
                    }else {
                        log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
                    }
                }

                int index = 0;
                DeleteResult[] deleteResults = connect.delete(ids);
                for (DeleteResult deleteResult : deleteResults) {
                    if (deleteResult.isSuccess()){
                        String typeName = idNameMap.get(ids[index]);
                        Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);

                        List<Map<String, Object>> maps = new ArrayList<>();

                        Map<String, Object> m1 = new HashMap<>();
                        m1.put("key", "is_delete");
                        m1.put("value", 1);

                        Map<String, Object> m2 = new HashMap<>();
                        m2.put("key", "all_data");
                        m2.put("value", JSON.toJSONString(objectMap));

                        maps.add(m1);
                        maps.add(m2);
                        customMapper.updateById(api, maps, ids[index]);

                        customMapper.deleteOne(typeName, ids[index]);

                    }else {
                        log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
                    }
                    index++;
                }
            }
            TimeUnit.MILLISECONDS.sleep(1);
        }catch (InterruptedException e){
            return ReturnT.FAIL;
        } catch (Exception e) {
            log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
        }

        return ReturnT.SUCCESS;
    }


    private ReturnT<String> autoCheckDeletedData(SalesforceParam param) throws Exception {
        String api = "DeleteEvent";
        PartnerConnection connect = salesforceTargetConnect.createConnect();

        String sql = "SystemModstamp > " + DateUtil.format(DateUtils.addDays(new Date(), -15), "yyyy-MM-dd HH:mm:ss");

        Integer count = customMapper.countBySQL(api, sql);

        int page = count%200 == 0 ? count/200 : (count/200) + 1;

        log.info("删除数据，总批次：{}",page);
        try{
            for (int i = 0; i < page; i++) {

                List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");

                log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());

                Map<String, String> idNameMap = new HashMap<>();
                String[] ids = new String[list.size()];

                for (int z = 0; z < list.size(); z++) {
                    Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
                    if (objectMap != null && objectMap.get("new_id") != null) {
                        ids[z] = objectMap.get("new_id").toString();
                        idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
                    }else {
                        log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
                    }
                }

                int index = 0;
                DeleteResult[] deleteResults = connect.delete(ids);
                for (DeleteResult deleteResult : deleteResults) {
                    if (deleteResult.isSuccess()){
                        String typeName = idNameMap.get(ids[index]);
                        Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);

                        List<Map<String, Object>> maps = new ArrayList<>();

                        Map<String, Object> m1 = new HashMap<>();
                        m1.put("key", "is_delete");
                        m1.put("value", 1);

                        Map<String, Object> m2 = new HashMap<>();
                        m2.put("key", "all_data");
                        m2.put("value", JSON.toJSONString(objectMap));

                        maps.add(m1);
                        maps.add(m2);
                        customMapper.updateById(api, maps, ids[index]);

                        customMapper.deleteOne(typeName, ids[index]);

                    }else {
                        log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
                    }
                    index++;
                }
            }
            TimeUnit.MILLISECONDS.sleep(1);
        }catch (InterruptedException e){
            return ReturnT.FAIL;
        } catch (Exception e) {
            log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
        }
        return ReturnT.SUCCESS;
    }


    /**
     * 参数转换qw
     *
     * @param param 参数
     * @return qw
     */
    private static QueryWrapper<DataBatch> getDataBatchQueryWrapper(SalesforceParam param) {
        QueryWrapper<DataBatch> qw = new QueryWrapper<>();
        if (StringUtils.isNotBlank(param.getApi())) {
            List<String> strings = DataUtil.toIdList(param.getApi());
            if (CollectionUtils.isNotEmpty(strings)) {
                qw.in("name", strings);
            }
        }
        // 只查询首次同步时间不为空，且首次sf num不为0的
        qw.isNotNull("first_sync_date")
                .ne("first_sf_num", 0)
                .orderByAsc("created_date", "name", "sync_start_date")
                .select("id", "name", "label", "sync_start_date", "sync_end_date", "first_db_num", "first_sf_num", "db_num", "sf_num", "sync_status");
        return qw;
    }


    @Override
    public ReturnT<String> insertSingle(SalesforceParam param) throws Exception {
        List<Future<?>> futures = Lists.newArrayList();
        try {
            if (StringUtils.isNotBlank(param.getApi())) {
                return insertSingleData(param, futures);
            } else {
                return autoInsertSingleData(param, futures);
            }
        }  catch (InterruptedException e) {
            throw e;
        }  catch (Throwable throwable) {
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
            log.error("insertSingle error", throwable);
            throw throwable;
        }
    }

    @Override
    public ReturnT<String> dumpContentFolderMemberJob(SalesforceParam param) throws Exception {
        String api = "ContentFolderMember";
        PartnerConnection partnerConnection = salesforceConnect.createConnect();

        List<Map<String, Object>> list = customMapper.list("Id", "ContentFolder", "new_id is not null order by Id ");
        if (list.isEmpty()){
            return ReturnT.SUCCESS;
        }
        DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
        List<String> fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList());
        Field[] dsrFields = dsr.getFields();
        try {
            for (Map<String, Object> map : list) {
                String contentFolderId = (String) map.get("Id");

                log.info("ContentFolderMember数据拉取！！！ contentFolderId：" + contentFolderId);

                String sql = "SELECT Id, ParentContentFolderId, ChildRecordId, IsDeleted, SystemModstamp, CreatedById, CreatedDate, LastModifiedById, LastModifiedDate FROM ContentFolderMember where ParentContentFolderId = '" + contentFolderId + "'";
                com.alibaba.fastjson2.JSONArray objects = null;
                QueryResult queryResult = partnerConnection.queryAll(sql);
                SObject[] records = queryResult.getRecords();
                objects = DataUtil.toJsonArray(records, dsrFields);
                commonService.saveOrUpdate(api, fields, records, objects, true);
            }
        } catch (Throwable e) {
            log.error("dumpContentFolderMemberJob error message:{}", e.getMessage());
            return ReturnT.FAIL;
        }

        return ReturnT.SUCCESS;
    }

    /**
     * 富文本图片替换
     * @param param 参数
     * @return 执行结果
     * @throws Exception 异常
     */
    @Override
    public ReturnT<String> richTextImageReplace(SalesforceParam param) throws Exception {
        log.info("richTextImageReplace execute start ..................");
        String beginDateStr = null;
        String endDateStr = null;

        if (param.getType() != 1){
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
                beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
                endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
            }
        }
        // 检测路径是否存在 不存在则创建
        File excel = new File(Const.SERVER_FILE_PATH + "/richTextImages" );
        if (!excel.exists()) {
            log.info("创建API目录: {}", excel.getAbsolutePath());
            boolean mkdir = excel.mkdir();
            if (!mkdir) {
                log.info("创建文件存储目录失败！");
            }
        }

        // 获取连接
        PartnerConnection sourceConnection = salesforceConnect.createConnect();
        PartnerConnection targetConnection = salesforceTargetConnect.createConnect();

        QueryWrapper<DataObject> qw = new QueryWrapper<>();
        if (StringUtils.isNotBlank(param.getApi())) {
            List<String> apis = DataUtil.toIdList(param.getApi());
            qw.in("name", apis);
        }
        qw.eq("data_work", true);
        List<DataObject> objectList = dataObjectService.list(qw);

        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);

        for (DataObject dataObject : objectList) {
            String api = dataObject.getName();
            QueryWrapper<DataField> qw1 = new QueryWrapper<>();
            qw1.eq("api", api);
            qw1.eq("isHtmlFormatted", true);
            List<DataField> fields = dataFieldService.list(qw1);

            for (DataField dataField : fields) {
                String field = dataField.getField();
                String sqlCondition = "new_id is not null and " + field + " like '%<img%refid%'";

                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
                    sqlCondition += " and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
                }

                // 获取记录总数
                Integer count = customMapper.countBySQL(api, "where " + sqlCondition);
                log.info("对象 {} 字段 {} 需要处理的富文本记录总数: {}", api, field, count);

                if (count == 0) {
                    log.info("对象 {} 字段 {} 没有需要处理的富文本记录", api, field);
                    continue;
                }

                // 分批处理，每批200条记录
                int pageSize = 200;
                int totalPages = (count + pageSize - 1) / pageSize;

                for (int page = 0; page < totalPages; page++) {

                    List<Map<String, Object>> records = customMapper.list("Id, new_id, " + field, api,
                            sqlCondition + " order by Id limit " + (page * pageSize) + "," + pageSize);

                    log.info("处理对象 {} 字段 {} 第 {}/{} 页，本页记录数: {}", api, field, (page + 1), totalPages, records.size());

                    SObject[] accounts = new SObject[records.size()];
                    int index = 0;

                    for (Map<String, Object> record : records) {
                        try {
                            String sourceId = (String) record.get("Id");
                            String targetId = (String) record.get("new_id");
                            String richTextContent = (String) record.get(field);

                            if (StringUtils.isBlank(richTextContent) || !richTextContent.contains("refid")) {
                                continue;
                            }

                            // 处理富文本中的图片
                            String updatedContent = processRichTextImages(richTextContent, sourceId, api, field, sourceConnection, targetConnection);

                            if (updatedContent == null){
                                return ReturnT.FAIL;
                            }

                            // 如果内容有变化，则更新目标组织中的记录
                            if (!richTextContent.equals(updatedContent)) {

                                // 更新目标组织中的记录
                                SObject sObject = new SObject();
                                sObject.setType(api);
                                sObject.setId(targetId);
                                sObject.setField(field, updatedContent);
                                sObject.setField("Id", targetId);
                                accounts[index] = sObject;
                                index ++;

                                List<Map<String, Object>> maps = new ArrayList<>();
                                Map<String, Object> linkMap = new HashMap<>();
                                linkMap.put("key", field + "_back");
                                linkMap.put("value", richTextContent);
                                maps.add(linkMap);
                                customMapper.updateByNewId(maps,api, targetId);

                            }
                        } catch (Exception e) {
                            log.error("处理记录 {} 的富文本内容时发生错误: {}", record.get("Id"), e.getMessage(), e);
                        }
                    }

                    if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                        printlnAccountsDetails(accounts,fields);
                    }
                    SaveResult[] saveResults = targetConnection.update(accounts);
                    for (SaveResult saveResult : saveResults) {
                        if (!saveResult.getSuccess()) {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 2);
                            maps.add(linkMap);
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", JSON.toJSONString(saveResult.getErrors()));
                            maps.add(linkMap1);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                            String format = String.format("数据更新 error, api name: %s, \ncause:\n%s",  api, JSON.toJSONString(saveResult));
                            log.info(format);
                        }else {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap = new HashMap<>();
                            linkMap.put("key", "is_update");
                            linkMap.put("value", 1);
                            maps.add(linkMap);
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", null);
                            maps.add(linkMap1);
                            customMapper.updateByNewId(maps,api, saveResult.getId());
                        }
                    }
                }
            }
        }

        log.info("richTextImageReplace execute end ..................");
        return ReturnT.SUCCESS;
    }

    /**
     * 处理富文本中的图片
     * @param richTextContent 原始富文本内容
     * @param sourceRecordId 源记录ID
     * @param api 对象API名称
     * @param field 字段名
     * @param sourceConnection 源组织连接
     * @param targetConnection 目标组织连接
     * @return 处理后的富文本内容
     * @throws Exception 异常
     */
    private String processRichTextImages(String richTextContent, String sourceRecordId, String api, String field,
                                         PartnerConnection sourceConnection, PartnerConnection targetConnection) throws Exception {
        // 匹配富文本中包含refid的img标签
        Pattern imgPattern = Pattern.compile("<img[^>]*src\\s*=\\s*\"[^\"]*refid=([^\"]+)(?:&quot;|&amp;|&|\\\"|\"|')?[^>]*/?>", Pattern.CASE_INSENSITIVE);        Matcher matcher = imgPattern.matcher(richTextContent);

        StringBuffer updatedContent = new StringBuffer();
        Map<String, String> imageIdMap = new ConcurrentHashMap<>();

        // 遍历所有匹配的img标签
        while (matcher.find()) {
            String imgTag = matcher.group(0);
            String refId = matcher.group(1);

            // 检查是否已经处理过这个refId
            if (imageIdMap.containsKey(refId)) {
                String newRefId = imageIdMap.get(refId);
                String updatedImgTag = imgTag.replace("refid=\"" + refId + "\"", "refid=\"" + newRefId + "\"");
                matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(updatedImgTag));
                continue;
            }

            try {
                // 构建源图片下载URL (参考Python脚本中的imageUrl构建方式)
                String downloadUrl = FILE_DOWNLOAD_URL +
                        String.format("/services/data/v55.0/sobjects/%s/%s/richTextImageFields/%s/%s",
                                api, sourceRecordId, field, refId);

                // 获取图片的alt属性作为标题
                String title = "image";
                Pattern altPattern = Pattern.compile("alt=\"([^\"]*)\"", Pattern.CASE_INSENSITIVE);
                Matcher altMatcher = altPattern.matcher(imgTag);
                if (altMatcher.find()) {
                    title = altMatcher.group(1);
                }
                    // 下载源图片文件
                String downloadImageFile = downloadImageFile(sourceConnection, downloadUrl, title);

                if (downloadImageFile != null) {

                    // 上传图片到目标组织
                    String newContentVersionId = uploadImageFile(targetConnection, downloadImageFile, title);

                    if (newContentVersionId != null) {
                        // 获取新的ContentDocumentId
                        Map<String, String> documentId = getDocumentId(targetConnection, newContentVersionId);
                        String newContentDocumentId = documentId.get("ContentDocumentId");
                        String newOwnId = documentId.get("OwnerId");

                        if (newContentDocumentId != null) {
                            imageIdMap.put(refId, newContentDocumentId);

                        // 构建新的img标签 (参考Python脚本中的richTextUrl构建方式)
                        String newImageUrl =
                                "/sfc/servlet.shepherd/version/download/" + newContentVersionId;
                        String updatedImgTag = "<img src=\"" + newImageUrl + "\" refid=\"" + newContentDocumentId + "\"/>";

                        matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(updatedImgTag));

                        // 记录替换日志
                        richTextLogService.logRichTextImageReplace(
                                api,
                                field,
                                newContentDocumentId,
                                newOwnId, // recordOwnerId 可以从记录中获取，但此处简化处理
                                richTextContent,
                                refId, // 原图片链接
                                newImageUrl, // 新图片链接
                                newContentDocumentId // 新文件ID
                            );
                            continue;
                        }
                    }
                }else {
                    return null;
                }

                // 如果找不到对应关系，保留原始标签
                matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(imgTag));
            } catch (Exception e) {
                log.error("处理富文本图片时发生错误，refId: {}", refId, e);
                matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(imgTag));
            }
        }
        matcher.appendTail(updatedContent);

        return updatedContent.toString();
    }

    /**
     * 从源组织下载图片文件
     * @param connection 源组织连接
     * @param downloadUrl 下载URL
     * @return 图片文件数据
     * @throws Exception 异常
     */
    private String downloadImageFile(PartnerConnection connection, String downloadUrl,String title) throws Exception {
        String imageServer = null;
        try {
            String token = connection.getSessionHeader().getSessionId();

            Map<String, String> headers = Maps.newHashMap();
            headers.put("Authorization", "Bearer " + token);
            headers.put("connection", "keep-alive");

            Response response = HttpUtil.doGet(downloadUrl, null, headers);

            if (response.body() != null && response.code() == 200){
                InputStream inputStream = response.body().byteStream();
                imageServer = dumpToRichTextImageServer(title, response, inputStream);
            }else {
                throw new RuntimeException("下载图片文件时发生错误，URL: " + downloadUrl + "；错误信息：" + response.message());
            }
        } catch (Exception e) {
            log.error("下载图片文件时发生错误，URL: {}", downloadUrl, e);
            return null;
        }
        return imageServer;
    }

    /**
     * 保存富文本图片到本地服务器
     * @param response
     * @param inputStream
     * @throws IOException
     */
    private String dumpToRichTextImageServer(String name, Response response, InputStream inputStream) throws IOException {
        String path = Const.SERVER_FILE_PATH + "/richTextImages/" + name;
        long offset = 0L;
        RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
        while (true) {
            try {
                // 保存到本地
                byte[] buf = new byte[8192];
                int len = 0;
                if (offset > 0) {
                    inputStream.skip(offset);
                    accessFile.seek(offset);
                }
                while ((len = inputStream.read(buf)) != -1) {
                    accessFile.write(buf, 0, len);
                    offset += len;
                }
                break;
            } catch (Exception e) {
                if (offset <= 0) {
                    throw e;
                }
                response.close();
            }
        }
        accessFile.close();
        return path;
    }

    /**
     * 上传图片文件到目标组织
     * @param connection 目标组织连接
=     * @param title 文件标题
     * @return 新的ContentVersion ID
     * @throws Exception 异常
     */
    private String uploadImageFile(PartnerConnection connection, String filePath, String title) throws Exception {
        String newId = null;
        try {
            int failCount = 0;

            String token = connection.getSessionHeader().getSessionId();

            String uploadUrl = FILE_UPLOAD_URL +
                    "/services/data/v55.0/sobjects/ContentVersion";

            HttpPost httpPost = new HttpPost(uploadUrl);

            httpPost.setHeader("Authorization", "Bearer " + token);
            httpPost.setHeader("connection", "keep-alive");

            com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
            credentialsJsonParam.put("title", title);
            credentialsJsonParam.put("pathOnClient", title);

            MultipartEntityBuilder builder = MultipartEntityBuilder.create();
            builder.addTextBody("entity_content", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
            builder.addBinaryBody("VersionData", new FileInputStream(filePath), ContentType.APPLICATION_OCTET_STREAM, title + ".jpg");
            HttpEntity entity = builder.build();
            httpPost.setEntity(entity);

            CloseableHttpClient httpClient = HttpClients.createDefault();

            CloseableHttpResponse response = null;
            String respContent = null;
            while (failCount < Const.MAX_FAIL_COUNT) {
                response = httpClient.execute(httpPost);
                if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400 && response.getEntity() != null) {
                    HttpEntity he = response.getEntity();
                    respContent = EntityUtils.toString(he, "UTF-8");
                    newId = String.valueOf(com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id"));
                    break;
                } else {
                    failCount++;
                    log.error("富文本图片替换图片文件上传失败，返回体信息：" + JSON.toJSONString(response));
                }
            }
        } catch (Exception e) {
            log.error("上传图片文件时发生错误，标题: {}", title, e);
            return null;
        }
        return newId;
    }

    /**
     * 获取ContentVersion ID
     * @param connection Salesforce连接
     * @param contentDocumentId ContentDocument ID
     * @return ContentVersion ID
     * @throws Exception 异常
     */
    private String getContentVersionId(PartnerConnection connection, String contentDocumentId) throws Exception {
        String query = "SELECT Id FROM ContentVersion WHERE ContentDocumentId = '" + contentDocumentId + "' ORDER BY CreatedDate DESC LIMIT 1";
        QueryResult result = connection.query(query);
        if (result.getRecords() != null && result.getRecords().length > 0) {
            return (String) result.getRecords()[0].getField("Id");
        }
        return null;
    }

    /**
     * 获取目标组织中的ContentDocumentId
     * @param connection 目标组织连接
     * @param contentVersionId ContentVersion ID
     * @return 目标组织中的ContentDocumentId
     * @throws Exception 异常
     */
    private String getTargetContentDocumentId(PartnerConnection connection, String contentVersionId) throws Exception {
        String query = "SELECT ContentDocumentId FROM ContentVersion WHERE Id = '" + contentVersionId + "'";
        QueryResult result = connection.query(query);
        if (result.getRecords() != null && result.getRecords().length > 0) {
            return (String) result.getRecords()[0].getField("ContentDocumentId");
        }
        return null;
    }

    /**
     * 获取ContentDocumentId和OwnerId
     * @param connection 连接
     * @param contentVersionId ContentVersion ID
     * @return 包含ContentDocumentId和OwnerId的Map
     * @throws Exception 异常
     */
    private Map<String, String> getDocumentId(PartnerConnection connection, String contentVersionId) throws Exception {
        String soql = "SELECT ContentDocumentId,OwnerId FROM ContentVersion WHERE Id = '" + contentVersionId + "'";
        QueryResult queryResult = connection.query(soql);
        Map<String, String> result = new HashMap<>();
        
        if (queryResult.getRecords() != null && queryResult.getRecords().length > 0) {
            SObject record = queryResult.getRecords()[0];
            result.put("ContentDocumentId", (String) record.getField("ContentDocumentId"));
            result.put("OwnerId", (String) record.getField("OwnerId"));
            return result;
        }
        return null;
    }

    @Override
    public void getNewIdByField(String api, String queryFields) throws Exception {
        List<String> stringList = DataUtil.toIdList(queryFields);
        String maxId = null;
        Field[] dsrFields;
        List<String> fields = Lists.newArrayList();
        PartnerConnection connect = salesforceTargetConnect.createConnect();

        DescribeSObjectResult dsr = connect.describeSObject(api);
        dsrFields = dsr.getFields();
        for (Field field : dsrFields) {
            // 不查询文件
            if ("base64".equalsIgnoreCase(field.getType().toString())) {
                continue;
            }
            fields.add(field.getName());
        }
        // 添加特定对象的关联字段
        addRelationFields(api, fields);

        String joined = StringUtils.join(fields, ",");
        com.alibaba.fastjson2.JSONArray objects;
        int count = 0;

        do {
            String sql = null;
            if (StringUtils.isNotEmpty(maxId)){
                sql = "select " + joined + " from "+ api + " Where Id > '"+ maxId+"' LIMIT 200";
            }else {
                sql = "select " + joined + " from "+ api +" LIMIT 200";
            }
            log.info("执行SQL查询: {}", sql);
            QueryResult queryResult = connect.queryAll(sql);
            if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
                log.info("查询结果为空，结束查询");
                break;
            }

            SObject[] records = queryResult.getRecords();
            objects = DataUtil.toJsonArray(records, dsrFields);
            maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);

            log.info("处理批次数据，记录数: {}", objects.size());
            // 批量处理记录
            processRecords(api, objects, stringList);

            int totalRecords = records.length;

            count += totalRecords;
            log.info("已处理记录总数: {}", count);

            if (totalRecords < 200) {
                log.info("无更多数据，反写NewId任务结束！！！");
                return;
            }

            TimeUnit.MILLISECONDS.sleep(1);
        } while (true);
    }

    private void addRelationFields(String api, List<String> fields) {
        if ("TaskRelation".equals(api) || "EventRelation".equals(api)) {
            fields.add("Relation.type");
        }
        if ("CollaborationGroupMember".equals(api)) {
            fields.add("Member.type");
        }
        if ("FeedAttachment".equals(api)) {
            fields.add("FeedEntity.type");
            fields.add("Record.type");
        }
    }

    private String buildQuerySql(String api, String joined, String maxId) {
        StringBuilder sqlBuilder = new StringBuilder("SELECT ");
        if (StringUtils.isNotEmpty(joined)) {
            sqlBuilder.append(joined).append(", ");
        }
        sqlBuilder.append("Id FROM ").append(api);
        if (StringUtils.isNotEmpty(maxId)) {
            sqlBuilder.append(" WHERE Id > '").append(maxId).append("'");
        }
        sqlBuilder.append(" LIMIT 200");
        return sqlBuilder.toString();
    }

    private void processRecords(String api, com.alibaba.fastjson2.JSONArray objects, List<String> stringList) throws Exception {
        for (int i = 0; i < objects.size(); i++) {
            JSONObject jsonObject = objects.getJSONObject(i);
            String id = jsonObject.getString(Const.ID);
            List<Map<String, Object>> maps = buildUpdateMaps(api, jsonObject, stringList);

            if (maps!=null && !maps.isEmpty()) {
                log.info("准备更新记录，API: {}, ID: {}, 匹配字段信息: {}", api, id, maps);
                customMapper.updateNewIdByFields(api, maps, id);
            }else if (maps == null){
                log.info("{}无需更新记录，数据ID: {}", api, id);
            }
        }
    }

    private List<Map<String, Object>> buildUpdateMaps(String api, JSONObject jsonObject, List<String> stringList) throws Exception {
        List<Map<String, Object>> maps = Lists.newArrayList();

        for (String key : stringList) {
            Map<String, Object> paramMap = null;

            switch (api) {
                case "AccountContactRelation":
                    paramMap = handleAccountContactRelation(key, jsonObject);
                    if (paramMap == null){
                        return null;
                    }
                    break;
                case "TaskRelation":
                case "EventRelation":
                    paramMap = handleTaskEventRelation(api, key, jsonObject);
                    if (paramMap == null){
                        log.info("{}处理失败，字段: {}, 数据ID: {}", api, key, jsonObject.getString("Id"));
                        return null;
                    }
                    break;
                case "CollaborationGroupMember":
                    paramMap = handleCollaborationGroupMember(key, jsonObject);
                    if (paramMap == null){
                        log.info("CollaborationGroupMember处理失败，字段: {}, 数据ID: {}", key, jsonObject.getString("Id"));
                        return null;
                    }
                    break;
                case "FeedAttachment":
                    paramMap = handleFeedAttachment(key, jsonObject);
                    if (paramMap == null){
                        log.info("FeedAttachment处理失败，字段: {}, 数据ID: {}", key, jsonObject.getString("Id"));
                        return null;
                    }
                    break;
                default:
                    paramMap = Maps.newHashMap();
                    paramMap.put("key", key);
                    paramMap.put("value", jsonObject.getString(key));
                    log.info("普通字段处理，API: {}, 字段: {}, 值: {}", api, key, jsonObject.getString(key));
                    break;
            }

            if (!paramMap.isEmpty()) {
                maps.add(paramMap);
            }
        }

        return maps;
    }

    private Map<String, Object> handleAccountContactRelation(String key, JSONObject jsonObject) throws Exception {
        switch (key) {
            case "ContactId":
                Map<String, Object> contactMap = getNewIdMap(key, "Contact", jsonObject.getString(key));
                if (contactMap == null){
                    log.info("AccountContactRelation，字段: {},关联对象段: {}, ContactId: {} 未找到对应的new_id", key,"Contact", jsonObject.getString(key));
                    return null;
                }
                return contactMap;
            case "AccountId":
                Map<String, Object> accountMap = getNewIdMap(key, "Account", jsonObject.getString(key));
                if (accountMap == null){
                    log.info("AccountContactRelation，字段: {},关联对象段: {}, AccountId: {} 未找到对应的new_id", key,"Account", jsonObject.getString(key));
                    return null;
                }
                return accountMap;
            default:
                return null;
        }
    }

    private Map<String, Object> handleTaskEventRelation(String api, String key, JSONObject jsonObject) throws Exception {
        switch (key) {
            case "TaskId":
                Map<String, Object> taskObjectMap = getNewIdMap(key, "Task", jsonObject.getString(key));
                if (taskObjectMap == null){
                    log.info("TaskEventRelation，字段: {},关联对象段: {}, TaskId: {} 未找到对应的new_id", key,"Task", jsonObject.getString(key));
                    return null;
                }
                return taskObjectMap;
            case "EventId":
                Map<String, Object> eventObjectMap = getNewIdMap(key, "Event", jsonObject.getString(key));
                if (eventObjectMap == null){
                    log.info("TaskEventRelation，字段: {},关联对象段: {}, EventId: {} 未找到对应的new_id", key,"Event", jsonObject.getString(key));
                    return null;
                }
                return eventObjectMap;
            case "RelationId":
                Map<String, Object> relationObjectMap = getNewIdMap(key, jsonObject.getString("Relation_Type"), jsonObject.getString(key));
                if (relationObjectMap == null){
                    log.info("TaskEventRelation，字段: {},关联对象段: {}, RelationId: {} 未找到对应的new_id", key,jsonObject.getString("Relation_Type"), jsonObject.getString(key));
                    return null;
                }
                return relationObjectMap;
            default:
                return null;
        }
    }

    private Map<String, Object> handleCollaborationGroupMember(String key, JSONObject jsonObject) throws Exception {
        switch (key) {
            case "CollaborationGroupId":
                Map<String, Object> groupObjectMap = getNewIdMap(key, "CollaborationGroup", jsonObject.getString(key));
                if (groupObjectMap == null){
                    log.info("CollaborationGroupMember，字段: {},关联对象段: {}, CollaborationGroupId: {} 未找到对应的new_id", key,"CollaborationGroup", jsonObject.getString(key));
                    return null;
                }
                return groupObjectMap;
            case "MemberId":
                Map<String, Object> memberObjectMap = getNewIdMap(key, jsonObject.getString("Member_Type"), jsonObject.getString(key));
                if (memberObjectMap == null){
                    log.info("CollaborationGroupMember，字段: {},关联对象段: {}, MemberId: {} 未找到对应的new_id", key,jsonObject.getString("Member_Type"), jsonObject.getString(key));
                    return null;
                }
                return memberObjectMap;
            default:
                return null;
        }
    }

    private Map<String, Object> handleFeedAttachment(String key, JSONObject jsonObject) throws Exception {
        switch (key) {
            case "FeedEntityId":
                Map<String, Object> feedEntityObjectMap = getNewIdMap(key, jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
                if (feedEntityObjectMap == null){
                    log.info("FeedAttachment，字段: {},关联对象段: {}, FeedEntityId: {} 未找到对应的new_id", key,jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
                    return null;
                }
                return feedEntityObjectMap;
            case "RecordId":
                Map<String, Object> recordObjectMap = getNewIdMap(key, jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
                if (recordObjectMap == null){
                    log.info("FeedAttachment，字段: {},关联对象段: {}, RecordId: {} 未找到对应的new_id", key,jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
                    return null;
                }
                return recordObjectMap;
            default:
                return null;
        }
    }

    private Map<String, Object> getNewIdMap(String key, String tableName, String id) throws Exception {
        Map<String, Object> m = customMapper.getByNewId("Id", tableName, id);
        if (m == null || m.get("Id") == null) {
            return null;
        }
        Map<String, Object> paramMap = Maps.newHashMap();
        paramMap.put("key", key);
        paramMap.put("value", m.get("Id"));
        return paramMap;
    }

    @Override
    public void getTaskNewId(SalesforceParam param) throws Exception {
        List<Map<String, Object>> mapsList = customMapper.list("*", "Task", "where What_Type = EmailMessage and new_id is null");

    }


    /**
     * 组装【单表】 一次性Insert 参数
     */
    public ReturnT<String> insertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
        List<String> apis = DataUtil.toIdList(param.getApi());

        String beginDateStr = null;
        String endDateStr = null;
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            Date beginDate = param.getBeginCreateDate();
            Date endDate = param.getEndCreateDate();
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        // 全量的时候 检测是否有自动任务锁住的表
        boolean isFull = CollectionUtils.isEmpty(param.getIds());
        if (isFull) {
            QueryWrapper<DataObject> qw = new QueryWrapper<>();
            qw.eq("data_lock", 1).in("name", apis);
            List<DataObject> list = dataObjectService.list(qw);
            if (CollectionUtils.isNotEmpty(list)) {
                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
                String message = "api:" + apiNames + " is locked";
                String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s",  apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
                EmailUtil.send("DataDump ERROR", format);
                return new ReturnT<>(500, message);
            }
        }

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
        for (String api : apis) {
            DataObject update = dataObjectService.getById(api);
            if (update == null) {
                log.info("对象 {} 不存在，无法执行一次性插入操作", api);
                continue;
            }
            try {
                if (!dataFieldService.hasCreatedDate(api)){
                    insertSingleShareData(api,partnerConnection,update);
                    continue;
                }
                List<SalesforceParam> salesforceParams = null;
                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                dbQw.eq("name", api);
                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
                    dbQw.eq("sync_start_date", beginDateStr);  // 等于开始时间
                    dbQw.eq("sync_end_date", endDateStr);    // 等于结束时间
                }
                List<DataBatch> list = dataBatchService.list(dbQw);
                AtomicInteger batch = new AtomicInteger(1);
                if (CollectionUtils.isNotEmpty(list)) {
                    salesforceParams = list.stream().map(t -> {
                        SalesforceParam salesforceParam = param.clone();
                        salesforceParam.setApi(t.getName());
                        salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                        salesforceParam.setEndCreateDate(t.getSyncEndDate());
                        salesforceParam.setBatch(batch.getAndIncrement());
                        return salesforceParam;
                    }).collect(Collectors.toList());
                }

                // 手动任务优先执行
                for (SalesforceParam salesforceParam : salesforceParams) {
                    if (salesforceParam.getIsSingleThread()){
                        insertSingleExecute(salesforceParam, partnerConnection,update);
                    }else {
                        Future<?> future = salesforceExecutor.execute(() -> {
                            try {
                                insertSingleExecute(salesforceParam, partnerConnection,update);
                            } catch (Throwable throwable) {
                                log.error("salesforceExecutor error", throwable);
                                throw new RuntimeException(throwable);
                            }
                        }, salesforceParam.getBatch(), 1);
                        futures.add(future);
                    }
                }
                // 等待当前所有线程执行完成
                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                update.setNeedUpdate(false);
            } catch (Exception e) {
                throw e;
            } finally {
                if (isFull) {
                    update.setName(api);
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
        }
        return ReturnT.SUCCESS;
    }


    /**
     * 组装【多】一次性Insert 参数
     */
    public  ReturnT<String> autoInsertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
        QueryWrapper<DataObject> qw = new QueryWrapper<>();
        qw.eq("data_work", 1)
                .eq("need_update", 1)
                .eq("data_lock", 0)
                .orderByAsc("data_index")
                .last(" limit 10");

        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
        while (true) {
            List<DataObject> dataObjects = dataObjectService.list(qw);
            if (CollectionUtils.isEmpty(dataObjects)) {
                break;
            }
            for (DataObject dataObject : dataObjects) {
                DataObject update = new DataObject();
                log.info("insertSingle api: {}", dataObject.getName());
                TimeUnit.MILLISECONDS.sleep(1);
                try {
                    String api = dataObject.getName();
                    update.setName(dataObject.getName());
                    update.setDataLock(1);
                    dataObjectService.updateById(update);

                    if (!dataFieldService.hasCreatedDate(api)){
                        insertSingleShareData(api,partnerConnection,dataObject);
                        update.setNeedUpdate(false);
                        dataObjectService.updateById(update);
                        continue;
                    }

                    List<SalesforceParam> salesforceParams = null;
                    QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                    dbQw.eq("name", api);
                    List<DataBatch> list = dataBatchService.list(dbQw);
                    AtomicInteger batch = new AtomicInteger(1);
                    if (CollectionUtils.isNotEmpty(list)) {
                        salesforceParams = list.stream().map(t -> {
                            SalesforceParam salesforceParam = param.clone();
                            salesforceParam.setApi(t.getName());
                            salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                            salesforceParam.setEndCreateDate(t.getSyncEndDate());
                            salesforceParam.setBatch(batch.getAndIncrement());
                            return salesforceParam;
                        }).collect(Collectors.toList());
                    }

                    // 手动任务优先执行
                    for (SalesforceParam salesforceParam : salesforceParams) {
                        if (salesforceParam.getIsSingleThread()){
                            insertSingleExecute(salesforceParam, partnerConnection,update);
                        }else {
                            Future<?> future = salesforceExecutor.execute(() -> {
                                try {
                                    insertSingleExecute(salesforceParam, partnerConnection,dataObject);
                                } catch (Throwable throwable) {
                                    log.error("salesforceExecutor error", throwable);
                                    throw new RuntimeException(throwable);
                                }
                            }, salesforceParam.getBatch(), 0);
                            futures.add(future);
                        }
                    }
                    // 等待当前所有线程执行完成
                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                    update.setNeedUpdate(false);
                } catch (InterruptedException e) {
                    throw e;
                } catch (Throwable e) {
                    throw new RuntimeException(e);
                } finally {
                    update.setDataLock(0);
                    dataObjectService.updateById(update);
                }
            }
            // 等待当前所有线程执行完成
            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            futures.clear();
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 执行一次性Insert数据
     */
    private void insertSingleExecute(SalesforceParam param, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);

        String api = param.getApi();
        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);
        List<DataField> list = dataFieldService.list(dbQw);

        String sql = "";
        String sql2 = "";

        String beginDateStr = null;
        String endDateStr = null;
        Date beginDate = param.getBeginCreateDate();
        Date endDate = param.getEndCreateDate();
        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
        }

        if (1 == param.getType()) {
            if (api.endsWith("Share")){
                sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
                sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000";
            }else {
                sql = "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
                sql2 = "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000";
            }
        }else {
            String updateDateField = dataFieldService.returnUpdateDateField(api);
            if (api.endsWith("Share")){
                sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' ";
                sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000";
            }else {
                sql = "where new_id is null and "+updateDateField+" >= '" + beginDateStr + "' ";
                sql2 = "new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000";
            }
        }
        //表内数据总量
        Integer count = customMapper.countBySQL(api, sql);

        if(count == 0){
            return;
        }

        //查询当前对象多态字段映射
        QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",1);
        List<LinkConfig> configs = linkConfigService.list(queryWrapper);
        Map<String, String> fieldMap = new HashMap<>();
        if (!configs.isEmpty())  {
            fieldMap = configs.stream()
                    .collect(Collectors.toMap(
                            LinkConfig::getField,      // Key提取器
                            LinkConfig::getLinkField,  // Value提取器
                            (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
                    ));
        }

        int targetCount = 0;

        int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;

        log.info("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,num, beginDateStr, endDateStr, api);

        for (int z = 0; z < num; z++) {

            List<Map<String, Object>> mapsList = customMapper.list("*", api,  sql2);

            log.info("总Insert数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,z, beginDateStr, endDateStr, api);

            int size = mapsList.size();

            //批量插入200一次
            int page = size%200 == 0 ? size/200 : (size/200) + 1;

            for (int i = 0; i < page; i++) {

                DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");
                int startIndex = 200 * i;
                int endIndex = Math.min(startIndex + 200, size);
                if (startIndex >= endIndex) {
                    break;
                }
                List<Map<String, Object>> data = mapsList.subList(startIndex, endIndex);

                int sized = data.size();
                SObject[] accounts = new SObject[sized];
                String[] ids = new String[sized];
                try {
                    for (int j = 1; j <= sized; j++) {
                        SObject account = new SObject();
                        account.setType(api);
                        String message = null;
                        //给对象赋值
                        for (DataField dataField : list) {
                            String field = dataField.getField();
                            String reference_to = dataField.getReferenceTo();

                            String value = String.valueOf(data.get(j - 1).get(field));
                            //根据旧sfid查找引用对象新sfid
                            if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) {
                                continue;
                            } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {

                                if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
                                    //判断reference_to内是否包含User字符串
                                    if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
                                        reference_to = "User";
                                    }
                                    //引用类型字段
                                    String linkfield = fieldMap.get(dataField.getField());
                                    if (StringUtils.isNotBlank(linkfield) ){
                                        if (data.get(j-1).get(linkfield)!=null){
                                            reference_to = data.get(j-1).get(linkfield).toString();
                                        }else {
                                            log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！");
                                        }
                                    }
                                    if (reference_to == null){
                                        continue;
                                    }
                                    Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
                                    if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
                                        account.setField(field, m.get("new_id"));
                                    }else {
                                        message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"异常！";
                                        log.info(message);
                                        break;
                                    }
                                }
                            } else {
                                if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                    account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
                                }else {
                                    account.setField(field, data.get(j - 1).get(field));
                                }
                            }
                        }
                        if (message!=null){
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", message);
                            maps.add(linkMap1);
                            customMapper.updateById(api, maps, data.get(j-1).get("Id").toString());
                            continue;
                        }
                        if (dataObject.getIsEditable()){
                            account.setField("old_owner_id__c", data.get(j - 1).get("OwnerId"));
                            account.setField("old_sfdc_id__c", data.get(j - 1).get("Id"));
                        }

                        ids[j-1] = data.get(j-1).get("Id").toString();
                        accounts[j-1] = account;
                        if (i*200+j == size){
                            break;
                        }
                    }
                    dataLog.setEndTime(new Date());
                    dataLogService.save(dataLog);

                    if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                        printlnAccountsDetails(accounts, list);
                    }

                    DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
                    SaveResult[] saveResults = partnerConnection.create(accounts);
                    dataLog1.setEndTime(new Date());
                    dataLogService.save(dataLog1);

                    DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");

                    int index = 0;
                    for (SaveResult saveResult : saveResults){
                        if (saveResult.getSuccess()) {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> m = new HashMap<>();
                            m.put("key", "new_id");
                            m.put("value", saveResult.getId());
                            maps.add(m);
                            customMapper.updateById(api, maps, ids[index]);
                            targetCount ++;
                        }else{
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", JSON.toJSONString(saveResult.getErrors()));
                            maps.add(linkMap1);
                            customMapper.updateById(api, maps, ids[index]);
                            log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult));
                        }
                        index++;
                    }
                    dataLog2.setEndTime(new Date());
                    dataLogService.save(dataLog2);

                    TimeUnit.MILLISECONDS.sleep(1);
                }catch (InterruptedException e){
                    return;
                } catch (Exception e) {
                    log.error("insertSingle error api:{}", api,e);
                }
            }
            if (z*10000+size >= count){
                break;
            }
        }

        SalesforceParam countParam = new SalesforceParam();
        countParam.setApi(api);
        countParam.setBeginCreateDate(beginDate);
        countParam.setEndCreateDate(DateUtils.addSeconds(endDate, -1));
        // 存在isDeleted 只查询IsDeleted为false的
        if (dataFieldService.hasDeleted(countParam.getApi())) {
            countParam.setIsDeleted(false);
        } else {
            // 不存在 过滤
            countParam.setIsDeleted(null);
        }
        Integer sfNum = 0;
        if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
            sfNum = commonService.countSfNum(partnerConnection, countParam);
        }

        UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
        updateQw.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
                .set("target_sf_num", count);
        dataBatchHistoryService.update(updateQw);

        UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
        updateQw2.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", endDate)
                .set("sf_add_num", sfNum);
        dataBatchService.update(updateQw2);

        UpdateWrapper<DataBatchHistory> updateQw3 = new UpdateWrapper<>();
        updateQw3.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
                .set("target_update_num", count);
        dataBatchHistoryService.update(updateQw3);

        UpdateWrapper<DataBatch> updateQw4 = new UpdateWrapper<>();
        updateQw4.eq("name", api)
                .eq("sync_start_date", beginDate)
                .eq("sync_end_date", endDate)
                .set("sf_update_num", targetCount);
        dataBatchService.update(updateQw4);
    }

    /**
     * 执行一次性Insert Share数据
     */
    private void insertSingleShareData(String api, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);

        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
        dbQw.eq("api", api);
        List<DataField> list = dataFieldService.list(dbQw);
        TimeUnit.MILLISECONDS.sleep(1);

        String sql = "";
        String sql2 = "";

        if (api.endsWith("Share")){
            sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null ";
            sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 10000";
        }else {
            sql = "where new_id is null ";
            sql2 = "new_id is null limit 10000";
        }

        //表内数据总量
        Integer count = customMapper.countBySQL(api, sql);

        if (count == 0) {
            return;
        }

        //查询当前对象多态字段映射
        QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
        List<LinkConfig> configs = linkConfigService.list(queryWrapper);
        Map<String, String> fieldMap = new HashMap<>();
        if (!configs.isEmpty())  {
            fieldMap = configs.stream()
                    .collect(Collectors.toMap(
                            LinkConfig::getField,      // Key提取器
                            LinkConfig::getLinkField,  // Value提取器
                            (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
                    ));
        }

        int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;

        log.info("总Insert数据 count:{}；总批次：{}; -api:{}；", count,num, api);

        for (int z = 0; z < num; z++) {

            List<Map<String, Object>> mapsList = customMapper.list("*", api, sql2);

            int size = mapsList.size();
            //批量插入200一次
            int page = size%200 == 0 ? size/200 : (size/200) + 1;

            log.info("执行api:{}, 执行批次:{}, 执行数据量:{}", api, z, size);

            for (int i = 0; i < page; i++) {
                try {
                    DataLog dataLog = new DataLog(api, null, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");

                    int startIndex = 200 * i;
                    int endIndex = Math.min(startIndex + 200, size);
                    if (startIndex >= endIndex) {
                        break;
                    }
                    List<Map<String, Object>> data = mapsList.subList(startIndex, endIndex);
                    int sized = data.size();

                    SObject[] accounts = new SObject[sized];
                    String[] ids = new String[sized];
                    for (int j = 1; j <= sized; j++) {
                        SObject account = new SObject();
                        account.setType(api);
                        String message = null;
                        //给对象赋值
                        for (DataField dataField : list) {
                            String field = dataField.getField();
                            String reference_to = dataField.getReferenceTo();

                            String value = String.valueOf(data.get(j - 1).get(field));
                            //根据旧sfid查找引用对象新sfid
                            if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) {
                                continue;
                            } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {

                                if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
                                    //判断reference_to内是否包含User字符串
                                    if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
                                        reference_to = "User";
                                    }
                                    //引用类型字段
                                    String linkfield = fieldMap.get(dataField.getField());
                                    if (StringUtils.isNotBlank(linkfield)){
                                        if (data.get(j-1).get(linkfield)!=null){
                                            reference_to = data.get(j-1).get(linkfield).toString();
                                        }else {
                                            log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！");
                                        }
                                    }
                                    if (reference_to == null){
                                        continue;
                                    }
                                    Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
                                    if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
                                        account.setField(field, m.get("new_id"));
                                    }else {
                                        message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"异常！";
                                        log.info(message);
                                        break;
                                    }
                                }
                            } else {
                                if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                    account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
                                }else {
                                    account.setField(field, data.get(j - 1).get(field));
                                }
                            }
                        }
                        if (message != null){
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> linkMap1 = new HashMap<>();
                            linkMap1.put("key", "error_message");
                            linkMap1.put("value", message);
                            maps.add(linkMap1);
                            customMapper.updateById(api, maps, data.get(j-1).get("Id").toString());
                            continue;
                        }
                        if (dataObject.getIsEditable()){
                            account.setField("old_owner_id__c", data.get(j - 1).get("OwnerId"));
                            account.setField("old_sfdc_id__c", data.get(j - 1).get("Id"));
                        }

                        ids[j-1] = data.get(j-1).get("Id").toString();
                        accounts[j-1] = account;
                        if (i*200+j == size){
                            break;
                        }
                    }
                    dataLog.setEndTime(new Date());
                    dataLogService.save(dataLog);

                    if (infoFlag != null && "1".equals(infoFlag.get("value"))){
                        printlnAccountsDetails(accounts, list);
                    }
                    DataLog dataLog1 = new DataLog(api, null, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
                    SaveResult[] saveResults = partnerConnection.create(accounts);
                    dataLog1.setEndTime(new Date());
                    dataLogService.save(dataLog1);

                    DataLog dataLog2 = new DataLog(api, null, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");

                    int index = 0;
                    for (SaveResult saveResult : saveResults){
                        if (saveResult.getSuccess()) {
                            List<Map<String, Object>> maps = new ArrayList<>();
                            Map<String, Object> m = new HashMap<>();
                            m.put("key", "new_id");
                            m.put("value", saveResult.getId());
                            maps.add(m);
                            customMapper.updateById(api, maps, ids[index]);
                            log.info("Created Success row with id " + saveResult.getId());
                        }else{
                            log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult));
                        }
                        index++;
                    }
                    dataLog2.setEndTime(new Date());
                    dataLogService.save(dataLog2);

                    TimeUnit.MILLISECONDS.sleep(1);
                }catch (InterruptedException e){
                    return;
                }catch (Exception e) {
                    log.error("insertSingle error api:{}", api, e);
                }
            }
            if (z*10000+size >= count){
                break;
            }
        }
    }
}
