【fix】 修复文件下载路径问题

(cherry picked from commit bde253f25d940ed2167d002e30ae648b57882ebe)

src/main/java/com/celnet/datadump/job/DataDumpNewJob.java

@@ -114,6 +114,27 @@ public class DataDumpNewJob {
         return commonBatchService.dumpBatch(param);
     }
 
+    /**
+     * 增量任务（大数据量）
+     *
+     * @param paramStr 参数json
+     * @return result
+     */
+    @XxlJob("dataDumpIncrementBatchJob")
+    public ReturnT<String> dataDumpIncrementBatchJob(String paramStr) throws Exception {
+        log.info("dataDumpIncrementBatchJob execute start ..................");
+        SalesforceParam param = new SalesforceParam();
+        try {
+            if (StringUtils.isNotBlank(paramStr)) {
+                param = JSON.parseObject(paramStr, SalesforceParam.class);
+            }
+        } catch (Throwable throwable) {
+            return new ReturnT<>(500, "参数解析失败！");
+        }
+
+        return commonBatchService.incrementBatch(param);
+    }
+
     /**
      * 存量任务（BigObject）
      *

src/main/java/com/celnet/datadump/service/CommonBatchService.java

@@ -17,4 +17,6 @@ public interface CommonBatchService {
     
     ReturnT<String> dumpBigObjectTargetV1(SalesforceParam param) throws Exception;
 
+    ReturnT<String> incrementBatch(SalesforceParam param) throws Exception;
+
 }

src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java

@@ -2,6 +2,7 @@ package com.celnet.datadump.service;
 
 import com.celnet.datadump.entity.DataObject;
 import com.celnet.datadump.param.DataDumpSpecialParam;
+import com.sforce.async.BulkConnection;
 import com.sforce.soap.partner.PartnerConnection;
 import com.xxl.job.core.biz.model.ReturnT;
 

src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java

@@ -13,6 +13,7 @@ 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.global.TypeCode;
 import com.celnet.datadump.mapper.CustomMapper;
 import com.celnet.datadump.param.DataDumpParam;
 import com.celnet.datadump.param.DataDumpSpecialParam;
@@ -95,7 +96,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
     private DataDumpSpecialService dataDumpSpecialService;
     @Autowired
     private DataBatchHistoryService dataBatchHistoryService;
-    
+    @Autowired
+    private DataReportService dataReportService;
     // 批次处理记录数，默认为10000
     private int batchProcessCount = 10000;
 
@@ -132,6 +134,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 if (result != null) {
                     return result;
                 }
+            }else {
+                autoBatchDump(param, futures);
             }
             return ReturnT.SUCCESS;
         } catch (Throwable throwable) {
@@ -141,6 +145,67 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         }
     }
 
+    @Override
+    public ReturnT<String> incrementBatch(SalesforceParam param) throws Exception {
+        QueryWrapper<DataObject> qw = new QueryWrapper<>();
+        if (StringUtils.isNotBlank(param.getApi())) {
+            List<String> apis = DataUtil.toIdList(param.getApi());
+            qw.in("name", apis);
+        }
+        qw.eq("need_update", true)
+                .isNotNull("last_update_date");
+        List<DataObject> list = dataObjectService.list(qw);
+        if (CollectionUtils.isEmpty(list)) {
+            return new ReturnT<>(500, ("表" + param.getApi() + "不存在或未开启更新"));
+        }
+        List<Future<?>> futures = Lists.newArrayList();
+        try {
+            DataReport dataReport = new DataReport();
+            dataReport.setType(TypeCode.INCREMENT);
+            dataReport.setApis(list.stream().map(DataObject::getName).collect(Collectors.joining(",")));
+            dataReportService.save(dataReport);
+            for (DataObject dataObject : list) {
+                //无创建时间对象
+                if (!dataFieldService.hasCreatedDate(dataObject.getName())) {
+                    DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam();
+                    dataDumpSpecialParam.setApi(dataObject.getName());
+                    Future<?> future = getDataBatch(dataDumpSpecialParam, salesforceConnect.createBulkConnect(),dataObject);
+                    // 等待当前所有线程执行完成
+                    salesforceExecutor.waitForFutures(future);
+                    continue;
+                }
+                Future<?> future = salesforceExecutor.execute(() -> {
+                    try {
+                        Date updateTime = new Date();
+                        SalesforceParam salesforceParam = new SalesforceParam();
+                        salesforceParam.setApi(dataObject.getName());
+                        salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate());
+                        salesforceParam.setType(2);
+                        // 更新字段值不为空 按更新字段里的字段校验
+                        if (StringUtils.isNotBlank(dataObject.getUpdateField())) {
+                            salesforceParam.setUpdateField(dataObject.getUpdateField());
+                        }
+                        dumpIncrementDataBatch(salesforceParam);
+                        dataObject.setLastUpdateDate(updateTime);
+                        dataObject.setNeedUpdate( false);
+                        dataObjectService.updateById(dataObject);
+
+                    } catch (Throwable throwable) {
+                        log.error("salesforceExecutor error", throwable);
+                        throw new RuntimeException(throwable);
+                    }
+                }, 0, 0);
+                futures.add(future);
+            }
+            // 等待当前所有线程执行完成
+            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
+            return ReturnT.SUCCESS;
+        } catch (Throwable throwable) {
+            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
+            throw throwable;
+        }
+    }
+
     @Override
     public ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception {
         try {
@@ -949,6 +1014,87 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         return ReturnT.SUCCESS;
     }
 
+    private void autoBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
+        QueryWrapper<DataObject> qw = new QueryWrapper<>();
+        qw.eq("data_work", 1)
+                .eq("data_lock", 0)
+                .orderByAsc("data_index")
+                .last(" limit 10");
+        while (true) {
+            List<DataObject> dataObjects = dataObjectService.list(qw);
+            if (CollectionUtils.isEmpty(dataObjects)) {
+                break;
+            }
+
+            // 根据参数获取sql
+            for (DataObject update : dataObjects) {
+                String api = update.getName();
+                TimeUnit.MILLISECONDS.sleep(1);
+                try {
+                    commonService.checkApi(api, true);
+                    if (!dataFieldService.hasCreatedDate(api)) {
+                        DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam();
+                        dataDumpSpecialParam.setApi(api);
+                        Future<?> future = getData(dataDumpSpecialParam, salesforceConnect.createBulkConnect());
+                        // 等待当前所有线程执行完成
+                        salesforceExecutor.waitForFutures(future);
+                        continue;
+                    }
+                    param.setApi(api);
+                    List<SalesforceParam> salesforceParams = null;
+                    update.setName(api);
+                    update.setDataLock(1);
+                    dataObjectService.updateById(update);
+                    QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
+                    dbQw.eq("name", api)
+                            .isNull("first_sync_date");
+                    if (param.getBeginDate() != null && param.getEndDate() != null) {
+                        dbQw.eq("sync_start_date", param.getBeginDate());
+                        dbQw.eq("sync_end_date", param.getEndDate());
+                    }
+                    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());
+                    } else {
+                        salesforceParams = DataUtil.splitTask(param);
+                    }
+                    // 手动任务优先执行
+                    for (SalesforceParam salesforceParam : salesforceParams) {
+                        Future<?> future = salesforceExecutor.execute(() -> {
+                            try {
+                                dumpBatchData(salesforceParam);
+                            } 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 (Throwable e) {
+                    log.error("manualDump error", e);
+                    throw new RuntimeException(e);
+                } finally {
+                    update.setName(api);
+                    update.setDataLock(0);
+                    dataObjectService.updateById(update);
+                }
+            }
+        }
+    }
+
     private BulkConnection dumpBatchData(SalesforceParam param) throws Throwable {
         String api = param.getApi();
         BulkConnection bulkConnect = null;
@@ -994,6 +1140,167 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         return bulkConnect;
     }
 
+    private BulkConnection dumpIncrementDataBatch(SalesforceParam param) throws Throwable {
+        String api = param.getApi();
+        BulkConnection bulkConnect = null;
+        try {
+            DataBatchHistory dataBatchHistory = new DataBatchHistory();
+            dataBatchHistory.setName(api);
+            dataBatchHistory.setStartDate(new Date());
+            dataBatchHistory.setSyncStartDate(param.getBeginCreateDate());
+            if (param.getEndCreateDate() != null) {
+                dataBatchHistory.setSyncEndDate(DateUtils.addSeconds(param.getEndCreateDate(), -1));
+            }
+            dataBatchHistory.setBatch(param.getBatch());
+            if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
+                log.info("NO.{} dump {}, date: {} ~ {} start", param.getBatch(),
+                        param.getApi(),
+                        DateFormatUtils.format(param.getBeginCreateDate(), "yyyy-MM-dd HH:mm:ss"),
+                        DateFormatUtils.format(param.getEndCreateDate(), "yyyy-MM-dd HH:mm:ss"));
+            }
+            bulkConnect = salesforceConnect.createBulkConnect();
+            // 存在isDeleted 只查询IsDeleted为false的
+            if (dataFieldService.hasDeleted(param.getApi())) {
+                param.setIsDeleted(false);
+            } else {
+                // 不存在 过滤
+                param.setIsDeleted(null);
+            }
+
+            int sfData = getAllBulkV1SfIncrementData(param, bulkConnect);
+
+            dataBatchHistory.setDbNum(sfData);
+            dataBatchHistory.setSfNum(sfData);
+
+            updateDataBatch(param, dataBatchHistory);
+        }catch (InterruptedException e){
+            return bulkConnect;
+        } catch (Throwable throwable) {
+            log.error("dataDumpJob error api:{}", api, throwable);
+            String type = param.getType() == 1 ? "存量" : "增量";
+            String format = String.format("%s数据迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", type, api, JSON.toJSONString(param, DataDumpParam.getFilter()), throwable);
+            EmailUtil.send("DataDump ERROR", format);
+            throw throwable;
+        }
+        return bulkConnect;
+    }
+
+    public Future<?> getDataBatch(DataDumpSpecialParam param, BulkConnection bulkConnect,DataObject dataObject) {
+        String api = param.getApi();
+        return salesforceExecutor.execute(() -> {
+            try {
+                // 检测表 不存在就生成 但不生成批次
+                commonService.checkApi(api, true);
+                int count = 0;
+                boolean hasMore = true;
+                QueryWrapper<DataField> wrapper = new QueryWrapper<>();
+                wrapper.eq("api", api);
+                List<DataField> list = dataFieldService.list(wrapper);
+
+                Map<String, Object> map = Maps.newHashMap();
+                SalesforceParam salesforceParam = new SalesforceParam();
+                salesforceParam.setApi(api);
+                salesforceParam.setIdField(param.getField());
+                String maxId = null;
+
+                List<String> fields = Lists.newArrayList();
+                for (DataField field : list) {
+                    if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){
+                        fields.add(field.getField());
+                    }
+                }
+                salesforceParam.setSelect(StringUtils.join(fields, ","));
+
+                while (hasMore) {
+                    int sfData = 0;
+                    salesforceParam.setMaxId(maxId);
+                    salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate());
+                    salesforceParam.setUpdateField(dataFieldService.returnUpdateDateField(dataObject.getName()));
+                    salesforceParam.setLimit(batchProcessCount);
+                    map.put("param", salesforceParam);
+                    JobInfo job = null;
+                    try {
+                        String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdShare", map);
+
+                        log.info("query sql: {}", sql);
+                        XxlJobLogger.log("query sql: {}", sql);
+
+                        job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
+                        log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
+
+                        BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8)));
+                        log.info("创建批次成功, 批次ID: {}", batchInfo.getId());
+
+                        int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
+                        log.info("批次处理完成, 处理记录数: {}", completionOne);
+
+                        if (completionOne == 0){
+                            log.info("无更多数据, 结束处理");
+                            BulkUtil.closeJob(bulkConnect, job.getId());
+                            break;
+                        }
+
+                        List<Map<String, Object>> batchRecords = new ArrayList<>();
+
+                        QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
+                        log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
+
+                        for (final String resultId : queryResultList.getResult()) {
+                            InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
+
+                            CSVReader rdr = new CSVReader(resultStream);
+                            List<String> headers = rdr.nextRecord();
+                            log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size());
+
+                            // 批量处理记录
+                            List<String> record;
+                            while ((record = rdr.nextRecord()) != null) {
+                                Map<String, Object> recordMap = new HashMap<>();
+                                for (int i = 0; i < headers.size() && i < record.size(); i++) {
+                                    recordMap.put(headers.get(i), record.get(i));
+                                }
+                                batchRecords.add(recordMap);
+                            }
+                        }
+                        if (!batchRecords.isEmpty()){
+                            log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
+                            sfData = saveOrUpdate(api, batchRecords, true);
+                            count += sfData;
+                            maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString();
+                            log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}", api, sfData, maxId);
+                        } else {
+                            log.info("批次无数据记录");
+                        }
+
+                        BulkUtil.closeJob(bulkConnect, job.getId());
+                        log.info("关闭作业成功, 作业ID: {}", job.getId());
+                    }catch (InterruptedException interruptedException){
+                        return ;
+                    } catch (Exception e) {
+                        log.error("处理查询结果时异常: ", e);
+                        throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
+                    }finally {
+                        if (sfData != batchProcessCount){
+                            hasMore = false;
+                            log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
+                        }
+                    }
+                }
+                UpdateWrapper<DataBatch> updateWrapper = new UpdateWrapper<>();
+                updateWrapper.eq("name", param.getApi());
+                updateWrapper.set("first_db_num", count);
+                updateWrapper.set("first_sf_num", count);
+                dataBatchService.update(updateWrapper);
+
+            } catch (Throwable throwable) {
+                String format = String.format("数据特殊表迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", api, JSON.toJSONString(param), throwable);
+                EmailUtil.send("DataDump ERROR", format);
+                log.error("dataDumpSpecial error ", throwable);
+                throw new RuntimeException(throwable);
+            }
+        }, 0, 0);
+    }
+
     public Future<?> getData(DataDumpSpecialParam param, BulkConnection bulkConnect) {
         String api = param.getApi();
         return salesforceExecutor.execute(() -> {
@@ -1108,6 +1415,132 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         }, 0, 0);
     }
 
+    private int getAllBulkV1SfIncrementData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable {
+
+        boolean hasMore = true;
+        QueryWrapper<DataField> wrapper = new QueryWrapper<>();
+        String api = param.getApi();
+        wrapper.eq("api", api);
+        List<DataField> list = dataFieldService.list(wrapper);
+
+        Map<String, Object> map = Maps.newHashMap();
+        String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField();
+        int count = 0;
+        int batch = 0;
+        Date lastCreatedDate = null;
+        String maxId = null;
+
+        List<String> fields = Lists.newArrayList();
+        for (DataField field : list) {
+            if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){
+                fields.add(field.getField());
+            }
+        }
+        DataObject dataObject = dataObjectService.getById(api);
+        if (dataObject != null && StringUtils.isNotBlank(dataObject.getExtraField())) {
+            fields.addAll(Arrays.asList(StringUtils.split(dataObject.getExtraField().replaceAll(StringUtils.SPACE, StringUtils.EMPTY), ",")));
+        }
+        param.setSelect(StringUtils.join(fields, ","));
+
+        log.info("开始处理API: {}, 类型: {}", api, param.getType() == 1 ? "按创建时间" : "按修改时间");
+
+        while (hasMore) {
+            int sfData = 0;
+            param.setMaxId(maxId);
+            param.setLimit(batchProcessCount);
+            // 获取创建时间
+            param.setTimestamp(lastCreatedDate);
+            map.put("param", param);
+            String sql;
+            JobInfo job = null;
+            try {
+                if (param.getType() == 1) {
+                    // type 1 按创建时间纬度
+                    sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.list", map);
+                } else {
+                    // type 2 按修改时间纬度
+                    sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listByModifyTime", map);
+                }
+                log.info("query sql: {}", sql);
+                XxlJobLogger.log("query sql: {}", sql);
+
+                DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
+
+                job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
+                log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
+
+                BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8)));
+                log.info("创建批次成功, 批次ID: {}", batchInfo.getId());
+
+                int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
+                log.info("批次处理完成, 处理记录数: {}", completionOne);
+
+                dataLogService.save(dataLog);
+
+                if (completionOne == 0){
+                    log.info("无更多数据, 结束处理");
+                    BulkUtil.closeJob(bulkConnect, job.getId());
+                    break;
+                }
+
+                List<Map<String, Object>> batchRecords = new ArrayList<>();
+
+                DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
+
+                QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
+                log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
+
+                for (final String resultId : queryResultList.getResult()) {
+                    InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
+
+                    CSVReader rdr = new CSVReader(resultStream);
+                    List<String> headers = rdr.nextRecord();
+                    log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size());
+
+                    // 批量处理记录
+                    List<String> record;
+                    while ((record = rdr.nextRecord()) != null) {
+                        Map<String, Object> recordMap = new HashMap<>();
+                        for (int i = 0; i < headers.size() && i < record.size(); i++) {
+                            recordMap.put(headers.get(i), record.get(i));
+                        }
+                        batchRecords.add(recordMap);
+                    }
+                }
+
+                dataLogService.save(dataLog1);
+
+                if (!batchRecords.isEmpty()){
+                    log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
+                    sfData = saveOrUpdate(api, batchRecords, true);
+                    count += sfData;
+                    maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString();
+                    lastCreatedDate = DateUtils.addHours(DateUtils.parseDate((String) batchRecords.get(batchRecords.size() - 1).get("CreatedDate"), Const.SF_DATE_FORMAT), 8);
+                    log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}, 当前批次最后创建时间:{}", api, sfData, maxId, lastCreatedDate);
+                } else {
+                    log.info("批次无数据记录");
+                }
+
+                BulkUtil.closeJob(bulkConnect, job.getId());
+                batch ++;
+                log.info("关闭作业成功, 作业ID: {}", job.getId());
+            }catch (InterruptedException interruptedException){
+                return 0;
+            } catch (Exception e) {
+                log.error("处理查询结果时异常: ", e);
+                throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
+            }finally {
+                if (sfData != batchProcessCount){
+                    hasMore = false;
+                    log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
+                }
+            }
+        }
+        log.info("API: {} 数据处理完成, 总处理记录数: {}", api, count);
+
+        return count;
+    }
+
     private int getAllBulkV1SfData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable {
 
         boolean hasMore = true;

src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java

@@ -286,6 +286,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_createable", 1)
+                .eq("is_nillable", 0)
+                .eq("is_defaulted_on_create", 0)
+                .or().eq("field", "Id")
+                .or().eq("field", "CreatedDate")
+                .or().eq("field", "CreatedById"));
         List<DataField> list = dataFieldService.list(dbQw);
 
         String beginDateStr = null;
@@ -346,10 +352,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
                 JSONObject account = new JSONObject();
                 for (DataField dataField : list) {
-                    if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField())
-                            || "Id".equals(dataField.getField())){
+                    if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){
                         // 转换为UTC时间并格式化
                         LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")),  inputFormatter);
@@ -369,6 +377,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         }
                         continue;
                     }
+
                     if (dataField.getIsCreateable() !=null && dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) {
                         if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){
                             //引用类型
@@ -419,7 +428,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     account.put("old_sfdc_id__c", data.get(j-1).get("Id"));
                 }
 
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j >= count){
                     break;
@@ -430,16 +438,21 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             JobInfo salesforceInsertJob = null;
 
+            if (insertList.isEmpty()){
+                log.info("当前批次无可获取SFID的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
+            }
+
             try {
                 DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
 
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob,ids);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
                 dataLog1.setEndTime(new Date());
+
                 dataLogService.save(dataLog1);
 
                 DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
@@ -533,52 +546,57 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         int retryCount = 0;
         final int maxRetries = 3;
         
-        while (retryCount <= maxRetries) {
+        while (retryCount < maxRetries) {
             try {
-                rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
-                List<String> resultHeader = rdr.nextRecord();
-                int resultCols = resultHeader.size();
-
-                List<String> row;
-                while ((row = rdr.nextRecord()) != null) {
-                    Map<String, String> resultInfo = new HashMap<String, String>();
-                    for (int i = 0; i < resultCols; i++) {
-                        resultInfo.put(resultHeader.get(i), row.get(i));
-                    }
-                    boolean insertStatus = Boolean.valueOf(resultInfo.get("Success"));
-                    boolean created = Boolean.valueOf(resultInfo.get("Created"));
-                    String id = resultInfo.get("Id");
-                    String error = resultInfo.get("Error");
-                    if (insertStatus && created) {
-                        List<Map<String, Object>> maps = new ArrayList<>();
-                        Map<String, Object> m = new HashMap<>();
-                        m.put("key", "new_id");
-                        m.put("value", id);
-                        maps.add(m);
-                        customMapper.updateById(api, maps, ids[index]);
-                    } else{
-                        List<Map<String, Object>> maps = new ArrayList<>();
-                        Map<String, Object> linkMap1 = new HashMap<>();
-                        linkMap1.put("key", "error_message");
-                        linkMap1.put("value", JSON.toJSONString(error));
-                        maps.add(linkMap1);
-                        customMapper.updateById(api, maps, ids[index]);
-                        log.info("Id:{},saveResults: {}",ids[index], error);
-                    }
-                    index ++;
-                }
+                InputStream batchResultStream = connection.getBatchResultStream(job.getId(), b.getId());
+                rdr = new CSVReader(batchResultStream);
                 break; // 成功执行，跳出重试循环
             }catch (Exception e) {
                 retryCount++;
                 log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api);
                 try {
-                    Thread.sleep(1000 * retryCount); // 简单的退避策略
+                    Thread.sleep(5000 * retryCount); // 简单的退避策略
                 } catch (InterruptedException ie) {
                     Thread.currentThread().interrupt();
                     throw new IOException("Interrupted while retrying", ie);
                 }
+                if (maxRetries <= retryCount){
+                    return 0;
+                }
             }
         }
+
+        List<String> resultHeader = rdr.nextRecord();
+        int resultCols = resultHeader.size();
+
+        List<String> row;
+        while ((row = rdr.nextRecord()) != null) {
+            Map<String, String> resultInfo = new HashMap<String, String>();
+            for (int i = 0; i < resultCols; i++) {
+                resultInfo.put(resultHeader.get(i), row.get(i));
+            }
+            boolean insertStatus = Boolean.valueOf(resultInfo.get("Success"));
+            boolean created = Boolean.valueOf(resultInfo.get("Created"));
+            String id = resultInfo.get("Id");
+            String error = resultInfo.get("Error");
+            if (insertStatus && created) {
+                List<Map<String, Object>> maps = new ArrayList<>();
+                Map<String, Object> m = new HashMap<>();
+                m.put("key", "new_id");
+                m.put("value", id);
+                maps.add(m);
+                customMapper.updateById(api, maps, ids[index]);
+            } else{
+                List<Map<String, Object>> maps = new ArrayList<>();
+                Map<String, Object> linkMap1 = new HashMap<>();
+                linkMap1.put("key", "error_message");
+                linkMap1.put("value", JSON.toJSONString(error));
+                maps.add(linkMap1);
+                customMapper.updateById(api, maps, ids[index]);
+                log.info("Id:{},saveResults: {}",ids[index], error);
+            }
+            index ++;
+        }
         return index;
     }
 
@@ -818,6 +836,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_updateable", 1)
+                .or().eq("field","Id"));
         List<DataField> list = dataFieldService.list(dbQw);
 
         String beginDateStr = null;
@@ -934,13 +954,17 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                             account.put(field, DataUtil.localBulkDataToSfData(dataField.getSfType(), String.valueOf(map.get(field))));
                         }else {
-                            account.put(field, value);
+                            if (map.get(field) == null){
+                                account.put(field, "#N/A");
+                            }else {
+                                account.put(field, value);
+                            }
                         }
                     }
                 }
                 
                 if (dataObject.getIsEditable()){
-                    account.put("old_owner_id__c", map.get("OwnerId"));
+                    account.put("old_owner_id__c", map.get("OwnerId") == null?"":map.get("OwnerId"));
                     account.put("old_sfdc_id__c", map.get("Id"));
                 }
                 updateList.add(account);
@@ -950,16 +974,21 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             JobInfo salesforceUpdateJob = null;
 
+            if (updateList.isEmpty()){
+                log.info("当前批次无可更新的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
+            }
+
             try {
                 DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
 
                 salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(updateList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceUpdateJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(updateList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceUpdateJob);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo);
 
                 dataLog1.setEndTime(new Date());
+
                 dataLogService.save(dataLog1);
 
                 DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
@@ -1145,8 +1174,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 update.setDataLock(1);
                 dataObjectService.updateById(update);
 
-                if (dataFieldService.hasCreatedDate(api)){
-                    insertSingleShareData(api,bulkConnection);
+                if (!dataFieldService.hasCreatedDate(api)){
+                    insertSingleShareData(api,bulkConnection,update);
                     continue;
                 }
 
@@ -1223,8 +1252,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     update.setDataLock(1);
                     dataObjectService.updateById(update);
 
-                    if (api.endsWith("Share") || "GroupMember".equals(api)){
-                        insertSingleShareData(api,bulkConnection);
+                    if (dataFieldService.hasCreatedDate(api)){
+                        insertSingleShareData(api,bulkConnection,dataObject);
                         continue;
                     }
 
@@ -1281,6 +1310,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_createable", 1)
+                .or().eq("field","Id"));
         List<DataField> list = dataFieldService.list(dbQw);
         TimeUnit.MILLISECONDS.sleep(1);
 
@@ -1293,16 +1324,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
         }
 
-        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 8000";
-        }else {
-            sql = "where new_id is null ";
-            sql2 = "new_id is null limit 8000";
-        }
+        String sql = "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
+        String sql2 = "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000";
 
         //表内数据总量
         Integer count = customMapper.countBySQL(api, sql);
@@ -1354,9 +1377,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
                 for (DataField dataField : list) {
 
-                    if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){
+                    if ("Owner_Type".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){
                         // 转换为UTC时间并格式化
                         LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")),  inputFormatter);
@@ -1405,7 +1431,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                 account.put(dataField.getField(), DataUtil.localBulkDataToSfData(dataField.getSfType(), data.get(j - 1).get(dataField.getField()).toString()));
                             }else {
-                                account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                if (data.get(j - 1).get(dataField.getField()) == null){
+                                    account.put(dataField.getField(), "#N/A");
+                                }else {
+                                    account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                }
                             }
                         }
                     }
@@ -1420,10 +1450,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     continue;
                 }
                 if (dataObject.getIsEditable()) {
-                    account.put("old_owner_id__c", data.get(j - 1).get("OwnerId"));
+                    account.put("old_owner_id__c", data.get(j - 1).get("OwnerId") == null? "":data.get(j - 1).get("OwnerId"));
                     account.put("old_sfdc_id__c", data.get(j - 1).get("Id"));
                 }
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j >= count){
                     break;
@@ -1433,14 +1462,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             JobInfo salesforceInsertJob = null;
 
             dataLog.setEndTime(new Date());
+
             dataLogService.save(dataLog);
 
+            if (insertList.isEmpty()){
+                log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
+            }
+
             try {
                 DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
 
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob,ids);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
@@ -1453,6 +1487,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
                 dataLog2.setEndTime(new Date());
+
                 dataLogService.save(dataLog2);
 
             } catch (Exception e) {
@@ -1482,7 +1517,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
     /**
      * 执行一次性Insert Share数据
      */
