kris/datai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
2e6f087732
datai/datai-scenes/datai-scene-salesforce/docs/reference-code/salesforce-pubsub-realtime-sync/DataImportNewServiceImpl.java
DESKTOP-368IV5S\Kris 55b053346d docs(salesforce): 新增技能文档和优化代码 
- 新增 11 个技能文档，指导 AI 完成特定开发任务
- 新增参考代码文件，用于代码分析和重构
- 优化 DataiIntegrationBatchServiceImpl.java，提取常量到 SalesforceConstants.java
- 移动 SessionManager.java 到 auth 模块
- 更新 docs/index.md，添加技能文档索引

主要变更：
- 新增技能文档：Bootstrap、SSOT Hub、信息架构、需求定义、架构决策、提示词资产化、上下文锚定、执行记录、变更归档、闭环复盘、测试提交
- 提取常量：批次处理、状态值、字段名、日志消息等
- 优化代码：改进代码结构和可读性

Related:
- Prompts: docs/prompts/03-创建工作流提示词.md
- Prompts: docs/prompts/07生成技能书提示词.md
2026-01-15 14:09:15 +08:00

4003 lines
191 KiB
Java
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package com.celnet.datadump.service.impl;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson2.JSONObject;
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.enums.FileType;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.DataDumpSpecialParam;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.*;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.soap.partner.*;
import com.sforce.soap.partner.sobject.SObject;
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 okhttp3.Response;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections4.Get;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@Service
@Slf4j
public class DataImportNewServiceImpl implements DataImportNewService {
@Autowired
private SalesforceTargetConnect salesforceTargetConnect;
@Autowired
private DataBatchHistoryService dataBatchHistoryService;
@Autowired
private SalesforceExecutor salesforceExecutor;
@Autowired
private SalesforceConnect salesforceConnect;
@Autowired
private DataObjectService dataObjectService;
@Autowired
private DataBatchService dataBatchService;
@Autowired
private DataFieldService dataFieldService;
@Autowired
private CustomMapper customMapper;
@Autowired
private CommonService commonService;
@Autowired
private LinkConfigService linkConfigService;
@Autowired
private DataLogService dataLogService;
@Autowired
private RichTextLogService richTextLogService;
private static final String TEMP_FILE_PATH = "verify/";
private static String FILE_UPLOAD_URL = "";
private static String FILE_DOWNLOAD_URL = "";
@PostConstruct
public void init() {
List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
for (Map<String, Object> map1 : poll) {
if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
FILE_UPLOAD_URL = (String) map1.get("value");
}
if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) {
FILE_DOWNLOAD_URL = (String) map1.get("value");
}
}
}
static {
// 检测路径是否存在 不存在则创建
File excel = new File(TEMP_FILE_PATH);
if (!excel.exists()) {
boolean mkdir = excel.mkdir();
}
}
/**
* Get返写个人客户联系人入口
*/
@Override
public void getPersonContact(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> result = manualGetPersonContact(param, futures);
if (result != null) {
}
}
} catch (Exception exception) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("immigration error", exception);
throw exception;
}
}
/**
* 组装Get执行参数
*/
public ReturnT<String> manualGetPersonContact(SalesforceParam param, List<Future<?>> futures) throws Exception {
String api = "Contact";
TimeUnit.MILLISECONDS.sleep(1);
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
return new ReturnT<>(500, "api:" + apiNames + " is locked");
}
}
PartnerConnection connect = salesforceTargetConnect.createConnect();
DataObject update = new DataObject();
TimeUnit.MILLISECONDS.sleep(1);
try {
List<SalesforceParam> salesforceParams = null;
update.setName(api);
update.setDataLock(1);
dataObjectService.updateById(update);
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
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) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
manualGetPersonContactId(salesforceParam, connect);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setDataWork(0);
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
log.error("manualImmigration error", e);
throw new RuntimeException(e);
} finally {
if (isFull) {
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
return null;
}
/**
* 获取newId,写入oldId
*/
private void manualGetPersonContactId(SalesforceParam param, PartnerConnection partnerConnection) throws Exception {
String api = param.getApi();
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
Field[] dsrFields = dsr.getFields();
//表内数据总量
Integer count = customMapper.countBySQL(api, "where new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
if (count == 0) {
return;
}
//批量插入200一次
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("总反写 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
for (int i = 0; i < page; i++) {
List<Map<String, Object>> data = customMapper.list("*", api, "new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 200");
int size = data.size();
log.info("总反写 count:{}；当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
SObject[] accounts = new SObject[size];
// 新客户ID，旧联系人ID，用于更新本地数据
Map<String,String> idMaps = new HashMap<>();
// 旧客户ID，新联系人ID，用于更新SF数据
Map<String,String> idMapa = new HashMap<>();
for (Map<String, Object> map : data) {
Map<String, Object> idMap = customMapper.getById("new_id", "Account", map.get("AccountId").toString());
if(idMap.get("new_id") != null && StringUtils.isNotEmpty(idMap.get("new_id").toString())){
// 新客户ID，旧联系人ID
idMaps.put(idMap.get("new_id").toString(),map.get("Id").toString());
}
}
String idStr = "(";
for (String ids : idMaps.keySet()) {
idStr += "'" + ids + "',"; // 拼接每个ID
}
if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束
idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号
}
idStr += ")"; // 添加右括号
try {
String sql = "SELECT Id,AccountId,Account.old_sfdc_id__c FROM Contact where AccountId in " + idStr ;
QueryResult queryResult = partnerConnection.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
SObject[] records = queryResult.getRecords();
com.alibaba.fastjson2.JSONArray objects = DataUtil.toJsonArray(records, dsrFields);
for (int z = 0; z < objects.size(); z++) {
JSONObject jsonObject = objects.getJSONObject(z);
String contactId = jsonObject.getString(Const.ID);
String accountId = jsonObject.getString("AccountId");
String oldAccountId = jsonObject.getString("Account_old_sfdc_id__c");
String id = idMaps.get(accountId);
List<Map<String, Object>> maps = Lists.newArrayList();
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "new_id");
paramMap.put("value", contactId);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
idMapa.put(oldAccountId, contactId);
}
TimeUnit.MILLISECONDS.sleep(1);
int index = 0;
for (Map<String, Object> map : data) {
SObject account = new SObject();
account.setType(api);
account.setField("old_owner_id__c", map.get("OwnerId"));
account.setField("old_sfdc_id__c", map.get("Id"));
account.setId(idMapa.get(map.get("AccountId").toString()));
accounts[index] = account;
index++;
}
SaveResult[] saveResults = partnerConnection.update(accounts);
for (SaveResult saveResult : saveResults) {
if (!saveResult.getSuccess()) {
log.info("-------------saveResults: {}", JSON.toJSONString(saveResult));
String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s, \n数据实体类:\n%s", api, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), JSON.toJSONString(saveResult));
EmailUtil.send("DataDump ERROR", format);
return;
}
}
} catch (Exception e) {
log.error("manualGetPersonContactId error api:{}", api, e);
throw e;
}
}
SalesforceParam countParam = new SalesforceParam();
countParam.setApi(api);
countParam.setBeginCreateDate(beginDate);
countParam.setEndCreateDate(DateUtils.addSeconds(endDate, -1));
// 存在isDeleted 只查询IsDeleted为false的
if (dataFieldService.hasDeleted(countParam.getApi())) {
countParam.setIsDeleted(false);
} else {
// 不存在 过滤
countParam.setIsDeleted(null);
}
Integer sfNum = 0;
// 不能 count Id
if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
sfNum = commonService.countSfNum(partnerConnection, countParam);
}
UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
updateQw.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
.set("target_sf_num", sfNum);
dataBatchHistoryService.update(updateQw);
UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
updateQw2.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", endDate)
.set("sf_add_num", sfNum);
dataBatchService.update(updateQw2);
}
/**
* 数据更新Update入口
*/
@Override
public ReturnT<String> immigrationUpdateNew(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
if (1 == param.getType()){
return maualupdateSfDataNew(param, futures);
}else {
return updateIncrementalSfDataNew(param, futures);
}
} else {
if (1 == param.getType()){
autoUpdateSfDataNew(param, futures);
}else {
autoUpdateIncrementalSfDataNew(param, futures);
}
}
return ReturnT.SUCCESS;
} catch (InterruptedException e) {
throw e;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("immigrationUpdate error", throwable);
throw throwable;
}
}
/**
* 组装【单表】【存量】Update参数
*/
public ReturnT<String> maualupdateSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis = DataUtil.toIdList(param.getApi());
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");
}
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
String message = "api:" + apiNames + " is locked";
String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
EmailUtil.send("DataDump ERROR", format);
return new ReturnT<>(500, message);
}
}
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
for (String api : apis) {
DataObject update = dataObjectService.getById(api);
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()){
UpdateSfDataNew(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
UpdateSfDataNew(salesforceParam, partnerConnection,update);
} 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 (Exception e) {
throw e;
} finally {
if (isFull) {
update.setNeedUpdate(false);
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
}
return ReturnT.SUCCESS;
}
/**
* 组装【单表】【增量】Update参数
*/
public ReturnT<String> updateIncrementalSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis = DataUtil.toIdList(param.getApi());
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
return new ReturnT<>(500, "api:" + apiNames + " is locked");
}
}
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
for (String api : apis) {
DataObject update = dataObjectService.getById(api);
try {
//无创建时间对象
if (!dataFieldService.hasCreatedDate(api)) {
UpdateSfShareDataNew(api, partnerConnection, update);
continue;
}
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
DataBatch t = dataBatchService.getOne(dbQw);
SalesforceParam salesforceParam = param.clone();
salesforceParam.setApi(t.getName());
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
salesforceParam.setEndCreateDate(t.getSyncEndDate());
salesforceParam.setBatch(new AtomicInteger(1).getAndIncrement());
if (param.getIsSingleThread()){
UpdateSfDataNew(salesforceParam, partnerConnection,update);
}else {
// 手动任务优先执行
Future<?> future = salesforceExecutor.execute(() -> {
try {
UpdateSfDataNew(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
update.setNeedUpdate(false);
} catch (Exception e) {
throw e;
} finally {
if (isFull) {
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
return ReturnT.SUCCESS;
}
/**
* 组装【多表】【存量】Update参数
*/
private void autoUpdateSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
String message = "api:" + apiNames + " is locked";
String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
EmailUtil.send("DataDump ERROR", format);
return ;
}
}
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");
}
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("need_update", 1)
.orderByAsc("data_index")
.last(" limit 10");
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
while (true) {
List<DataObject> dataObjects = dataObjectService.list(qw);
//判断dataObjects是否为空
if (CollectionUtils.isEmpty(dataObjects)) {
return;
}
for (DataObject dataObject : dataObjects) {
TimeUnit.MILLISECONDS.sleep(1);
DataObject update = new DataObject();
update.setName(dataObject.getName());
update.setDataLock(1);
dataObjectService.updateById(update);
TimeUnit.MILLISECONDS.sleep(1);
try {
String api = dataObject.getName();
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()){
UpdateSfDataNew(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
UpdateSfDataNew(salesforceParam, partnerConnection, dataObject);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
} catch (Exception e) {
throw e;
} finally {
update.setNeedUpdate(false);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
}
}
/**
* 组装【多表】【增量】Update参数
*/
private void autoUpdateIncrementalSfDataNew(SalesforceParam param, List<Future<?>> futures) throws Exception {
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
String message = "api:" + apiNames + " is locked";
String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
EmailUtil.send("DataDump ERROR", format);
return ;
}
}
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("need_update", 1)
.orderByAsc("data_index")
.last(" limit 10");
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
while (true) {
List<DataObject> dataObjects = dataObjectService.list(qw);
//判断dataObjects是否为空
if (CollectionUtils.isEmpty(dataObjects)) {
return;
}
for (DataObject dataObject : dataObjects) {
TimeUnit.MILLISECONDS.sleep(1);
DataObject update = new DataObject();
update.setName(dataObject.getName());
update.setDataLock(1);
dataObjectService.updateById(update);
TimeUnit.MILLISECONDS.sleep(1);
try {
String api = dataObject.getName();
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
DataBatch t = dataBatchService.getOne(dbQw);
AtomicInteger batch = new AtomicInteger(1);
SalesforceParam salesforceParam = param.clone();
salesforceParam.setApi(t.getName());
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
salesforceParam.setEndCreateDate(t.getSyncEndDate());
salesforceParam.setBatch(batch.getAndIncrement());
if (param.getIsSingleThread()){
UpdateSfDataNew(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
UpdateSfDataNew(salesforceParam, partnerConnection,dataObject);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
} catch (Exception e) {
throw e;
} finally {
update.setNeedUpdate(false);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
}
}
/**
* 执行Update更新数据
*/
private void UpdateSfDataNew(SalesforceParam param, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+ SystemConfigCode.INFO_FLAG+"'").get(0);
String api = param.getApi();
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> list = dataFieldService.list(dbQw);
String sql = "";
String sql2 = "";
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
if (1 == param.getType()) {
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000 ";
}else {
sql = "where new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000 ";
}
}else {
String updateDateField = dataFieldService.returnUpdateDateField(api);
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000 ";
}else {
sql = "where new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000 ";
}
}
//表内数据总量
Integer count = customMapper.countBySQL(api, sql);
if(count == 0){
log.info("没有需要更新的数据，API: {}，开始时间: {}，结束时间: {}", api, beginDateStr, endDateStr);
return;
}
//查询当前对象多态字段映射
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
List<LinkConfig> configs = linkConfigService.list(queryWrapper);
Map<String, String> fieldMap = new HashMap<>();
if (!configs.isEmpty()) {
fieldMap = configs.stream()
.collect(Collectors.toMap(
LinkConfig::getField, // Key提取器
LinkConfig::getLinkField, // Value提取器
(oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
));
}
int targetCount = 0;
int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;
log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
for (int z = 0; z < num; z++){
List<Map<String, Object>> mapsList = customMapper.list("*", api, sql2);
int size = mapsList.size();
log.info("总批次:{}；当前批次：{}；当前批次数据量:{}；-开始时间:{}；-结束时间:{}；-api:{}；", num,z,size, beginDateStr, endDateStr, api);
//批量插入200一次
int page = size%200 == 0 ? size/200 : (size/200) + 1;
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，查询并组装第" + (z*50 + i) + "批数据", "QueryDB");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
log.info("当前批次：{} 结束!! -开始时间:{}；-结束时间:{}；-api:{}；", z, beginDateStr, endDateStr, api);
break;
}
List<Map<String, Object>> mapList = mapsList.subList(startIndex, endIndex);
SObject[] accounts = new SObject[mapList.size()];
int j = 0;
try {
for (Map<String, Object> map : mapList) {
//需要置空的参数
List<String> fieldsToNull = new ArrayList<>();
SObject account = new SObject();
account.setType(api);
String message = null;
//给对象赋值
for (DataField dataField : list) {
String field = dataField.getField();
String reference_to = dataField.getReferenceTo();
String value = String.valueOf(map.get(field));
//根据旧sfid查找引用对象新sfid
if (field.equals("Id")) {
account.setId(String.valueOf(map.get("new_id")));
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
continue;
} else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {
if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
//判断reference_to内是否包含User字符串
if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
reference_to = "User";
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield)){
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
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){
continue;
}
Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
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;
}
}
} else {
if(StringUtils.isNotBlank(dataField.getSfType())){
if (map.get(field) != null){
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
fieldsToNull.add(field);
}
}
}
}
if (message != null){
continue;
}
if (!fieldsToNull.isEmpty()){
String[] fieldsToNullArray = fieldsToNull.toArray(new String[0]);
account.setField("fieldsToNull", fieldsToNullArray);
}
if (dataObject.getIsEditable()){
account.setField("old_owner_id__c", map.get("OwnerId"));
account.setField("old_sfdc_id__c", map.get("Id"));
}
accounts[j++] = account;
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts,list);
}
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，更新SF第" + (z*50 + i) + "批数据", "UpdateSF");
SaveResult[] saveResults = partnerConnection.update(accounts);
for (SaveResult saveResult : saveResults) {
if (!saveResult.getSuccess()) {
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", JSON.toJSONString(saveResult.getErrors()));
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, saveResult.getId());
String format = String.format("数据更新 error, api name: %s, \ncause:\n%s", api, JSON.toJSONString(saveResult));
log.info(format);
}else {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 1);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", null);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, saveResult.getId());
targetCount ++;
}
}
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException interruptedException){
return;
}catch (Throwable e) {
log.info(JSON.toJSONString(e));
}
}
}
UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
updateQw.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
.set("target_update_num", num);
dataBatchHistoryService.update(updateQw);
UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
updateQw2.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", endDate)
.set("sf_update_num", targetCount);
dataBatchService.update(updateQw2);
log.info("数据更新完成，API: {}，总更新数: {}，开始时间: {}，结束时间: {}", api, targetCount, beginDateStr, endDateStr);
}
/**
* 执行Update更新数据
*/
private void UpdateSfShareDataNew(String api, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> list = dataFieldService.list(dbQw);
String sql = "";
String sql2 = "";
String beginDateStr = null;
String endDateStr = null;
String updateDateField = dataFieldService.returnUpdateDateField(api);
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
}else {
sql = "where new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
}
//表内数据总量
Integer count = customMapper.countBySQL(api, sql);
if(count == 0){
return;
}
//查询当前对象多态字段映射
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
List<LinkConfig> configs = linkConfigService.list(queryWrapper);
Map<String, String> fieldMap = new HashMap<>();
if (!configs.isEmpty()) {
fieldMap = configs.stream()
.collect(Collectors.toMap(
LinkConfig::getField, // Key提取器
LinkConfig::getLinkField, // Value提取器
(oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
));
}
int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;
log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
for (int z = 0; z < num; z++){
List<Map<String, Object>> mapsList = customMapper.list("*", api, sql2+ z * 10000 + ",10000");
int size = mapsList.size();
log.info("总Update数据 count:{}；当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,z, beginDateStr, endDateStr, api);
//批量插入200一次
int page = size%200 == 0 ? count/200 : (count/200) + 1;
for (int i = 0; i < page; i++) {
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
break;
}
List<Map<String, Object>> mapList = mapsList.subList(startIndex, endIndex);
SObject[] accounts = new SObject[mapList.size()];
int j = 0;
try {
for (Map<String, Object> map : mapList) {
SObject account = new SObject();
account.setType(api);
String message = null;
//给对象赋值
for (DataField dataField : list) {
String field = dataField.getField();
String reference_to = dataField.getReferenceTo();
String value = String.valueOf(map.get(field));
//根据旧sfid查找引用对象新sfid
if (field.equals("Id")) {
account.setId(String.valueOf(map.get("new_id")));
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
continue;
} else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {
if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
//判断reference_to内是否包含User字符串
if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
reference_to = "User";
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield)){
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
}
Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
account.setField(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"异常！";
log.info(message);
break;
}
}
} else {
if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
account.setField(field, map.get(field));
}
}
}
if (message!=null){
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateById(api, maps, map.get("Id").toString());
continue;
}
if (dataObject.getIsEditable()){
account.setField("old_owner_id__c", map.get("OwnerId"));
account.setField("old_sfdc_id__c", map.get("Id"));
}
accounts[j++] = account;
}
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts,list);
}
SaveResult[] saveResults = partnerConnection.update(accounts);
for (SaveResult saveResult : saveResults) {
if (!saveResult.getSuccess()) {
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", JSON.toJSONString(saveResult.getErrors()));
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, saveResult.getId());
String format = String.format("数据更新 error, api name: %s, \ncause:\n%s", api, JSON.toJSONString(saveResult));
log.info(format);
}else {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 1);
maps.add(linkMap);
customMapper.updateByNewId(maps,api, saveResult.getId());
}
}
} catch (Throwable e) {
log.info(JSON.toJSONString(e));
}
}
}
}
/**
* 打印SF交互数据明细
*/
public void printlnAccountsDetails(SObject[] accounts,List<DataField> list) {
for (int i = 0; i < accounts.length; i++) {
SObject account = accounts[i];
System.out.println("--- 对象数据[" + i + "] ---");
if (account == null){
continue;
}
// 获取对象所有字段名
for (DataField dataField : list) {
try {
Object value = account.getField(dataField.getField());
System.out.println(dataField.getField() + ": " + (value != null ? value.toString() : "null"));
} catch (Exception e) {
System.out.println(dataField.getField() + ": [权限不足或字段不存在]");
}
}
System.out.println("old_owner_id__c: " + account.getField("old_owner_id__c"));
System.out.println("old_sfdc_id__c: " + account.getField("old_sfdc_id__c"));
}
}
/**
* 返回SF交互数据错误明细
*/
public Map<String,String> returnErrorAccountsDetails(SObject[] accounts,List<DataField> list,String errorId) {
HashMap<String, String> map = new HashMap<>();
for (int i = 0; i < accounts.length; i++) {
SObject account = accounts[i];
if (account == null){
continue;
}
if (errorId != null && (errorId.equals(account.getId()) || errorId.equals(account.getField("Id")))){
for (DataField dataField : list) {
try {
Object value = account.getField(dataField.getField());
map.put(dataField.getField(),String.valueOf(value));
System.out.println(dataField.getField() + ": " + value);
} catch (Exception e) {
System.out.println(dataField.getField() + ": [权限不足或字段不存在]");
}
}
map.put("old_owner_id__c",String.valueOf(account.getField("old_owner_id__c")));
map.put("old_sfdc_id__c",String.valueOf(account.getField("old_sfdc_id__c")));
}
}
return map;
}
/**
* 获取DocumentLink
*/
@Override
public ReturnT<String> dumpDocumentLinkJob(String paramStr) throws Exception {
String api = "ContentDocumentLink";
PartnerConnection partnerConnection = salesforceConnect.createConnect();
Integer count = customMapper.countBySQL("ContentDocument", "where new_id is not null");
log.info("ContentDocument Total size：" + count);
int page = count%200 == 0 ? count/200 : (count/200) + 1;
int totalSize = 0;
for (int i = 0; i < page; i++) {
List<Map<String, Object>> list = customMapper.list("Id", "ContentDocument", "new_id is not null order by Id limit " + i * 200 + ",200");
if (list.isEmpty()){
break;
}
totalSize = totalSize +list.size();
log.info("dumpContentDocumentLink By ContentDocument Now size：" + totalSize);
DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
List<String> fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList());
Field[] dsrFields = dsr.getFields();
try {
for (Map<String, Object> map : list) {
String contentDocumentId = (String) map.get("Id");
String sql = "SELECT Id, LinkedEntityId, LinkedEntity.Type, ContentDocumentId, Visibility, ShareType, SystemModstamp, IsDeleted FROM ContentDocumentLink where ContentDocumentId = '" + contentDocumentId + "'";
com.alibaba.fastjson2.JSONArray objects = null;
QueryResult queryResult = partnerConnection.queryAll(sql);
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
commonService.saveOrUpdate(api, fields, records, objects, true);
}
} catch (Throwable e) {
log.error("dumpDocumentLinkJob error message:{}", e.getMessage());
return ReturnT.FAIL;
}
}
log.info("dumpDocumentLink Success !!! ");
return ReturnT.SUCCESS;
}
/**
* 推送DocumentLink
*/
@Override
public ReturnT<String> uploadDocumentLinkJob(String paramStr) throws Exception {
String api = "ContentDocumentLink";
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> dataFields = dataFieldService.list(dbQw);
PartnerConnection connection = salesforceTargetConnect.createConnect();
List<Map<String, Object>> list = customMapper.list("Id", "ContentDocument", "new_id is not null");
try {
if (list != null && !list.isEmpty()) {
//表内数据总量
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++) {
List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit 200");
SObject[] accounts = new SObject[linkList.size()];
String[] ids = new String[linkList.size()];
int index = 0;
for (Map<String, Object> map : linkList) {
String linkedEntityId = (String) map.get("LinkedEntityId");
String id = (String) map.get("Id");
String contentDocumentId = (String) map.get("ContentDocumentId");
String linkedEntityType = (String) map.get("LinkedEntity_Type");
String shareType = (String) map.get("ShareType");
String Visibility = (String) map.get("Visibility");
// dataObject查询
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("name", linkedEntityType);
List<DataObject> objects = dataObjectService.list(qw);
if (!objects.isEmpty()) {
Map<String, Object> dMap = customMapper.getById("new_id", "ContentDocument", contentDocumentId);
Map<String, Object> lMap = customMapper.getById("new_id", linkedEntityType, linkedEntityId);
SObject account = new SObject();
account.setType(api);
if (dMap != null){
account.setField("ContentDocumentId", dMap.get("new_id").toString());
}else {
String errorMessage = String.format("ContentDocumentLink Id: %s，对应的ContentDocumentId: %s 数据不存在！", id, contentDocumentId);
log.info(errorMessage);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", errorMessage);
maps.add(linkMap1);
customMapper.updateById(api, maps, id);
continue;
}
if (lMap != null){
account.setField("LinkedEntityId", lMap.get("new_id").toString());
}else {
String errorMessage = String.format("ContentDocumentLink Id: %s，对应的LinkedEntityId: %s 数据不存在！", id, linkedEntityId);
log.info(errorMessage);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", errorMessage);
maps.add(linkMap1);
customMapper.updateById(api, maps, id);
continue;
}
account.setField("ShareType", shareType);
account.setField("Visibility", Visibility);
ids[index] = id;
accounts[index] = account;
index++;
}
}
try {
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts,dataFields);
}
SaveResult[] saveResults = connection.create(accounts);
;
for (int j = 0; j < saveResults.length; j++) {
if (saveResults[j].getSuccess()) {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m = new HashMap<>();
m.put("key", "new_id");
m.put("value", saveResults[j].getId());
maps.add(m);
customMapper.updateById(api, maps, ids[j]);
log.info("ContentDocumentLink Id: {}，对应的new_id: {} 更新成功！ " , ids[j], saveResults[j].getId());
}else{
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", JSON.toJSONString(saveResults[j].getErrors()));
maps.add(linkMap1);
customMapper.updateById(api, maps, ids[j]);
log.error("Id:{},saveResults: {}",ids[j], JSON.toJSONString(saveResults[j]));
}
}
} catch (Exception e) {
log.error("uploadDocumentLinkJob error message:{}", e.getMessage());
throw new RuntimeException(e);
}
}
}
} catch (Exception e) {
log.error("uploadDocumentLink error api:{}, data:{}", api, com.alibaba.fastjson2.JSON.toJSONString(list), e);
return ReturnT.FAIL;
}
log.info("uploadDocumentLink Success !!! ");
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> dumpFileNew(SalesforceParam param) throws Exception {
String downloadUrl = null;
QueryWrapper<DataObject> wrapper = new QueryWrapper<>();
if (StringUtils.isNotBlank(param.getApi())) {
wrapper.in("name",DataUtil.toIdList(param.getApi()));
}else {
wrapper.isNotNull("blob_field");
}
List<DataObject> objectList = dataObjectService.list(wrapper);
if (objectList.isEmpty()){
log.info("没有对象存在文件二进制字段！不进行文件下载");
return ReturnT.FAIL;
}
List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
for (Map<String, Object> map1 : poll) {
if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) {
downloadUrl = (String) map1.get("value");
}
}
if (StringUtils.isEmpty(downloadUrl)) {
EmailUtil.send("DumpFile ERROR", "文件下载失败！下载地址未配置");
return ReturnT.FAIL;
}
PartnerConnection connect = salesforceConnect.createConnect();
List<Future<?>> futures = Lists.newArrayList();
for (DataObject dataObject : objectList) {
DataObject update = new DataObject();
log.info("dump file api:{}, field:{}", dataObject.getName(), dataObject.getBlobField());
try {
String api = dataObject.getName();
update.setName(dataObject.getName());
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
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());
}
if (Const.FILE_TYPE == FileType.SERVER) {
Date defaultBeginDate = DataUtil.DEFAULT_BEGIN_DATE;
Calendar calendar = Calendar.getInstance();
int currentYear = calendar.get(Calendar.YEAR);
calendar.setTime(defaultBeginDate);
int beginYear = calendar.get(Calendar.YEAR);
// 检测路径是否存在 不存在则创建
File baseDir = new File(Const.SERVER_FILE_PATH + "/" + api);
if (!baseDir.exists()) {
baseDir.mkdirs();
}
// 创建从开始年份到当前年份的子目录
for (int year = beginYear; year <= currentYear; year++) {
File yearDir = new File(Const.SERVER_FILE_PATH + "/" + api + "-" + year);
if (!yearDir.exists()) {
yearDir.mkdir();
}
}
}
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
String finalDownloadUrl = downloadUrl;
Future<?> future = salesforceExecutor.execute(() -> {
try {
saveFile(salesforceParam, connect, finalDownloadUrl,dataObject.getName(),dataObject.getBlobField());
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setDataWork(0);
} catch (InterruptedException e) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
throw e;
} catch (Throwable e) {
throw new RuntimeException(e);
} finally {
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
return ReturnT.SUCCESS;
}
/**
* 下载文件
*/
private void saveFile(SalesforceParam param, PartnerConnection Connection ,String downloadUrl, String api, String field) throws Exception {
String extraSql = "";
if (dataFieldService.hasDeleted(api)) {
extraSql += " AND IsDeleted = false ";
}
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
String token = Connection.getSessionHeader().getSessionId();
String name = "Name";
if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
name = "Title,FileExtension";
}
Map<String, String> headers = Maps.newHashMap();
headers.put("Authorization", "Bearer " + token);
headers.put("connection", "keep-alive");
Integer count = customMapper.countBySQL(api, "where is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql );
if (count == 0) {
return;
}
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("api：{}；总文件数 count:{}；总批次 ：{}；-开始时间:{}；-结束时间:{}；",api, count, page,beginDateStr, endDateStr);
//总插入数
for (int i = 0; i < page; i++) {
log.info("api：{} 批次:{}；-开始时间:{}；-结束时间:{}；",api, i, beginDateStr, endDateStr);
// 获取未存储的附件id
List<Map<String, Object>> list = customMapper.list("Id,CreatedDate, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit 200");
for (Map<String, Object> map : list) {
int failCount = 0;
String id = null;
// 上传完毕 更新附件信息
List<Map<String, Object>> maps = Lists.newArrayList();
try {
String fileName;
id = (String) map.get("Id");
if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
fileName = map.get("Title") + "." + map.get("FileExtension");
}else {
fileName = (String) map.get("Name");
}
String year = map.get("CreatedDate").toString().substring(0, 4);
// 判断路径是否为空
if (StringUtils.isNotBlank(fileName)) {
String filePath = api + "-" + year + "/" + id + "_" + fileName;
// 拼接url
String url = downloadUrl + String.format(Const.SF_FILE_URL, api, id, field);
log.info("文件下载请求地址：{}",url);
while (failCount < Const.MAX_FAIL_COUNT){
Response response = HttpUtil.doGet(url, null, headers);
if (response.body() != null && response.code() == 200) {
InputStream inputStream = response.body().byteStream();
dumpToServer(headers, id, filePath, url, response, inputStream);
Map<String, Object> paramMap = Maps.newHashMap();
if ("Document".equals(api)) {
paramMap.put("key", "localUrl");
} else {
paramMap.put("key", "url");
}
paramMap.put("value", filePath);
maps.add(paramMap);
inputStream.close();
break;
}else {
failCount++;
log.info("文件下载失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + response.message());
}
}
if (failCount == Const.MAX_FAIL_COUNT) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_dump");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
}else {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_dump");
paramMap.put("value", 1);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
TimeUnit.MILLISECONDS.sleep(1);
}
}
} catch (InterruptedException interruptedException){
log.info("文件下载手动终止！");
return;
} catch (Exception e) {
log.error("文件下载失败！, id: {}, 错误信息：{}", id ,e.getMessage());
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_dump");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
EmailUtil.send("File Dump ERROR", "文件下载失败！, id: "+ id + "，错误信息：" + e.getMessage());
}
}
}
}
/**
* 下载文件到服务器
*
* @param headers 请求头
* @param id id
* @param filePath 文件路径
* @param url 链接
* @param response 响应
* @param inputStream 流
* @throws IOException exception
*/
private void dumpToServer(Map<String, String> headers, String id, String filePath, String url, Response response, InputStream inputStream) throws IOException {
String path = Const.SERVER_FILE_PATH + "/" + filePath;
log.info("--------文件名称："+path);
long offset = 0L;
RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
while (true) {
try {
TimeUnit.MILLISECONDS.sleep(1);
// 保存到本地
byte[] buf = new byte[8192];
int len = 0;
if (offset > 0) {
inputStream.skip(offset);
accessFile.seek(offset);
}
while ((len = inputStream.read(buf)) != -1) {
accessFile.write(buf, 0, len);
offset += len;
}
break;
}catch (InterruptedException interruptedException){
return;
}catch (Exception e) {
if (offset <= 0) {
throw e;
}
System.out.println("file dump to server EOF ERROR try to reconnect");
response.close();
response = HttpUtil.doGet(url, null, headers);
assert response.body() != null;
inputStream = response.body().byteStream();
}
}
accessFile.close();
}
@Override
public ReturnT<String> uploadFileNew(SalesforceParam param, DataObject dataObject) throws Exception {
String uploadUrl = null;
List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
for (Map<String, Object> map1 : poll) {
if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
uploadUrl = (String) map1.get("value");
}
}
if (StringUtils.isBlank(uploadUrl)) {
EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
return ReturnT.FAIL;
}
if (StringUtils.isEmpty(dataObject.getBlobField())){
log.info("没有对象存在文件二进制字段！不进行文件上传");
}
PartnerConnection connect = salesforceTargetConnect.createConnect();
List<Future<?>> futures = Lists.newArrayList();
DataObject update = new DataObject();
log.info("dump file api:{}, field:{}", dataObject.getName(), dataObject.getBlobField());
try {
String api = dataObject.getName();
update.setName(dataObject.getName());
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
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) {
String finalDownloadUrl = uploadUrl;
Future<?> future = salesforceExecutor.execute(() -> {
try {
uploadFile(salesforceParam, connect, finalDownloadUrl,dataObject.getName());
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setDataWork(0);
} catch (InterruptedException e) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
throw e;
} catch (Throwable e) {
throw new RuntimeException(e);
} finally {
update.setDataLock(0);
dataObjectService.updateById(update);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
return ReturnT.SUCCESS;
}
/**
* 上传Attachment
*/
private void uploadAttachment(SalesforceParam param, PartnerConnection connection ,String uploadUrl, String api) throws Exception {
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
String token = connection.getSessionHeader().getSessionId();
Map<String, String> headers = Maps.newHashMap();
headers.put("Authorization", "Bearer " + token);
headers.put("connection", "keep-alive");
Integer count = customMapper.countBySQL(api, "where is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'");
if (count == 0) {
return;
}
int page = count%1000 == 0 ? count/1000 : (count/1000) + 1;
log.info("总文件数 count:{}；总批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
//总插入数
for (int i = 0; i < page; i++) {
log.info("总文件数 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_Type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000");
for (Map<String, Object> map : list) {
int failCount = 0;
String id = null;
// 上传完毕 更新附件信息
List<Map<String, Object>> maps = Lists.newArrayList();
CloseableHttpResponse response = null;
String respContent = null;
try {
id = (String) map.get("Id");
String url_fileName = (String) map.get("url");
String parentId = (String) map.get("ParentId");
String parentType = (String) map.get("Parent_Type");
// 判断路径是否为空
if (StringUtils.isNotBlank(url_fileName)) {
String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
File file = new File(filePath);
boolean exists = file.exists();
if (!exists) {
log.info("文件不存在");
}
String fileName = (String) map.get("Name");
log.info("文件名称：" + fileName);
// dataObject查询
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("name", parentType);
List<DataObject> objects = dataObjectService.list(qw);
if (objects.isEmpty()) {
log.info("关联对象不存在： {}", parentType);
}
Map<String, Object> lMap = customMapper.getById("new_id",parentType, parentId);
// 拼接url
String url = uploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
HttpPost httpPost = new HttpPost(url);
httpPost.setHeader("Authorization", "Bearer " + token);
httpPost.setHeader("connection", "keep-alive");
com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
credentialsJsonParam.put("parentId", lMap.get("new_id"));
credentialsJsonParam.put("Name", fileName);
MultipartEntityBuilder builder = MultipartEntityBuilder.create();
builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
builder.addBinaryBody("Body", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
HttpEntity entity = builder.build();
httpPost.setEntity(entity);
CloseableHttpClient httpClient = HttpClients.createDefault();
while (failCount < Const.MAX_FAIL_COUNT) {
response = httpClient.execute(httpPost);
if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400) {
HttpEntity he = response.getEntity();
if (he != null) {
respContent = EntityUtils.toString(he, "UTF-8");
String newId = com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id").toString();
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "new_id");
paramMap.put("value", newId);
maps.add(paramMap);
}
break;
} else {
failCount++;
log.error("文件上传失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + JSON.toJSONString(response));
}
}
if (failCount == Const.MAX_FAIL_COUNT) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
}else {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 1);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
TimeUnit.MILLISECONDS.sleep(1);
}
}
} catch (InterruptedException interruptedException){
return;
} catch (Exception e) {
if (response != null) {
try {
response.close();
} catch (IOException re) {
log.error("exception message", re);
}
}
log.error("文件上传失败！, id: {}, 错误信息：{}", id ,JSON.toJSONString(e));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
EmailUtil.send("File Upload ERROR", "文件上传失败！, id: "+ id + "，错误信息：" + JSON.toJSONString(e));
}
}
}
}
/**
* 上传文件
*/
private void uploadFile(SalesforceParam param, PartnerConnection connection ,String uploadUrl, String api) throws Exception {
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
String token = connection.getSessionHeader().getSessionId();
Map<String, String> headers = Maps.newHashMap();
headers.put("Authorization", "Bearer " + token);
headers.put("connection", "keep-alive");
Integer count = customMapper.countBySQL("ContentDocument", "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'");
if (count == 0) {
return;
}
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("总文件数 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
//总插入数
for (int i = 0; i < page; i++) {
// 获取未存储的附件id
List<Map<String, Object>> documentList = customMapper.list("Id " , "ContentDocument", " new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 200 + ",200");
for (Map<String, Object> documentMap : documentList) {
String documentId = (String) documentMap.get("Id");
// 获取未存储的附件id
List<Map<String, Object>> list = customMapper.list("Id, url, PathOnClient, Title, ContentDocumentId, VersionNumber", api, "ContentDocumentId = '" + documentId + "' and new_id is null ORDER BY VersionNumber ASC");
if (CollectionUtils.isEmpty(list)) {
continue;
}
String finalUploadUrl = uploadUrl;
String newDocumentId = null;
for (Map<String, Object> map : list) {
int failCount = 0;
String id = null;
// 上传完毕 更新附件信息
List<Map<String, Object>> maps = Lists.newArrayList();
CloseableHttpResponse response = null;
String respContent = null;
try {
id = (String) map.get("Id");
String url_fileName = (String) map.get("url");
String fileName = (String) map.get("PathOnClient");
String title = (String) map.get("Title");
String oldDocumentId = (String) map.get("ContentDocumentId");
Integer versionNumber = Integer.valueOf(map.get("VersionNumber").toString());
log.info("文件名称：" + fileName);
// 判断路径是否为空
if (StringUtils.isNotBlank(url_fileName)) {
String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
File file = new File(filePath);
boolean exists = file.exists();
if (!exists) {
log.info("文件不存在");
break;
}
// 拼接url
String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
HttpPost httpPost = new HttpPost(url);
httpPost.setHeader("Authorization", "Bearer " + token);
httpPost.setHeader("connection", "keep-alive");
com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
credentialsJsonParam.put("title", title);
credentialsJsonParam.put("pathOnClient", fileName);
if (newDocumentId != null) {
credentialsJsonParam.put("ContentDocumentId", newDocumentId);
}
MultipartEntityBuilder builder = MultipartEntityBuilder.create();
builder.addTextBody("entity_content", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
builder.addBinaryBody("VersionData", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
HttpEntity entity = builder.build();
httpPost.setEntity(entity);
CloseableHttpClient httpClient = HttpClients.createDefault();
while (failCount < Const.MAX_FAIL_COUNT) {
response = httpClient.execute(httpPost);
if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400 && response.getEntity() != null) {
HttpEntity he = response.getEntity();
respContent = EntityUtils.toString(he, "UTF-8");
String newId = String.valueOf(com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id"));
if (StringUtils.isBlank(newId)) {
log.error("文件上传错误，返回实体信息：" + respContent);
break;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "new_id");
paramMap.put("value", newId);
maps.add(paramMap);
if (versionNumber == 1) {
// 更新documentId
String dId = commonService.getDocumentId(connection, newId);
List<Map<String, Object>> dList = new ArrayList<>();
Map<String, Object> dMap = Maps.newHashMap();
dMap.put("key", "new_id");
dMap.put("value", dId);
dList.add(dMap);
customMapper.updateById("ContentDocument", dList, oldDocumentId);
newDocumentId = dId;
}
break;
} else {
failCount++;
log.error("文件上传失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + JSON.toJSONString(response));
}
}
if (failCount == Const.MAX_FAIL_COUNT) {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
}else {
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 1);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
TimeUnit.MILLISECONDS.sleep(1);
}
}
} catch (Exception e) {
if (response != null) {
try {
response.close();
} catch (IOException re) {
log.error("exception message", re);
}
}
log.error("文件上传失败！, id: {}, 错误信息：{}", id ,JSON.toJSONString(e));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");
paramMap.put("value", 2);
maps.add(paramMap);
customMapper.updateById(api, maps, id);
EmailUtil.send("File Upload ERROR", "文件上传失败！, id: "+ id + "，错误信息：" + JSON.toJSONString(e));
}
}
}
}
}
/**
* updateLinkType入口
*/
@Override
public ReturnT<String> updateLinkTypeJob(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (param.getType() == 1){
return updateLinkType(param, futures);
}else {
return updateLinkTypeBatch(param,futures);
}
} catch (InterruptedException e) {
throw e;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("updateLinkTypeJob error", throwable);
throw throwable;
}
}
/**
* 【batch逻辑】组装updateLinkType执行参数
*/
public ReturnT<String> updateLinkTypeBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis;
if (StringUtils.isBlank(param.getApi())) {
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("is_link","1").eq("is_create",1);
apis = linkConfigService.list(queryWrapper).stream().map(LinkConfig::getApi).collect(Collectors.toList());
} else {
apis = DataUtil.toIdList(param.getApi());
}
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");
}
QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
wrapper.in("api", apis);
Map<String, String> resultMap = dataObjectService.list().stream()
.collect(Collectors.toMap(
DataObject::getKeyPrefix,
DataObject::getName,
(existing, replacement) -> replacement));
for (String api : apis) {
try {
if (!dataFieldService.hasCreatedDate(api)){
log.info("当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
EmailUtil.send("DataDump ERROR", "当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
break;
}
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
if (linkConfigs.isEmpty()){
continue;
}
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw1 = new QueryWrapper<>();
dbQw1.eq("name", api);
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
dbQw1.eq("sync_start_date", beginDateStr); // 等于开始时间
dbQw1.eq("sync_end_date", endDateStr); // 等于结束时间
}
List<DataBatch> list = dataBatchService.list(dbQw1);
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());
}
Set<String> safeSet = new HashSet<>();
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
updateLinkBatch(salesforceParam, resultMap,safeSet,linkConfigs);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
if (!safeSet.isEmpty()){
String format = String.format("更新关联类型 error, api name: %s, \ncause:\n%s 前三位编码对象不存在！！", api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
EmailUtil.send("DataDump ERROR", format);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
} catch (Exception e) {
throw e;
}
}
return ReturnT.SUCCESS;
}
/**
* 执行数据updateLinkType
*/
private void updateLinkBatch(SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
String sql = "CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
String allFiled = "Id";
if (linkConfigs.size() == 1){
for (LinkConfig linkConfig : linkConfigs) {
sql = sql + " and " + linkConfig.getField() + " is not null";
allFiled = allFiled + "," + linkConfig.getField();
}
}else {
String info = null;
for (LinkConfig linkConfig : linkConfigs) {
if (StringUtils.isBlank(info)){
info = linkConfig.getField() + " is not null" ;
}else {
info = info + " or " + linkConfig.getField() + " is not null";
}
allFiled = allFiled + "," + linkConfig.getField();
}
sql = sql +" and ("+ info +")";
}
// 表内数据总量
Integer count = customMapper.countBySQL(param.getApi(), "where " + sql);
log.info("表api：{} 存在" +count+ "条需更新数据！开始时间:{},结束时间：{}", param.getApi(), beginDateStr, endDateStr);
if (count >0 ) {
int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
for (int i = 0; i < page; i++) {
log.info("表api：{}，批次：{}，单批次数：2000，开始时间：{}，结束时间：{}，执行数据更新!", param.getApi(), i,beginDateStr, endDateStr);
List<Map<String, Object>> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000");
for (int j = 1; j <= mapList.size(); j++) {
List<Map<String, Object>> updateMapList = new ArrayList<>();
Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){
String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", config.getLinkField());
paramMap.put("value", type);
updateMapList.add(paramMap);
if (StringUtils.isBlank(type)){
safeSet.add(map.get(config.getField()).toString().substring(0, 3));
}else {
}
}
}
if (!updateMapList.isEmpty()) {
customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
}
}
}
}
}
/**
* 组装updateLinkType执行参数
*/
public ReturnT<String> updateLinkType(SalesforceParam param, List<Future<?>> futures) throws Exception {
Set<String> apis;
if (StringUtils.isBlank(param.getApi())) {
apis = linkConfigService.list().stream().map(LinkConfig::getApi).collect(Collectors.toSet());
} else {
apis = DataUtil.toIdSet(param.getApi());
}
//查询所有创建字段的关联类型
QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
wrapper.in("api", apis).eq("is_link",1).eq("is_create",1);
Map<String, String> resultMap = dataObjectService.list().stream()
.collect(Collectors.toMap(
DataObject::getKeyPrefix,
DataObject::getName,
(existing, replacement) -> replacement));
for (String api : apis) {
//邮件发送前三位编码不存在的对象
Set<String> safeSet = new HashSet<>();
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", api).eq("is_link",1).eq("is_create",1);
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
if (linkConfigs.isEmpty()){
continue;
}
Future<?> future = salesforceExecutor.execute(() -> {
try {
updateLink(api,param, resultMap,safeSet,linkConfigs);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, 0, 1);
futures.add(future);
if (!safeSet.isEmpty()){
String format = String.format("更新关联类型 error, api name: %s, \ncause:\n%s 前三位编码对象不存在！！", api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
EmailUtil.send("DataDump ERROR", format);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
return ReturnT.SUCCESS;
}
/**
* 执行数据updateLinkType
*/
private void updateLink(String api,SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {
String beginDateStr = null;
Date beginDate = param.getBeginModifyDate();
if (param.getBeginModifyDate() != null ){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
}
String sql1 = null;
String sql2 = null;
if (beginDateStr != null){
String updateDateField = dataFieldService.returnUpdateDateField(api);
sql1 = "where " + updateDateField + " >= '" + beginDateStr + "'";
sql2 = updateDateField + " >= '"+ beginDateStr +"' " ;
}else {
sql1 = "where 1=1";
sql2 = "1=1 " ;
}
String allFiled = "Id";
if (linkConfigs.size() == 1){
for (LinkConfig linkConfig : linkConfigs) {
sql1 = sql1 + " and " + linkConfig.getField() + " is not null";
sql2 = sql2 + " and " + linkConfig.getField() + " is not null";
allFiled = allFiled + "," + linkConfig.getField();
}
}else {
String info = null;
for (LinkConfig linkConfig : linkConfigs) {
if (StringUtils.isBlank(info)){
info = linkConfig.getField() + " is not null" ;
}else {
info = info + " or " + linkConfig.getField() + " is not null";
}
allFiled = allFiled + "," + linkConfig.getField();
}
sql1 = sql1 +" and ("+ info +")";
sql2 = sql2 +" and ("+ info +")";
}
// 表内数据总量
Integer count = customMapper.countBySQL(api, sql1);
log.info("表api：{} 存在" +count+ "条数据！", api);
if (count >0 ) {
int page = count % 10000 == 0 ? count / 10000 : (count / 10000) + 1;
for (int i = 0; i < page; i++) {
log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 10000);
List<Map<String, Object>> mapList = customMapper.list(allFiled, api, sql2 +" order by Id limit "+ + i * 10000 + ",10000" );
for (int j = 1; j <= mapList.size(); j++) {
List<Map<String, Object>> updateMapList = new ArrayList<>();
Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){
String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", config.getLinkField());
paramMap.put("value", type);
updateMapList.add(paramMap);
if (StringUtils.isBlank(type)){
safeSet.add(map.get(config.getField()).toString().substring(0, 3));
}
}
}
if (!updateMapList.isEmpty()){
customMapper.updateById(api, updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
}
}
}
}
}
/**
* 校验删除数据入口
*/
@Override
public ReturnT<String> checkDeletedData(SalesforceParam param) throws Exception {
if (param.getType() == 1){
return manualCheckDeletedData(param);
}else {
return autoCheckDeletedData(param);
}
}
/**
* 删除目标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";
PartnerConnection connect = salesforceTargetConnect.createConnect();
if ((param.getEndCreateDate() == null || param.getBeginCreateDate() == null) && param.getIds().isEmpty()){
log.info("手动删除数据必须指定参数：开始与结束时间/Ids !!!!");
return ReturnT.FAIL;
}
String sql = "";
if (!param.getIds().isEmpty()){
sql = "where Record in (" + param.getIds().stream().map(id -> "'" + id + "'").collect(Collectors.joining(",")) + ")";
}else {
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
String beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
String endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
sql = "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
}
Integer count = customMapper.countBySQL(api, sql);
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("删除数据，总批次：{}",page);
try{
for (int i = 0; i < page; i++) {
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}else {
log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
}
}
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
customMapper.deleteOne(typeName, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
}
index++;
}
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return ReturnT.FAIL;
} catch (Exception e) {
log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
}
return ReturnT.SUCCESS;
}
private ReturnT<String> autoCheckDeletedData(SalesforceParam param) throws Exception {
String api = "DeleteEvent";
PartnerConnection connect = salesforceTargetConnect.createConnect();
String sql = "SystemModstamp > " + DateUtil.format(DateUtils.addDays(new Date(), -15), "yyyy-MM-dd HH:mm:ss");
Integer count = customMapper.countBySQL(api, sql);
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("删除数据，总批次：{}",page);
try{
for (int i = 0; i < page; i++) {
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}else {
log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
}
}
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
customMapper.deleteOne(typeName, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
}
index++;
}
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return ReturnT.FAIL;
} catch (Exception e) {
log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
}
return ReturnT.SUCCESS;
}
/**
* 参数转换qw
*
* @param param 参数
* @return qw
*/
private static QueryWrapper<DataBatch> getDataBatchQueryWrapper(SalesforceParam param) {
QueryWrapper<DataBatch> qw = new QueryWrapper<>();
if (StringUtils.isNotBlank(param.getApi())) {
List<String> strings = DataUtil.toIdList(param.getApi());
if (CollectionUtils.isNotEmpty(strings)) {
qw.in("name", strings);
}
}
// 只查询首次同步时间不为空，且首次sf num不为0的
qw.isNotNull("first_sync_date")
.ne("first_sf_num", 0)
.orderByAsc("created_date", "name", "sync_start_date")
.select("id", "name", "label", "sync_start_date", "sync_end_date", "first_db_num", "first_sf_num", "db_num", "sf_num", "sync_status");
return qw;
}
@Override
public ReturnT<String> insertSingle(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
return insertSingleData(param, futures);
} else {
return autoInsertSingleData(param, futures);
}
} catch (InterruptedException e) {
throw e;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("insertSingle error", throwable);
throw throwable;
}
}
@Override
public ReturnT<String> dumpContentFolderMemberJob(SalesforceParam param) throws Exception {
String api = "ContentFolderMember";
PartnerConnection partnerConnection = salesforceConnect.createConnect();
List<Map<String, Object>> list = customMapper.list("Id", "ContentFolder", "new_id is not null order by Id ");
if (list.isEmpty()){
return ReturnT.SUCCESS;
}
DescribeSObjectResult dsr = partnerConnection.describeSObject(api);
List<String> fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList());
Field[] dsrFields = dsr.getFields();
try {
for (Map<String, Object> map : list) {
String contentFolderId = (String) map.get("Id");
log.info("ContentFolderMember数据拉取！！！ contentFolderId：" + contentFolderId);
String sql = "SELECT Id, ParentContentFolderId, ChildRecordId, IsDeleted, SystemModstamp, CreatedById, CreatedDate, LastModifiedById, LastModifiedDate FROM ContentFolderMember where ParentContentFolderId = '" + contentFolderId + "'";
com.alibaba.fastjson2.JSONArray objects = null;
QueryResult queryResult = partnerConnection.queryAll(sql);
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
commonService.saveOrUpdate(api, fields, records, objects, true);
}
} catch (Throwable e) {
log.error("dumpContentFolderMemberJob error message:{}", e.getMessage());
return ReturnT.FAIL;
}
return ReturnT.SUCCESS;
}
/**
* 富文本图片替换
* @param param 参数
* @return 执行结果
* @throws Exception 异常
*/
@Override
public ReturnT<String> richTextImageReplace(SalesforceParam param) throws Exception {
log.info("richTextImageReplace execute start ..................");
String beginDateStr = null;
String endDateStr = null;
if (param.getType() != 1){
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
}
// 检测路径是否存在 不存在则创建
File excel = new File(Const.SERVER_FILE_PATH + "/richTextImages" );
if (!excel.exists()) {
log.info("创建API目录: {}", excel.getAbsolutePath());
boolean mkdir = excel.mkdir();
if (!mkdir) {
log.info("创建文件存储目录失败！");
}
}
// 获取连接
PartnerConnection sourceConnection = salesforceConnect.createConnect();
PartnerConnection targetConnection = salesforceTargetConnect.createConnect();
QueryWrapper<DataObject> qw = new QueryWrapper<>();
if (StringUtils.isNotBlank(param.getApi())) {
List<String> apis = DataUtil.toIdList(param.getApi());
qw.in("name", apis);
}
qw.eq("data_work", true);
List<DataObject> objectList = dataObjectService.list(qw);
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
for (DataObject dataObject : objectList) {
String api = dataObject.getName();
QueryWrapper<DataField> qw1 = new QueryWrapper<>();
qw1.eq("api", api);
qw1.eq("isHtmlFormatted", true);
List<DataField> fields = dataFieldService.list(qw1);
for (DataField dataField : fields) {
String field = dataField.getField();
String sqlCondition = "new_id is not null and " + field + " like '%<img%refid%'";
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
sqlCondition += " and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
}
// 获取记录总数
Integer count = customMapper.countBySQL(api, "where " + sqlCondition);
log.info("对象 {} 字段 {} 需要处理的富文本记录总数: {}", api, field, count);
if (count == 0) {
log.info("对象 {} 字段 {} 没有需要处理的富文本记录", api, field);
continue;
}
// 分批处理，每批200条记录
int pageSize = 200;
int totalPages = (count + pageSize - 1) / pageSize;
for (int page = 0; page < totalPages; page++) {
List<Map<String, Object>> records = customMapper.list("Id, new_id, " + field, api,
sqlCondition + " order by Id limit " + (page * pageSize) + "," + pageSize);
log.info("处理对象 {} 字段 {} 第 {}/{} 页，本页记录数: {}", api, field, (page + 1), totalPages, records.size());
SObject[] accounts = new SObject[records.size()];
int index = 0;
for (Map<String, Object> record : records) {
try {
String sourceId = (String) record.get("Id");
String targetId = (String) record.get("new_id");
String richTextContent = (String) record.get(field);
if (StringUtils.isBlank(richTextContent) || !richTextContent.contains("refid")) {
continue;
}
// 处理富文本中的图片
String updatedContent = processRichTextImages(richTextContent, sourceId, api, field, sourceConnection, targetConnection);
if (updatedContent == null){
return ReturnT.FAIL;
}
// 如果内容有变化，则更新目标组织中的记录
if (!richTextContent.equals(updatedContent)) {
// 更新目标组织中的记录
SObject sObject = new SObject();
sObject.setType(api);
sObject.setId(targetId);
sObject.setField(field, updatedContent);
sObject.setField("Id", targetId);
accounts[index] = sObject;
index ++;
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", field + "_back");
linkMap.put("value", richTextContent);
maps.add(linkMap);
customMapper.updateByNewId(maps,api, targetId);
}
} catch (Exception e) {
log.error("处理记录 {} 的富文本内容时发生错误: {}", record.get("Id"), e.getMessage(), e);
}
}
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts,fields);
}
SaveResult[] saveResults = targetConnection.update(accounts);
for (SaveResult saveResult : saveResults) {
if (!saveResult.getSuccess()) {
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", JSON.toJSONString(saveResult.getErrors()));
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, saveResult.getId());
String format = String.format("数据更新 error, api name: %s, \ncause:\n%s", api, JSON.toJSONString(saveResult));
log.info(format);
}else {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap = new HashMap<>();
linkMap.put("key", "is_update");
linkMap.put("value", 1);
maps.add(linkMap);
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", null);
maps.add(linkMap1);
customMapper.updateByNewId(maps,api, saveResult.getId());
}
}
}
}
}
log.info("richTextImageReplace execute end ..................");
return ReturnT.SUCCESS;
}
/**
* 处理富文本中的图片
* @param richTextContent 原始富文本内容
* @param sourceRecordId 源记录ID
* @param api 对象API名称
* @param field 字段名
* @param sourceConnection 源组织连接
* @param targetConnection 目标组织连接
* @return 处理后的富文本内容
* @throws Exception 异常
*/
private String processRichTextImages(String richTextContent, String sourceRecordId, String api, String field,
PartnerConnection sourceConnection, PartnerConnection targetConnection) throws Exception {
// 匹配富文本中包含refid的img标签
Pattern imgPattern = Pattern.compile("<img[^>]*src\\s*=\\s*\"[^\"]*refid=([^\"]+)(?:&quot;|&amp;|&|\\\"|\"|')?[^>]*/?>", Pattern.CASE_INSENSITIVE); Matcher matcher = imgPattern.matcher(richTextContent);
StringBuffer updatedContent = new StringBuffer();
Map<String, String> imageIdMap = new ConcurrentHashMap<>();
// 遍历所有匹配的img标签
while (matcher.find()) {
String imgTag = matcher.group(0);
String refId = matcher.group(1);
// 检查是否已经处理过这个refId
if (imageIdMap.containsKey(refId)) {
String newRefId = imageIdMap.get(refId);
String updatedImgTag = imgTag.replace("refid=\"" + refId + "\"", "refid=\"" + newRefId + "\"");
matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(updatedImgTag));
continue;
}
try {
// 构建源图片下载URL (参考Python脚本中的imageUrl构建方式)
String downloadUrl = FILE_DOWNLOAD_URL +
String.format("/services/data/v55.0/sobjects/%s/%s/richTextImageFields/%s/%s",
api, sourceRecordId, field, refId);
// 获取图片的alt属性作为标题
String title = "image";
Pattern altPattern = Pattern.compile("alt=\"([^\"]*)\"", Pattern.CASE_INSENSITIVE);
Matcher altMatcher = altPattern.matcher(imgTag);
if (altMatcher.find()) {
title = altMatcher.group(1);
}
// 下载源图片文件
String downloadImageFile = downloadImageFile(sourceConnection, downloadUrl, title);
if (downloadImageFile != null) {
// 上传图片到目标组织
String newContentVersionId = uploadImageFile(targetConnection, downloadImageFile, title);
if (newContentVersionId != null) {
// 获取新的ContentDocumentId
Map<String, String> documentId = getDocumentId(targetConnection, newContentVersionId);
String newContentDocumentId = documentId.get("ContentDocumentId");
String newOwnId = documentId.get("OwnerId");
if (newContentDocumentId != null) {
imageIdMap.put(refId, newContentDocumentId);
// 构建新的img标签 (参考Python脚本中的richTextUrl构建方式)
String newImageUrl =
"/sfc/servlet.shepherd/version/download/" + newContentVersionId;
String updatedImgTag = "<img src=\"" + newImageUrl + "\" refid=\"" + newContentDocumentId + "\"/>";
matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(updatedImgTag));
// 记录替换日志
richTextLogService.logRichTextImageReplace(
api,
field,
newContentDocumentId,
newOwnId, // recordOwnerId 可以从记录中获取，但此处简化处理
richTextContent,
refId, // 原图片链接
newImageUrl, // 新图片链接
newContentDocumentId // 新文件ID
);
continue;
}
}
}else {
return null;
}
// 如果找不到对应关系，保留原始标签
matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(imgTag));
} catch (Exception e) {
log.error("处理富文本图片时发生错误，refId: {}", refId, e);
matcher.appendReplacement(updatedContent, Matcher.quoteReplacement(imgTag));
}
}
matcher.appendTail(updatedContent);
return updatedContent.toString();
}
/**
* 从源组织下载图片文件
* @param connection 源组织连接
* @param downloadUrl 下载URL
* @return 图片文件数据
* @throws Exception 异常
*/
private String downloadImageFile(PartnerConnection connection, String downloadUrl,String title) throws Exception {
String imageServer = null;
try {
String token = connection.getSessionHeader().getSessionId();
Map<String, String> headers = Maps.newHashMap();
headers.put("Authorization", "Bearer " + token);
headers.put("connection", "keep-alive");
Response response = HttpUtil.doGet(downloadUrl, null, headers);
if (response.body() != null && response.code() == 200){
InputStream inputStream = response.body().byteStream();
imageServer = dumpToRichTextImageServer(title, response, inputStream);
}else {
throw new RuntimeException("下载图片文件时发生错误，URL: " + downloadUrl + "；错误信息：" + response.message());
}
} catch (Exception e) {
log.error("下载图片文件时发生错误，URL: {}", downloadUrl, e);
return null;
}
return imageServer;
}
/**
* 保存富文本图片到本地服务器
* @param response
* @param inputStream
* @throws IOException
*/
private String dumpToRichTextImageServer(String name, Response response, InputStream inputStream) throws IOException {
String path = Const.SERVER_FILE_PATH + "/richTextImages/" + name;
long offset = 0L;
RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
while (true) {
try {
// 保存到本地
byte[] buf = new byte[8192];
int len = 0;
if (offset > 0) {
inputStream.skip(offset);
accessFile.seek(offset);
}
while ((len = inputStream.read(buf)) != -1) {
accessFile.write(buf, 0, len);
offset += len;
}
break;
} catch (Exception e) {
if (offset <= 0) {
throw e;
}
response.close();
}
}
accessFile.close();
return path;
}
/**
* 上传图片文件到目标组织
* @param connection 目标组织连接
= * @param title 文件标题
* @return 新的ContentVersion ID
* @throws Exception 异常
*/
private String uploadImageFile(PartnerConnection connection, String filePath, String title) throws Exception {
String newId = null;
try {
int failCount = 0;
String token = connection.getSessionHeader().getSessionId();
String uploadUrl = FILE_UPLOAD_URL +
"/services/data/v55.0/sobjects/ContentVersion";
HttpPost httpPost = new HttpPost(uploadUrl);
httpPost.setHeader("Authorization", "Bearer " + token);
httpPost.setHeader("connection", "keep-alive");
com.alibaba.fastjson.JSONObject credentialsJsonParam = new com.alibaba.fastjson.JSONObject();
credentialsJsonParam.put("title", title);
credentialsJsonParam.put("pathOnClient", title);
MultipartEntityBuilder builder = MultipartEntityBuilder.create();
builder.addTextBody("entity_content", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
builder.addBinaryBody("VersionData", new FileInputStream(filePath), ContentType.APPLICATION_OCTET_STREAM, title + ".jpg");
HttpEntity entity = builder.build();
httpPost.setEntity(entity);
CloseableHttpClient httpClient = HttpClients.createDefault();
CloseableHttpResponse response = null;
String respContent = null;
while (failCount < Const.MAX_FAIL_COUNT) {
response = httpClient.execute(httpPost);
if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400 && response.getEntity() != null) {
HttpEntity he = response.getEntity();
respContent = EntityUtils.toString(he, "UTF-8");
newId = String.valueOf(com.alibaba.fastjson.JSONObject.parseObject(respContent).get("id"));
break;
} else {
failCount++;
log.error("富文本图片替换图片文件上传失败，返回体信息：" + JSON.toJSONString(response));
}
}
} catch (Exception e) {
log.error("上传图片文件时发生错误，标题: {}", title, e);
return null;
}
return newId;
}
/**
* 获取ContentVersion ID
* @param connection Salesforce连接
* @param contentDocumentId ContentDocument ID
* @return ContentVersion ID
* @throws Exception 异常
*/
private String getContentVersionId(PartnerConnection connection, String contentDocumentId) throws Exception {
String query = "SELECT Id FROM ContentVersion WHERE ContentDocumentId = '" + contentDocumentId + "' ORDER BY CreatedDate DESC LIMIT 1";
QueryResult result = connection.query(query);
if (result.getRecords() != null && result.getRecords().length > 0) {
return (String) result.getRecords()[0].getField("Id");
}
return null;
}
/**
* 获取目标组织中的ContentDocumentId
* @param connection 目标组织连接
* @param contentVersionId ContentVersion ID
* @return 目标组织中的ContentDocumentId
* @throws Exception 异常
*/
private String getTargetContentDocumentId(PartnerConnection connection, String contentVersionId) throws Exception {
String query = "SELECT ContentDocumentId FROM ContentVersion WHERE Id = '" + contentVersionId + "'";
QueryResult result = connection.query(query);
if (result.getRecords() != null && result.getRecords().length > 0) {
return (String) result.getRecords()[0].getField("ContentDocumentId");
}
return null;
}
/**
* 获取ContentDocumentId和OwnerId
* @param connection 连接
* @param contentVersionId ContentVersion ID
* @return 包含ContentDocumentId和OwnerId的Map
* @throws Exception 异常
*/
private Map<String, String> getDocumentId(PartnerConnection connection, String contentVersionId) throws Exception {
String soql = "SELECT ContentDocumentId,OwnerId FROM ContentVersion WHERE Id = '" + contentVersionId + "'";
QueryResult queryResult = connection.query(soql);
Map<String, String> result = new HashMap<>();
if (queryResult.getRecords() != null && queryResult.getRecords().length > 0) {
SObject record = queryResult.getRecords()[0];
result.put("ContentDocumentId", (String) record.getField("ContentDocumentId"));
result.put("OwnerId", (String) record.getField("OwnerId"));
return result;
}
return null;
}
@Override
public void getNewIdByField(String api, String queryFields) throws Exception {
List<String> stringList = DataUtil.toIdList(queryFields);
String maxId = null;
Field[] dsrFields;
List<String> fields = Lists.newArrayList();
PartnerConnection connect = salesforceTargetConnect.createConnect();
DescribeSObjectResult dsr = connect.describeSObject(api);
dsrFields = dsr.getFields();
for (Field field : dsrFields) {
// 不查询文件
if ("base64".equalsIgnoreCase(field.getType().toString())) {
continue;
}
fields.add(field.getName());
}
// 添加特定对象的关联字段
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";
}
log.info("执行SQL查询: {}", sql);
QueryResult queryResult = connect.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
log.info("查询结果为空，结束查询");
break;
}
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
log.info("处理批次数据，记录数: {}", objects.size());
// 批量处理记录
processRecords(api, objects, stringList);
int totalRecords = records.length;
count += totalRecords;
log.info("已处理记录总数: {}", 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)) {
fields.add("Member.type");
}
if ("FeedAttachment".equals(api)) {
fields.add("FeedEntity.type");
fields.add("Record.type");
}
}
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()) {
log.info("准备更新记录，API: {}, ID: {}, 匹配字段信息: {}", api, id, maps);
customMapper.updateNewIdByFields(api, maps, id);
}else if (maps == null){
log.info("{}无需更新记录，数据ID: {}", api, id);
}
}
}
private List<Map<String, Object>> buildUpdateMaps(String api, JSONObject jsonObject, List<String> stringList) throws Exception {
List<Map<String, Object>> maps = Lists.newArrayList();
for (String key : stringList) {
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){
log.info("{}处理失败，字段: {}, 数据ID: {}", api, key, jsonObject.getString("Id"));
return null;
}
break;
case "CollaborationGroupMember":
paramMap = handleCollaborationGroupMember(key, jsonObject);
if (paramMap == null){
log.info("CollaborationGroupMember处理失败，字段: {}, 数据ID: {}", key, jsonObject.getString("Id"));
return null;
}
break;
case "FeedAttachment":
paramMap = handleFeedAttachment(key, jsonObject);
if (paramMap == null){
log.info("FeedAttachment处理失败，字段: {}, 数据ID: {}", key, jsonObject.getString("Id"));
return null;
}
break;
default:
paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value", jsonObject.getString(key));
log.info("普通字段处理，API: {}, 字段: {}, 值: {}", api, key, jsonObject.getString(key));
break;
}
if (!paramMap.isEmpty()) {
maps.add(paramMap);
}
}
return maps;
}
private Map<String, Object> handleAccountContactRelation(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "ContactId":
Map<String, Object> contactMap = getNewIdMap(key, "Contact", jsonObject.getString(key));
if (contactMap == null){
log.info("AccountContactRelation，字段: {},关联对象段: {}, ContactId: {} 未找到对应的new_id", key,"Contact", jsonObject.getString(key));
return null;
}
return contactMap;
case "AccountId":
Map<String, Object> accountMap = getNewIdMap(key, "Account", jsonObject.getString(key));
if (accountMap == null){
log.info("AccountContactRelation，字段: {},关联对象段: {}, AccountId: {} 未找到对应的new_id", key,"Account", jsonObject.getString(key));
return null;
}
return accountMap;
default:
return null;
}
}
private Map<String, Object> handleTaskEventRelation(String api, String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "TaskId":
Map<String, Object> taskObjectMap = getNewIdMap(key, "Task", jsonObject.getString(key));
if (taskObjectMap == null){
log.info("TaskEventRelation，字段: {},关联对象段: {}, TaskId: {} 未找到对应的new_id", key,"Task", jsonObject.getString(key));
return null;
}
return taskObjectMap;
case "EventId":
Map<String, Object> eventObjectMap = getNewIdMap(key, "Event", jsonObject.getString(key));
if (eventObjectMap == null){
log.info("TaskEventRelation，字段: {},关联对象段: {}, EventId: {} 未找到对应的new_id", key,"Event", jsonObject.getString(key));
return null;
}
return eventObjectMap;
case "RelationId":
Map<String, Object> relationObjectMap = getNewIdMap(key, jsonObject.getString("Relation_Type"), jsonObject.getString(key));
if (relationObjectMap == null){
log.info("TaskEventRelation，字段: {},关联对象段: {}, RelationId: {} 未找到对应的new_id", key,jsonObject.getString("Relation_Type"), jsonObject.getString(key));
return null;
}
return relationObjectMap;
default:
return null;
}
}
private Map<String, Object> handleCollaborationGroupMember(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "CollaborationGroupId":
Map<String, Object> groupObjectMap = getNewIdMap(key, "CollaborationGroup", jsonObject.getString(key));
if (groupObjectMap == null){
log.info("CollaborationGroupMember，字段: {},关联对象段: {}, CollaborationGroupId: {} 未找到对应的new_id", key,"CollaborationGroup", jsonObject.getString(key));
return null;
}
return groupObjectMap;
case "MemberId":
Map<String, Object> memberObjectMap = getNewIdMap(key, jsonObject.getString("Member_Type"), jsonObject.getString(key));
if (memberObjectMap == null){
log.info("CollaborationGroupMember，字段: {},关联对象段: {}, MemberId: {} 未找到对应的new_id", key,jsonObject.getString("Member_Type"), jsonObject.getString(key));
return null;
}
return memberObjectMap;
default:
return null;
}
}
private Map<String, Object> handleFeedAttachment(String key, JSONObject jsonObject) throws Exception {
switch (key) {
case "FeedEntityId":
Map<String, Object> feedEntityObjectMap = getNewIdMap(key, jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
if (feedEntityObjectMap == null){
log.info("FeedAttachment，字段: {},关联对象段: {}, FeedEntityId: {} 未找到对应的new_id", key,jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
return null;
}
return feedEntityObjectMap;
case "RecordId":
Map<String, Object> recordObjectMap = getNewIdMap(key, jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
if (recordObjectMap == null){
log.info("FeedAttachment，字段: {},关联对象段: {}, RecordId: {} 未找到对应的new_id", key,jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
return null;
}
return recordObjectMap;
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) {
return null;
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", key);
paramMap.put("value", m.get("Id"));
return paramMap;
}
@Override
public void getTaskNewId(SalesforceParam param) throws Exception {
List<Map<String, Object>> mapsList = customMapper.list("*", "Task", "where What_Type = EmailMessage and new_id is null");
}
/**
* 组装【单表】 一次性Insert 参数
*/
public ReturnT<String> insertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis = DataUtil.toIdList(param.getApi());
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");
}
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isNotEmpty(list)) {
String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
String message = "api:" + apiNames + " is locked";
String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
EmailUtil.send("DataDump ERROR", format);
return new ReturnT<>(500, message);
}
}
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
for (String api : apis) {
DataObject update = dataObjectService.getById(api);
if (update == null) {
log.info("对象 {} 不存在，无法执行一次性插入操作", api);
continue;
}
try {
if (!dataFieldService.hasCreatedDate(api)){
insertSingleShareData(api,partnerConnection,update);
continue;
}
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 (salesforceParam.getIsSingleThread()){
insertSingleExecute(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setNeedUpdate(false);
} catch (Exception e) {
throw e;
} finally {
if (isFull) {
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
}
return ReturnT.SUCCESS;
}
/**
* 组装【多】一次性Insert 参数
*/
public ReturnT<String> autoInsertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("data_work", 1)
.eq("need_update", 1)
.eq("data_lock", 0)
.orderByAsc("data_index")
.last(" limit 10");
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
while (true) {
List<DataObject> dataObjects = dataObjectService.list(qw);
if (CollectionUtils.isEmpty(dataObjects)) {
break;
}
for (DataObject dataObject : dataObjects) {
DataObject update = new DataObject();
log.info("insertSingle api: {}", dataObject.getName());
TimeUnit.MILLISECONDS.sleep(1);
try {
String api = dataObject.getName();
update.setName(dataObject.getName());
update.setDataLock(1);
dataObjectService.updateById(update);
if (!dataFieldService.hasCreatedDate(api)){
insertSingleShareData(api,partnerConnection,dataObject);
update.setNeedUpdate(false);
dataObjectService.updateById(update);
continue;
}
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
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 (salesforceParam.getIsSingleThread()){
insertSingleExecute(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,dataObject);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setNeedUpdate(false);
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
throw new RuntimeException(e);
} finally {
update.setDataLock(0);
dataObjectService.updateById(update);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
futures.clear();
}
return ReturnT.SUCCESS;
}
/**
* 执行一次性Insert数据
*/
private void insertSingleExecute(SalesforceParam param, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
String api = param.getApi();
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> list = dataFieldService.list(dbQw);
String sql = "";
String sql2 = "";
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
if (1 == param.getType()) {
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000";
}else {
sql = "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 10000";
}
}else {
String updateDateField = dataFieldService.returnUpdateDateField(api);
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000";
}else {
sql = "where new_id is null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000";
}
}
//表内数据总量
Integer count = customMapper.countBySQL(api, sql);
if(count == 0){
return;
}
//查询当前对象多态字段映射
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",1);
List<LinkConfig> configs = linkConfigService.list(queryWrapper);
Map<String, String> fieldMap = new HashMap<>();
if (!configs.isEmpty()) {
fieldMap = configs.stream()
.collect(Collectors.toMap(
LinkConfig::getField, // Key提取器
LinkConfig::getLinkField, // Value提取器
(oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
));
}
int targetCount = 0;
int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;
log.info("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,num, beginDateStr, endDateStr, api);
for (int z = 0; z < num; z++) {
List<Map<String, Object>> mapsList = customMapper.list("*", api, sql2);
log.info("总Insert数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,z, beginDateStr, endDateStr, api);
int size = mapsList.size();
//批量插入200一次
int page = size%200 == 0 ? size/200 : (size/200) + 1;
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
break;
}
List<Map<String, Object>> data = mapsList.subList(startIndex, endIndex);
int sized = data.size();
SObject[] accounts = new SObject[sized];
String[] ids = new String[sized];
try {
for (int j = 1; j <= sized; j++) {
SObject account = new SObject();
account.setType(api);
String message = null;
//给对象赋值
for (DataField dataField : list) {
String field = dataField.getField();
String reference_to = dataField.getReferenceTo();
String value = String.valueOf(data.get(j - 1).get(field));
//根据旧sfid查找引用对象新sfid
if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) {
continue;
} else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {
if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
//判断reference_to内是否包含User字符串
if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
reference_to = "User";
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) ){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
}
Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
account.setField(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"异常！";
log.info(message);
break;
}
}
} else {
if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
account.setField(field, data.get(j - 1).get(field));
}
}
}
if (message!=null){
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateById(api, maps, data.get(j-1).get("Id").toString());
continue;
}
if (dataObject.getIsEditable()){
account.setField("old_owner_id__c", data.get(j - 1).get("OwnerId"));
account.setField("old_sfdc_id__c", data.get(j - 1).get("Id"));
}
ids[j-1] = data.get(j-1).get("Id").toString();
accounts[j-1] = account;
if (i*200+j == size){
break;
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts, list);
}
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
SaveResult[] saveResults = partnerConnection.create(accounts);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");
int index = 0;
for (SaveResult saveResult : saveResults){
if (saveResult.getSuccess()) {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m = new HashMap<>();
m.put("key", "new_id");
m.put("value", saveResult.getId());
maps.add(m);
customMapper.updateById(api, maps, ids[index]);
targetCount ++;
}else{
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", JSON.toJSONString(saveResult.getErrors()));
maps.add(linkMap1);
customMapper.updateById(api, maps, ids[index]);
log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult));
}
index++;
}
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return;
} catch (Exception e) {
log.error("insertSingle error api:{}", api,e);
}
}
if (z*10000+size >= count){
break;
}
}
SalesforceParam countParam = new SalesforceParam();
countParam.setApi(api);
countParam.setBeginCreateDate(beginDate);
countParam.setEndCreateDate(DateUtils.addSeconds(endDate, -1));
// 存在isDeleted 只查询IsDeleted为false的
if (dataFieldService.hasDeleted(countParam.getApi())) {
countParam.setIsDeleted(false);
} else {
// 不存在 过滤
countParam.setIsDeleted(null);
}
Integer sfNum = 0;
if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
sfNum = commonService.countSfNum(partnerConnection, countParam);
}
UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
updateQw.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
.set("target_sf_num", count);
dataBatchHistoryService.update(updateQw);
UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
updateQw2.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", endDate)
.set("sf_add_num", sfNum);
dataBatchService.update(updateQw2);
UpdateWrapper<DataBatchHistory> updateQw3 = new UpdateWrapper<>();
updateQw3.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
.set("target_update_num", count);
dataBatchHistoryService.update(updateQw3);
UpdateWrapper<DataBatch> updateQw4 = new UpdateWrapper<>();
updateQw4.eq("name", api)
.eq("sync_start_date", beginDate)
.eq("sync_end_date", endDate)
.set("sf_update_num", targetCount);
dataBatchService.update(updateQw4);
}
/**
* 执行一次性Insert Share数据
*/
private void insertSingleShareData(String api, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> list = dataFieldService.list(dbQw);
TimeUnit.MILLISECONDS.sleep(1);
String sql = "";
String sql2 = "";
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null ";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 10000";
}else {
sql = "where new_id is null ";
sql2 = "new_id is null limit 10000";
}
//表内数据总量
Integer count = customMapper.countBySQL(api, sql);
if (count == 0) {
return;
}
//查询当前对象多态字段映射
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
List<LinkConfig> configs = linkConfigService.list(queryWrapper);
Map<String, String> fieldMap = new HashMap<>();
if (!configs.isEmpty()) {
fieldMap = configs.stream()
.collect(Collectors.toMap(
LinkConfig::getField, // Key提取器
LinkConfig::getLinkField, // Value提取器
(oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
));
}
int num = count%10000 == 0 ? count/10000 : (count/10000) + 1;
log.info("总Insert数据 count:{}；总批次：{}; -api:{}；", count,num, api);
for (int z = 0; z < num; z++) {
List<Map<String, Object>> mapsList = customMapper.list("*", api, sql2);
int size = mapsList.size();
//批量插入200一次
int page = size%200 == 0 ? size/200 : (size/200) + 1;
log.info("执行api:{}, 执行批次:{}, 执行数据量:{}", api, z, size);
for (int i = 0; i < page; i++) {
try {
DataLog dataLog = new DataLog(api, null, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
break;
}
List<Map<String, Object>> data = mapsList.subList(startIndex, endIndex);
int sized = data.size();
SObject[] accounts = new SObject[sized];
String[] ids = new String[sized];
for (int j = 1; j <= sized; j++) {
SObject account = new SObject();
account.setType(api);
String message = null;
//给对象赋值
for (DataField dataField : list) {
String field = dataField.getField();
String reference_to = dataField.getReferenceTo();
String value = String.valueOf(data.get(j - 1).get(field));
//根据旧sfid查找引用对象新sfid
if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) {
continue;
} else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {
if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
//判断reference_to内是否包含User字符串
if (dataField.getReferenceTo() != null && (dataField.getReferenceTo().contains(",User") || dataField.getReferenceTo().contains("User,"))) {
reference_to = "User";
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield)){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
}
Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
if (m!= null && !m.isEmpty() && m.get("new_id") != null) {
account.setField(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"异常！";
log.info(message);
break;
}
}
} else {
if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
account.setField(field, data.get(j - 1).get(field));
}
}
}
if (message != null){
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", message);
maps.add(linkMap1);
customMapper.updateById(api, maps, data.get(j-1).get("Id").toString());
continue;
}
if (dataObject.getIsEditable()){
account.setField("old_owner_id__c", data.get(j - 1).get("OwnerId"));
account.setField("old_sfdc_id__c", data.get(j - 1).get("Id"));
}
ids[j-1] = data.get(j-1).get("Id").toString();
accounts[j-1] = account;
if (i*200+j == size){
break;
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts, list);
}
DataLog dataLog1 = new DataLog(api, null, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
SaveResult[] saveResults = partnerConnection.create(accounts);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, null, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");
int index = 0;
for (SaveResult saveResult : saveResults){
if (saveResult.getSuccess()) {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m = new HashMap<>();
m.put("key", "new_id");
m.put("value", saveResult.getId());
maps.add(m);
customMapper.updateById(api, maps, ids[index]);
log.info("Created Success row with id " + saveResult.getId());
}else{
log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult));
}
index++;
}
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return;
}catch (Exception e) {
log.error("insertSingle error api:{}", api, e);
}
}
if (z*10000+size >= count){
break;
}
}
}
}
Reference in New Issue View Git Blame Copy Permalink