kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 bigObject数据推送、删除目标org数据、反写newId

Browse Source
This commit is contained in:
Kris 2025-09-25 10:19:57 +08:00
parent fd881694aa
commit 03fa708ed7
21 changed files with 941 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.run/data-dump（准时达） .run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（准时达） " type="docker-deploy" factoryName="dockerfile" server-name="准时达docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /data/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --ulimit nofile=65536:65536 --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

14
.run/data-dump（格凌) .run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（格凌) " type="docker-deploy" factoryName="dockerfile" server-name="格凌docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /data/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --ulimit nofile=65536:65536 --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

2
.run/data-dump（城博测试）.run.xml → .run/data-dump（雨花石测试）.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（城博测试）" type="docker-deploy" factoryName="dockerfile" server-name="城博测试Docekr">
<configuration default="false" name="data-dump（雨花石测试）" type="docker-deploy" factoryName="dockerfile" server-name="雨花石测试Docekr">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />

2
DockerfileBase
View File

@ -1,5 +1,5 @@
FROM --platform=linux/amd64 centos:7
MAINTAINER Red
LABEL maintainer="Kris"
#nginx部署
ADD nginx-1.22.1.tar.gz /usr/local/

15
src/main/java/com/celnet/datadump/controller/JobController.java
View File

@ -363,17 +363,30 @@ public class JobController {
case "Contact":
dataImportNewService.getPersonContact(param);
case "User":
dataImportNewService.getNewIdByField(api,"Username");
case "BrandTemplate":
case "EmailTemplate":
case "Folder":
dataImportNewService.getNewIdByField(api,"DeveloperName");
case "Group":
dataImportNewService.getNewIdByField(api,"DeveloperName,Type");
case "RecordType":
dataImportNewService.getNewIdByField(api,"DeveloperName,SObjectType");
case "AccountContactRelation":
dataImportNewService.getNewIdByField(api,"AccountId,ContactId,IsPrimary");
case "TaskRelation":
dataImportNewService.getNewIdByField(api,"TaskId,RelationId,RelationType");
case "EventRelation":
dataImportNewService.getNewIdByField(api,"EventId,RelationId,RelationType");
case "CollaborationGroupMember":
dataImportNewService.getNewIdByField(api,"CollaborationGroupId,MemberId,CollaborationRole");
case "FeedAttachment":
dataImportNewService.getNewIdByField(api,null);
dataImportNewService.getNewIdByField(api,"FeedItemId,AttachmentId");
case "Task":
dataImportNewService.getTaskNewId(param);
case "PermissionSet":
dataImportNewService.getNewIdByField(api,"Name");
default:
log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi());
}

21
src/main/java/com/celnet/datadump/job/DataDumpJob.java
View File