-    private void insertSingleShareData(String api, BulkConnection bulkConnection) throws Exception {
+    private void insertSingleShareData(String api, BulkConnection bulkConnection, DataObject dataObject) throws Exception {
 
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
@@ -1492,8 +1527,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String beginDateStr = null;
         String endDateStr = null;
 
+        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 8000";
+        }else {
+            sql = "where new_id is null ";
+            sql2 = "new_id is null limit 8000";
+        }
+
         //表内数据总量
-        Integer count = customMapper.countBySQL(api, "where new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant')");
+        Integer count = customMapper.countBySQL(api, sql);
 
         if (count == 0) {
             return;
@@ -1522,7 +1568,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
 
-            List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') order by Id asc limit " + i * 10000 + ",10000");
+            List<JSONObject> data = customMapper.listJsonObject("*", api, sql2);
             int size = data.size();
 
             log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -1537,9 +1583,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
 
                 for (DataField dataField : list) {
-                    if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){
+                    if ("Owner_Type".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if (dataField.getIsCreateable() !=null && dataField.getIsCreateable()) {
                         if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){
                             //引用类型
@@ -1574,7 +1623,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                 account.put(dataField.getField(), DataUtil.localBulkDataToSfData(dataField.getSfType(), data.get(j - 1).get(dataField.getField()).toString()));
                             }else {
-                                account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                if (data.get(j - 1).get(dataField.getField()) == null){
+                                    account.put(dataField.getField(), "#N/A");
+                                }else {
+                                    account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                }
                             }
                         }
                     }
@@ -1590,7 +1643,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     continue;
                 }
 
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j == count){
                     break;
@@ -1603,14 +1655,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             JobInfo salesforceInsertJob = null;
 
+            if (insertList.isEmpty()){
+                log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
+            }
+
             try {
                 DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
 
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                List<BatchInfo> batchInfos = CsvConverterUtil.writeToCsvNew(insertList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list,dataObject.getIsEditable(), bulkConnection, salesforceInsertJob,ids);
 
-                BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
+                BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
                 dataLog1.setEndTime(new Date());
 
@@ -1618,16 +1674,16 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
                 DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
 
-                checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
+                checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
                 dataLog2.setEndTime(new Date());
 
                 dataLogService.save(dataLog2);
+
             } catch (Exception e) {
-                log.error("insertSingleShareData error api:{}", api, e);
+                log.error("createdNewIdBatch error api:{}", api, e);
                 throw e;
             }finally {
-
                 BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
             }
         }

