data-dump/src/main/java/com/celnet/datadump/job/DataDumpNewJob.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);
    }


}