From 36c2caad228ed14614551801dec183b4cc1dc834 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 9 Sep 2025 11:10:53 +0800 Subject: [PATCH] =?UTF-8?q?[feat]=20=20bulk=20api=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .run/data-dump（城博测试）.run.xml | 14 + .run/data-dump（康蒂思）.run.xml | 14 + file/BlukV1_750e2000004GIDCAA4.csv | 1 + pom.xml | 7 + .../datadump/controller/JobController.java | 42 +- .../celnet/datadump/job/DataDumpNewJob.java | 2 +- .../celnet/datadump/mapper/CustomMapper.java | 41 + .../datadump/service/CommonService.java | 2 + .../service/impl/CommonBatchServiceImpl.java | 803 ++++++++++++++---- .../service/impl/CommonServiceImpl.java | 328 ++++++- .../impl/DataDumpSpecialServiceImpl.java | 8 +- .../impl/DataImportBatchServiceImpl.java | 8 +- .../impl/DataImportNewServiceImpl.java | 83 +- .../service/impl/DataImportServiceImpl.java | 23 +- .../service/impl/FileServiceImpl.java | 2 + .../com/celnet/datadump/util/BulkUtil.java | 275 +++++- .../com/celnet/datadump/util/DataUtil.java | 71 ++ .../datadump/util/HttpTransportImpl.java | 429 ++++++++++ .../datadump/util/HttpTransportInterface.java | 37 + src/main/resources/mapper/CustomMapper.xml | 62 +- .../resources/mapper/SalesforceMapper.xml | 35 +- 21 files changed, 2013 insertions(+), 274 deletions(-) create mode 100644 .run/data-dump（城博测试）.run.xml create mode 100644 .run/data-dump（康蒂思）.run.xml create mode 100644 file/BlukV1_750e2000004GIDCAA4.csv create mode 100644 src/main/java/com/celnet/datadump/util/HttpTransportImpl.java create mode 100644 src/main/java/com/celnet/datadump/util/HttpTransportInterface.java diff --git a/.run/data-dump（城博测试）.run.xml b/.run/data-dump（城博测试）.run.xml new file mode 100644 index 0000000..60cda99 --- /dev/null +++ b/.run/data-dump（城博测试）.run.xml @@ -0,0 +1,14 @@ + + + + + + + + + \ No newline at end of file diff --git a/.run/data-dump（康蒂思）.run.xml b/.run/data-dump（康蒂思）.run.xml new file mode 100644 index 0000000..080dc3b --- /dev/null +++ b/.run/data-dump（康蒂思）.run.xml @@ -0,0 +1,14 @@ + + + + + + + + + \ No newline at end of file diff --git a/file/BlukV1_750e2000004GIDCAA4.csv b/file/BlukV1_750e2000004GIDCAA4.csv new file mode 100644 index 0000000..3eec75d --- /dev/null +++ b/file/BlukV1_750e2000004GIDCAA4.csv @@ -0,0 +1 @@ +752e2000002qkjy \ No newline at end of file diff --git a/pom.xml b/pom.xml index 3ad6b55..6b5bc4a 100644 --- a/pom.xml +++ b/pom.xml @@ -36,6 +36,13 @@ + + + org.apache.httpcomponents + httpclient + 4.5.14 + + org.springframework.boot spring-boot-starter diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java index 5c9809c..206cf7b 100644 --- a/src/main/java/com/celnet/datadump/controller/JobController.java +++ b/src/main/java/com/celnet/datadump/controller/JobController.java @@ -48,6 +48,8 @@ public class JobController { private DataImportBatchService dataImportBatchService; @Autowired private DataImportNewService dataImportNewService; + @Autowired + private CommonBatchService commonBatchService; @PostMapping("/fileTransform") @ApiOperation("附件解析") @@ -118,15 +120,20 @@ public class JobController { /** * 创建表结构 * - * @param param 参数 * @return result */ @PostMapping("/createApi") @ApiOperation("创建表结构") - public Result createApi(@RequestBody SalesforceParam param) throws Exception { - if (param.getType() == null) { - return Result.fail("参数缺失！"); + public Result createApi(String paramStr) throws Exception { + log.info("dataImportBatchJob execute start .................."); + SalesforceParam param = new SalesforceParam(); + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, SalesforceParam.class); } + param.setType(1); + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); ReturnT returnT = commonService.createApi(param); if (returnT.getCode() == ReturnT.SUCCESS_CODE) { return Result.success(); @@ -224,6 +231,33 @@ public class JobController { return dataImportBatchService.immigrationBatch(param); } + /** + * bulk批量大数据生成newSFID + * @param paramStr + * @author kris + * @return + * @throws Exception + */ + @PostMapping("/dataDumpManualBatchJob") + @ApiOperation("存量任务（大数据量）") + @LogServiceAnnotation(operateType = OperateTypeConstant.TYPE_INSERT, remark = "存量任务（大数据量）") + public ReturnT dataDumpManualBatchJob(String paramStr) throws Exception { + log.info("dataImportBatchJob 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.setType(1); + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + return commonBatchService.dumpBatch(param); + } + /** * bulk批量更新大数据量数据 * @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 7db6881..7d7d5ef 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -91,7 +91,7 @@ public class DataDumpNewJob { } /** - * 存量任务（大批量） + * 存量任务（大数据量） * * @param paramStr 参数json * @return result diff --git a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java index abe1cf1..a05895f 100644 --- a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java +++ b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java @@ -57,6 +57,47 @@ public interface CustomMapper { */ public void createTable(@Param("tableName") String tableName, @Param("tableComment") String tableComment, @Param("maps") List> maps, @Param("index") List> index); + /** + * 创建RANGE分区表 + * + * @param tableName 表名 + * @param tableComment 表注释 + * @param maps 字段定义列表 + * @param index 索引定义列表 + * @param partitionKey 分区键 + * @param partitions 分区定义列表 + */ + void createRangePartitionTable(@Param("tableName") String tableName, + @Param("tableComment") String tableComment, + @Param("maps") List> maps, + @Param("index") List> index, + @Param("partitionKey") String partitionKey, + @Param("partitions") List> partitions); + + /** + * 在指定分区插入单条记录 + * + * @param tableName 表名 + * @param partitionName 分区名 + * @param maps 插入字段映射 + */ + void saveToPartition(@Param("tableName") String tableName, + @Param("partitionName") String partitionName, + @Param("maps") List> maps); + + /** + * 在指定分区根据ID更新记录 + * + * @param tableName 表名 + * @param partitionName 分区名 + * @param maps 更新字段映射 + * @param id 记录ID + */ + void updateToPartition(@Param("tableName") String tableName, + @Param("partitionName") String partitionName, + @Param("maps") List> maps, + @Param("id") String id); + /** * 创建字段 * @param tableName diff --git a/src/main/java/com/celnet/datadump/service/CommonService.java b/src/main/java/com/celnet/datadump/service/CommonService.java index d9091d3..4ae07b9 100644 --- a/src/main/java/com/celnet/datadump/service/CommonService.java +++ b/src/main/java/com/celnet/datadump/service/CommonService.java @@ -27,6 +27,8 @@ public interface CommonService { void checkApi(String apiName, Boolean createBatch) throws Exception; + void checkPartitionedApi(String apiName, Boolean createBatch) throws Exception; + /** * 创建表 * @param param 参数 diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index 5cbd9e2..66fa038 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -1,45 +1,62 @@ package com.celnet.datadump.service.impl; import com.alibaba.fastjson2.JSON; + import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.celnet.datadump.config.SalesforceConnect; import com.celnet.datadump.config.SalesforceExecutor; -import com.celnet.datadump.entity.DataBatch; -import com.celnet.datadump.entity.DataBatchHistory; -import com.celnet.datadump.entity.DataObject; -import com.celnet.datadump.entity.DataReport; +import com.celnet.datadump.entity.*; import com.celnet.datadump.global.Const; +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.BulkUtil; import com.celnet.datadump.util.DataUtil; import com.celnet.datadump.util.EmailUtil; import com.celnet.datadump.util.SqlUtil; import com.google.common.collect.Lists; import com.google.common.collect.Maps; -import com.sforce.async.BulkConnection; + +import com.opencsv.CSVReaderHeaderAware; +import com.opencsv.exceptions.CsvValidationException; +import com.sforce.async.*; +import com.sforce.soap.partner.DescribeSObjectResult; +import com.sforce.soap.partner.Field; import com.sforce.soap.partner.PartnerConnection; import com.sforce.soap.partner.QueryResult; import com.sforce.soap.partner.sobject.SObject; +import com.sforce.ws.parser.XmlInputStream; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.log.XxlJobLogger; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.time.DateFormatUtils; import org.apache.commons.lang3.time.DateUtils; +import org.apache.http.Header; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +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 java.util.Date; -import java.util.List; -import java.util.Map; +import java.io.*; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.*; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; +import static org.apache.commons.lang.CharEncoding.UTF_8; + @Service @Slf4j public class CommonBatchServiceImpl implements CommonBatchService { @@ -53,185 +70,617 @@ public class CommonBatchServiceImpl implements CommonBatchService { @Autowired private CommonService commonService; @Autowired + private DataFieldService dataFieldService; + @Autowired private SalesforceConnect salesforceConnect; @Autowired - private DataFieldService dataFieldService; + private CustomMapper customMapper; + @Autowired + private DataLogService dataLogService; + @Autowired + private DataDumpSpecialService dataDumpSpecialService; + @Autowired + private DataBatchHistoryService dataBatchHistoryService; @Override public ReturnT dumpBatch(SalesforceParam param) throws Exception { -// List> futures = Lists.newArrayList(); -// try { -// if (StringUtils.isNotBlank(param.getApi())) { -// // 手动任务 -// ReturnT result = manualBatchDump(param, futures); -// if (result != null) { -// return result; -// } -// } else { -// // 自动任务 -// autoDump(param, futures); -// } -// return ReturnT.SUCCESS; -// } catch (Throwable throwable) { -// salesforceExecutor.remove(futures.toArray(new Future[]{})); -// log.error("dump error", throwable); -// throw throwable; -// } + List> futures = Lists.newArrayList(); + try { + if (StringUtils.isNotBlank(param.getApi())) { + // 手动任务 + ReturnT result = manualBatchDump(param, futures); + if (result != null) { + return result; + } + } + return ReturnT.SUCCESS; + } catch (Throwable throwable) { + salesforceExecutor.remove(futures.toArray(new Future[]{})); + log.error("dump error", throwable); + throw throwable; + } + } + + private ReturnT manualBatchDump(SalesforceParam param, List> futures) throws InterruptedException { + List apis; + apis = DataUtil.toIdList(param.getApi()); + String join = StringUtils.join(apis, ","); + log.info("dump apis: {}", join); + XxlJobLogger.log("dump apis: {}", join); + + // 根据参数获取sql + for (String api : apis) { + DataObject update = new DataObject(); + TimeUnit.MILLISECONDS.sleep(1); + try { + commonService.checkApi(api, true); + if (!dataFieldService.hasCreatedDate(api)) { + DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam(); + dataDumpSpecialParam.setApi(api); + Future future = dataDumpSpecialService.getData(dataDumpSpecialParam, salesforceConnect.createConnect()); + // 等待当前所有线程执行完成 + salesforceExecutor.waitForFutures(future); + continue; + } + param.setApi(api); + List salesforceParams = null; + update.setName(api); + update.setDataLock(1); + dataObjectService.updateById(update); + QueryWrapper 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 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); + } + } return ReturnT.SUCCESS; } -// private ReturnT manualBatchDump(SalesforceParam param, List> futures) throws InterruptedException { -// List apis = DataUtil.toIdList(param.getApi()); -// String join = StringUtils.join(apis, ","); -// log.info("dump apis: {}", join); -// -// // 根据参数获取sql -// for (String api : apis) { -// DataObject update = new DataObject(); -// TimeUnit.MILLISECONDS.sleep(1); -// try { -// // 判断当前表是否存在 -// commonService.checkApi(api, true); -// -//// if (!hasCreatedDate(api)) { -//// DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam(); -//// dataDumpSpecialParam.setApi(api); -//// Future future = dataDumpSpecialService.getData(dataDumpSpecialParam, salesforceConnect.createConnect()); -//// // 等待当前所有线程执行完成 -//// salesforceExecutor.waitForFutures(future); -//// continue; -//// } -// -// List salesforceParams = null; -// update.setName(api); -// update.setDataLock(1); -// dataObjectService.updateById(update); -// QueryWrapper dbQw = new QueryWrapper<>(); -// dbQw.eq("name", api) -// .isNull("first_sync_date"); -// -// List 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); -// } -// } -// return null; -// } -// -// -// private void dumpBatchData(SalesforceParam param) throws Throwable { -// String api = param.getApi(); -// BulkConnection bulkConnect = null; + private BulkConnection dumpBatchData(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 = getAllBulkV1SfData(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; + } + + + private int getAllBulkV1SfData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable { + + boolean hasMore = true; + QueryWrapper wrapper = new QueryWrapper<>(); + String api = param.getApi(); + wrapper.eq("api", api); + List list = dataFieldService.list(wrapper); + + Map map = Maps.newHashMap(); + String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField(); + int count = 0; + Date lastCreatedDate = null; + String maxId = null; + + List 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(10000); + // 获取创建时间 + 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); + + 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> 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 headers = rdr.nextRecord(); + log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size()); + + // 批量处理记录 + List record; + while ((record = rdr.nextRecord()) != null) { + Map 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(); + 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()); + 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 != 10000){ + hasMore = false; + log.info("当前批次数据不足10000条, 结束循环"); + } + } + } + log.info("API: {} 数据处理完成, 总处理记录数: {}", api, count); + + return count; + } + + /** + * 处理查询结果 + * + + * @throws AsyncApiException 异步API异常 + */ +// private int processQueryV1Results(BulkConnection connection, JobInfo job, String api) throws AsyncApiException { +// int totalProcessedRecords = 0; +// int batch = 0; // 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); -// } -// // 若count数量过多 可能导致超时出不来结果 对该任务做进一步拆分 -// int failCount = 0; -// boolean isSuccess = false; -// while (failCount <= Const.MAX_FAIL_COUNT) { -// try { -// // 不能 count Id -// if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){ -// dataBatchHistory.setSfNum(commonService.countSfNum(connect, param)); -// } -// isSuccess = true; -// break; -// } catch (Throwable throwable) { -// failCount++; -// } -// } -// // 不成功 做任务拆分 -// if (!isSuccess) { -// if (splitTask(param)) { -// return connect; -// } -// } -// getAllSfData(param, connect, dataReport); -// updateDataBatch(param, dataBatchHistory); -// } 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 connect; -// } +// BatchInfoList batchInfoList = connection.getBatchInfoList(job.getId()); +// BatchInfo[] batchInfos = batchInfoList.getBatchInfo(); // -// /** -// * 统计salesforce数据量 -// * -// * @param connect connect -// * @param param 参数 -// * @return sf统计数量 -// */ -// public Integer countSfNum(PartnerConnection connect, SalesforceParam param) throws Exception { -// Map map = Maps.newHashMap(); -// map.put("param", param); -// String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.count", map); -// log.info("count sql: {}", sql); -// XxlJobLogger.log("count sql: {}", sql); -// QueryResult queryResult = connect.queryAll(sql); -// SObject record = queryResult.getRecords()[0]; -// Integer num = (Integer) record.getField("num"); -// log.info("count sf num: {}", num); -// XxlJobLogger.log("count sf num: {}", num); -// return num; +// for (BatchInfo batchInfo : batchInfos) { +// if (batchInfo.getState() == BatchStateEnum.Completed) { +// +// InputStream resultStream = connection.getBatchResultStream(job.getId(), batchInfo.getId()); +// +// // 获取批次结果 +// CSVReader rdr = new CSVReader(resultStream); +// List headers = rdr.nextRecord(); +// +// // 批量处理记录 +// List> batchRecords = new ArrayList<>(); +// List record; +// while ((record = rdr.nextRecord()) != null) { +// Map 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); +// } +// DataLog dataLog = new DataLog(api, null, new Date(), null, "Bulk数据拉取，拉取第" + batch + "批数据", "InsertDB"); +//// totalProcessedRecords += processBatchRecords(job.getObject(), batchRecords); +// dataLog.setEndTime(new Date()); +// dataLogService.save(dataLog); +// +// } else { +// log.error("批次 {} 的批量 {} 状态异常: {}", job.getId(), batchInfo.getId(), batchInfo.getState()); +// } +// } +// } catch (Exception e) { +// throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown); +// } +// return totalProcessedRecords; // } + + + private int getAllSfData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable { + String v2JobId = null; + try { + String api = param.getApi(); + Map map = Maps.newHashMap(); + // 获取对象字段信息 + PartnerConnection partnerConnection = salesforceConnect.createConnect(); + DescribeSObjectResult dsr = partnerConnection.describeSObject(api); + Field[] dsrFields = dsr.getFields(); + List fields = Lists.newArrayList(); + + for (Field field : dsrFields) { + // 不查询文件 + if ("base64".equalsIgnoreCase(field.getType().toString())) { + continue; + } + fields.add(field.getName()); + } + + 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, ",")); + map.put("param", param); + String sql; + 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); + +// job = createBulkJob(bulkConnect, api, sql); + + // 创建Bulk API作业 +// job = BulkUtil.createQueryJob1(api,sql,bulkConnect,job); +// +// List batchInfoList = new ArrayList<>(); +// +// BatchInfo batch = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8))); +// +// batchInfoList.add( batch); + String restEndpoint = bulkConnect.getConfig().getRestEndpoint(); + + String url = restEndpoint.substring(0, restEndpoint.indexOf("services/")); + + v2JobId = BulkUtil.createBulkV2Job(url,sql, bulkConnect); + + BulkUtil.waitForBulkV2JobCompletion(url,v2JobId, bulkConnect); + + // 处理查询结果 + return processQueryResults(url,bulkConnect, v2JobId,null, api); + + }catch (Exception e){ + log.error("getAllSfData error", e); + } + return 0; + } + + + private JobInfo createBulkJob(BulkConnection connection, String objectName, String query) throws AsyncApiException { + JobInfo job = new JobInfo(); + job.setObject(objectName); + job.setOperation(OperationEnum.queryAll); + job.setContentType(ContentType.CSV); + job = connection.createJob(job); + + // 关闭作业 + job.setState(JobStateEnum.Closed); + + return job; + } + + + /** + * 处理查询结果 + * + * @param connection Bulk连接 + * @throws AsyncApiException 异步API异常 + */ + private int processQueryResults(String url,BulkConnection connection, String jobId ,String locator,String api) throws AsyncApiException { + int totalProcessedRecords = 0; + try { + String newLocator = null; + int numberOfRecords = 0; + boolean hasMoreRecords = true; + + while (hasMoreRecords) { + + CloseableHttpClient httpClient = HttpClients.createDefault(); + String resultsUrl = url + "services/data/v56.0/jobs/query/" + jobId + "/results"; + if (locator != null && !locator.isEmpty()) { + resultsUrl += "?locator=" + java.net.URLEncoder.encode(locator, "UTF-8"); + } + HttpGet httpGet = new HttpGet(resultsUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + httpGet.setHeader("Content-Type", "application/json"); + httpGet.setHeader("Accept", "text/csv"); + httpGet.setHeader("X-PrettyPrint", "1"); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + Header locatorHeader = response.getFirstHeader("Sforce-Locator"); + Header numRecordsHeader = response.getFirstHeader("Sforce-NumberOfRecords"); + newLocator = (locatorHeader != null) ? locatorHeader.getValue() : null; + numberOfRecords = (numRecordsHeader != null) ? Integer.parseInt(numRecordsHeader.getValue()) : 0; + totalProcessedRecords += numberOfRecords; + String csvResponse = EntityUtils.toString(response.getEntity()); +// writeExtractionForServerStream(csvResponse,api); + // 检查是否还有更多记录 + if (numberOfRecords == 0 || newLocator == null || newLocator.isEmpty()) { + hasMoreRecords = false; + } else { + locator = newLocator; // 设置下一次迭代的locator + } + } else { + String errorBody = EntityUtils.toString(response.getEntity()); + throw new RuntimeException("Failed to get query job results, status code: " + statusCode + ", error: " + errorBody); + } + } finally { + response.close(); + httpClient.close(); + } + } + + }catch (Exception e){ + log.error("processQueryResults error", e); + } + return totalProcessedRecords; + } + +// public int writeExtractionForServerStream(String csvData,String api) throws IOException { +// +//// int totalProcessedRecords = 0; +//// try { +//// CSVReader rdr = new CSVReader(new StringReader(csvData)); +//// rdr.setMaxCharsInFile(Integer.MAX_VALUE); +//// rdr.setMaxRowsInFile(Integer.MAX_VALUE); +//// List record ; +//// List> batchRecords = new ArrayList<>(); +//// while ((record = rdr.nextRecord()) != null) { +//// Map recordMap = new HashMap<>(); +//// for (int i = 0; i < record.size(); i++) { +//// recordMap.put(record.get(i), record.get(i)); +//// } +//// batchRecords.add(recordMap); +//// } +// +//// totalProcessedRecords += processBatchRecords(api, batchRecords); +// +//// }catch (Exception e){ +//// throw new IOException("处理结果时异常: " + e.getMessage()); +//// } +//// return totalProcessedRecords; +// } + + + /** + * 插入或更新数据 + * + * @param api 表名 + * @param insert 不存在是否插入 + * @return 处理的记录数 + */ + public Integer saveOrUpdate(String api,List> batchRecords, boolean insert) throws Throwable { + // 根据id查询数据库 取出已存在数据的id + List ids = batchRecords.stream() + .map(record -> record.get("Id").toString()) + .filter(Objects::nonNull) + .collect(Collectors.toList()); + DataObject one = dataObjectService.getById(api); + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", api); + List listed = dataFieldService.list(dbQw); + List existsIds = customMapper.getIds(api, ids); + int processedCount = 0; + + for ( Map jsonObject : batchRecords) { + try { + // update + String id = jsonObject.get(Const.ID).toString(); + List> maps = Lists.newArrayList(); + for (DataField field : listed) { + if (StringUtils.isNotEmpty(field.getSfType())) { + // ContentVersion表，插入更新时把title中的/替换为- + if ("ContentVersion".equals(one.getName()) && "Title".equals(field.getField())){ + String title = StringUtils.replace(String.valueOf(jsonObject.get("Title")), "/", "-"); + Map paramMap = Maps.newHashMap(); + paramMap.put("key", "Title"); + paramMap.put("value", title); + maps.add(paramMap); + continue; + } + + // Document,Attachment，插入更新时把Name中的/替换为- + if (("Document".equals(one.getName()) || "Attachment".equals(one.getName())) && "Name".equals(field.getField())){ + String name = StringUtils.replace(String.valueOf(jsonObject.get("Name")), "/", "-"); + Map paramMap = Maps.newHashMap(); + paramMap.put("key", "Name"); + paramMap.put("value", name); + maps.add(paramMap); + continue; + } + + Map paramMap = Maps.newHashMap(); + paramMap.put("key", field.getField()); + paramMap.put("value",DataUtil.convertMysqlValue(field.getSfType(),jsonObject.get(field.getField()))); + maps.add(paramMap); + } + } + // 附件表 插入更新时把is_dump置为false + if (StringUtils.isNotBlank(one.getBlobField())) { + Map paramMap = Maps.newHashMap(); + paramMap.put("key", "is_dump"); + paramMap.put("value", false); + + Map paramMap2 = Maps.newHashMap(); + paramMap2.put("key", "is_upload"); + paramMap2.put("value", false); + maps.add(paramMap); + maps.add(paramMap2); + } + + Map paramMap3 = Maps.newHashMap(); + paramMap3.put("key", "is_update"); + paramMap3.put("value", 0); + maps.add(paramMap3); + + if (existsIds.contains(id)) { + customMapper.updateById(api, maps, id); + } else { + if (insert) { + customMapper.save(api, maps); + } + } + processedCount++; + TimeUnit.MILLISECONDS.sleep(1); + } catch (InterruptedException e) { + throw e; + } catch (Throwable throwable) { + if (throwable.toString().contains("interrupt")) { + log.error("may interrupt error:", throwable); + throw new InterruptedException(); + } + throwable.addSuppressed(new Exception("saveOrUpdate error data:" + JSON.toJSONString(jsonObject))); + throw throwable; + } + } + return processedCount; + } + + /** + * 批次存在 且首次时间为0或者为空 赋值、保存执行记录 + * + * @param param 参数 + * @param dataBatchHistory 执行记录 + */ + private void updateDataBatch(SalesforceParam param, DataBatchHistory dataBatchHistory) { + if (dataBatchHistory == null) { + return; + } + dataBatchHistory.setEndDate(new Date()); + Integer num = customMapper.count(param); + dataBatchHistory.setDbNum(num); + if (dataBatchHistory.getSfNum() != null) { + dataBatchHistory.setSyncStatus(dataBatchHistory.getSfNum().equals(dataBatchHistory.getDbNum()) ? 1 : 0); + } + dataBatchHistory.setCost(DataUtil.calTime(dataBatchHistory.getEndDate(), dataBatchHistory.getStartDate())); + QueryWrapper qw = new QueryWrapper<>(); + qw.eq("name", param.getApi()) + .eq("sync_start_date", param.getBeginCreateDate()) + .eq("sync_end_date", param.getEndCreateDate()) + .isNull("first_sync_date"); + DataBatch dataBatch = dataBatchService.getOne(qw); + if (dataBatch == null) { + return; + } + dataBatch.setFirstSfNum(dataBatchHistory.getSfNum()); + dataBatch.setFirstDbNum(dataBatchHistory.getDbNum()); + dataBatch.setFirstSyncDate(dataBatchHistory.getStartDate()); + dataBatch.setSyncStatus(dataBatchHistory.getSyncStatus()); + dataBatchService.update(dataBatch, qw); + + log.info("count db num: {}", num); + XxlJobLogger.log("count db num: {}", num); + dataBatchHistoryService.save(dataBatchHistory); + log.info("dataDumpJob done api:{}, count:{}", dataBatchHistory.getName(), num); + } + } diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java index fd26f32..9935c95 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java @@ -721,7 +721,10 @@ public class CommonServiceImpl implements CommonService { objects = DataUtil.toJsonArray(records, dsrFields); // 获取最大修改时间和等于该修改时间的数据id { + // 获取当前批次最后一条记录的时间戳，用于下一批次的查询条件 Date maxDate = objects.getJSONObject(objects.size() - 1).getDate(dateName); + // 在当前批次中找出所有具有相同时间戳的记录，并获取其中ID最大的那条记录的ID + // 这个ID将作为下一批次查询的起始位置，避免重复处理相同时间戳的数据 maxId = objects.stream() .map(t -> (JSONObject) t) .filter(t -> maxDate.equals(t.getDate(dateName))) @@ -742,6 +745,7 @@ public class CommonServiceImpl implements CommonService { XxlJobLogger.log("dump success count: {}, timestamp: {}", count, format); failCount = 0; objects = null; + batch ++; } catch (InterruptedException e) { throw e; } catch (Throwable throwable) { @@ -870,7 +874,7 @@ public class CommonServiceImpl implements CommonService { DataObject one = dataObjectService.getById(api); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); - List listed = dataFieldService.list(dbQw); + List list = dataFieldService.list(dbQw).stream().map(DataField::getField).collect(Collectors.toList()); List existsIds = customMapper.getIds(api, ids); for (int i = 0; i < objects.size(); i++) { JSONObject jsonObject = objects.getJSONObject(i); @@ -878,10 +882,10 @@ public class CommonServiceImpl implements CommonService { // update String id = jsonObject.getString(Const.ID); List> maps = Lists.newArrayList(); - for (DataField field : listed) { - if (StringUtils.isNotEmpty(field.getSfType())) { + for (String key : list) { + if (fields.stream().anyMatch(key::equalsIgnoreCase)) { // ContentVersion表，插入更新时把title中的/替换为- - if ("ContentVersion".equals(one.getName()) && "Title".equals(field.getField())){ + if ("ContentVersion".equals(one.getName()) && "Title".equals(key)){ String title = StringUtils.replace(String.valueOf(jsonObject.get("Title")), "/", "-"); Map paramMap = Maps.newHashMap(); paramMap.put("key", "Title"); @@ -891,7 +895,7 @@ public class CommonServiceImpl implements CommonService { } // Document,Attachment，插入更新时把Name中的/替换为- - if (("Document".equals(one.getName()) || "Attachment".equals(one.getName())) && "Name".equals(field.getField())){ + if (("Document".equals(one.getName()) || "Attachment".equals(one.getName())) && "Name".equals(key)){ String name = StringUtils.replace(String.valueOf(jsonObject.get("Name")), "/", "-"); Map paramMap = Maps.newHashMap(); paramMap.put("key", "Name"); @@ -899,30 +903,20 @@ public class CommonServiceImpl implements CommonService { maps.add(paramMap); continue; } - + if ("is_dump".equals( key) || "is_upload".equals( key)){ + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value", false); + maps.add(paramMap); + continue; + } Map paramMap = Maps.newHashMap(); - paramMap.put("key", field.getField()); - paramMap.put("value", jsonObject.get(field.getField())); + paramMap.put("key", key); + paramMap.put("value", jsonObject.get(key)); maps.add(paramMap); } } - // 附件表 插入更新时把is_dump置为false - if (StringUtils.isNotBlank(one.getBlobField())) { - Map paramMap = Maps.newHashMap(); - paramMap.put("key", "is_dump"); - paramMap.put("value", false); - Map paramMap2 = Maps.newHashMap(); - paramMap2.put("key", "is_upload"); - paramMap2.put("value", false); - maps.add(paramMap); - maps.add(paramMap2); - } - - Map paramMap3 = Maps.newHashMap(); - paramMap3.put("key", "is_update"); - paramMap3.put("value", 0); - maps.add(paramMap3); if (existsIds.contains(id)) { customMapper.updateById(api, maps, id); @@ -946,6 +940,7 @@ public class CommonServiceImpl implements CommonService { return existsIds.size(); } + private final ReentrantLock reentrantLock = new ReentrantLock(); /** @@ -957,7 +952,6 @@ public class CommonServiceImpl implements CommonService { @Override public void checkApi(String apiName, Boolean createBatch) throws Exception { // 加个锁 避免重复执行创建api - reentrantLock.lock(); log.info("check api:{}", apiName); QueryWrapper queryWrapper = new QueryWrapper<>(); @@ -1183,8 +1177,288 @@ public class CommonServiceImpl implements CommonService { update.setKeyPrefix(dsr.getKeyPrefix()); dataObjectService.saveOrUpdate(update); } - } finally { - reentrantLock.unlock(); + }catch (Exception e){ + EmailUtil.send("DataDump ERROR", e.getMessage()); + } + } + + /** + * 创建分区表结构 + * @param apiName + * @param createBatch + * @throws Exception + */ + public void checkPartitionedApi(String apiName, Boolean createBatch) throws Exception { + // 加个锁 避免重复执行创建api + log.info("check api:{}", apiName); + + QueryWrapper queryWrapper = new QueryWrapper<>(); + queryWrapper.eq("name",apiName); + long count = metaclassConfigService.count(queryWrapper); + + try { + boolean hasCreatedDate = false; + + PartnerConnection connection = salesforceConnect.createConnect(); + DataObject dataObject = dataObjectService.getById(apiName); + Date now = new Date(); + if (StringUtils.isBlank(customMapper.checkTable(apiName))) { + log.info("api:{} doesn't exist create", apiName); + // 构建字段 + List> list = Lists.newArrayList(); + DescribeSObjectResult dsr = connection.describeSObject(apiName); + String label = dsr.getLabel(); + List fieldList = Lists.newArrayList(); + List fields = Lists.newArrayList(); + String blobField = null; + List dataPicklists = Lists.newArrayList(); + for (Field field : dsr.getFields()) { + // 过滤字段 + if (Const.TABLE_FILTERS.contains(field.getName())) { + continue; + } + if (Const.CREATED_DATE.equalsIgnoreCase(field.getName())) { + hasCreatedDate = true; + } + Map map = Maps.newHashMap(); + String sfType = field.getType().toString(); + if ("base64".equalsIgnoreCase(sfType)) { + blobField = field.getName(); + } + map.put("type", DataUtil.fieldTypeToMysql(field)); + // 英文' 转换为 \' + map.put("comment", field.getLabel().replaceAll("'", "\\\\'")); + map.put("name", field.getName()); + list.add(map); + DataField dataField = new DataField(); + dataField.setApi(apiName); + dataField.setSfType(sfType); + dataField.setField(field.getName()); + dataField.setName(field.getLabel()); + dataField.setIsCreateable(field.getCreateable()); + dataField.setIsUpdateable(field.getUpdateable()); + dataField.setIsNillable(field.getNillable()); + dataField.setIsDefaultedOnCreate(field.getDefaultedOnCreate()); + String join = null; + // 会有非常多映射关系的字段在 这里置空不管 + if (field.getReferenceTo().length <= 3) { + join = StringUtils.join(field.getReferenceTo(), ","); + } else { + log.info("referenceTo too long set null, api:{}, field:{}, reference to:{}", apiName, field.getName(), StringUtils.join(field.getReferenceTo(), ",")); + } + if (field.getReferenceTo().length > 1){ + if (field.getName().contains("Id") && !"OwnerId".equals(field.getName()) && !"UserOrGroupId".equals(field.getName())){ + LinkConfig linkConfig = new LinkConfig(); + linkConfig.setApi(apiName); + linkConfig.setLabel(label); + linkConfig.setField(field.getName()); + linkConfig.setLinkField(StringUtils.replace(field.getName(), "Id", "_Type")); + linkConfigService.save(linkConfig); + } + } + // picklist保存到picklist表 + if ("picklist".equalsIgnoreCase(sfType)) { + join = "data_picklist"; + PicklistEntry[] picklistValues = field.getPicklistValues(); + if (ArrayUtils.isNotEmpty(picklistValues)) { + for (PicklistEntry picklistValue : picklistValues) { + DataPicklist dataPicklist = new DataPicklist(); + dataPicklist.setApi(apiName); + dataPicklist.setField(field.getName()); + dataPicklist.setLabel(picklistValue.getLabel()); + dataPicklist.setValue(picklistValue.getValue()); + dataPicklists.add(dataPicklist); + } + } + } + dataField.setReferenceTo(join); + fieldList.add(dataField); + fields.add(field.getName()); + } + // 存在ownerId字段 + String extraField = ""; + if (dataObject != null && StringUtils.isNotBlank(dataObject.getExtraField())) { + extraField = dataObject.getExtraField(); + } + if (fields.stream().anyMatch(Const.OWNER_ID::equalsIgnoreCase)) { + if (StringUtils.isNotBlank(extraField)) { + extraField += ",Owner.Type"; + } else { + extraField = "Owner.Type"; + } + } + // 是否存在额外字段 配置上表 + if (StringUtils.isNotBlank(extraField)) { + Set extras = Arrays.stream(StringUtils.split(extraField.replaceAll(StringUtils.SPACE, StringUtils.EMPTY), ",")).collect(Collectors.toSet()); + for (String extra : extras) { + String fieldName = extra.replaceAll("\\.", "_"); + DataField dataField = new DataField(); + dataField.setApi(apiName); + dataField.setField(fieldName); + dataField.setName(extra); + fieldList.add(dataField); + fields.add(fieldName); + Map map = Maps.newHashMap(); + map.put("type", "varchar(255)"); + map.put("comment", extra); + map.put("name", fieldName); + list.add(map); + } + } + // 存在附件的表 作特殊处理 构建两个字段 + if (StringUtils.isNotBlank(blobField)) { + fileExtraFieldBuild(apiName, list, fieldList, fields); + } + log.info("api {} not exist, create..", apiName); + //新增一个用来存储新sfid的字段 + Map map = Maps.newHashMap(); + map.put("type", "varchar(255)"); + map.put("comment", "新sfid"); + map.put("name", "new_id"); + list.add(map); + + fields.add("new_id"); + + // 构建索引 + List> index = Lists.newArrayList(); + for (String tableIndex : Const.TABLE_INDEX) { + if (!fields.contains(tableIndex)) { + continue; + } + Map createDateMap = Maps.newHashMap(); + createDateMap.put("field", tableIndex); + createDateMap.put("name", String.format("IDX_%s_%s", apiName, tableIndex)); + index.add(createDateMap); + } + + // 构建分区结构，按object上的时间字段构建分区 + List> partitions = new ArrayList<>(); + + // 查找对象上的时间字段作为分区键 + String partitionKey = "CreatedDate"; // 默认使用创建时间字段 + for (Field field : dsr.getFields()) { + if ("datetime".equalsIgnoreCase(field.getType().toString()) || + "date".equalsIgnoreCase(field.getType().toString())) { + partitionKey = field.getName(); + break; + } + } + + // 添加默认分区，按照日期进行分区 + Date defaultBeginDate = DataUtil.DEFAULT_BEGIN_DATE; + Calendar calendar = Calendar.getInstance(); + int currentYear = calendar.get(Calendar.YEAR); + + Calendar startCalendar = Calendar.getInstance(); + startCalendar.setTime(defaultBeginDate); + int startYear = startCalendar.get(Calendar.YEAR); + + // 创建从开始年份到当前年份+1的分区，每个年份一个分区 + for (int year = startYear; year <= currentYear + 1; year++) { + Map partition = new HashMap<>(); + partition.put("name", "p" + year); + partition.put("value", String.format("'%s-01-01 00:00:00'", year + 1)); // datetime类型的值 + partitions.add(partition); + } + + // 添加一个未来分区 + Map futurePartition = new HashMap<>(); + futurePartition.put("name", "p" + (currentYear + 2)); + futurePartition.put("value", "'9999-12-31 23:59:59'"); + partitions.add(futurePartition); + + //DeleteEvent 回收站新增是否删除，数据内同字段 + if ("DeleteEvent".equals(apiName)){ + Map map1 = Maps.newHashMap(); + map1.put("type", "tinyint(1) DEFAULT 0"); + map1.put("comment", "是否删除"); + map1.put("name", "is_delete"); + list.add(map1); + + Map map2 = Maps.newHashMap(); + map2.put("type", "text"); + map2.put("comment", "完整数据"); + map2.put("name", "all_data"); + list.add(map2); + } + //是否更新 + Map map3 = Maps.newHashMap(); + map3.put("type", "tinyint(1) DEFAULT 0"); + map3.put("comment", "是否更新"); + map3.put("name", "is_update"); + list.add(map3); + + //是否更新 + Map map4 = Maps.newHashMap(); + map4.put("type", "text"); + map4.put("comment", "更新错误信息"); + map4.put("name", "error_message"); + list.add(map4); + // 创建RANGE分区表，按年份分区 + customMapper.createRangePartitionTable( + apiName, + label, + list, + index, + partitionKey, // 使用动态查找的分区键 + partitions + ); + // 生成字段映射 + QueryWrapper delfieldQw = new QueryWrapper<>(); + delfieldQw.eq("api", apiName); + dataFieldService.remove(delfieldQw); + dataFieldService.saveBatch(fieldList); + // 生成picklist + QueryWrapper deldQw = new QueryWrapper<>(); + deldQw.eq("api", apiName); + dataPicklistService.remove(deldQw); + dataPicklistService.saveBatch(dataPicklists); + // 生成batch + Date startDate = DataUtil.DEFAULT_BEGIN_DATE; + // 取当天零点 作为最大时间 + Date endCreateDate = DateUtils.parseDate(DateFormatUtils.format(now, "yyyy-MM-dd"), "yyyy-MM-dd"); + Date lastDay = null; + QueryWrapper delQw = new QueryWrapper<>(); + delQw.eq("name", apiName); + dataBatchService.remove(delQw); + if (createBatch && Const.BATCH_FILTERS.stream().noneMatch(t -> apiName.indexOf(t) > 0)) { + // 匹配的不生成 + DataBatch one = new DataBatch(); + one.setFirstDbNum(0); + one.setFirstSfNum(0); + one.setDbNum(0); + one.setSfNum(0); + one.setName(apiName); + one.setLabel(label); + if (hasCreatedDate) { + List> bachList = customMapper.list("code,value","system_config","code ='"+BATCH_TYPE+"'"); + String batchType = bachList.get(0).get("value").toString(); + do { + lastDay = getLastDay(batchType, endCreateDate, startDate); + DataBatch dataBatch = one.clone(); + dataBatch.setSyncStartDate(startDate); + dataBatch.setSyncEndDate(lastDay); + dataBatchService.save(dataBatch); + startDate = lastDay; + } while (lastDay.compareTo(endCreateDate) < 0); + } else { + // 没创建时间 开始结束时间为空 + dataBatchService.save(one); + } + } + DataObject update = new DataObject(); + update.setLabel(label); + update.setName(apiName); + update.setLastUpdateDate(endCreateDate); + update.setBlobField(blobField); + if (count>0 || apiName.endsWith("Share")){ + update.setIsEditable(false); + } + update.setKeyPrefix(dsr.getKeyPrefix()); + dataObjectService.saveOrUpdate(update); + } + }catch (Exception e){ + EmailUtil.send("DataDump ERROR", e.getMessage()); } } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java index 2d0cb96..770e271 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java @@ -9,6 +9,7 @@ import com.celnet.datadump.config.SalesforceExecutor; import com.celnet.datadump.entity.DataBatch; import com.celnet.datadump.entity.DataLog; import com.celnet.datadump.entity.DataObject; +import com.celnet.datadump.entity.DataPicklist; import com.celnet.datadump.global.Const; import com.celnet.datadump.param.DataDumpSpecialParam; import com.celnet.datadump.param.SalesforceParam; @@ -34,10 +35,7 @@ import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; -import java.util.Date; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -66,6 +64,8 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService { private SalesforceExecutor salesforceExecutor; @Autowired private DataLogService dataLogService; + @Autowired + private DataPicklistServiceImpl dataPicklistServiceImpl; @Override public ReturnT dataDumpSpecial(DataDumpSpecialParam param) throws Throwable { 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 c2462e2..1d54334 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -410,12 +410,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } } - account.put("old_owner_id__c", data.get(j-1).get("OwnerId")); - account.put("old_sfdc_id__c", data.get(j-1).get("Id")); + if (dataObject.getIsEditable()){ + account.put("old_owner_id__c", 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*5000+j == count){ + if (i*5000+j >= count){ break; } } 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 88f3f12..70146ca 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -809,6 +809,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { List fieldsToNull = new ArrayList<>(); SObject account = new SObject(); account.setType(api); + String message = null; //给对象赋值 for (DataField dataField : list) { String field = dataField.getField(); @@ -843,8 +844,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (m != null && !m.isEmpty()) { account.setField(field, m.get("new_id")); }else { - String message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; - EmailUtil.send("DataDump ERROR", message); + message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; log.info(message); break; } @@ -859,6 +859,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { } } } + if (message != null){ + continue; + } if (!fieldsToNull.isEmpty()){ String[] fieldsToNullArray = fieldsToNull.toArray(new String[0]); @@ -888,13 +891,11 @@ public class DataImportNewServiceImpl implements DataImportNewService { maps.add(linkMap); Map linkMap1 = new HashMap<>(); linkMap1.put("key", "error_message"); - linkMap1.put("value", saveResult.getErrors()); + linkMap1.put("value", JSON.toJSONString(saveResult.getErrors())); maps.add(linkMap1); customMapper.updateByNewId(maps,api, saveResult.getId()); - Map map = returnErrorAccountsDetails(accounts, list, saveResult.getId()); - 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),JSON.toJSONString(map)); + String format = String.format("数据更新 error, api name: %s, \ncause:\n%s", api, JSON.toJSONString(saveResult)); log.info(format); - return; }else { List> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); @@ -949,32 +950,15 @@ public class DataImportNewServiceImpl implements DataImportNewService { String beginDateStr = null; String endDateStr = null; + String updateDateField = dataFieldService.returnUpdateDateField(api); if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and new_id is null "; - sql2 = "RowCause = 'Manual' and new_id is null limit 10000"; + sql = "where RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; }else { - sql = "where new_id is null "; - sql2 = "new_id is null limit 10000"; + 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 "; } - if (1 == 1) { - if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; - sql2 = "RowCause = 'Manual' and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; - }else { - sql = "where new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; - sql2 = "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; - } - }else { - String updateDateField = dataFieldService.returnUpdateDateField(api); - if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; - sql2 = "RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; - }else { - sql = "where new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; - sql2 = "new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; - } - } //表内数据总量 Integer count = customMapper.countBySQL(api, sql); @@ -997,8 +981,6 @@ public class DataImportNewServiceImpl implements DataImportNewService { } - int targetCount = 0; - int num = count%10000 == 0 ? count/10000 : (count/10000) + 1; log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api); @@ -1089,12 +1071,25 @@ public class DataImportNewServiceImpl implements DataImportNewService { SaveResult[] saveResults = partnerConnection.update(accounts); for (SaveResult saveResult : saveResults) { if (!saveResult.getSuccess()) { - Map map = returnErrorAccountsDetails(accounts, list, saveResult.getId()); - String format = String.format("数据更新 error, api name: %s, \n cause:\n%s, \n数据实体类:\n%s", api, com.alibaba.fastjson2.JSON.toJSONString(DataDumpParam.getFilter()), JSON.toJSONString(saveResult),JSON.toJSONString(map)); - EmailUtil.send("DataDump ERROR", format); - return; + List> maps = new ArrayList<>(); + Map linkMap = new HashMap<>(); + linkMap.put("key", "is_update"); + linkMap.put("value", 2); + maps.add(linkMap); + Map 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 { - targetCount ++; + List> maps = new ArrayList<>(); + Map linkMap = new HashMap<>(); + linkMap.put("key", "is_update"); + linkMap.put("value", 1); + maps.add(linkMap); + customMapper.updateByNewId(maps,api, saveResult.getId()); } } @@ -2976,6 +2971,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { 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(); @@ -3008,7 +3004,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (m != null && !m.isEmpty()) { account.setField(field, m.get("new_id")); }else { - String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; + message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; log.info(message); break; } @@ -3021,6 +3017,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { } } } + if (message!=null){ + 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")); @@ -3057,6 +3056,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { customMapper.updateById(api, maps, ids[index]); targetCount ++; }else{ + List> maps = new ArrayList<>(); + Map 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++; @@ -3195,6 +3200,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { 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(); @@ -3227,7 +3233,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (m != null && !m.isEmpty()) { account.setField(field, m.get("new_id")); }else { - String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; + message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; log.info(message); break; } @@ -3240,6 +3246,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { } } } + if (message != null){ + 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")); diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java index 0cee5a7..2a33678 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java @@ -325,6 +325,7 @@ public class DataImportServiceImpl implements DataImportService { try { for (int j = 1; j <= sized; j++) { + String message = null; SObject account = new SObject(); account.setType(api); //找出sf对象必填字段，并且给默认值 @@ -351,8 +352,11 @@ public class DataImportServiceImpl implements DataImportService { Map CreatedByIdMap = customMapper.getById("new_id", "User", mapList.get(j-1).get("CreatedById").toString()); if(CreatedByIdMap != null ){ account.setField("CreatedById", CreatedByIdMap.get("new_id")); + }else { + message = "CreatedById:" + mapList.get(j-1).get("CreatedById") + "不存在"; + log.info( message); + break; } - continue; } if (dataField.getIsCreateable() != null && dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) { @@ -365,11 +369,12 @@ public class DataImportServiceImpl implements DataImportService { if (mapList.get(j - 1).get(linkfield)!=null){ reference = mapList.get(j - 1).get(linkfield).toString(); }else { - log.info("对象类型:" + api + "的数据:"+ mapList.get(j - 1).get("Id") +"的关联类型不存在！"); + message ="对象类型:" + api + "的数据:"+ mapList.get(j - 1).get("Id") +"的关联类型不存在！"; + log.info(message); } } if (reference == null){ - continue; + break; } List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); @@ -396,6 +401,10 @@ public class DataImportServiceImpl implements DataImportService { } } + if (message != null){ + continue; + } + if (dataObject.getIsEditable()){ account.setField("old_owner_id__c", mapList.get(j - 1).get("OwnerId")); account.setField("old_sfdc_id__c", mapList.get(j - 1).get("Id")); @@ -432,7 +441,13 @@ public class DataImportServiceImpl implements DataImportService { maps.add(m); customMapper.updateById(api, maps, ids[index]); }else{ - log.error("-------------saveResults: {}", JSON.toJSONString(saveResult)); + List> maps = new ArrayList<>(); + Map 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++; } diff --git a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java index 2bd3e02..a3deac7 100644 --- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java @@ -42,6 +42,7 @@ import org.apache.http.util.EntityUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import springfox.documentation.spring.web.json.Json; import java.io.*; import java.net.URLEncoder; @@ -558,6 +559,7 @@ public class FileServiceImpl implements FileService { break; } } else { + log.info("文件上传错误，返回：" + JSON.toJSONString(response)); throw new RuntimeException(); } } diff --git a/src/main/java/com/celnet/datadump/util/BulkUtil.java b/src/main/java/com/celnet/datadump/util/BulkUtil.java index a7a1bc4..c955b78 100644 --- a/src/main/java/com/celnet/datadump/util/BulkUtil.java +++ b/src/main/java/com/celnet/datadump/util/BulkUtil.java @@ -1,21 +1,47 @@ package com.celnet.datadump.util; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; import com.sforce.async.*; import com.sforce.soap.partner.PartnerConnection; import com.sforce.ws.ConnectionException; import com.sforce.ws.ConnectorConfig; +import com.sforce.ws.bind.CalendarCodec; +import org.apache.http.Header; +import org.apache.http.HttpEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; + import java.io.*; + import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Paths; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Set; +import java.util.*; + +import static com.sforce.async.BulkConnection.JSON_CONTENT_TYPE; public class BulkUtil { + private static final String URI_STEM_QUERY = "query/"; + private static final String URI_STEM_INGEST = "ingest/"; + private static final String AUTH_HEADER = "Authorization"; + private static final String REQUEST_CONTENT_TYPE_HEADER = "Content-Type"; + private static final String ACCEPT_CONTENT_TYPES_HEADER = "ACCEPT"; + private static final String AUTH_HEADER_VALUE_PREFIX = "Bearer "; + private static final String UTF_8 = StandardCharsets.UTF_8.name(); + private static final String INGEST_RESULTS_SUCCESSFUL = "successfulResults"; + private static final String INGEST_RESULTS_UNSUCCESSFUL = "failedResults"; + private static final String INGEST_RECORDS_UNPROCESSED = "unprocessedrecords"; + public static final String SFORCE_CALL_OPTIONS_HEADER = "Sforce-Call-Options"; + + public static void closeJob(BulkConnection connection, String jobId) throws AsyncApiException { JobInfo job = new JobInfo(); @@ -24,6 +50,33 @@ public class BulkUtil { connection.updateJob(job); } + public static int awaitCompletionOne(BulkConnection connection, JobInfo job, + BatchInfo bi) + throws AsyncApiException { + long sleepTime = 0L; + int numberRecordsProcessed = 0; + Set incomplete = new HashSet(); + incomplete.add(bi.getId()); + while (!incomplete.isEmpty()) { + try { + Thread.sleep(sleepTime); + } catch (InterruptedException e) {} + System.out.println("Awaiting results..." + incomplete.size()); + sleepTime = 10000L; + BatchInfo[] statusList = + connection.getBatchInfoList(job.getId()).getBatchInfo(); + for (BatchInfo b : statusList) { + if (b.getState() == BatchStateEnum.Completed + || b.getState() == BatchStateEnum.Failed) { + if (incomplete.remove(b.getId())) { + System.out.println("BATCH STATUS:\n" + b); + } + } + numberRecordsProcessed += b.getNumberRecordsProcessed(); + } + } + return numberRecordsProcessed; + } /** * Wait for a job to complete by polling the Bulk API. @@ -32,7 +85,6 @@ public class BulkUtil { * BulkConnection used to check results. * @param job * The job awaiting completion. - * @param batchInfoList * List of batches for this job. * @throws AsyncApiException */ @@ -83,6 +135,123 @@ public class BulkUtil { return job; } + /** + * 创建Bulk API 2.0查询作业 + ** @param soqlQuery SOQL查询语句 + * @return 作业ID + * @throws Exception 创建作业时发生错误 + */ + public static String createBulkV2Job(String url,String soqlQuery,BulkConnection connection) throws Exception { + // 构建创建查询作业的 JSON 请求体 + String jsonRequest = String.format("{\"operation\":\"query\", \"query\":\"%s\"}", soqlQuery.replace("\"", "\\\"")); + + CloseableHttpClient httpClient = HttpClients.createDefault(); + HttpPost httpPost = new HttpPost(url + "services/data/v56.0/jobs/query"); + httpPost.setHeader("Content-Type", "application/json"); + httpPost.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + httpPost.setEntity(new StringEntity(jsonRequest, "UTF-8")); + + CloseableHttpResponse response = httpClient.execute(httpPost); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + // 从响应体中解析出 jobId + String responseBody = EntityUtils.toString(response.getEntity()); + // 简单的JSON解析提取id，实际项目中建议使用JSON库如Jackson + String jobId = responseBody.split("\"id\":\"")[1].split("\"")[0]; + System.out.println("Bulk API 2.0 Query job created with ID: " + jobId); + return jobId; + } else { + String errorBody = EntityUtils.toString(response.getEntity()); + throw new RuntimeException("Failed to create Bulk API 2.0 query job, status code: " + statusCode + ", error: " + errorBody); + } + } finally { + response.close(); + httpClient.close(); + } + } + + /** + * Wait for a Bulk API 2.0 job to complete by polling the job status. + * + * @param jobId The ID of the job to check. + * @param connection BulkConnection used to get session info. + * @throws Exception If an error occurs during the polling process. + */ + public static void waitForBulkV2JobCompletion(String url, String jobId, BulkConnection connection) throws Exception { + CloseableHttpClient httpClient = HttpClients.createDefault(); + String jobStatusUrl = url + "services/data/v56.0/jobs/query/" + jobId; + boolean jobCompleted = false; + + while (!jobCompleted) { + HttpGet httpGet = new HttpGet(jobStatusUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + String responseBody = EntityUtils.toString(response.getEntity()); + // 简单解析响应体中的状态字段，实际项目中建议使用JSON库如Jackson + if (responseBody.contains("\"state\":\"JobComplete\"")) { + System.out.println("Job completed successfully."); + jobCompleted = true; + } else if (responseBody.contains("\"state\":\"Failed\"") || responseBody.contains("\"state\":\"Aborted\"")) { + throw new RuntimeException("Job failed or was aborted. Response: " + responseBody); + } else { + System.out.println("Job is still running. Waiting..."); + Thread.sleep(10000); // 等待10秒后再次检查 + } + } else { + String errorBody = EntityUtils.toString(response.getEntity()); + throw new RuntimeException("Failed to get job status, status code: " + statusCode + ", error: " + errorBody); + } + } finally { + response.close(); + } + } + httpClient.close(); + } + + /** + * 获取 Bulk API 2.0 查询作业的结果 + * + * @param url Salesforce实例URL + * @param jobId 作业ID + * @param connection BulkConnection used to get session info. + * @return 查询结果的字符串内容 + * @throws Exception 如果在获取结果过程中发生错误 + */ + public static String getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection) throws Exception { + CloseableHttpClient httpClient = HttpClients.createDefault(); + String resultsUrl = url + "/services/data/v65.0/jobs/query/" + jobId + "/results" ; + HttpPost httpGet = new HttpPost(resultsUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + httpGet.setHeader("Content-Type", "application/json"); + httpGet.setHeader("Accept", "text/csv"); + httpGet.setHeader("X-PrettyPrint", "1"); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + + Header locatorHeader = response.getFirstHeader("Sforce-Locator"); + Header numRecordsHeader = response.getFirstHeader("Sforce-NumberOfRecords"); + String newLocator = (locatorHeader != null) ? locatorHeader.getValue() : null; + int numberOfRecords = (numRecordsHeader != null) ? Integer.parseInt(numRecordsHeader.getValue()) : 0; + + return EntityUtils.toString(response.getEntity()); + } else { + String errorBody = EntityUtils.toString(response.getEntity()); + throw new RuntimeException("Failed to get query job results, status code: " + statusCode + ", error: " + errorBody); + } + } finally { + response.close(); + httpClient.close(); + } + } + /** @@ -118,6 +287,102 @@ public class BulkUtil { return connection; } + + + public static JobInfo createQueryJob(String api, String soqlQuery,BulkConnection connection) throws Exception { + JobInfo job = new JobInfo(); + job.setOperation(OperationEnum.queryAll); + ContentType jobContentType = ContentType.CSV; + job.setContentType(jobContentType); + ConcurrencyMode jobConcurrencyMode = ConcurrencyMode.Parallel; + job.setConcurrencyMode(jobConcurrencyMode); + job.setObject(soqlQuery); + + String endpoint = connection.getConfig().getRestEndpoint() + "job/"; + HashMap headers = getHeaders(JSON_CONTENT_TYPE, JSON_CONTENT_TYPE, connection.getConfig().getSessionId()); + + HashMap requestBodyMap = new HashMap(); + requestBodyMap.put("operation", job.getOperation().toString()); + requestBodyMap.put("object", job.getObject()); + requestBodyMap.put("contentType", job.getContentType().toString()); + requestBodyMap.put("lineEnding", "LF"); + requestBodyMap.put("concurrencyMode", job.getConcurrencyMode().toString()); + + HttpTransportImpl transport = HttpTransportImpl.getInstance(); + OutputStream out = transport.connect(endpoint, headers, true, HttpTransportInterface.SupportedHttpMethodType.POST); + + ObjectMapper mapper = new ObjectMapper(); + mapper.setDateFormat(CalendarCodec.getDateFormat()); + String requestContent = mapper.writeValueAsString(requestBodyMap); + out.write(requestContent.getBytes(UTF_8)); + out.close(); + + InputStream in = transport.getContent(); + boolean successfulRequest = transport.isSuccessful(); + + if (!successfulRequest) { + throw new AsyncApiException("Failed to create query job", AsyncExceptionCode.Unknown); + } + + ObjectMapper responseMapper = new ObjectMapper(); + responseMapper.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true); + responseMapper.setTimeZone(TimeZone.getTimeZone("GMT")); + return responseMapper.readValue(in, JobInfo.class); + } + + public static JobInfo createQueryJob1(String api, String soqlQuery,BulkConnection connection) throws Exception { + JobInfo job = new JobInfo(); + InputStream in = null; + job.setOperation(OperationEnum.queryAll); + ContentType jobContentType = ContentType.CSV; + job.setContentType(jobContentType); + ConcurrencyMode jobConcurrencyMode = ConcurrencyMode.Parallel; + job.setConcurrencyMode(jobConcurrencyMode); + job.setObject(soqlQuery); + String endpoint =connection.getConfig().getRestEndpoint()+ "job/"; + String requestURL = constructRequestURL(endpoint, job.getId()); + HashMap headers = null; + headers = getHeaders(JSON_CONTENT_TYPE, JSON_CONTENT_TYPE,connection.getConfig().getSessionId()); + HashMap requestBodyMap = new HashMap(); + ContentType type = job.getContentType(); + requestBodyMap.put("object", job.getObject()); + requestBodyMap.put("contentType", type.toString()); + requestBodyMap.put("lineEnding", "LF"); + OutputStream out; + HttpTransportImpl transport = HttpTransportImpl.getInstance(); + out = transport.connect(requestURL, headers, true, HttpTransportInterface.SupportedHttpMethodType.POST); + ObjectMapper mapper = new ObjectMapper(); + mapper.setDateFormat(CalendarCodec.getDateFormat()); + String requestContent = mapper.writeValueAsString(requestBodyMap); + out.write(requestContent.getBytes(UTF_8)); + out.close(); + in = transport.getContent(); + boolean successfulRequest = true; + successfulRequest = transport.isSuccessful(); + JobInfo result = null; + ObjectMapper mapper1 = new ObjectMapper(); + mapper1.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true); + // By default, ObjectMapper generates Calendar instances with UTC TimeZone. + // Here, override that to "GMT" to better match the behavior of the WSC XML parser. + mapper1.setTimeZone(TimeZone.getTimeZone("GMT")); + return mapper1.readValue(in, JobInfo.class); + } + + + public static HashMap getHeaders(String requestContentType, String acceptContentType,String seesionId) { + HashMap newMap = new HashMap(); + String authHeaderValue = AUTH_HEADER_VALUE_PREFIX + seesionId; + newMap.put(REQUEST_CONTENT_TYPE_HEADER, requestContentType); + newMap.put(ACCEPT_CONTENT_TYPES_HEADER, acceptContentType); + newMap.put(AUTH_HEADER, authHeaderValue); + newMap.put(SFORCE_CALL_OPTIONS_HEADER, "client=data-dump, batchSize=5000"); + return newMap; + } + + private static String constructRequestURL(String urlString,String jobId) { + urlString += "query/" + jobId + "/"; + return urlString; + } /** * Create and upload batches using a CSV file. * The file into the appropriate size batch files. diff --git a/src/main/java/com/celnet/datadump/util/DataUtil.java b/src/main/java/com/celnet/datadump/util/DataUtil.java index bdcae02..57bd415 100644 --- a/src/main/java/com/celnet/datadump/util/DataUtil.java +++ b/src/main/java/com/celnet/datadump/util/DataUtil.java @@ -11,6 +11,7 @@ import com.celnet.datadump.param.SalesforceParam; import com.celnet.datadump.global.SystemConfigCode; import com.celnet.datadump.service.SystemConfigService; import com.google.common.collect.Lists; +import com.sforce.async.CSVReader; import com.sforce.soap.partner.Field; import com.sforce.soap.partner.sobject.SObject; import com.sforce.ws.bind.XmlObject; @@ -28,7 +29,10 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; +import java.io.IOException; +import java.io.InputStream; import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; import java.text.ParseException; import java.text.SimpleDateFormat; import java.time.*; @@ -102,6 +106,9 @@ public class DataUtil { case "currency": case "percent": int scale = field.getScale(); + if (scale < 4) { + scale = 4; + } result = "double(" + 18 + "," + scale + ")"; break; case "base64": @@ -370,6 +377,48 @@ public class DataUtil { return jsonObject; } + /** + * 根据字段类型将值转换为MySQL类型并返回 + * + * @param type 字段类型 + * @param value 字段值 + * @return 转换后的值 + * @throws ParseException 日期解析异常 + */ + public static Object convertMysqlValue( String type, Object value) throws ParseException { + if (value == null) { + return null; + } + switch (type) { + case "date": + return DateUtils.parseDate((String) value, "yyyy-MM-dd"); + case "datetime": + return SF_DATE_FORMAT.parse((String) value); + case "time": + return SF_TIME_FORMAT.parse((String) value); + case "boolean": + return BooleanUtils.toBoolean((String) value); + case "long": + case "int": + return Integer.parseInt((String) value); + case "double": + case "currency": + case "percent": + return Double.parseDouble((String) value); + case "base64": + // 文件不处理 统一存oss或者server + return null; + default: + if (value instanceof GregorianCalendar) { + return DateFormatUtils.format(((GregorianCalendar) value), "yyyy-MM-dd HH:mm:ss"); + } else if (value instanceof Time) { + return DateFormatUtils.format((((Time) value).getTimeInMillis()), "HH:mm:ss"); + } else { + return String.valueOf(value); + } + } + } + /** * sObject 数组转 JSONArray @@ -537,4 +586,26 @@ public class DataUtil { } } + public static JSONArray toJsonObjectArray(InputStream resultStream) throws IOException { + JSONArray jsonArray = new JSONArray(); + + try{ + CSVReader reader = new CSVReader(resultStream); + CSVReader rdr = new CSVReader(resultStream); + List headers = rdr.nextRecord(); + + List record; + while ((record = reader.nextRecord()) != null) { + JSONObject jsonObject = new JSONObject(); + for (int i = 0; i < headers.size() && i < record.size(); i++) { + jsonObject.put(headers.get(i), record.get(i)); + } + jsonArray.add(jsonObject); + } + }catch (Exception e){ + log.error("exception message", e); + } + + return jsonArray; + } } diff --git a/src/main/java/com/celnet/datadump/util/HttpTransportImpl.java b/src/main/java/com/celnet/datadump/util/HttpTransportImpl.java new file mode 100644 index 0000000..33aa010 --- /dev/null +++ b/src/main/java/com/celnet/datadump/util/HttpTransportImpl.java @@ -0,0 +1,429 @@ +package com.celnet.datadump.util; + +import com.sforce.async.AsyncApiException; +import com.sforce.ws.ConnectorConfig; +import com.sforce.ws.tools.VersionInfo; +import com.sforce.ws.transport.LimitingOutputStream; +import com.sforce.ws.transport.MessageHandlerOutputStream; +import org.apache.commons.io.IOUtils; +import org.apache.http.Header; +import org.apache.http.HttpEntity; +import org.apache.http.HttpHost; +import org.apache.http.HttpResponse; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.Credentials; +import org.apache.http.auth.NTCredentials; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.entity.UrlEncodedFormEntity; +import org.apache.http.client.methods.*; +import org.apache.http.client.protocol.HttpClientContext; +import org.apache.http.entity.BufferedHttpEntity; +import org.apache.http.entity.ByteArrayEntity; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.InputStreamEntity; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.client.ProxyAuthenticationStrategy; +import org.apache.http.message.BasicNameValuePair; +import org.apache.logging.log4j.Logger; + +import java.io.*; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.URL; +import java.net.UnknownHostException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +/** + * This class implements the Transport interface for WSC with HttpClient in order to properly work + * with NTLM proxies. The existing JdkHttpTransport in WSC does not work with NTLM proxies when + * compiled on Java 1.6 + * + * @author Jeff Lai + * @since 25.0.2 + */ +public class HttpTransportImpl implements HttpTransportInterface { + + + private static final String AUTH_HEADER_VALUE_PREFIX = "Bearer "; + private static final String AUTH_HEADER_FOR_JSON = "Authorization"; + private static final String AUTH_HEADER_FOR_XML = "X-SFDC-Session"; + private static final String USER_AGENT_HEADER = "User-Agent"; + + private static ConnectorConfig currentConnectorConfig = null; + private boolean successful; + private HttpRequestBase httpMethod = null; + private OutputStream output; + private ByteArrayOutputStream entityByteOut; + private static CloseableHttpClient currentHttpClient = null; + private static long serverInvocationCount = 0; + private HttpResponse httpResponse; + private static final HttpTransportImpl singletonTransportInstance = new HttpTransportImpl(); + + @Override + public synchronized void setConfig(ConnectorConfig newConfig) { + if (!canReuseHttpClient(currentConnectorConfig, newConfig) + && currentHttpClient != null) { + closeHttpClient(); + } + currentConnectorConfig = newConfig; + if (currentConnectorConfig != null && currentHttpClient == null) { + try { + initializeHttpClient(); + } catch (UnknownHostException e) { + } + } + } + + private boolean canReuseHttpClient(ConnectorConfig config1, ConnectorConfig config2) { + + if (config1 == config2) { + return true; + } else if (config1 == null || config2 == null) { + // one of the configs is null, other isn't. They can't be equal. + return false; + } + + InetSocketAddress proxy1Address = (InetSocketAddress)config1.getProxy().address(); + InetSocketAddress proxy2Address = (InetSocketAddress)config2.getProxy().address(); + + if ((proxy1Address == null && proxy2Address != null) + || (proxy1Address != null && proxy2Address == null)) { + return false; + } else if (proxy1Address != null && proxy2Address != null) { + String field1, field2; + field1 = config1.getProxyUsername() == null ? "" : config1.getProxyUsername(); + field2 = config2.getProxyUsername() == null ? "" : config2.getProxyUsername(); + if (field1.compareTo(field2) != 0) { + return false; + } + + field1 = config1.getProxyPassword() == null ? "" : config1.getProxyPassword(); + field2 = config2.getProxyPassword() == null ? "" : config2.getProxyPassword(); + if (field1.compareTo(field2) != 0) { + return false; + } + + field1 = config1.getNtlmDomain() == null ? "" : config1.getNtlmDomain(); + field2 = config2.getNtlmDomain() == null ? "" : config2.getNtlmDomain(); + if (field1.compareTo(field2) != 0) { + return false; + } + + field1 = proxy1Address.getHostName() == null ? "" : proxy1Address.getHostName(); + field2 = proxy2Address.getHostName() == null ? "" : proxy2Address.getHostName(); + if (field1.compareTo(field2) != 0) { + return false; + } + + int intField1 = proxy1Address.getPort(); + int intField2 = proxy2Address.getPort(); + if (intField1 != intField2) { + return false; + } + } + return true; + } + + private synchronized void initializeHttpClient() throws UnknownHostException { + closeHttpClient(); + httpMethod = null; + HttpClientBuilder httpClientBuilder = HttpClientBuilder.create(); + httpClientBuilder = httpClientBuilder.useSystemProperties(); + + + if (currentConnectorConfig != null + && currentConnectorConfig.getProxy() != null + && currentConnectorConfig.getProxy().address() != null) { + String proxyUser = currentConnectorConfig.getProxyUsername() == null ? "" : currentConnectorConfig.getProxyUsername(); + String proxyPassword = currentConnectorConfig.getProxyPassword() == null ? "" : currentConnectorConfig.getProxyPassword(); + + InetSocketAddress proxyAddress = (InetSocketAddress) currentConnectorConfig.getProxy().address(); + HttpHost proxyHost = new HttpHost(proxyAddress.getHostName(), proxyAddress.getPort(), "http"); + httpClientBuilder.setProxy(proxyHost); + + CredentialsProvider credentialsprovider = new BasicCredentialsProvider(); + AuthScope scope = new AuthScope(proxyAddress.getHostName(), proxyAddress.getPort(), null, null); + + Credentials credentials; + credentials = new UsernamePasswordCredentials(proxyUser, proxyPassword); + + // based on answer at https://stackoverflow.com/questions/6962047/apache-httpclient-4-1-proxy-authentication + credentialsprovider.setCredentials(scope, credentials); + httpClientBuilder.setDefaultCredentialsProvider(credentialsprovider); + httpClientBuilder.setProxyAuthenticationStrategy(new ProxyAuthenticationStrategy()); + currentHttpClient = httpClientBuilder.build(); + } + if (currentHttpClient == null) { + currentHttpClient = httpClientBuilder.build(); + } + } + + @Override + public synchronized InputStream getContent() throws IOException { + serverInvocationCount++; + if (this.httpMethod instanceof HttpEntityEnclosingRequestBase + && ((HttpEntityEnclosingRequestBase)this.httpMethod).getEntity() == null) { + byte[] entityBytes = entityByteOut.toByteArray(); + HttpEntity entity = new ByteArrayEntity(entityBytes); + currentConnectorConfig.setUseChunkedPost(false); + ((HttpEntityEnclosingRequestBase)this.httpMethod).setEntity(entity); + } + InputStream input = new ByteArrayInputStream(new byte[1]); + HttpClientContext context = HttpClientContext.create(); + RequestConfig config = RequestConfig.custom().setExpectContinueEnabled(currentConnectorConfig.useChunkedPost()).build(); + context.setRequestConfig(config); + + if (currentConnectorConfig.getNtlmDomain() != null && !currentConnectorConfig.getNtlmDomain().equals("")) { + // need to send a HEAD request to trigger NTLM authentication + try (CloseableHttpResponse ignored = currentHttpClient.execute(new HttpHead("http://salesforce.com"))) { + } catch (Exception ex) { + throw ex; + } + } + + try (CloseableHttpResponse response = currentHttpClient.execute(this.httpMethod, context)) { + successful = true; + httpResponse = response; + if (response.getStatusLine().getStatusCode() > 399) { + successful = false; + if (response.getStatusLine().getStatusCode() == 407) { + throw new RuntimeException(response.getStatusLine().getStatusCode() + " " + response.getStatusLine().getReasonPhrase()); + } + } + // copy input stream data into a new input stream because releasing the connection will close the input stream + ByteArrayOutputStream bOut = new ByteArrayOutputStream(); + if (response.getEntity() != null) { + try (InputStream inStream = response.getEntity().getContent()) { + IOUtils.copy(inStream, bOut); + input = new ByteArrayInputStream(bOut.toByteArray()); + if (response.containsHeader("Content-Encoding") && response.getHeaders("Content-Encoding")[0].getValue().equals("gzip")) { + input = new GZIPInputStream(input); + } + } + } + bOut.close(); + } + return input; + } + + public HttpResponse getHttpResponse() { + return this.httpResponse; + } + + @Override + public boolean isSuccessful() { + return successful; + } + + @Override + public OutputStream connect(String url, String soapAction) throws IOException { + if (soapAction == null) { + soapAction = ""; + } + + HashMap header = new HashMap(); + + header.put("SOAPAction", "\"" + soapAction + "\""); + header.put("Content-Type", "text/xml; charset=" + StandardCharsets.UTF_8.name()); + header.put("Accept", "text/xml"); + + return connect(url, header); + } + + @Override + public OutputStream connect(String endpoint, HashMap httpHeaders) throws IOException { + return connect(endpoint, httpHeaders, true); + } + + @Override + public OutputStream connect(String endpoint, HashMap httpHeaders, boolean enableCompression) throws IOException { + return connect(endpoint, httpHeaders, enableCompression, SupportedHttpMethodType.POST); + } + + @Override + public OutputStream connect(String endpoint, HashMap httpHeaders, boolean enableCompression, + SupportedHttpMethodType httpMethod) throws IOException { + return doConnect(endpoint, httpHeaders, enableCompression, httpMethod, null, null); + } + + @Override + public void connect(String endpoint, HashMap httpHeaders, boolean enableCompression, + SupportedHttpMethodType httpMethod, InputStream contentInputStream, String contentEncoding) + throws IOException { + doConnect(endpoint, httpHeaders, enableCompression, httpMethod, contentInputStream, contentEncoding); + } + + public static long getServerInvocationCount() { + return serverInvocationCount; + } + + public static void resetServerInvocationCount() { + serverInvocationCount = 0; + } + + private OutputStream doConnect(String endpoint, + HashMap httpHeaders, + boolean enableCompression, + SupportedHttpMethodType httpMethodType, + InputStream requestInputStream, + String contentTypeStr) throws IOException { + configureHttpMethod(endpoint, httpHeaders, enableCompression, httpMethodType, + requestInputStream, contentTypeStr); + if (requestInputStream != null) { + // Request content is in an input stream. + // Caller won't be using an output stream to write request content to. + return null; + } + entityByteOut = new ByteArrayOutputStream(); + output = entityByteOut; + + if (currentConnectorConfig.getMaxRequestSize() > 0) { + output = new LimitingOutputStream(currentConnectorConfig.getMaxRequestSize(), output); + } + + if (enableCompression && currentConnectorConfig.isCompression()) { + output = new GZIPOutputStream(output); + } + + if (currentConnectorConfig.isTraceMessage()) { + output = currentConnectorConfig.teeOutputStream(output); + } + + if (currentConnectorConfig.hasMessageHandlers()) { + URL url = new URL(endpoint); + output = new MessageHandlerOutputStream(currentConnectorConfig, url, output); + } + + return output; + } + + private void configureHttpMethod( + String endpoint, + HashMap httpHeaders, + boolean enableCompression, + SupportedHttpMethodType httpMethodType, + InputStream requestInputStream, + String contentTypeStr + ) throws IOException { + switch (httpMethodType) { + case GET : + this.httpMethod = new HttpGet(endpoint); + break; + case PATCH : + this.httpMethod = new HttpPatch(endpoint); + break; + case PUT : + this.httpMethod = new HttpPut(endpoint); + break; + case DELETE : + this.httpMethod = new HttpDelete(endpoint); + break; + default: + this.httpMethod = new HttpPost(endpoint); + } + if (httpHeaders != null) { + for (String name : httpHeaders.keySet()) { + this.httpMethod.addHeader(name, httpHeaders.get(name)); + } + } + Map connectorHeaders = currentConnectorConfig.getHeaders(); + if (connectorHeaders != null) { + for (String name : connectorHeaders.keySet()) { + if (httpHeaders == null || !httpHeaders.containsKey(name)) { + this.httpMethod.addHeader(name, connectorHeaders.get(name)); + } + } + } + setAuthAndClientHeadersForHttpMethod(); + if (enableCompression && currentConnectorConfig.isCompression()) { + this.httpMethod.addHeader("Content-Encoding", "gzip"); + this.httpMethod.addHeader("Accept-Encoding", "gzip"); + } + if (requestInputStream != null) { + // caller has pre-specified input stream + ContentType contentType = ContentType.DEFAULT_TEXT; + if (contentTypeStr != null) { + contentType = ContentType.create(contentTypeStr); + } + BufferedHttpEntity entity = new BufferedHttpEntity(new InputStreamEntity(requestInputStream, contentType)); + currentConnectorConfig.setUseChunkedPost(true); + if (this.httpMethod instanceof HttpEntityEnclosingRequestBase) { + ((HttpEntityEnclosingRequestBase)this.httpMethod).setEntity(entity); + } + } + } + + public InputStream simplePost( + String endpoint, + HashMap httpHeaders, + BasicNameValuePair[] inputs) throws IOException { + configureHttpMethod(endpoint, httpHeaders, + false, SupportedHttpMethodType.POST, null, null); + UrlEncodedFormEntity entity = new UrlEncodedFormEntity(Arrays.asList(inputs)); + if (this.httpMethod instanceof HttpPost) { + ((HttpPost)this.httpMethod).setEntity(entity); + return this.getContent(); + } + return null; + } + + private void setAuthAndClientHeadersForHttpMethod() { + if (this.httpMethod != null + && currentConnectorConfig.getSessionId() != null + ) { + String authSessionId = currentConnectorConfig.getSessionId(); + Header authHeaderValForXML = this.httpMethod.getFirstHeader(AUTH_HEADER_FOR_XML); + Header authHeaderValForJSON = this.httpMethod.getFirstHeader(AUTH_HEADER_FOR_JSON); + + if (authHeaderValForXML != null) { + authSessionId = authHeaderValForXML.getValue(); + } else if (authHeaderValForJSON != null) { + authSessionId = authHeaderValForJSON.getValue(); + } + if (authHeaderValForXML == null) { + this.httpMethod.addHeader(AUTH_HEADER_FOR_XML, AUTH_HEADER_VALUE_PREFIX + authSessionId); + } + if (authHeaderValForJSON == null) { + this.httpMethod.addHeader(AUTH_HEADER_FOR_JSON, AUTH_HEADER_VALUE_PREFIX + authSessionId); + } + } + Header userAgentHeaderVal = this.httpMethod.getFirstHeader(USER_AGENT_HEADER); + if (userAgentHeaderVal == null) { + this.httpMethod.addHeader(USER_AGENT_HEADER, VersionInfo.info()); + } + + } + + public static void closeHttpClient() { + if (currentHttpClient != null) { + try { + currentHttpClient.close(); + } catch (IOException ex) { + // do nothing + } + currentHttpClient = null; + } + } + + + + public InputStream httpGet(String urlStr) throws IOException, AsyncApiException{ + InputStream in = null; + connect(urlStr, null, false, SupportedHttpMethodType.GET, null, null); + in = getContent(); + return in; + } + + public static HttpTransportImpl getInstance() { + return singletonTransportInstance; + } +} \ No newline at end of file diff --git a/src/main/java/com/celnet/datadump/util/HttpTransportInterface.java b/src/main/java/com/celnet/datadump/util/HttpTransportInterface.java new file mode 100644 index 0000000..5fd8d4f --- /dev/null +++ b/src/main/java/com/celnet/datadump/util/HttpTransportInterface.java @@ -0,0 +1,37 @@ + +package com.celnet.datadump.util; + +import com.sforce.async.AsyncApiException; +import com.sforce.ws.transport.Transport; +import org.apache.http.HttpResponse; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.HashMap; + +/** + * This interface defines a Transport. + * + * @author http://cheenath.com + * @version 1.0 + * @since 1.0 Nov 30, 2005 + */ +public interface HttpTransportInterface extends Transport { + enum SupportedHttpMethodType { + PUT, + POST, + PATCH, + DELETE, + GET + } + int PROXY_AUTHENTICATION_REQUIRED = 407; + OutputStream connect(String endpoint, HashMap httpHeaders, boolean enableCompression, + SupportedHttpMethodType httpMethod) throws IOException; + + void connect(String endpoint, HashMap httpHeaders, boolean enableCompression, + SupportedHttpMethodType httpMethod, InputStream contentInputStream, String contentEncoding) throws IOException; + + InputStream httpGet(String urlStr) throws IOException, AsyncApiException; + HttpResponse getHttpResponse(); +} diff --git a/src/main/resources/mapper/CustomMapper.xml b/src/main/resources/mapper/CustomMapper.xml index c1e17cd..1923816 100644 --- a/src/main/resources/mapper/CustomMapper.xml +++ b/src/main/resources/mapper/CustomMapper.xml @@ -35,6 +35,32 @@ ) COMMENT = '${tableComment}'; + + + CREATE TABLE `${tableName}` ( + + `${map.name}` ${map.type} comment '${map.comment}', + + + + + UNIQUE INDEX `${map.name}`(`${map.field}`), + + + INDEX `${map.name}`(`${map.field}`), + + + + PRIMARY KEY (`id`,`CreatedDate`) + ) COMMENT = '${tableComment}' + PARTITION BY RANGE COLUMNS(`${partitionKey}`) + ( + + PARTITION ${partition.name} VALUES LESS THAN (${partition.value}) + + ) + + ALTER TABLE `${tableName}` ADD COLUMN `${fieldName}` varchar(255) @@ -51,6 +77,27 @@ + + + INSERT INTO `${tableName}` PARTITION (${partitionName}) + + `${map.key}` + + VALUES + + #{map.value} + + + + + + UPDATE `${tableName}` PARTITION (${partitionName}) + SET + + `${map.key}` = #{map.value} + + WHERE id = #{id} + UPDATE @@ -78,13 +125,14 @@ - UPDATE `${tableName}` - SET new_id = #{newId} - - - ${map.key} = #{map.value} - - + UPDATE + `${tableName}` + SET + + ${map.key} = #{map.value} + + WHERE + new_id = #{newId} diff --git a/src/main/resources/mapper/SalesforceMapper.xml b/src/main/resources/mapper/SalesforceMapper.xml index e2b1e2c..4c5e66d 100644 --- a/src/main/resources/mapper/SalesforceMapper.xml +++ b/src/main/resources/mapper/SalesforceMapper.xml @@ -26,7 +26,12 @@ - AND RowCause = 'Manual' + AND RowCause NOT IN ( + 'Owner', 'Rule', 'Territory', 'Team', + 'ImplicitChild', 'ImplicitParent', 'TerritoryRule', + 'ImplicitCallCenter', 'PortalRole', 'Portal', + 'ImplicitPerson', 'ImplicitGrant' + ) @@ -70,7 +75,12 @@ - AND RowCause = 'Manual' + AND RowCause NOT IN ( + 'Owner', 'Rule', 'Territory', 'Team', + 'ImplicitChild', 'ImplicitParent', 'TerritoryRule', + 'ImplicitCallCenter', 'PortalRole', 'Portal', + 'ImplicitPerson', 'ImplicitGrant' + ) @@ -111,7 +121,12 @@ - AND RowCause = 'Manual' + AND RowCause NOT IN ( + 'Owner', 'Rule', 'Territory', 'Team', + 'ImplicitChild', 'ImplicitParent', 'TerritoryRule', + 'ImplicitCallCenter', 'PortalRole', 'Portal', + 'ImplicitPerson', 'ImplicitGrant' + ) AND CreatedDate < #{param.endCreateDate} @@ -136,7 +151,12 @@ - AND RowCause = 'Manual' + AND RowCause NOT IN ( + 'Owner', 'Rule', 'Territory', 'Team', + 'ImplicitChild', 'ImplicitParent', 'TerritoryRule', + 'ImplicitCallCenter', 'PortalRole', 'Portal', + 'ImplicitPerson', 'ImplicitGrant' + ) @@ -162,7 +182,12 @@ - AND RowCause = 'Manual' + AND RowCause NOT IN ( + 'Owner', 'Rule', 'Territory', 'Team', + 'ImplicitChild', 'ImplicitParent', 'TerritoryRule', + 'ImplicitCallCenter', 'PortalRole', 'Portal', + 'ImplicitPerson', 'ImplicitGrant' + ) AND SystemModstamp >= #{param.beginModifyDate}