src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java

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

src/main/java/com/celnet/datadump/util/BulkUtil.java

@@ -165,6 +165,7 @@ public class BulkUtil {
         job.setObject(sobjectType);
         job.setOperation(operation);
         job.setContentType(ContentType.CSV);
+        job.setConcurrencyMode(ConcurrencyMode.Serial);
         job = connection.createJob(job);
         return job;
     }

src/main/java/com/celnet/datadump/util/CsvConverterUtil.java

@@ -192,13 +192,11 @@ public class CsvConverterUtil {
 
         if (isEditable){
             header = Stream.concat(
-                    fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
-                            .map(DataField::getField),
-                    Stream.of("Id","old_owner_id__c", "old_sfdc_id__c")
+                    fields.stream().map(DataField::getField),
+                    Stream.of("old_owner_id__c", "old_sfdc_id__c")
             ).toArray(String[]::new);
         }else {
-            header = fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
-                    .map(DataField::getField).toArray(String[]::new);
+            header = fields.stream().map(DataField::getField).toArray(String[]::new);
         }
 
         BatchInfo batch = null;
@@ -240,6 +238,81 @@ public class CsvConverterUtil {
         return batch;
     }
 
+    public static BatchInfo writeToCsvOne(List<JSONObject> jsonList,
+                                             String fileName,
+                                             List<DataField> fields,
+                                             Boolean isEditable,
+                                             BulkConnection connection,
+                                             JobInfo jobInfo,String[] ids) {
+
+        // 1. 创建目标目录（不存在则创建）
+        File targetDir = FileUtil.mkdir("data-dump/dataFile");
+
+        // 2. 构建完整文件路径
+        String fullPath = targetDir.getAbsolutePath()  + File.separator  + fileName + ".csv";
+
+        CsvWriter csvWriter = CsvUtil.getWriter(fullPath, CharsetUtil.CHARSET_UTF_8, false);
+
+        String[] header = null;
+
+        File file = null;
+
+        if (isEditable){
+            header = Stream.concat(
+                    fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+                            .map(DataField::getField),
+                    Stream.of("old_owner_id__c", "old_sfdc_id__c")
+            ).toArray(String[]::new);
+        }else {
+            header = fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+                    .map(DataField::getField).toArray(String[]::new);
+        }
+
+        BatchInfo batch = null;
+
+        try {
+
+            // 2. 写入表头（必须使用 String[]）
+            csvWriter.writeHeaderLine(header);
+
+            int index = 0;
+
+            // 遍历数据列表
+            for (JSONObject jsonObject : jsonList) {
+                // 按表头顺序获取值
+                String[] row = new String[header.length];
+                for (int i = 0; i < header.length;  i++) {
+                    // 将每个值转换为字符串（如果为null则转为空字符串）
+                    Object value = jsonObject.get(header[i]);
+                    row[i] = String.valueOf(value);
+                }
+                ids[index] = jsonObject.get("Id").toString();
+
+                csvWriter.writeLine(row);
+
+                index ++;
+            }
+
+            // 关闭writer（在try-with-resources中可省略，但这里我们显式关闭）
+            csvWriter.close();
+
+            file = new File(fullPath);
+
+            InputStream inputStream = Files.newInputStream(file.toPath());
+
+            batch = connection.createBatchFromStream(jobInfo, inputStream);
+
+        } catch (IOException e) {
+            log.error("CSV文件操作失败: {}", fullPath, e);
+            file.deleteOnExit();
+        } catch (AsyncApiException e) {
+            log.error("Bulk API上传失败: {}", e.getMessage(), e);
+            file.deleteOnExit();
+        }
+
+        return batch;
+    }
+
     public static String exportToCsv(List<Map<String, Object>> data, String fileName) throws IOException {
         // 1. 创建目标目录（不存在则创建）
         File targetDir = FileUtil.mkdir("data-dump/dataFile");