diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java index 07cdfa7..c4e2bf1 100644 --- a/src/main/java/com/celnet/datadump/controller/JobController.java +++ b/src/main/java/com/celnet/datadump/controller/JobController.java @@ -248,6 +248,32 @@ public class JobController { return dataImportBatchService.immigrationUpdateBatch(param); } + /** + * 一次性insert（大数据量） + * @param paramStr + * @author kris + * @return + * @throws Exception + */ + @PostMapping("/insertSingleBatchJob") + @ApiOperation("一次性insert（大数据量）") + @LogServiceAnnotation(operateType = OperateTypeConstant.TYPE_INSERT, remark = "一次性insert（大数据量）") + public ReturnT insertSingleBatchJob(String paramStr) throws Exception { + log.info("insertSingleBatchJob execute start .................."); + SalesforceParam param = new SalesforceParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, SalesforceParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + return dataImportBatchService.insertSingleBatch(param); + } + /** * 返写个人联系人ID * @param paramStr diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index d0f07eb..8d9d54d 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -294,7 +294,27 @@ public class DataDumpNewJob { } catch (Throwable throwable) { return new ReturnT<>(500, "参数解析失败！"); } - param.setType(2); + return commonService.incrementNew(param); } + + /** + * 校验删除数据 + * @param paramStr 参数json + * @return result + */ + @XxlJob("checkDeletedDataJob") + public ReturnT checkDeletedDataJob(String paramStr) throws Exception { + log.info("checkDeletedDataJob 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 dataImportNewService.checkDeletedData(param); + } } diff --git a/src/main/java/com/celnet/datadump/service/DataImportNewService.java b/src/main/java/com/celnet/datadump/service/DataImportNewService.java index 1946b17..253a524 100644 --- a/src/main/java/com/celnet/datadump/service/DataImportNewService.java +++ b/src/main/java/com/celnet/datadump/service/DataImportNewService.java @@ -46,4 +46,7 @@ public interface DataImportNewService { * @throws Exception exception */ ReturnT updateLinkTypeJob(SalesforceParam param) throws Exception; + + ReturnT checkDeletedData(SalesforceParam param) throws Exception; + } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java index c5b7150..16e67a2 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -284,7 +284,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List list = dataFieldService.list(dbQw); - TimeUnit.MILLISECONDS.sleep(1); String beginDateStr = null; String endDateStr = null; @@ -413,7 +412,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { //写入csv文件 - String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString()); + String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,false); JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); @@ -693,7 +692,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List list = dataFieldService.list(dbQw); - TimeUnit.MILLISECONDS.sleep(1); String beginDateStr = null; String endDateStr = null; @@ -746,6 +744,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { //根据旧sfid查找引用对象新sfid if (field.equals("Id")) { account.put("Id",String.valueOf(map.get("new_id"))); + } else if (!DataUtil.isUpdate(field) || (dataField.getIsCreateable() != null && !dataField.getIsCreateable())) { continue; } else if ( "WhoId".equals(field) ||"WhatId".equals(field)){ @@ -794,7 +793,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { //写入csv文件 - String fullPath = CsvConverterUtil.writeToCsv(updateList, UUID.randomUUID().toString()); + String fullPath = CsvConverterUtil.writeToCsv(updateList, UUID.randomUUID().toString(),list,dataObject.getIsEditable()); JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update); @@ -1063,8 +1062,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { List list = dataFieldService.list(dbQw); TimeUnit.MILLISECONDS.sleep(1); QueryWrapper obQw = new QueryWrapper<>(); - dbQw.eq("api", api); - DataObject dataObject = dataObjectService.list(obQw).get(0); + obQw.eq("name", api); + DataObject dataObject = dataObjectService.getOne(obQw); String beginDateStr = null; String endDateStr = null; @@ -1193,7 +1192,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { //写入csv文件 - String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString()); + String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,dataObject.getIsEditable()); JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); @@ -1208,7 +1207,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { new File(fullPath).delete(); } catch (Exception e) { - log.error("manualCreatedNewId error api:{}", api, e); + log.error("insertSingleData error api:{}", api, e); throw e; } } @@ -1338,7 +1337,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { //写入csv文件 - String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString()); + String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,false); JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); @@ -1353,7 +1352,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { new File(fullPath).delete(); } catch (Exception e) { - log.error("manualCreatedNewId error api:{}", api, e); + log.error("insertSingleShareData error api:{}", api, e); throw e; } } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java index 366bf99..ab71fd2 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -1689,6 +1689,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { } } + /** * 【batch逻辑】组装updateLinkType执行参数 */ @@ -1910,4 +1911,15 @@ public class DataImportNewServiceImpl implements DataImportNewService { } + + /** + * 校验删除数据入口 + */ + @Override + public ReturnT checkDeletedData(SalesforceParam param) throws Exception { + + + return ReturnT.SUCCESS; + } + } diff --git a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java index de85b86..01b3a78 100644 --- a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java +++ b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java @@ -8,6 +8,8 @@ import cn.hutool.core.text.csv.CsvWriteConfig; import cn.hutool.core.text.csv.CsvWriter; import cn.hutool.core.util.CharsetUtil; import cn.hutool.json.JSONObject; +import com.celnet.datadump.entity.DataField; +import com.celnet.datadump.entity.DataObject; import com.opencsv.CSVReader; import com.opencsv.exceptions.CsvException; import org.apache.commons.csv.CSVFormat; @@ -16,6 +18,7 @@ import org.apache.commons.csv.CSVPrinter; import java.io.*; import java.nio.charset.StandardCharsets; import java.util.*; +import java.util.stream.Stream; /** * csv格式转换工具类 @@ -26,7 +29,7 @@ public class CsvConverterUtil { * 将JSONObject（代表一行）或JSONArray（代表多行）写入CSV文件 * @param jsonList JSON数据 */ - public static String writeToCsv(List jsonList, String fileName) { + public static String writeToCsv(List jsonList, String fileName, List list, Boolean isEditable) { // 1. 创建目标目录（不存在则创建） File targetDir = FileUtil.mkdir("data-dump/dataFile"); @@ -36,7 +39,18 @@ public class CsvConverterUtil { CsvWriter csvWriter = CsvUtil.getWriter(fullPath, CharsetUtil.CHARSET_GBK, false); - String[] header = jsonList.get(0).keySet().toArray(new String[0]); + String[] header = null; + + if (isEditable){ + header = Stream.concat( + list.stream().filter(DataField::getIsCreateable) + .map(DataField::getField), + Stream.of("old_owner_id__c", "old_sfdc_id__c") + ).toArray(String[]::new); + }else { + header = list.stream().filter(DataField::getIsCreateable) + .map(DataField::getField).toArray(String[]::new); + } // 2. 写入表头（必须使用 String[]） csvWriter.writeHeaderLine(header); diff --git a/src/main/java/com/celnet/datadump/util/DataUtil.java b/src/main/java/com/celnet/datadump/util/DataUtil.java index 185fcd1..2db2ea8 100644 --- a/src/main/java/com/celnet/datadump/util/DataUtil.java +++ b/src/main/java/com/celnet/datadump/util/DataUtil.java @@ -31,9 +31,7 @@ import javax.annotation.PostConstruct; import java.math.BigDecimal; import java.text.ParseException; import java.text.SimpleDateFormat; -import java.time.LocalDateTime; -import java.time.ZoneId; -import java.time.ZonedDateTime; +import java.time.*; import java.time.format.DateTimeFormatter; import java.util.*; @@ -432,9 +430,9 @@ public class DataUtil { case "date": return data+"T08:00:00Z"; case "datetime": - LocalDateTime localDateTime = LocalDateTime.parse(data, DateTimeFormatter.ISO_LOCAL_DATE_TIME); - ZonedDateTime zonedDateTime = localDateTime.atZone(ZoneId.systemDefault()); - return zonedDateTime.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME); + LocalDateTime localDateTime = LocalDateTime.parse(data); + OffsetDateTime offsetDateTime = localDateTime.atOffset(ZoneOffset.ofHours(0)); + return offsetDateTime.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME); case "time": return adjustHour(data); default: