533 lines
18 KiB
Java
533 lines
18 KiB
Java
package com.celnet.datadump.job;
|
||
|
||
import com.alibaba.fastjson.JSON;
|
||
import com.celnet.datadump.config.SalesforceConnect;
|
||
import com.celnet.datadump.entity.DataObject;
|
||
import com.celnet.datadump.param.SalesforceParam;
|
||
import com.celnet.datadump.service.*;
|
||
import com.celnet.datadump.util.DataUtil;
|
||
import com.xxl.job.core.biz.model.ReturnT;
|
||
import com.xxl.job.core.handler.annotation.XxlJob;
|
||
import com.xxl.job.core.log.XxlJobLogger;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.apache.commons.lang3.StringUtils;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.stereotype.Component;
|
||
|
||
import java.util.List;
|
||
|
||
/**
|
||
* 迁移任务 (新)
|
||
* 2024/06/12
|
||
* kris
|
||
*/
|
||
@Component
|
||
@Slf4j
|
||
public class DataDumpNewJob {
|
||
|
||
@Autowired
|
||
private CommonService commonService;
|
||
|
||
@Autowired
|
||
private DataImportNewService dataImportNewService;
|
||
|
||
@Autowired
|
||
private DataImportBatchService dataImportBatchService;
|
||
|
||
@Autowired
|
||
private DataObjectService dataObjectService;
|
||
|
||
@Autowired
|
||
private CommonBatchService commonBatchService;
|
||
|
||
|
||
/**
|
||
* bulk批量大数据生成newSFID
|
||
* @param paramStr
|
||
* @author kris
|
||
* @return
|
||
* @throws Exception
|
||
*/
|
||
@XxlJob("dataImportBatchJob")
|
||
public ReturnT<String> dataImportBatchJob(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 dataImportBatchService.immigrationBatch(param);
|
||
}
|
||
|
||
/**
|
||
* bulk批量大数据更新数据
|
||
* @param paramStr
|
||
* @author kris
|
||
* @return
|
||
* @throws Exception
|
||
*/
|
||
@XxlJob("dataUpdateBatchJob")
|
||
public ReturnT<String> dataUpdateBatchJob(String paramStr) throws Exception {
|
||
log.info("dataUpdateBatchJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
// 参数转换
|
||
param.setBeginCreateDate(param.getBeginDate());
|
||
param.setEndCreateDate(param.getEndDate());
|
||
return dataImportBatchService.immigrationUpdateBatch(param);
|
||
}
|
||
|
||
/**
|
||
* 存量任务（大数据量）
|
||
*
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("dataDumpManualBatchJob")
|
||
public ReturnT<String> dataDumpManualBatchJob(String paramStr) throws Exception {
|
||
log.info("dataDumpManualBatchJob 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.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
|
||
* @return result
|
||
*/
|
||
@XxlJob("getNewId")
|
||
public ReturnT<String> getNewId(String paramStr) throws Exception {
|
||
log.info("getNewId 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());
|
||
|
||
if (StringUtils.isBlank(param.getApi())) {
|
||
return new ReturnT<>(500, "api不能为空！");
|
||
}
|
||
|
||
List<String> apis = DataUtil.toIdList(param.getApi());
|
||
for (String api : apis) {
|
||
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");
|
||
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逻辑！！！" , api);
|
||
break;
|
||
}
|
||
}
|
||
return ReturnT.SUCCESS;
|
||
|
||
}
|
||
|
||
/**
|
||
* 数据更新
|
||
*/
|
||
@XxlJob("dataUpdateNewJob")
|
||
public ReturnT<String> dataUpdateNewJob(String paramStr) throws Exception {
|
||
log.info("dataUpdateNewJob 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 dataImportNewService.immigrationUpdateNew(param);
|
||
}
|
||
|
||
/**
|
||
* 一次性插入数据
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("insertSingleJob")
|
||
public ReturnT<String> insertSingleJob(String paramStr) throws Exception {
|
||
log.info("insertSingleJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
|
||
return dataImportNewService.insertSingle(param);
|
||
}
|
||
|
||
/**
|
||
* 拉取文件关联表
|
||
* @return result
|
||
*/
|
||
@XxlJob("dumpDocumentLinkJob")
|
||
public ReturnT<String> dumpDocumentLinkJob(String paramStr) throws Exception{
|
||
log.info("dumpDocumentLinkJob execute start ..................");
|
||
|
||
return dataImportNewService.dumpDocumentLinkJob(paramStr);
|
||
}
|
||
|
||
/**
|
||
* 推送文件关联表
|
||
* @return result
|
||
*/
|
||
@XxlJob("uploadDocumentLinkJob")
|
||
public ReturnT<String> uploadDocumentLinkJob(String paramStr) throws Exception{
|
||
log.info("uploadDocumentLinkJob execute start ..................");
|
||
|
||
return dataImportNewService.uploadDocumentLinkJob(paramStr);
|
||
}
|
||
|
||
|
||
/**
|
||
* 创建关联字段
|
||
* @return result
|
||
*/
|
||
@XxlJob("createLinkTypeFieldJob")
|
||
public ReturnT<String> createLinkTypeJob(String paramStr) throws Exception{
|
||
log.info("createLinkTypeFieldJob 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.createLinkTypeField(param);
|
||
}
|
||
|
||
/**
|
||
* 更新关联类型
|
||
* @return result
|
||
*/
|
||
@XxlJob("updateLinkTypeJob")
|
||
public ReturnT<String> updateLinkTypeJob(String paramStr) throws Exception{
|
||
log.info("updateLinkTypeJob execute start ..................");
|
||
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
|
||
return dataImportNewService.updateLinkTypeJob(param);
|
||
}
|
||
|
||
/**
|
||
* 一次性插入数据（大数据量）
|
||
* @param paramStr
|
||
* @author kris
|
||
* @return
|
||
* @throws Exception
|
||
*/
|
||
@XxlJob("insertSingleBatchJob")
|
||
public ReturnT<String> insertSingleBatchJob(String paramStr) throws Exception {
|
||
log.info("insertSingleBatchJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
// 参数转换
|
||
param.setBeginCreateDate(param.getBeginDate());
|
||
param.setEndCreateDate(param.getEndDate());
|
||
return dataImportBatchService.insertSingleBatch(param);
|
||
}
|
||
|
||
|
||
/**
|
||
* 文件下载（新）
|
||
* @param paramStr
|
||
* @author kris
|
||
* @return
|
||
* @throws Exception
|
||
*/
|
||
@XxlJob("dumpFileNewJob")
|
||
public ReturnT<String> dumpFileNewJob(String paramStr) throws Exception {
|
||
log.info("dumpFileNewJob 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 dataImportNewService.dumpFileNew(param);
|
||
}
|
||
|
||
/**
|
||
* 文件上传（新）
|
||
* @param paramStr
|
||
* @author kris
|
||
* @return
|
||
* @throws Exception
|
||
*/
|
||
@XxlJob("uploadFileNewJob")
|
||
public ReturnT<String> uploadFileNewJob(String paramStr) throws Exception {
|
||
log.info("uploadFileNewJob 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());
|
||
|
||
for (String api : DataUtil.toIdList(param.getApi())) {
|
||
DataObject dataObject = dataObjectService.getById(api);
|
||
String blobField = dataObject.getBlobField();
|
||
if (StringUtils.isBlank(blobField)) {
|
||
log.error("api:{} does not have blob field", api);
|
||
XxlJobLogger.log("api:{} does not have blob field", api);
|
||
}
|
||
if (api.equals("ContentVersion")){
|
||
return dataImportNewService.uploadFileNew(param,dataObject);
|
||
}{
|
||
log.info("当前文件对象：{}，不存在文件上传逻辑！！！" ,api);
|
||
}
|
||
}
|
||
return ReturnT.SUCCESS;
|
||
}
|
||
|
||
/**
|
||
* 增量任务（新）
|
||
*
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("dataDumpIncrementNewJob")
|
||
public ReturnT<String> dataDumpIncrementNewJob(String paramStr) throws Exception {
|
||
log.info("dataDumpIncrementNewJob 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.incrementNew(param);
|
||
}
|
||
|
||
/**
|
||
* 校验删除数据
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("checkDeletedDataJob")
|
||
public ReturnT<String> checkDeletedDataJob(String paramStr) throws Exception {
|
||
log.info("checkDeletedDataJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
|
||
return dataImportNewService.checkDeletedData(param);
|
||
}
|
||
|
||
/**
|
||
* 删除目标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 拉取
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("dumpContentFolderMemberJob")
|
||
public ReturnT<String> dumpContentFolderMemberJob(String paramStr) throws Exception {
|
||
log.info("dumpContentFolderMemberJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
|
||
return dataImportNewService.dumpContentFolderMemberJob(param);
|
||
}
|
||
/**
|
||
* 富文本图片替换任务
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("richTextImageReplaceJob")
|
||
public ReturnT<String> richTextImageReplaceJob(String paramStr) throws Exception {
|
||
log.info("richTextImageReplaceJob execute start ..................");
|
||
SalesforceParam param = new SalesforceParam();
|
||
try {
|
||
if (StringUtils.isNotBlank(paramStr)) {
|
||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||
}
|
||
} catch (Throwable throwable) {
|
||
return new ReturnT<>(500, "参数解析失败！");
|
||
}
|
||
|
||
return dataImportNewService.richTextImageReplace(param);
|
||
}
|
||
|
||
|
||
} |