301 lines
9.9 KiB
Java
301 lines
9.9 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.CommonService;
|
||
import com.celnet.datadump.service.DataImportBatchService;
|
||
import com.celnet.datadump.service.DataImportNewService;
|
||
import com.celnet.datadump.service.DataObjectService;
|
||
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;
|
||
|
||
/**
|
||
* 迁移任务 (新)
|
||
* 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;
|
||
|
||
|
||
/**
|
||
* 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("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.setBeginCreateDate(param.getBeginDate());
|
||
param.setEndCreateDate(param.getEndDate());
|
||
return dataImportBatchService.immigrationUpdateBatch(param);
|
||
}
|
||
|
||
/**
|
||
* 写入个人客户联系人old_id，返写new_id
|
||
* @param paramStr 参数json
|
||
* @return result
|
||
*/
|
||
@XxlJob("getPersonContactJob")
|
||
public ReturnT<String> getPersonContactJob(String paramStr) throws Exception {
|
||
log.info("getPersonContactJob 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.getPersonContact(param);
|
||
}
|
||
|
||
/**
|
||
* 数据更新
|
||
*/
|
||
@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);
|
||
}
|
||
|
||
/**
|
||
* 拉取文件关联表
|
||
* @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 commonService.updateLinkType(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);
|
||
}
|
||
return dataImportNewService.uploadFileNew(param,dataObject);
|
||
}
|
||
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, "参数解析失败！");
|
||
}
|
||
param.setType(2);
|
||
return commonService.incrementNew(param);
|
||
}
|
||
}
|