@ -48,6 +48,27 @@ public class DataDumpJob {
return commonService.createApi(param);
}
/**
* 创建bigobject api
* @param paramStr
* @return
* @throws Exception
*/
@XxlJob("createBigObjectApiJob")
public ReturnT<String> createBigObjectApiJob(String paramStr) throws Exception {
log.info("createBigObjectApiJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
return commonService.createBigObjectApi(param);
}
/**
* 存量任务
*

89
src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
View File

@ -114,6 +114,54 @@ public class DataDumpNewJob {
return commonBatchService.dumpBatch(param);
}
/**
* 存量任务（BigObject）
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataDumpBigObjectJob")
public ReturnT<String> dataDumpBigObjectJob(String paramStr) throws Exception {
log.info("dataDumpBigObjectJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
// 参数转换
param.setBeginCreateDate(param.getBeginDate());
param.setEndCreateDate(param.getEndDate());
return commonBatchService.dumpBigObject(param);
}
/**
* 导入BigObject数据到目标ORG
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataImportBigObjectJob")
public ReturnT<String> dataImportBigObjectJob(String paramStr) throws Exception {
log.info("dataImportBigObjectJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
// 参数转换
param.setBeginCreateDate(param.getBeginDate());
param.setEndCreateDate(param.getEndDate());
return commonBatchService.importBigObject(param);
}
/**
* 返写NewId
* @param paramStr 参数json
@ -143,30 +191,46 @@ public class DataDumpNewJob {
switch (api){
case "Contact":
dataImportNewService.getPersonContact(param);
break;
case "User":
dataImportNewService.getNewIdByField(api,"Username");
break;
case "BrandTemplate":
case "EmailTemplate":
case "Folder":
dataImportNewService.getNewIdByField(api,"DeveloperName");
break;
case "Group":
dataImportNewService.getNewIdByField(api,"DeveloperName,Type");
break;
case "RecordType":
dataImportNewService.getNewIdByField(api,"DeveloperName,SObjectType");
break;
case "AccountContactRelation":
dataImportNewService.getNewIdByField(api,"AccountId,ContactId,IsPrimary");
break;
case "TaskRelation":
dataImportNewService.getNewIdByField(api,"TaskId,RelationId,RelationType");
break;
case "EventRelation":
dataImportNewService.getNewIdByField(api,"EventId,RelationId,RelationType");
break;
case "CollaborationGroupMember":
dataImportNewService.getNewIdByField(api,"CollaborationGroupId,MemberId,CollaborationRole");
break;
case "FeedAttachment":
dataImportNewService.getNewIdByField(api,"FeedItemId,AttachmentId");
break;
case "Task":
dataImportNewService.getTaskNewId(param);
break;
case "PermissionSet":
case "Organization":
dataImportNewService.getNewIdByField(api,"Name");
break;
default:
log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi());
log.info("当前对象：{}，不存在反写NewId逻辑！！！" , api);
break;
}
}
return ReturnT.SUCCESS;
@ -404,6 +468,27 @@ public class DataDumpNewJob {
return dataImportNewService.checkDeletedData(param);
}
/**
* 删除目标ORG中的数据
* @param paramStr 参数json
* @return result
*/
@XxlJob("deleteTargetOrgDataJob")
public ReturnT<String> deleteTargetOrgDataJob(String paramStr) throws Exception {
log.info("deleteTargetOrgDataJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
log.error("参数解析失败！", throwable);
return new ReturnT<>(500, "参数解析失败！");
}
// 调用新实现的删除方法
return dataImportNewService.deleteTargetOrgData(param);
}
/**
* ContentFolderMember 拉取
@ -424,4 +509,6 @@ public class DataDumpNewJob {
return dataImportNewService.dumpContentFolderMemberJob(param);
}
}

11
src/main/java/com/celnet/datadump/mapper/CustomMapper.java
View File

@ -4,6 +4,7 @@ import com.celnet.datadump.param.SalesforceParam;
import cn.hutool.json.JSONObject;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.MapKey;
import java.util.Collection;
import java.util.List;
@ -134,6 +135,14 @@ public interface CustomMapper {
*/
public void updateByNewId(@Param("maps") List<Map<String, Object>> maps, @Param("tableName") String api, @Param("newId") String newId);
/**
* 根据map中的字段条件更新数据的new_id字段
* @param tableName 表名
* @param maps 包含条件字段的map
*/
public void updateNewIdByFields(@Param("tableName") String tableName,
@Param("maps") List<Map<String, Object>> maps,
@Param("newId") String newId);
/**
* 插入方法
*
@ -199,4 +208,4 @@ public interface CustomMapper {
public void deleteOne(@Param("tableName") String tableName, @Param("id") String id);
}
}

5
src/main/java/com/celnet/datadump/service/CommonBatchService.java
View File

@ -7,5 +7,8 @@ public interface CommonBatchService {
ReturnT<String> dumpBatch(SalesforceParam param) throws Exception;
ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception;
ReturnT<String> importBigObject(SalesforceParam param) throws Exception;
}
}

2
src/main/java/com/celnet/datadump/service/CommonService.java
View File

@ -37,6 +37,8 @@ public interface CommonService {
*/
ReturnT<String> createApi(SalesforceParam param) throws Exception;
ReturnT<String> createBigObjectApi(SalesforceParam param) throws Exception;
/**
* 获取所有对象
* @return

9
src/main/java/com/celnet/datadump/service/DataImportNewService.java
View File

@ -22,6 +22,13 @@ public interface DataImportNewService {
ReturnT<String> checkDeletedData(SalesforceParam param) throws Exception;
/**
* 删除目标ORG中的数据
* @param param 参数
* @return 执行结果
* @throws Exception 异常
*/
ReturnT<String> deleteTargetOrgData(SalesforceParam param) throws Exception;
ReturnT<String> insertSingle(SalesforceParam param) throws Exception;
@ -31,4 +38,4 @@ public interface DataImportNewService {
void getTaskNewId(SalesforceParam param) throws Exception;
}
}

278
src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
View File

