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> poll = customMapper.list("code,value","org_config",null); for (Map 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> futures = Lists.newArrayList(); try { if (StringUtils.isNotBlank(param.getApi())) { // 手动任务 ReturnT 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 manualGetPersonContact(SalesforceParam param, List> futures) throws Exception { String api = "Contact"; TimeUnit.MILLISECONDS.sleep(1); // 全量的时候 检测是否有自动任务锁住的表 boolean isFull = CollectionUtils.isEmpty(param.getIds()); if (isFull) { QueryWrapper qw = new QueryWrapper<>(); qw.eq("data_lock", 1); List 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 salesforceParams = null; update.setName(api); update.setDataLock(1); dataObjectService.updateById(update); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); List 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 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> 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 idMaps = new HashMap<>(); // 旧客户ID，新联系人ID，用于更新SF数据 Map idMapa = new HashMap<>(); for (Map map : data) { Map 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> maps = Lists.newArrayList(); Map 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 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 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 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 immigrationUpdateNew(SalesforceParam param) throws Exception { List> 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 maualupdateSfDataNew(SalesforceParam param, List> futures) throws Exception { List 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 qw = new QueryWrapper<>(); qw.eq("data_lock", 1).in("name", apis); List 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 salesforceParams = null; QueryWrapper 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 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 updateIncrementalSfDataNew(SalesforceParam param, List> futures) throws Exception { List apis = DataUtil.toIdList(param.getApi()); // 全量的时候 检测是否有自动任务锁住的表 boolean isFull = CollectionUtils.isEmpty(param.getIds()); if (isFull) { QueryWrapper qw = new QueryWrapper<>(); qw.eq("data_lock", 1).in("name", apis); List 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 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> futures) throws Exception { // 全量的时候 检测是否有自动任务锁住的表 boolean isFull = CollectionUtils.isEmpty(param.getIds()); if (isFull) { QueryWrapper qw = new QueryWrapper<>(); qw.eq("data_lock", 1); List 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 qw = new QueryWrapper<>(); qw.eq("need_update", 1) .orderByAsc("data_index") .last(" limit 10"); PartnerConnection partnerConnection = salesforceTargetConnect.createConnect(); while (true) { List 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 salesforceParams = null; QueryWrapper 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 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> futures) throws Exception { // 全量的时候 检测是否有自动任务锁住的表 boolean isFull = CollectionUtils.isEmpty(param.getIds()); if (isFull) { QueryWrapper qw = new QueryWrapper<>(); qw.eq("data_lock", 1); List 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 qw = new QueryWrapper<>(); qw.eq("need_update", 1) .orderByAsc("data_index") .last(" limit 10"); PartnerConnection partnerConnection = salesforceTargetConnect.createConnect(); while (true) { List 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 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 infoFlag = customMapper.list("code,value","system_config","code ='"+ SystemConfigCode.INFO_FLAG+"'").get(0); String api = param.getApi(); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List 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 queryWrapper = new QueryWrapper<>(); queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); List configs = linkConfigService.list(queryWrapper); Map 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> 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> mapList = mapsList.subList(startIndex, endIndex); SObject[] accounts = new SObject[mapList.size()]; int j = 0; try { for (Map map : mapList) { //需要置空的参数 List 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 2); maps.add(linkMap); Map 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 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 2); maps.add(linkMap); Map 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 2); maps.add(linkMap); Map 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 1); maps.add(linkMap); Map 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 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 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 infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List 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 queryWrapper = new QueryWrapper<>(); queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); List configs = linkConfigService.list(queryWrapper); Map 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> 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> mapList = mapsList.subList(startIndex, endIndex); SObject[] accounts = new SObject[mapList.size()]; int j = 0; try { for (Map 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 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 2); maps.add(linkMap); Map 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> maps = new ArrayList<>(); Map 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 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 returnErrorAccountsDetails(SObject[] accounts,List list,String errorId) { HashMap 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 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> 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 fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList()); Field[] dsrFields = dsr.getFields(); try { for (Map 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 uploadDocumentLinkJob(String paramStr) throws Exception { String api = "ContentDocumentLink"; Map infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List dataFields = dataFieldService.list(dbQw); PartnerConnection connection = salesforceTargetConnect.createConnect(); List> 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> 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 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 qw = new QueryWrapper<>(); qw.eq("name", linkedEntityType); List objects = dataObjectService.list(qw); if (!objects.isEmpty()) { Map dMap = customMapper.getById("new_id", "ContentDocument", contentDocumentId); Map 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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 dumpFileNew(SalesforceParam param) throws Exception { String downloadUrl = null; QueryWrapper wrapper = new QueryWrapper<>(); if (StringUtils.isNotBlank(param.getApi())) { wrapper.in("name",DataUtil.toIdList(param.getApi())); }else { wrapper.isNotNull("blob_field"); } List objectList = dataObjectService.list(wrapper); if (objectList.isEmpty()){ log.info("没有对象存在文件二进制字段！不进行文件下载"); return ReturnT.FAIL; } List> poll = customMapper.list("code,value","org_config",null); for (Map 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> 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 salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); List 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 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> list = customMapper.list("Id,CreatedDate, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit 200"); for (Map map : list) { int failCount = 0; String id = null; // 上传完毕 更新附件信息 List> 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 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 paramMap = Maps.newHashMap(); paramMap.put("key", "is_dump"); paramMap.put("value", 2); maps.add(paramMap); customMapper.updateById(api, maps, id); }else { Map 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 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 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 uploadFileNew(SalesforceParam param, DataObject dataObject) throws Exception { String uploadUrl = null; List> poll = customMapper.list("code,value","org_config",null); for (Map 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> 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 salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); List 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 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> 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 map : list) { int failCount = 0; String id = null; // 上传完毕 更新附件信息 List> 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 qw = new QueryWrapper<>(); qw.eq("name", parentType); List objects = dataObjectService.list(qw); if (objects.isEmpty()) { log.info("关联对象不存在： {}", parentType); } Map 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 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 paramMap = Maps.newHashMap(); paramMap.put("key", "is_upload"); paramMap.put("value", 2); maps.add(paramMap); customMapper.updateById(api, maps, id); }else { Map 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 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 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> documentList = customMapper.list("Id " , "ContentDocument", " new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 200 + ",200"); for (Map documentMap : documentList) { String documentId = (String) documentMap.get("Id"); // 获取未存储的附件id List> 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 map : list) { int failCount = 0; String id = null; // 上传完毕 更新附件信息 List> 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 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> dList = new ArrayList<>(); Map 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 paramMap = Maps.newHashMap(); paramMap.put("key", "is_upload"); paramMap.put("value", 2); maps.add(paramMap); customMapper.updateById(api, maps, id); }else { Map 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 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 updateLinkTypeJob(SalesforceParam param) throws Exception { List> 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 updateLinkTypeBatch(SalesforceParam param, List> futures) throws Exception { List apis; if (StringUtils.isBlank(param.getApi())) { QueryWrapper 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 wrapper = new QueryWrapper<>(); wrapper.in("api", apis); Map 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 dbQw = new QueryWrapper<>(); dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1); List linkConfigs = linkConfigService.list(dbQw); if (linkConfigs.isEmpty()){ continue; } List salesforceParams = null; QueryWrapper 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 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 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 resultMap,Set safeSet,List 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> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000"); for (int j = 1; j <= mapList.size(); j++) { List> updateMapList = new ArrayList<>(); Map 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 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 updateLinkType(SalesforceParam param, List> futures) throws Exception { Set apis; if (StringUtils.isBlank(param.getApi())) { apis = linkConfigService.list().stream().map(LinkConfig::getApi).collect(Collectors.toSet()); } else { apis = DataUtil.toIdSet(param.getApi()); } //查询所有创建字段的关联类型 QueryWrapper wrapper = new QueryWrapper<>(); wrapper.in("api", apis).eq("is_link",1).eq("is_create",1); Map resultMap = dataObjectService.list().stream() .collect(Collectors.toMap( DataObject::getKeyPrefix, DataObject::getName, (existing, replacement) -> replacement)); for (String api : apis) { //邮件发送前三位编码不存在的对象 Set safeSet = new HashSet<>(); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api).eq("is_link",1).eq("is_create",1); List 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 resultMap,Set safeSet,List 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> mapList = customMapper.list(allFiled, api, sql2 +" order by Id limit "+ + i * 10000 + ",10000" ); for (int j = 1; j <= mapList.size(); j++) { List> updateMapList = new ArrayList<>(); Map 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 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 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 deleteTargetOrgData(SalesforceParam param) throws Exception { log.info("deleteTargetOrgData execute start .................."); QueryWrapper wrapper = new QueryWrapper<>(); if (StringUtils.isNotBlank(param.getApi())) { wrapper.in("name", DataUtil.toIdList(param.getApi())); } else { wrapper.eq("data_work", 1); } List 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 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> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200"); log.info("删除数据，当前批次：{}，批次数：{}", i,list.size()); Map idNameMap = new HashMap<>(); String[] ids = new String[list.size()]; for (int z = 0; z < list.size(); z++) { Map 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 objectMap = customMapper.getById("*", typeName, ids[index]); List> maps = new ArrayList<>(); Map m1 = new HashMap<>(); m1.put("key", "is_delete"); m1.put("value", 1); Map 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 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> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200"); log.info("删除数据，当前批次：{}，批次数：{}", i,list.size()); Map idNameMap = new HashMap<>(); String[] ids = new String[list.size()]; for (int z = 0; z < list.size(); z++) { Map 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 objectMap = customMapper.getById("*", typeName, ids[index]); List> maps = new ArrayList<>(); Map m1 = new HashMap<>(); m1.put("key", "is_delete"); m1.put("value", 1); Map 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 getDataBatchQueryWrapper(SalesforceParam param) { QueryWrapper qw = new QueryWrapper<>(); if (StringUtils.isNotBlank(param.getApi())) { List 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 insertSingle(SalesforceParam param) throws Exception { List> 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 dumpContentFolderMemberJob(SalesforceParam param) throws Exception { String api = "ContentFolderMember"; PartnerConnection partnerConnection = salesforceConnect.createConnect(); List> 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 fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList()); Field[] dsrFields = dsr.getFields(); try { for (Map 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 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 qw = new QueryWrapper<>(); if (StringUtils.isNotBlank(param.getApi())) { List apis = DataUtil.toIdList(param.getApi()); qw.in("name", apis); } qw.eq("data_work", true); List objectList = dataObjectService.list(qw); Map infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); for (DataObject dataObject : objectList) { String api = dataObject.getName(); QueryWrapper qw1 = new QueryWrapper<>(); qw1.eq("api", api); qw1.eq("isHtmlFormatted", true); List fields = dataFieldService.list(qw1); for (DataField dataField : fields) { String field = dataField.getField(); String sqlCondition = "new_id is not null and " + field + " like '%> 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 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 2); maps.add(linkMap); Map 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> maps = new ArrayList<>(); Map linkMap = new HashMap<>(); linkMap.put("key", "is_update"); linkMap.put("value", 1); maps.add(linkMap); Map 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("]*src\\s*=\\s*\"[^\"]*refid=([^\"]+)(?:"|&|&|\\\"|\"|')?[^>]*/?>", Pattern.CASE_INSENSITIVE); Matcher matcher = imgPattern.matcher(richTextContent); StringBuffer updatedContent = new StringBuffer(); Map 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 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 = ""; 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 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 getDocumentId(PartnerConnection connection, String contentVersionId) throws Exception { String soql = "SELECT ContentDocumentId,OwnerId FROM ContentVersion WHERE Id = '" + contentVersionId + "'"; QueryResult queryResult = connection.query(soql); Map 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 stringList = DataUtil.toIdList(queryFields); String maxId = null; Field[] dsrFields; List 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 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 stringList) throws Exception { for (int i = 0; i < objects.size(); i++) { JSONObject jsonObject = objects.getJSONObject(i); String id = jsonObject.getString(Const.ID); List> 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> buildUpdateMaps(String api, JSONObject jsonObject, List stringList) throws Exception { List> maps = Lists.newArrayList(); for (String key : stringList) { Map 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 handleAccountContactRelation(String key, JSONObject jsonObject) throws Exception { switch (key) { case "ContactId": Map 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 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 handleTaskEventRelation(String api, String key, JSONObject jsonObject) throws Exception { switch (key) { case "TaskId": Map 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 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 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 handleCollaborationGroupMember(String key, JSONObject jsonObject) throws Exception { switch (key) { case "CollaborationGroupId": Map 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 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 handleFeedAttachment(String key, JSONObject jsonObject) throws Exception { switch (key) { case "FeedEntityId": Map 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 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 getNewIdMap(String key, String tableName, String id) throws Exception { Map m = customMapper.getByNewId("Id", tableName, id); if (m == null || m.get("Id") == null) { return null; } Map paramMap = Maps.newHashMap(); paramMap.put("key", key); paramMap.put("value", m.get("Id")); return paramMap; } @Override public void getTaskNewId(SalesforceParam param) throws Exception { List> mapsList = customMapper.list("*", "Task", "where What_Type = EmailMessage and new_id is null"); } /** * 组装【单表】 一次性Insert 参数 */ public ReturnT insertSingleData(SalesforceParam param, List> futures) throws Exception { List 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 qw = new QueryWrapper<>(); qw.eq("data_lock", 1).in("name", apis); List 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 salesforceParams = null; QueryWrapper 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 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 autoInsertSingleData(SalesforceParam param, List> futures) throws Exception { QueryWrapper 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 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 salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); List 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 infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); String api = param.getApi(); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List 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 queryWrapper = new QueryWrapper<>(); queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",1); List configs = linkConfigService.list(queryWrapper); Map 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> 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> 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 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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 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 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 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 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 infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List 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 queryWrapper = new QueryWrapper<>(); queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); List configs = linkConfigService.list(queryWrapper); Map 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> 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> 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 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> maps = new ArrayList<>(); Map 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> maps = new ArrayList<>(); Map 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; } } } }