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 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 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 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 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 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 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 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,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逻辑!!!" , api); break; } } return ReturnT.SUCCESS; } /** * 数据更新 */ @XxlJob("dataUpdateNewJob") public ReturnT 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 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 dumpDocumentLinkJob(String paramStr) throws Exception{ log.info("dumpDocumentLinkJob execute start .................."); return dataImportNewService.dumpDocumentLinkJob(paramStr); } /** * 推送文件关联表 * @return result */ @XxlJob("uploadDocumentLinkJob") public ReturnT uploadDocumentLinkJob(String paramStr) throws Exception{ log.info("uploadDocumentLinkJob execute start .................."); return dataImportNewService.uploadDocumentLinkJob(paramStr); } /** * 创建关联字段 * @return result */ @XxlJob("createLinkTypeFieldJob") public ReturnT 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 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 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 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 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 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 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 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 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 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); } }