@ -8,6 +8,7 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.*;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.mapper.CustomMapper;
@ -33,6 +34,7 @@ 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 com.xxl.job.core.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
@ -67,6 +69,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
@Autowired
private SalesforceExecutor salesforceExecutor;
@Autowired
private SalesforceTargetConnect salesforceTargetConnect;
@Autowired
private DataObjectService dataObjectService;
@Autowired
private DataBatchService dataBatchService;
@ -104,6 +108,276 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
}
@Override
public ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> result = manualBigObjectDump(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;
}
}
@Override
public ReturnT<String> importBigObject(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> result = manualBigObjectImport(param, futures);
if (result != null) {
return result;
}
}
return ReturnT.SUCCESS;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("import error", throwable);
throw throwable;
}
}
private ReturnT<String> manualBigObjectDump(SalesforceParam param, List<Future<?>> futures) throws Exception {
String api = param.getApi();
BulkConnection bulkConnect = salesforceConnect.createBulkConnect();
String beginDateStr = null;
String endDateStr = null;
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
try {
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
dbQw.eq("sync_start_date", beginDateStr); // 等于开始时间
dbQw.eq("sync_end_date", endDateStr); // 等于结束时间
}
List<DataBatch> list = dataBatchService.list(dbQw);
AtomicInteger batch = new AtomicInteger(1);
if (CollectionUtils.isNotEmpty(list)) {
salesforceParams = list.stream().map(t -> {
SalesforceParam salesforceParam = param.clone();
salesforceParam.setApi(t.getName());
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
salesforceParam.setEndCreateDate(t.getSyncEndDate());
salesforceParam.setBatch(batch.getAndIncrement());
return salesforceParam;
}).collect(Collectors.toList());
}
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
if (param.getIsSingleThread()){
manualBigObject(salesforceParam, bulkConnect);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
manualBigObject(salesforceParam, bulkConnect);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
}catch (InterruptedException interruptedException){
return ReturnT.FAIL;
}catch (Exception e) {
throw e;
}
return ReturnT.SUCCESS;
}
private void manualBigObject(SalesforceParam param, BulkConnection bulkConnect) throws Exception {
String sql = "select * from " + param.getApi() + " where CreatedDate >= '" + param.getBeginCreateDate() + "' and CreatedDate <= '" + param.getEndCreateDate() + "'";
log.info("BigObject query SQL: {}", sql);
XxlJobLogger.log("BigObject query SQL: {}", sql);
// 获取Salesforce实例URL
String restEndpoint = bulkConnect.getConfig().getRestEndpoint();
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
// 创建Bulk API 2.0查询作业
String jobId = BulkUtil.createBulkV2Job(url, sql, bulkConnect);
log.info("Created Bulk API 2.0 job with ID: {}", jobId);
// 等待作业完成
BulkUtil.waitForBulkV2JobCompletion(url, jobId, bulkConnect);
// 获取查询结果
String csvResult = BulkUtil.getBulkV2QueryJobResults(url, jobId, bulkConnect);
// 保存到本地文件
String fileName = param.getApi() + "_" + System.currentTimeMillis() + ".csv";
String filePath = "bigobject/" + fileName;
// 确保目录存在
File directory = new File("bigobject");
if (!directory.exists()) {
directory.mkdirs();
}
// 写入文件
try (FileWriter fileWriter = new FileWriter(filePath);
BufferedWriter bufferedWriter = new BufferedWriter(fileWriter)) {
bufferedWriter.write(csvResult);
}
log.info("BigObject data saved to file: {}", filePath);
XxlJobLogger.log("BigObject data saved to file: {}", filePath);
}
private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future<?>> futures) throws Exception {
String api = param.getApi();
// 创建目标ORG连接
BulkConnection targetBulkConnect = salesforceTargetConnect.createBulkConnect();
String beginDateStr = null;
String endDateStr = null;
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
try {
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
dbQw.eq("sync_start_date", beginDateStr); // 等于开始时间
dbQw.eq("sync_end_date", endDateStr); // 等于结束时间
}
List<DataBatch> list = dataBatchService.list(dbQw);
AtomicInteger batch = new AtomicInteger(1);
if (CollectionUtils.isNotEmpty(list)) {
salesforceParams = list.stream().map(t -> {
SalesforceParam salesforceParam = param.clone();
salesforceParam.setApi(t.getName());
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
salesforceParam.setEndCreateDate(t.getSyncEndDate());
salesforceParam.setBatch(batch.getAndIncrement());
return salesforceParam;
}).collect(Collectors.toList());
}
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
if (param.getIsSingleThread()){
manualImportBigObject(salesforceParam, targetBulkConnect);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
manualImportBigObject(salesforceParam, targetBulkConnect);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
}catch (InterruptedException interruptedException){
return ReturnT.FAIL;
}catch (Exception e) {
throw e;
}
return ReturnT.SUCCESS;
}
private void manualImportBigObject(SalesforceParam param, BulkConnection targetBulkConnect) throws Exception {
String api = param.getApi();
// 构造CSV文件路径 - 使用与导出时相同的文件名格式（但需要知道确切的时间戳）
// 这里应该查找最新的相关文件，而不是随机生成一个文件名
File directory = new File("bigobject");
if (!directory.exists() || !directory.isDirectory()) {
log.error("BigObject目录不存在: {}", directory.getAbsolutePath());
throw new FileNotFoundException("BigObject目录不存在: " + directory.getAbsolutePath());
}
// 查找匹配的CSV文件（最近的）
File[] files = directory.listFiles((dir, name) ->
name.startsWith(api + "_") && name.endsWith(".csv"));
if (files == null || files.length == 0) {
log.error("未找到API {} 的CSV文件", api);
throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
}
// 选择最新的文件
File csvFile = Arrays.stream(files)
.max(Comparator.comparingLong(File::lastModified))
.orElse(null);
if (csvFile == null) {
log.error("未找到API {} 的CSV文件", api);
throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
}
String filePath = csvFile.getAbsolutePath();
log.info("找到CSV文件: {}", filePath);
log.info("开始导入BigObject数据到目标ORG: {}", api);
// 创建Bulk API作业
JobInfo job = BulkUtil.createJob(targetBulkConnect, api, OperationEnum.insert);
log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
// 从CSV文件创建批次
List<BatchInfo> batchInfos = BulkUtil.createBatchesFromCSVFile(targetBulkConnect, job, filePath);
log.info("创建批次成功, 批次数: {}", batchInfos.size());
// 等待作业完成
BulkUtil.awaitCompletion(targetBulkConnect, job, batchInfos);
log.info("数据导入完成");
// 检查作业状态
job = targetBulkConnect.getJobStatus(job.getId());
if (job.getState() == JobStateEnum.Failed) {
log.error("Bulk作业失败, 作业ID: {}", job.getId());
throw new RuntimeException("Bulk作业失败 ");
}
// 关闭作业
BulkUtil.closeJob(targetBulkConnect, job.getId());
log.info("关闭作业成功, 作业ID: {}", job.getId());
// 记录导入统计信息
int totalRecords = batchInfos.stream()
.mapToInt(BatchInfo::getNumberRecordsProcessed)
.sum();
log.info("成功导入 {} 条记录到 {}", totalRecords, api);
}
private ReturnT<String> manualBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
List<String> apis;
apis = DataUtil.toIdList(param.getApi());
@ -739,6 +1013,10 @@ public class CommonBatchServiceImpl implements CommonBatchService {
maps.add(paramMap3);
if (existsIds.contains(id)) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_update");
paramMap.put("value",0);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
} else {
if (insert) {

71
src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
View File

@ -464,12 +464,12 @@ public class CommonServiceImpl implements CommonService {
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setDataWork(0);
} catch (Throwable e) {
log.error("manualDump error", e);
throw new RuntimeException(e);
} finally {
if (isFull) {
update.setDataWork(0);
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
@ -921,6 +921,10 @@ public class CommonServiceImpl implements CommonService {
if (existsIds.contains(id)) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_update");
paramMap.put("value",0);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
} else {
if (insert) {
@ -1599,6 +1603,71 @@ public class CommonServiceImpl implements CommonService {
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> createBigObjectApi(SalesforceParam param) throws Exception {
String api = param.getApi();
PartnerConnection connection = salesforceConnect.createConnect();
DescribeSObjectResult dsr = connection.describeSObject(api);
DataObject dataObject = dataObjectService.getById(api);
dataObject.setLabel(dsr.getLabel());
dataObjectService.updateById(dataObject);
Date startDate = dataObject.getLastUpdateDate() == null ? DataUtil.DEFAULT_BEGIN_DATE : dataObject.getLastUpdateDate();
Date endCreateDate = DateUtils.parseDate(DateFormatUtils.format(new Date(), "yyyy-MM-dd"), "yyyy-MM-dd");
// 按年生成批次
DataBatch one = new DataBatch();
one.setFirstDbNum(0);
one.setFirstSfNum(0);
one.setDbNum(0);
one.setSfNum(0);
one.setName(api);
one.setLabel(dsr.getLabel());
// 按年份划分开始时间和结束时间创建批次
Calendar startCalendar = Calendar.getInstance();
startCalendar.setTime(startDate);
int startYear = startCalendar.get(Calendar.YEAR);
Calendar endCalendar = Calendar.getInstance();
endCalendar.setTime(endCreateDate);
int endYear = endCalendar.get(Calendar.YEAR);
// 优化批次创建逻辑，确保正确的日期范围
for (int year = startYear; year <= endYear; year++) {
Date batchStartDate;
Date batchEndDate;
// 设置批次开始时间为当年1月1日 00:00:00
batchStartDate = DateUtils.parseDate(year + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
if (year == endYear) {
// 结束年份使用实际结束日期
batchEndDate = endCreateDate;
} else {
int nextYear = year + 1;
// 其他年份结束时间为当年12月31日 00:00:00
batchEndDate = DateUtils.parseDate(nextYear + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
}
// 确保批次开始时间不早于实际开始时间
if (batchStartDate.before(startDate)) {
batchStartDate = startDate;
}
DataBatch dataBatch = one.clone();
dataBatch.setSyncStartDate(batchStartDate);
dataBatch.setSyncEndDate(batchEndDate);
dataBatchService.save(dataBatch);
}
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> getAllApi() throws Exception {
PartnerConnection partnerConnection = salesforceConnect.createConnect();

25
src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
View File

@ -826,7 +826,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
List<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 8000 + ",8000");
List<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000");
log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@ -860,7 +860,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！";
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 2);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
}
}
if (reference_to == null){
@ -871,6 +882,16 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
account.put(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！";
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 2);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
break;
}

449
src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
View File

@ -836,7 +836,18 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！";
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 2);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
}
}
if (reference_to == null){
@ -847,6 +858,16 @@ public class DataImportNewServiceImpl implements DataImportNewService {
account.setField(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！";
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 2);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
break;
}
@ -1214,7 +1235,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
try {
if (list != null && !list.isEmpty()) {
//表内数据总量
Integer count = customMapper.countBySQL(api, "where ShareType = 'V' and new_id = '0'");
Integer count = customMapper.countBySQL(api, "where ShareType = 'V' ");
//批量插入200一次
int page = count % 200 == 0 ? count / 200 : (count / 200) + 1;
for (int i = 0; i < page; i++) {
@ -2271,6 +2292,89 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
/**
* 删除目标ORG中的数据
* @param param 参数
* @return 执行结果
* @throws Exception 异常
*/
@Override
public ReturnT<String> deleteTargetOrgData(SalesforceParam param) throws Exception {
log.info("deleteTargetOrgData execute start ..................");
QueryWrapper<DataObject> wrapper = new QueryWrapper<>();
if (StringUtils.isNotBlank(param.getApi())) {
wrapper.in("name", DataUtil.toIdList(param.getApi()));
} else {
wrapper.eq("data_work", 1);
}
List<DataObject> objectList = dataObjectService.list(wrapper);
if (objectList.isEmpty()) {
log.info("没有对象需要删除数据！!!");
return ReturnT.FAIL;
}
PartnerConnection connect = salesforceTargetConnect.createConnect();
try {
for (DataObject dataObject : objectList) {
String api = dataObject.getName();
boolean moreData = true;
DescribeSObjectResult dsr = connect.describeSObject(api);
Field[] dsrFields = dsr.getFields();
String sql = "select Id from " + api + " Limit 200";
log.info("执行SQL：{}", sql);
while (moreData){
int size = 0;
QueryResult queryResult = connect.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
SObject[] records = queryResult.getRecords();
com.alibaba.fastjson2.JSONArray objects = DataUtil.toJsonArray(records, dsrFields);
String[] ids = new String[objects.size()];
for (int i = 0; i < objects.size(); i++) {
JSONObject jsonObject = objects.getJSONObject(i);
String id = jsonObject.getString("Id");
ids[i] = id;
size ++;
}
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()) {
log.info("成功删除目标ORG中的数据，ID: {}，类型: {}", deleteResult.getId(), dataObject.getLabel());
} else {
log.error("ID:{}删除目标ORG数据失败：{}", deleteResult.getId(), deleteResult.getErrors());
}
}
if (size < 200){
moreData = false;
}
}
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException interruptedException){
return ReturnT.FAIL;
}catch (Exception e){
log.info("目标ORG数据删除失败！", e);
}
log.info("目标ORG数据删除完成！");
return ReturnT.SUCCESS;
}
private ReturnT<String> manualCheckDeletedData(SalesforceParam param) throws Exception {
String api = "DeleteEvent";
@ -2505,15 +2609,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
@Override
public void getNewIdByField(String api, String queryFields) throws Exception {
List<String> stringList = DataUtil.toIdList(queryFields);
// 检测表 不存在就生成 但不生成批次
commonService.checkApi(api, true);
Map<String, Object> map = Maps.newHashMap();
SalesforceParam salesforceParam = new SalesforceParam();
salesforceParam.setApi(api);
salesforceParam.setIdField("Id");
String maxId = null;
Field[] dsrFields;
List<String> fields = Lists.newArrayList();
PartnerConnection connect = salesforceConnect.createConnect();
@ -2526,178 +2623,190 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
fields.add(field.getName());
}
if ("TaskRelation".equals(api) || "EventRelation".equals(api)){
// 添加特定对象的关联字段
addRelationFields(api, fields);
String joined = StringUtils.join(fields, ",");
com.alibaba.fastjson2.JSONArray objects;
int count = 0;
do {
String sql = null;
if (StringUtils.isNotEmpty(maxId)){
sql = "select " + joined + " from "+ api + " Where Id > '"+ maxId+"' LIMIT 200";
}else {
sql = "select " + joined + " from "+ api +" LIMIT 200";
}
QueryResult queryResult = connect.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
// 批量处理记录
processRecords(api, objects, stringList);
int totalRecords = records.length;
count += totalRecords;
log.info("getNewId success count: {}", count);
if (totalRecords < 200) {
log.info("无更多数据，反写NewId任务结束！！！");
return;
}
TimeUnit.MILLISECONDS.sleep(1);
} while (true);
}
private void addRelationFields(String api, List<String> fields) {
if ("TaskRelation".equals(api) || "EventRelation".equals(api)) {
fields.add("Relation.type");
}
if ("CollaborationGroupMember".equals(api) || "MemberId".equals(api)){
if ("CollaborationGroupMember".equals(api)) {
fields.add("Member.type");
}
if ("FeedAttachment".equals(api) || "FeedEntityId".equals(api)){
if ("FeedAttachment".equals(api)) {
fields.add("FeedEntity.type");
}
if ("FeedAttachment".equals(api) || "RecordId".equals(api)){
fields.add("Record.type");
}
salesforceParam.setSelect(StringUtils.join(fields, ","));
com.alibaba.fastjson2.JSONArray objects = null;
int count = 0, failCount = 0;
// 遍历
while (true) {
try {
// 判断是否存在要排除的id
salesforceParam.setMaxId(maxId);
map.put("param", salesforceParam);
String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map);
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);
QueryResult queryResult = connect.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
}
// 更新new_id
for (int i = 0; i < objects.size(); i++) {
JSONObject jsonObject = objects.getJSONObject(i);
try {
Set<String> keys = jsonObject.keySet();
// update
String id = jsonObject.getString(Const.ID);
List<Map<String, Object>> maps = Lists.newArrayList();
for (String key : keys) {
if ("AccountContactRelation".equals(api) && "ContactId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", "Contact", jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("AccountContactRelation".equals(api) && "AccountId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", "Account", jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("TaskRelation".equals(api) && "TaskId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", "Task", jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("EventRelation".equals(api) && "EventId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", "Event", jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if (("EventRelation".equals(api) || "TaskRelation".equals(api)) && "RelationId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Relation_Type"), jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("CollaborationGroupMember".equals(api) && "CollaborationGroupId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", "CollaborationGroup", jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("CollaborationGroupMember".equals(api) && "MemberId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Member_Type"), jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("FeedAttachment".equals(api) && "FeedEntityId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if ("FeedAttachment".equals(api) && "RecordId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
if (m == null){
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value",m.get("Id"));
maps.add(paramMap);
continue;
}
if (stringList.contains(key)) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value", jsonObject.getString(key));
maps.add(paramMap);
}
}
customMapper.updateByNewId(maps, api, id);
} catch (Throwable throwable) {
if (throwable.toString().contains("interrupt")) {
log.error("may interrupt error:", throwable);
throw new InterruptedException();
}
log.info(JSON.toJSONString(throwable));
}
}
count += records.length;
log.info("getNewId success count: {}", count);
failCount = 0;
objects = null;
} catch (InterruptedException e) {
throw e;
} catch (Throwable throwable) {
failCount++;
log.error("getNewId error api:{}, data:{}", api, com.alibaba.fastjson2.JSON.toJSONString(objects), throwable);
if (failCount > Const.MAX_FAIL_COUNT) {
log.info("getNewId error data:" + com.alibaba.fastjson2.JSON.toJSONString(objects));
}
private String buildQuerySql(String api, String joined, String maxId) {
StringBuilder sqlBuilder = new StringBuilder("SELECT ");
if (StringUtils.isNotEmpty(joined)) {
sqlBuilder.append(joined).append(", ");
}
sqlBuilder.append("Id FROM ").append(api);
if (StringUtils.isNotEmpty(maxId)) {
sqlBuilder.append(" WHERE Id > '").append(maxId).append("'");
}
sqlBuilder.append(" LIMIT 200");
return sqlBuilder.toString();
}
private void processRecords(String api, com.alibaba.fastjson2.JSONArray objects, List<String> stringList) throws Exception {
for (int i = 0; i < objects.size(); i++) {
JSONObject jsonObject = objects.getJSONObject(i);
String id = jsonObject.getString(Const.ID);
List<Map<String, Object>> maps = buildUpdateMaps(api, jsonObject, stringList);
if (maps!=null && !maps.isEmpty()) {
customMapper.updateNewIdByFields(api, maps, id);
}
}
}
// 3、拼接sql，更新本地数据库new_id
private List<Map<String, Object>> buildUpdateMaps(String api, JSONObject jsonObject, List<String> stringList) throws Exception {
List<Map<String, Object>> maps = Lists.newArrayList();
Set<String> keys = jsonObject.keySet();
for (String key : keys) {
Map<String, Object> paramMap = null;
switch (api) {
case "AccountContactRelation":
paramMap = handleAccountContactRelation(key, jsonObject);
if (paramMap == null){
return null;
}
break;
case "TaskRelation":
case "EventRelation":
paramMap = handleTaskEventRelation(api, key, jsonObject);
if (paramMap == null){
return null;
}
break;
case "CollaborationGroupMember":
paramMap = handleCollaborationGroupMember(key, jsonObject);
if (paramMap == null){
return null;
}
break;
case "FeedAttachment":
paramMap = handleFeedAttachment(key, jsonObject);
if (paramMap == null){
return null;
}
break;
default:
if (stringList.contains(key)) {
paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value", jsonObject.getString(key));
}
break;
}
if (paramMap != null) {
maps.add(paramMap);
}
}
return maps;
}
private Map<String, Object> handleAccountContactRelation(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "ContactId":
return getNewIdMap(key, "Contact", jsonObject.getString(key));
case "AccountId":
return getNewIdMap(key, "Account", jsonObject.getString(key));
default:
return null;
}
}
private Map<String, Object> handleTaskEventRelation(String api, String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "TaskId":
return getNewIdMap(key, "Task", jsonObject.getString(key));
case "EventId":
return getNewIdMap(key, "Event", jsonObject.getString(key));
case "RelationId":
return getNewIdMap(key, jsonObject.getString("Relation_Type"), jsonObject.getString(key));
default:
return null;
}
}
private Map<String, Object> handleCollaborationGroupMember(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "CollaborationGroupId":
return getNewIdMap(key, "CollaborationGroup", jsonObject.getString(key));
case "MemberId":
return getNewIdMap(key, jsonObject.getString("Member_Type"), jsonObject.getString(key));
default:
return null;
}
}
private Map<String, Object> handleFeedAttachment(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "FeedEntityId":
return getNewIdMap(key, jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
case "RecordId":
return getNewIdMap(key, jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
default:
return null;
}
}
private Map<String, Object> getNewIdMap(String key, String tableName, String id) throws Exception {
Map<String, Object> m = customMapper.getByNewId("Id", tableName, id);
if (m == null || m.get("Id") == null) {
log.info("关联对象未找到，跳过处理。字段: {}, 表名: {}, ID: {}", key, tableName, id);
return null; // 返回null表示跳过该字段
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value", m.get("Id"));
return paramMap;
}
@Override

8
src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
View File

@ -386,6 +386,7 @@ public class DataImportServiceImpl implements DataImportService {
//如果必填lookup字段没有值，跳过
dataObject.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1);
dataObjectService.updateById(dataObject);
log.info("引用对象：" + reference + "不存在数据！请检查这个对象详情！！！");
return;
}else{
account.setField(dataField.getField(), referenceMap.get(0).get("new_id"));
@ -455,12 +456,15 @@ public class DataImportServiceImpl implements DataImportService {
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException interruptedException){
return;
} catch (Exception e) {
log.info(JSON.toJSONString(e));
log.info(JSON.toJSONString(e.getMessage()));
}
}
if (z*10000+size >= count){
if (z * 10000 + size >= count) {
log.info("数据处理完成，共处理 {} 条记录，当前批次结束", count);
break;
}
}

11
src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
View File

@ -629,7 +629,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
sourceRow.add(sourceField.isNillable());
sourceRow.add(sourceField.isUnique());
sourceRow.add(sourceField.isPolymorphicForeignKey() ? "是" : "否");
sourceRow.add(sourceField.getReferenceTo().length> 2 ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
sourceRow.add((sourceField.getReferenceTo()!=null && sourceField.getReferenceTo().length> 2) ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
// 写入Excel
excelWriter.write(Collections.singletonList(sourceRow), fieldTypeSheet);
@ -652,7 +652,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
targetRow.add(targetField.isNillable());
targetRow.add(targetField.isUnique());
targetRow.add(targetField.isPolymorphicForeignKey() ? "是" : "否");
targetRow.add(targetField.getReferenceTo().length> 2 ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", sourceField.getReferenceTo()));
// 写入Excel
excelWriter.write(Collections.singletonList(targetRow), fieldTypeSheet);
@ -672,14 +672,13 @@ public class DataVerifyServiceImpl implements DataVerifyService {
PartnerConnection sourceConn, PartnerConnection targetConn,Map<String, Map<String, Field>> fieldMap) {
String startDate = DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss");
String endDate = DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss");
log.info("开始处理数据批次: 批次ID={} 批次开始时间={} 批次结束时间={} 对象API={}", batch.getId(),startDate,endDate, objectApi);
log.info("开始处理数据批次: 批次ID={} 批次总量(大于十万只校验十万)={} 批次开始时间={} 批次结束时间={} 对象API={}", batch.getId(),batch.getFirstSfNum(),startDate,endDate, objectApi);
// 分页处理记录
int page = 0;
int totalRecordNum = 0;
int targetRecordNum = 0;
int sourceRecordNum = 0;
int errorFieldNum = 0;
int sorceOnlyNum = 0;
int targetOnlyNum = 0;
boolean isError = false;
@ -700,7 +699,6 @@ public class DataVerifyServiceImpl implements DataVerifyService {
List<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size();
log.info("数据总量: {} 子批次: 批次数={}, 记录数={}, 累计记录数={}",batch.getFirstDbNum(), (page*50 + i), idList.size(), totalRecordNum);
List<String> ids = Lists.newArrayList();
List<String> newIds = Lists.newArrayList();
@ -714,14 +712,11 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 批量查询源和目标数据
Map<String, SObject> sourceRecords = queryRecords(sourceConn, objectApi, ids);
sourceRecordNum += sourceRecords.size();
log.info("源系统查询完成，获取到{}条记录", sourceRecords.size());
Map<String, SObject> targetRecords = queryRecords(targetConn, objectApi, newIds);
targetRecordNum += targetRecords.size();
log.info("目标系统查询完成，获取到{}条记录", targetRecords.size());
// 比对记录
log.info("开始比对{}条记录的字段值", idLinks.size());
for (String id : idLinks.keySet()) {
String newId = idLinks.get(id);
SObject sourceRecord = sourceRecords.get(id);

66
src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java
View File

@ -693,8 +693,17 @@ public class FileServiceImpl implements FileService {
qw.eq("name", parentType);
List<DataObject> objects = dataObjectService.list(qw);
if (objects.isEmpty()) {
log.info( "文件ID：" + id + "关联对象不存在： {}", parentType);
EmailUtil.send("DataDump ERROR", "文件ID：" + id + "关联对象"+ parentType+"不存在");
String format = "文件ID：" + id + "关联对象不存在：" + parentType;
log.info(format);
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
Map<String, Object> paramMap1 = Maps.newHashMap();
paramMap1.put("key", "error_message");
paramMap1.put("value",format);
maps.add(paramMap1);
customMapper.updateById(api, maps, id);
break;
}
@ -702,7 +711,15 @@ public class FileServiceImpl implements FileService {
if(lMap == null) {
String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
Map<String, Object> paramMap1 = Maps.newHashMap();
paramMap1.put("key", "error_message");
paramMap1.put("value",format);
maps.add(paramMap1);
customMapper.updateById(api, maps, id);
break;
}
@ -710,7 +727,15 @@ public class FileServiceImpl implements FileService {
if(uMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + ownerId + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
Map<String, Object> paramMap1 = Maps.newHashMap();
paramMap1.put("key", "error_message");
paramMap1.put("value",format);
maps.add(paramMap1);
customMapper.updateById(api, maps, id);
break;
}
@ -718,7 +743,15 @@ public class FileServiceImpl implements FileService {
if(cMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + createdById + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
Map<String, Object> paramMap1 = Maps.newHashMap();
paramMap1.put("key", "error_message");
paramMap1.put("value",format);
maps.add(paramMap1);
customMapper.updateById(api, maps, id);
break;
}
@ -755,7 +788,18 @@ public class FileServiceImpl implements FileService {
maps.add(paramMap);
}
} else {
log.info("文件ID：" + id + "上传失败，返回信息：" + EntityUtils.toString(response.getEntity(), "UTF-8"));
{
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
Map<String, Object> paramMap1 = Maps.newHashMap();
paramMap1.put("key", "error_message");
paramMap1.put("value", response == null?"上传失败" :EntityUtils.toString(response.getEntity(), "UTF-8"));
maps.add(paramMap1);
}
customMapper.updateById(api, maps, id);
log.info("---------文件ID：" + id + "上传失败--------" );
throw new RuntimeException();
}
}
@ -770,12 +814,10 @@ public class FileServiceImpl implements FileService {
log.error("upload file error, id: {}", id, throwable);
failCount++;
if (Const.MAX_FAIL_COUNT < failCount) {
{
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
break;
}

16
src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java
View File

@ -23,6 +23,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static com.celnet.datadump.global.SystemConfigCode.EXECUTOR_SIZE;
/**
* org配置表 服务实现类
*/
@ -73,7 +75,7 @@ public class OrgConfigServiceImpl extends ServiceImpl<OrgConfigMapper, OrgConfig
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
} catch (ConnectionException e) {
String message = "源ORG连接配置错误！,\n地址：" + sourceOrgUrl;
String message = "源ORG连接配置错误！,\n地址：" + sourceOrgUrl + "\n错误信息：\n" + e.getMessage() ;;
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
@ -112,7 +114,7 @@ public class OrgConfigServiceImpl extends ServiceImpl<OrgConfigMapper, OrgConfig
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
} catch (ConnectionException e) {
String message = "目标ORG连接配置错误！,\n地址：" + targetOrgUrl;
String message = "目标ORG连接配置错误！,\n地址：" + targetOrgUrl + "\n错误信息：\n" + e.getMessage() ;
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
@ -120,10 +122,12 @@ public class OrgConfigServiceImpl extends ServiceImpl<OrgConfigMapper, OrgConfig
}
if (flag){
String message = "源ORG与目标ORG连接配置成功！";
String format = String.format("ORG连接成功！, \ncause:\n%s , \n源ORG地址:\n%s , \n目标ORG地址:\n%s ,请检查文件上传下载地址：\n ,文件上传地址：\n%s, 文件下载地址：\n%s",
message,sourceOrgUrl,targetOrgUrl,uploadUrl,dumpUrl);
EmailUtil.send("DataDump ERROR", format);
List<Map<String, Object>> list = customerMapper.list("code,value","system_config","code ='DEFAULT_START_DATE'");
List<Map<String, Object>> list1 = customerMapper.list("code,value","system_config","code ='BATCH_TYPE'");
String format = String.format("源ORG与目标ORG连接配置成功！ \n源ORG地址:\n%s , \n目标ORG地址:\n%s \n---请检查文件上传下载地址---\n 文件上传地址：\n%s \n文件下载地址：\n%s \n---请检查批次开始时间（不对要重启服务）---\n%s \n---请检查批次类型（不对要重启服务）---\n%s",
sourceOrgUrl,targetOrgUrl,uploadUrl,dumpUrl,list.get(0).get("value"),list1.get(0).get("value"));
EmailUtil.send("校验ORG配置", format);
}
}

10
src/main/resources/mapper/CustomMapper.xml
View File

@ -138,6 +138,16 @@
new_id = #{newId}
</update>
<update id="updateNewIdByFields">
UPDATE `${tableName}`
SET new_id = #{newId}
<where>
<foreach item="map" collection="maps" open="" separator=" AND " close="">
${map.key} = #{map.value}
</foreach>
</where>
</update>
<insert id="save">
insert into
`${tableName}`

31
src/main/resources/mapper/SalesforceMapper.xml
View File

@ -167,6 +167,37 @@
</if>
</select>
<select id="listNewId">
select Id,${param.select} from ${param.api}
<where>
<if test="param != null">
<if test="param.maxId != null">
AND id > #{param.maxId}
</if>
<if test="param.isDeleted != null">
AND IsDeleted = #{param.isDeleted}
</if>
<if test="param.sql != null">
AND ${param.sql}
</if>
<!-- 判断 param.api 是否以 "Share" 结尾 -->
<if test="param.api != null and param.api.endsWith('Share')">
AND RowCause NOT IN (
'Owner', 'Rule', 'Territory', 'Team',
'ImplicitChild', 'ImplicitParent', 'TerritoryRule',
'ImplicitCallCenter', 'PortalRole', 'Portal',
'ImplicitPerson', 'ImplicitGrant'
)
</if>
</if>
</where>
<if test="param != null">
<if test="param.idField != null">
order by ${param.idField} asc
</if>
</if>
</select>
<select id="listOrderByIdNew">
select ${param.select} from ${param.api}
<where>