From 12e4a5a5e5b8081f21e7448746eaedb21c5eab66 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 19 Aug 2025 11:28:14 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=9120250819=E7=89=88?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../datadump/controller/JobController.java | 36 +- .../celnet/datadump/job/DataDumpNewJob.java | 48 +- .../celnet/datadump/mapper/CustomMapper.java | 9 + .../datadump/mapper/SalesforceMapper.java | 3 + .../datadump/param/DataDumpByParentParam.java | 8 + .../datadump/service/CommonService.java | 7 - .../service/DataDumpSpecialService.java | 4 + .../datadump/service/DataFieldService.java | 6 + .../service/DataImportNewService.java | 9 +- .../service/impl/CommonServiceImpl.java | 77 +- .../impl/DataDumpByParentServiceImpl.java | 19 +- .../impl/DataDumpSpecialServiceImpl.java | 76 + .../service/impl/DataFieldServiceImpl.java | 50 + .../impl/DataImportBatchServiceImpl.java | 133 +- .../impl/DataImportNewServiceImpl.java | 1218 ++++++++++++----- .../service/impl/DataImportServiceImpl.java | 218 +-- .../service/impl/FileServiceImpl.java | 23 +- .../com/celnet/datadump/util/DataUtil.java | 15 + src/main/resources/mapper/CustomMapper.xml | 22 + .../resources/mapper/SalesforceMapper.xml | 32 + 20 files changed, 1359 insertions(+), 654 deletions(-) diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java index f041a60..5c9809c 100644 --- a/src/main/java/com/celnet/datadump/controller/JobController.java +++ b/src/main/java/com/celnet/datadump/controller/JobController.java @@ -19,6 +19,8 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import java.util.List; + /** * @author Red * @description @@ -297,17 +299,33 @@ public class JobController { // 参数转换 param.setBeginCreateDate(param.getBeginDate()); param.setEndCreateDate(param.getEndDate()); - switch (param.getApi()){ - case "Contact": - return dataImportNewService.getPersonContact(param); - case "User": -// return dataImportNewService.getUser(param); - return ReturnT.FAIL; - default: - log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi()); - return ReturnT.FAIL; + + if (StringUtils.isBlank(param.getApi())) { + return new ReturnT<>(500, "api不能为空！"); } + List apis = DataUtil.toIdList(param.getApi()); + for (String api : apis) { + switch (api){ + case "Contact": + dataImportNewService.getPersonContact(param); + case "User": + case "BrandTemplate": + case "EmailTemplate": + case "Folder": + case "Group": + case "RecordType": + case "AccountContactRelation": + case "TaskRelation": + case "EventRelation": + case "CollaborationGroupMember": + case "FeedAttachment": + dataImportNewService.getNewIdByField(api,null); + default: + log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi()); + } + } + return ReturnT.SUCCESS; } /** diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index 629c2ec..b86092c 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -17,6 +17,8 @@ import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import java.util.List; + /** * 迁移任务 (新) * 2024/06/12 @@ -89,13 +91,13 @@ public class DataDumpNewJob { } /** - * 写入个人客户联系人old_id，返写new_id + * 返写NewId * @param paramStr 参数json * @return result */ - @XxlJob("getPersonContactJob") - public ReturnT getPersonContactJob(String paramStr) throws Exception { - log.info("getPersonContactJob execute start .................."); + @XxlJob("getNewId") + public ReturnT getNewId(String paramStr) throws Exception { + log.info("getNewId execute start .................."); SalesforceParam param = new SalesforceParam(); try { if (StringUtils.isNotBlank(paramStr)) { @@ -108,7 +110,43 @@ public class DataDumpNewJob { param.setBeginCreateDate(param.getBeginDate()); param.setEndCreateDate(param.getEndDate()); - return dataImportNewService.getPersonContact(param); + if (StringUtils.isBlank(param.getApi())) { + return new ReturnT<>(500, "api不能为空！"); + } + + List apis = DataUtil.toIdList(param.getApi()); + for (String api : apis) { + switch (api){ + case "Contact": + dataImportNewService.getPersonContact(param); + case "User": + dataImportNewService.getNewIdByField(api,"Username"); + case "BrandTemplate": + case "EmailTemplate": + case "Folder": + dataImportNewService.getNewIdByField(api,"DeveloperName"); + case "Group": + dataImportNewService.getNewIdByField(api,"DeveloperName,Type"); + case "RecordType": + dataImportNewService.getNewIdByField(api,"DeveloperName,SObjectType"); + case "AccountContactRelation": + dataImportNewService.getNewIdByField(api,"AccountId,ContactId,IsPrimary"); + case "TaskRelation": + dataImportNewService.getNewIdByField(api,"TaskId,RelationId,RelationType"); + case "EventRelation": + dataImportNewService.getNewIdByField(api,"EventId,RelationId,RelationType"); + case "CollaborationGroupMember": + dataImportNewService.getNewIdByField(api,"CollaborationGroupId,MemberId,CollaborationRole"); + case "FeedAttachment": + dataImportNewService.getNewIdByField(api,"FeedItemId,AttachmentId"); + case "Task": + dataImportNewService.getTaskNewId(param); + default: + log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi()); + } + } + return ReturnT.SUCCESS; + } /** diff --git a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java index 1a0f2a8..83ae80b 100644 --- a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java +++ b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java @@ -82,6 +82,14 @@ public interface CustomMapper { */ public void update(@Param("maps") List> maps, @Param("tableName") String api, @Param("sql") String sql); + /** + * 根据字段进行更新 + * @param maps + * @param api + * @param newId + */ + public void updateByField(@Param("maps") List> maps, @Param("tableName") String api, @Param("newId") String newId); + /** * 插入方法 * @@ -139,6 +147,7 @@ public interface CustomMapper { */ public Map getById(@Param("select") String select, @Param("api") String api, @Param("id") String id); + public Map getByNewId(@Param("select") String select, @Param("api") String api, @Param("id") String id); public Integer countBySQL(@Param("api") String api, @Param("sql") String sql); diff --git a/src/main/java/com/celnet/datadump/mapper/SalesforceMapper.java b/src/main/java/com/celnet/datadump/mapper/SalesforceMapper.java index a6746d3..8bcacaf 100644 --- a/src/main/java/com/celnet/datadump/mapper/SalesforceMapper.java +++ b/src/main/java/com/celnet/datadump/mapper/SalesforceMapper.java @@ -33,4 +33,7 @@ public interface SalesforceMapper { */ public void listOrderById(@Param("param") SalesforceParam param); + public void listOrderByIdNew(@Param("param") SalesforceParam param); + + } diff --git a/src/main/java/com/celnet/datadump/param/DataDumpByParentParam.java b/src/main/java/com/celnet/datadump/param/DataDumpByParentParam.java index 3d35dac..87d6856 100644 --- a/src/main/java/com/celnet/datadump/param/DataDumpByParentParam.java +++ b/src/main/java/com/celnet/datadump/param/DataDumpByParentParam.java @@ -2,6 +2,8 @@ package com.celnet.datadump.param; import lombok.Data; +import java.util.Date; + /** * @author Red * @description @@ -28,4 +30,10 @@ public class DataDumpByParentParam { * 父类字段 */ private String parentField; + + private Date beginModifyDate; + + private String type; + + } diff --git a/src/main/java/com/celnet/datadump/service/CommonService.java b/src/main/java/com/celnet/datadump/service/CommonService.java index be1c7bf..d9091d3 100644 --- a/src/main/java/com/celnet/datadump/service/CommonService.java +++ b/src/main/java/com/celnet/datadump/service/CommonService.java @@ -68,12 +68,5 @@ public interface CommonService { */ ReturnT createLinkTypeField(SalesforceParam param) throws Exception; - /** - * 更新关联类型 - * @param param 参数 - * @return ReturnT - * @throws Exception exception - */ - ReturnT updateLinkType(SalesforceParam param) throws Exception; } diff --git a/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java b/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java index f1e0ab9..c74a030 100644 --- a/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java +++ b/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java @@ -1,5 +1,6 @@ package com.celnet.datadump.service; +import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.param.DataDumpSpecialParam; import com.sforce.soap.partner.PartnerConnection; import com.xxl.job.core.biz.model.ReturnT; @@ -15,4 +16,7 @@ public interface DataDumpSpecialService { ReturnT dataDumpSpecial(DataDumpSpecialParam param) throws Throwable; Future getData(DataDumpSpecialParam param, PartnerConnection connect); + + Future getDataNew(DataDumpSpecialParam param, PartnerConnection connect, DataObject dataObject); + } diff --git a/src/main/java/com/celnet/datadump/service/DataFieldService.java b/src/main/java/com/celnet/datadump/service/DataFieldService.java index 55f8b92..c767723 100644 --- a/src/main/java/com/celnet/datadump/service/DataFieldService.java +++ b/src/main/java/com/celnet/datadump/service/DataFieldService.java @@ -22,4 +22,10 @@ public interface DataFieldService extends IService { public void tableRelation(); + String returnDateField(String api); + + String returnUpdateDateField(String api); + + Boolean hasCreatedDate(String api); + } diff --git a/src/main/java/com/celnet/datadump/service/DataImportNewService.java b/src/main/java/com/celnet/datadump/service/DataImportNewService.java index 4f5fc28..c9abf3c 100644 --- a/src/main/java/com/celnet/datadump/service/DataImportNewService.java +++ b/src/main/java/com/celnet/datadump/service/DataImportNewService.java @@ -4,12 +4,9 @@ import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.param.SalesforceParam; import com.xxl.job.core.biz.model.ReturnT; -import java.util.List; -import java.util.concurrent.Future; - public interface DataImportNewService { - ReturnT getPersonContact(SalesforceParam param) throws Exception; + void getPersonContact(SalesforceParam param) throws Exception; ReturnT immigrationUpdateNew(SalesforceParam param) throws Exception; @@ -31,4 +28,8 @@ public interface DataImportNewService { ReturnT dumpContentFolderMemberJob(SalesforceParam param) throws Exception; + void getNewIdByField(String api, String queryFields) throws Exception; + + void getTaskNewId(SalesforceParam param) throws Exception; + } diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java index 8d21871..07757ca 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java @@ -214,6 +214,15 @@ public class CommonServiceImpl implements CommonService { dataReport.setApis(list.stream().map(DataObject::getName).collect(Collectors.joining(","))); dataReportService.save(dataReport); for (DataObject dataObject : list) { + //无创建时间对象 + if (!hasCreatedDate(dataObject.getName())) { + DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam(); + dataDumpSpecialParam.setApi(dataObject.getName()); + Future future = dataDumpSpecialService.getDataNew(dataDumpSpecialParam, salesforceConnect.createConnect(),dataObject); + // 等待当前所有线程执行完成 + salesforceExecutor.waitForFutures(future); + continue; + } Future future = salesforceExecutor.execute(() -> { try { Date updateTime = new Date(); @@ -226,7 +235,6 @@ public class CommonServiceImpl implements CommonService { salesforceParam.setUpdateField(dataObject.getUpdateField()); } dumpDataNew(salesforceParam, dataReport,dataObject); - dataObject.setLastUpdateDate(updateTime); dataObjectService.updateById(dataObject); @@ -988,7 +996,7 @@ public class CommonServiceImpl implements CommonService { log.info("referenceTo too long set null, api:{}, field:{}, reference to:{}", apiName, field.getName(), StringUtils.join(field.getReferenceTo(), ",")); } if (field.getReferenceTo().length > 1){ - if (field.getName().contains("Id") && !"OwnerId".equals(field.getName())){ + if (field.getName().contains("Id") && !"OwnerId".equals(field.getName()) && !"UserOrGroupId".equals(field.getName())){ LinkConfig linkConfig = new LinkConfig(); linkConfig.setApi(apiName); linkConfig.setLabel(label); @@ -1134,7 +1142,7 @@ public class CommonServiceImpl implements CommonService { update.setName(apiName); update.setLastUpdateDate(endCreateDate); update.setBlobField(blobField); - if (count>0){ + if (count>0 || apiName.endsWith("Share")){ update.setIsEditable(false); } update.setKeyPrefix(dsr.getKeyPrefix()); @@ -1202,6 +1210,8 @@ public class CommonServiceImpl implements CommonService { return one != null; } + + /** * 存在附件的表 特殊构建 * @@ -1293,7 +1303,7 @@ public class CommonServiceImpl implements CommonService { //表内数据总量 Integer count = customMapper.countBySQL(api, "where ShareType = 'V' and new_id = '0'"); - //批量插入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 asc limit 200"); @@ -1416,64 +1426,7 @@ public class CommonServiceImpl implements CommonService { linkConfigService.updateById(config); } }catch (Exception e){ - log.error(e.getMessage()); - } - return ReturnT.SUCCESS; - } - - @Override - public ReturnT updateLinkType(SalesforceParam param) throws Exception { - try { - List apis; - if (StringUtils.isBlank(param.getApi())) { - apis = dataObjectService.list().stream().map(DataObject::getName).collect(Collectors.toList()); - } else { - apis = DataUtil.toIdList(param.getApi()); - } - QueryWrapper wrapper = new QueryWrapper<>(); - wrapper.eq("is_link","1").eq("is_create",1); - if (!apis.isEmpty()){ - wrapper.in("api", apis); - } - Map resultMap = dataObjectService.list().stream() - .collect(Collectors.toMap( - DataObject::getKeyPrefix, - DataObject::getName, - (existing, replacement) -> replacement)); - - List list = linkConfigService.list(wrapper); - for (LinkConfig config : list) { - - //若是该对象已经存在数据，则遍历写入值 - Integer dbNum = customMapper.countBySQL(config.getApi(),null); - log.info("表api：{} 存在" +dbNum+ "条数据！赋值当前表多态类型字段：{}", config.getApi(),config.getLinkField()); - - if (dbNum >0 ) { - int page = dbNum % 2000 == 0 ? dbNum / 2000 : (dbNum / 2000) + 1; - - for (int i = 0; i < page; i++) { - - log.info("表api：{}，字段：{}，批次：{}，执行数据更新!", config.getApi(),config.getLinkField(), i); - - List> mapList = customMapper.list("*", config.getApi(), "1=1 order by Id limit " + i * 2000 + ",2000"); - - List> updateMapList = new ArrayList<>(); - for (int j = 1; j <= mapList.size(); j++) { - Map map = mapList.get(j - 1); - 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); - customMapper.updateById(config.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id"))); - } - } - } - } - } - }catch (Exception e){ - log.error(e.getMessage()); + log.error(JSON.toJSONString(e)); } return ReturnT.SUCCESS; } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataDumpByParentServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataDumpByParentServiceImpl.java index 4501ec6..eddc132 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataDumpByParentServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataDumpByParentServiceImpl.java @@ -3,13 +3,16 @@ package com.celnet.datadump.service.impl; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.celnet.datadump.config.SalesforceConnect; +import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.global.Const; import com.celnet.datadump.param.DataDumpByParentParam; import com.celnet.datadump.param.SalesforceParam; import com.celnet.datadump.mapper.CustomMapper; import com.celnet.datadump.service.CommonService; import com.celnet.datadump.service.DataDumpByParentService; +import com.celnet.datadump.service.DataObjectService; import com.celnet.datadump.util.DataUtil; import com.celnet.datadump.util.EmailUtil; import com.celnet.datadump.util.SqlUtil; @@ -29,6 +32,7 @@ import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import java.util.Date; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -48,6 +52,8 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService { private CommonService commonService; @Autowired private SalesforceConnect salesforceConnect; + @Autowired + private DataObjectService dataObjectService; @Override public ReturnT dataDumpByParent(DataDumpByParentParam param) throws Throwable { @@ -62,6 +68,15 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService { // 检测子表 不存在就生成 但不生成批次 commonService.checkApi(param.getApi(), false); // 遍历父表 + if (!param.getType().equals("1")) { + QueryWrapper qw = new QueryWrapper<>(); + qw.eq("name", param.getApi()); + DataObject update = dataObjectService.getOne(qw); + param.setBeginModifyDate(update.getLastUpdateDate()); + update.setLastUpdateDate(new Date()); + dataObjectService.updateById(update); + } + while (true) { List strings = customMapper.listById(param.getParentField(), param.getParentApi(), "order by " + param.getParentField() + " asc limit " + page + "," + size); if (CollectionUtils.isEmpty(strings)) { @@ -72,6 +87,7 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService { Map map = Maps.newHashMap(); SalesforceParam salesforceParam = new SalesforceParam(); salesforceParam.setApi(param.getApi()); + salesforceParam.setBeginModifyDate(param.getBeginModifyDate()); salesforceParam.setSql(param.getField() + " IN ('" + StringUtils.join(strings, "','") + "')"); String maxId = null; Field[] dsrFields; @@ -93,11 +109,12 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService { int count = 0, failCount = 0; // 遍历子表 while (true) { + try { // 判断是否存在要排除的id salesforceParam.setMaxId(maxId); map.put("param", salesforceParam); - String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map); + String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdNew", map); log.info("query sql: {}", sql); XxlJobLogger.log("query sql: {}", sql); QueryResult queryResult = connect.queryAll(sql); diff --git a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java index d671b93..1adc945 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java @@ -219,6 +219,82 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService { }, 0, 0); } + @Override + public Future getDataNew(DataDumpSpecialParam param, PartnerConnection connect,DataObject dataObject) { + String api = param.getApi(); + return salesforceExecutor.execute(() -> { + try { + Map map = Maps.newHashMap(); + SalesforceParam salesforceParam = new SalesforceParam(); + salesforceParam.setApi(api); + salesforceParam.setIdField(param.getField()); + String maxId = null; + Field[] dsrFields; + List fields = Lists.newArrayList(); + // 获取sf字段 + { + DescribeSObjectResult dsr = connect.describeSObject(api); + dsrFields = dsr.getFields(); + for (Field field : dsrFields) { + // 不查询文件 + if ("base64".equalsIgnoreCase(field.getType().toString())) { + continue; + } + fields.add(field.getName()); + } + salesforceParam.setSelect(StringUtils.join(fields, ",")); + } + JSONArray objects = null; + int count = 0, failCount = 0; + // 遍历 + while (true) { + try { + // 判断是否存在要排除的id + salesforceParam.setMaxId(maxId); + salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate()); + map.put("param", salesforceParam); + String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdNew", map); + log.info("query sql: {}", sql); + XxlJobLogger.log("query sql: {}", sql); + QueryResult queryResult = connect.queryAll(sql); + if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) { + break; + } + SObject[] records = queryResult.getRecords(); + objects = DataUtil.toJsonArray(records, dsrFields); + maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID); + // 存储更新 + commonService.saveOrUpdate(api, fields, records, objects, true); + count += records.length; + TimeUnit.MILLISECONDS.sleep(1); + log.info("dump success count: {}", count); + XxlJobLogger.log("dump success count: {}", count); + failCount = 0; + objects = null; + } catch (InterruptedException e) { + throw e; + } catch (Throwable throwable) { + failCount++; + log.error("dataDumpSpecial error api:{}, data:{}", api, JSON.toJSONString(objects), throwable); + if (failCount > Const.MAX_FAIL_COUNT) { + throwable.addSuppressed(new Exception("dataDumpSpecial error data:" + JSON.toJSONString(objects))); + throw throwable; + } + TimeUnit.MINUTES.sleep(1); + } + } + updateDataBatch(connect, api); + dataObject.setLastUpdateDate(new Date()); + dataObjectService.updateById(dataObject); + } catch (Throwable throwable) { + String format = String.format("数据特殊表迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", api, JSON.toJSONString(param), throwable); + EmailUtil.send("DataDump ERROR", format); + log.error("dataDumpSpecial error ", throwable); + throw new RuntimeException(throwable); + } + }, 0, 0); + } + /** * 更新batch * diff --git a/src/main/java/com/celnet/datadump/service/impl/DataFieldServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataFieldServiceImpl.java index 3ce39cc..12e5ee2 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataFieldServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataFieldServiceImpl.java @@ -5,6 +5,7 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.celnet.datadump.entity.DataField; +import com.celnet.datadump.global.Const; import com.celnet.datadump.mapper.DataFieldMapper; import com.celnet.datadump.service.DataFieldService; import org.springframework.beans.factory.annotation.Autowired; @@ -23,6 +24,7 @@ import java.util.*; @Service public class DataFieldServiceImpl extends ServiceImpl implements DataFieldService { + @Override public Boolean hasDeleted(String api) { QueryWrapper qw = new QueryWrapper<>(); @@ -60,7 +62,55 @@ public class DataFieldServiceImpl extends ServiceImpl queryWrapper = new QueryWrapper<>(); + queryWrapper.eq("api", api); + List list = list(queryWrapper); + Set fieldNames = new HashSet<>(); + for (DataField dataField : list) { + fieldNames.add(dataField.getField()); + } + + if (fieldNames.contains("CreatedDate")) { + return "CreatedDate"; + } else if (fieldNames.contains("SystemModstamp")) { + return "SystemModstamp"; + } else if (fieldNames.contains("LastModifiedDate")) { + return "LastModifiedDate"; + } else { + return ""; + } + } + + @Override + public String returnUpdateDateField(String api) { + QueryWrapper queryWrapper = new QueryWrapper<>(); + queryWrapper.eq("api", api); + List list = list(queryWrapper); + + Set fieldNames = new HashSet<>(); + for (DataField dataField : list) { + fieldNames.add(dataField.getField()); + } + + if (fieldNames.contains("SystemModstamp")) { + return "SystemModstamp"; + } else if (fieldNames.contains("LastModifiedDate")) { + return "LastModifiedDate"; + } else { + return ""; + } + } + + @Override + public Boolean hasCreatedDate(String api) { + QueryWrapper qw = new QueryWrapper<>(); + qw.eq("api", api).eq("field", Const.CREATED_DATE).last("limit 1"); + DataField one = getOne(qw); + return one != null; + } public void getTableLevel(int[][] connections,List list) { diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java index 47df882..a4daef4 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -376,7 +376,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (StringUtils.isNotBlank(linkfield)){ reference = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null; } - if (reference == null){ continue; } @@ -419,8 +418,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } JobInfo salesforceInsertJob = null; - JobInfo salesforceQueryJob = null; - try { salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); @@ -428,40 +425,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); - List resultIds = checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); - - String idStr = "("; - for (String id : resultIds) { - idStr += "'" + id + "',"; // 拼接每个ID - } - if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束 - idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号 - } - idStr += ")"; // 添加右括号 - - String query = "SELECT Id,old_sfdc_id__c FROM " + api + " where Id in " + idStr; - - salesforceQueryJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.queryAll); - - ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes()); - - BatchInfo info = bulkConnection.createBatchFromStream(salesforceQueryJob, bout); - - List batchQueryInfos = new ArrayList<>(); - - batchQueryInfos.add(info); - - BulkUtil.awaitCompletion(bulkConnection, salesforceQueryJob, batchQueryInfos); - - sfNum = sfNum + checkQueryResultsNew(bulkConnection, salesforceQueryJob, info, api); + sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); } catch (Exception e) { log.error("createdNewIdBatch error api:{}", api, e); throw e; }finally { BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); - - BulkUtil.closeJob(bulkConnection, salesforceQueryJob.getId()); } } @@ -484,10 +454,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { /** * 读写Insert结果 */ - public List checkInsertResults(BulkConnection connection, JobInfo job, - List batchInfoList,String api,String[] ids) + public int checkInsertResults(BulkConnection connection, JobInfo job, + List batchInfoList,String api,String[] ids) throws AsyncApiException, IOException { - ArrayList strings = new ArrayList<>(); + int index = 0; // batchInfoList was populated when batches were created and submitted for (BatchInfo b : batchInfoList) { CSVReader rdr = @@ -497,7 +467,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { List row; while ((row = rdr.nextRecord()) != null) { - Map resultInfo = new HashMap<>(); + Map resultInfo = new HashMap(); for (int i = 0; i < resultCols; i++) { resultInfo.put(resultHeader.get(i), row.get(i)); } @@ -506,13 +476,23 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { String id = resultInfo.get("Id"); String error = resultInfo.get("Error"); if (insertStatus && created) { - strings.add(id); + List> maps = new ArrayList<>(); + Map m = new HashMap<>(); + m.put("key", "new_id"); + m.put("value", id); + maps.add(m); + customMapper.updateById(api, maps, ids[index]); + log.info("Created Success row with id " + id); + } else if (!insertStatus) { + log.error("Created Fail with error: " + error); } + index ++; } } - return strings; + return index; } + public Integer checkQueryResultsNew(BulkConnection connection, JobInfo job, BatchInfo batchInfo,String api) throws AsyncApiException, IOException { int num = 0; // 获取查询结果的 ResultId (对于Query类型，每个Batch可能包含多个ResultId) @@ -787,7 +767,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { log.error("总Update数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); for (int i = 0; i < page; i++) { - List> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 10000 + ",10000"); + + List> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 5000 + ",5000"); log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); @@ -812,18 +793,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } 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)){ - reference_to = map.get(linkfield)!=null?map.get(linkfield).toString():null; + if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); } if (reference_to == null){ continue; } - //判断reference_to内是否包含User字符串 - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } Map m = customMapper.getById("new_id", reference_to, value); if (m != null && !m.isEmpty()) { account.put(field, m.get("new_id")); @@ -863,7 +844,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { sfNum = sfNum + checkUpdateResults(bulkConnection, salesforceUpdateJob, batchInfos,api); } catch (Throwable e) { - log.info(e.getMessage()); + log.info(JSON.toJSONString(e)); throw e; }finally { @@ -1199,20 +1180,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){ //引用类型 String reference_to = dataField.getReferenceTo(); + //判断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)){ - reference_to = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null; + if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); } - if (reference_to == null){ continue; } - - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } + Map m = customMapper.getById("new_id", reference_to, data.get(j - 1).get(dataField.getField()).toString()); if (m != null && !m.isEmpty()) { account.put(dataField.getField(), m.get("new_id")); @@ -1243,50 +1223,20 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceInsertJob = null; - JobInfo salesforceQueryJob = null; - try { - salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); List batchInfos = CsvConverterUtil.writeToCsvNew(insertList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceInsertJob); BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); - List resultIds = checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); - - String idStr = "("; - for (String id : resultIds) { - idStr += "'" + id + "',"; // 拼接每个ID - } - if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束 - idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号 - } - idStr += ")"; // 添加右括号 - - String query = "SELECT Id,old_sfdc_id__c FROM " + api + " where Id in " + idStr; - - salesforceQueryJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.queryAll); - - ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes()); - - BatchInfo info = bulkConnection.createBatchFromStream(salesforceQueryJob, bout); - - List batchQueryInfos = new ArrayList<>(); - - batchQueryInfos.add(info); - - BulkUtil.awaitCompletion(bulkConnection, salesforceQueryJob, batchQueryInfos); - - sfNum = sfNum + checkQueryResultsNew(bulkConnection, salesforceQueryJob, info, api); + sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); } catch (Exception e) { - log.error("insertSingleData error api:{}", api, e); + log.error("createdNewIdBatch error api:{}", api, e); throw e; }finally { BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); - - BulkUtil.closeJob(bulkConnection, salesforceQueryJob.getId()); } } @@ -1369,20 +1319,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){ //引用类型 String reference_to = dataField.getReferenceTo(); + //判断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)){ - reference_to = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null; + if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); } - if (reference_to == null){ continue; } - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } Map m = customMapper.getById("new_id", reference_to, data.get(j - 1).get(dataField.getField()).toString()); if (m != null && !m.isEmpty()) { account.put(dataField.getField(), m.get("new_id")); diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java index c9bef2d..b8cb7cd 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -1,16 +1,13 @@ package com.celnet.datadump.service.impl; -import cn.hutool.core.lang.UUID; import com.alibaba.excel.EasyExcel; import com.alibaba.excel.ExcelWriter; -import com.alibaba.excel.util.FileUtils; 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.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.celnet.datadump.config.SalesforceConnect; import com.celnet.datadump.config.SalesforceExecutor; import com.celnet.datadump.config.SalesforceTargetConnect; @@ -20,16 +17,12 @@ 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.DataVerifyParam; +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.async.BatchInfo; -import com.sforce.async.BulkConnection; -import com.sforce.async.JobInfo; -import com.sforce.async.OperationEnum; import com.sforce.soap.partner.*; import com.sforce.soap.partner.sobject.SObject; import com.xxl.job.core.biz.model.ReturnT; @@ -54,14 +47,11 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.*; -import java.text.ParseException; -import java.text.SimpleDateFormat; 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.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; @Service @@ -111,17 +101,15 @@ public class DataImportNewServiceImpl implements DataImportNewService { * Get返写个人客户联系人入口 */ @Override - public ReturnT getPersonContact(SalesforceParam param) throws Exception { + 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) { - return result; } } - return ReturnT.SUCCESS; } catch (Exception exception) { salesforceExecutor.remove(futures.toArray(new Future[]{})); log.error("immigration error", exception); @@ -327,8 +315,11 @@ public class DataImportNewServiceImpl implements DataImportNewService { // 不存在 过滤 countParam.setIsDeleted(null); } - // sf count - Integer sfNum = commonService.countSfNum(partnerConnection, countParam); + 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) @@ -354,7 +345,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { try { if (StringUtils.isNotBlank(param.getApi())) { if (1 == param.getType()){ - return updateSfDataNew(param, futures); + return maualupdateSfDataNew(param, futures); }else { return updateIncrementalSfDataNew(param, futures); } @@ -378,7 +369,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { /** * 组装【单表】【存量】Update参数 */ - public ReturnT updateSfDataNew(SalesforceParam param, List> futures) throws Exception { + public ReturnT maualupdateSfDataNew(SalesforceParam param, List> futures) throws Exception { List apis = DataUtil.toIdList(param.getApi()); String beginDateStr = null; @@ -484,6 +475,11 @@ public class DataImportNewServiceImpl implements DataImportNewService { 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); @@ -492,20 +488,21 @@ public class DataImportNewServiceImpl implements DataImportNewService { salesforceParam.setBeginCreateDate(t.getSyncStartDate()); salesforceParam.setEndCreateDate(t.getSyncEndDate()); salesforceParam.setBatch(new AtomicInteger(1).getAndIncrement()); - - // 手动任务优先执行 - 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); - + 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 { @@ -681,15 +678,20 @@ public class DataImportNewServiceImpl implements DataImportNewService { salesforceParam.setEndCreateDate(t.getSyncEndDate()); salesforceParam.setBatch(batch.getAndIncrement()); - 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); + 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 { @@ -708,10 +710,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { * 执行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); + Map infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); String api = param.getApi(); - TimeUnit.MILLISECONDS.sleep(1); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List list = dataFieldService.list(dbQw); @@ -737,12 +738,13 @@ public class DataImportNewServiceImpl implements DataImportNewService { sql2 = "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; } }else { + String updateDateField = dataFieldService.returnUpdateDateField(api); if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and new_id is not null and LastModifiedDate >= '" + beginDateStr + "' "; - sql2 = "RowCause = 'Manual' and new_id is not null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; }else { - sql = "where new_id is not null and LastModifiedDate >= '" + beginDateStr + "' "; - sql2 = "new_id is not null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; } } //表内数据总量 @@ -768,97 +770,113 @@ public class DataImportNewServiceImpl implements DataImportNewService { int targetCount = 0; - //批量插入200一次 - int page = count%200 == 0 ? count/200 : (count/200) + 1; - log.info("总Update数据 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, page,beginDateStr, endDateStr, api); + int num = count%10000 == 0 ? count/10000 : (count/10000) + 1; - for (int i = 0; i < page; i++) { + log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api); - log.info("当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；当前时间：{}",i,beginDateStr, endDateStr, api,); + for (int z = 0; z < num; z++){ - List> mapList = customMapper.list("*", api, sql2+ i * 200 + ",200"); - SObject[] accounts = new SObject[mapList.size()]; - int j = 0; - try { - for (Map map : mapList) { - SObject account = new SObject(); - account.setType(api); - //给对象赋值 - for (DataField dataField : list) { - String field = dataField.getField(); - String reference_to = dataField.getReferenceTo(); + List> mapsList = customMapper.list("*", api, sql2+ z * 10000 + ",10000"); + int size = mapsList.size(); + log.info("总Update数据 count:{}；当前批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,z, beginDateStr, endDateStr, api); - 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.getIsCreateable() != null && !dataField.getIsCreateable())) { - continue; - } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) { + //批量插入200一次 + int page = size%200 == 0 ? count/200 : (count/200) + 1; - if (!"null".equals(value) && StringUtils.isNotEmpty(value)) { - //引用类型字段 - String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield)){ - reference_to = map.get(linkfield)!=null?map.get(linkfield).toString():null; - } - if (reference_to == null){ + 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); + //给对象赋值 + 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.getIsCreateable() != null && !dataField.getIsCreateable())) { continue; - } -// log.info("----------" + dataField.getField() + "的引用类型 ------------" + reference_to); - //判断reference_to内是否包含User字符串 - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } - Map m = customMapper.getById("new_id", reference_to, value); - if (m != null && !m.isEmpty()) { - account.setField(field, m.get("new_id")); - }else { - String message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; - EmailUtil.send("DataDump ERROR", message); - log.info(message); - return; - } - } - } 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 (dataObject.getIsEditable()){ - account.setField("old_owner_id__c", map.get("OwnerId")); - account.setField("old_sfdc_id__c", map.get("Id")); - } - accounts[j++] = account; - } + } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) { - if (infoFlag != null && "1".equals(infoFlag.get("value"))){ - printlnAccountsDetails(accounts,list); - } - SaveResult[] saveResults = partnerConnection.update(accounts); - for (SaveResult saveResult : saveResults) { - if (!saveResult.getSuccess()) { - Map map = returnErrorAccountsDetails(accounts, list, saveResult.getId()); - 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),JSON.toJSONString(map)); - EmailUtil.send("DataDump ERROR", format); - return; - }else { - targetCount ++; + 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) && map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); + } + if (reference_to == null){ + continue; + } + Map m = customMapper.getById("new_id", reference_to, value); + if (m != null && !m.isEmpty()) { + account.setField(field, m.get("new_id")); + }else { + String message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; + EmailUtil.send("DataDump ERROR", message); + 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 (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()) { + Map map = returnErrorAccountsDetails(accounts, list, saveResult.getId()); + 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),JSON.toJSONString(map)); + EmailUtil.send("DataDump ERROR", format); + return; + }else { + targetCount ++; + } + } + + } catch (Throwable e) { + log.info(JSON.toJSONString(e)); } - } catch (Throwable e) { - throw 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", targetCount); + .set("target_update_num", num); dataBatchHistoryService.update(updateQw); UpdateWrapper updateQw2 = new UpdateWrapper<>(); @@ -869,6 +887,175 @@ public class DataImportNewServiceImpl implements DataImportNewService { dataBatchService.update(updateQw2); } + /** + * 执行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; + + if (api.endsWith("Share")){ + sql = "where RowCause = 'Manual' and new_id is null "; + sql2 = "RowCause = 'Manual' and new_id is null limit 10000"; + }else { + sql = "where new_id is null "; + sql2 = "new_id is null limit 10000"; + } + + if (1 == 1) { + if (api.endsWith("Share")){ + sql = "where RowCause = 'Manual' and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; + sql2 = "RowCause = 'Manual' and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; + }else { + sql = "where new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; + sql2 = "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; + } + }else { + String updateDateField = dataFieldService.returnUpdateDateField(api); + if (api.endsWith("Share")){ + sql = "where RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "RowCause = 'Manual' and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; + }else { + sql = "where new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "new_id is not null 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 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+ 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); + //给对象赋值 + 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.getIsCreateable() != null && !dataField.getIsCreateable())) { + 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) && map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); + } + if (reference_to == null){ + continue; + } + Map m = customMapper.getById("new_id", reference_to, value); + if (m != null && !m.isEmpty()) { + account.setField(field, m.get("new_id")); + }else { + String message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; + EmailUtil.send("DataDump ERROR", message); + 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 (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()) { + Map map = returnErrorAccountsDetails(accounts, list, saveResult.getId()); + String format = String.format("数据更新 error, api name: %s, \n cause:\n%s, \n数据实体类:\n%s", api, com.alibaba.fastjson2.JSON.toJSONString(DataDumpParam.getFilter()), JSON.toJSONString(saveResult),JSON.toJSONString(map)); + EmailUtil.send("DataDump ERROR", format); + return; + }else { + targetCount ++; + } + } + + } catch (Throwable e) { + log.info(JSON.toJSONString(e)); + } + } + } + + } + /** * 打印SF交互数据明细 */ @@ -1052,6 +1239,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { List objectList = dataObjectService.list(wrapper); if (objectList.isEmpty()){ log.info("没有对象存在文件二进制字段！不进行文件下载"); + return ReturnT.FAIL; } List> poll = customMapper.list("code,value","org_config",null); @@ -1172,7 +1360,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { return; } - int page = count%1000 == 0 ? count/1000 : (count/1000) + 1; + int page = count%200 == 0 ? count/200 : (count/200) + 1; log.info("api：{}；总文件数 count:{}；总批次 ：{}；-开始时间:{}；-结束时间:{}；",api, count, page,beginDateStr, endDateStr); @@ -1182,7 +1370,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { log.info("api：{} 批次:{}；-开始时间:{}；-结束时间:{}；",api, i, beginDateStr, endDateStr); // 获取未存储的附件id - List> list = customMapper.list("Id, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit " + i * 1000 + ",1000"); + List> list = customMapper.list("Id, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit 200"); for (Map map : list) { int failCount = 0; @@ -1217,6 +1405,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { } paramMap.put("value", filePath); maps.add(paramMap); + inputStream.close(); break; }else { failCount++; @@ -1324,7 +1513,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { log.info("没有对象存在文件二进制字段！不进行文件上传"); } - PartnerConnection connect = salesforceConnect.createConnect(); + PartnerConnection connect = salesforceTargetConnect.createConnect(); List> futures = Lists.newArrayList(); @@ -1551,7 +1740,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { return; } - int page = count%1000 == 0 ? count/1000 : (count/1000) + 1; + int page = count%200 == 0 ? count/200 : (count/200) + 1; log.info("总文件数 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); @@ -1559,7 +1748,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { 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 * 1000 + ",1000"); + List> documentList = customMapper.list("Id " , "ContentDocument", " new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 200 + ",200"); for (Map documentMap : documentList) { @@ -1638,7 +1827,6 @@ public class DataImportNewServiceImpl implements DataImportNewService { dMap.put("value", dId); dList.add(dMap); customMapper.updateById("ContentDocument", dList, oldDocumentId); - newDocumentId = dId; } break; @@ -1870,21 +2058,23 @@ public class DataImportNewServiceImpl implements DataImportNewService { * 组装updateLinkType执行参数 */ public ReturnT updateLinkType(SalesforceParam param, List> futures) throws Exception { - List apis; + Set apis; if (StringUtils.isBlank(param.getApi())) { - apis = dataObjectService.list().stream().map(DataObject::getName).collect(Collectors.toList()); + apis = linkConfigService.list().stream().map(LinkConfig::getApi).collect(Collectors.toSet()); } else { - apis = DataUtil.toIdList(param.getApi()); + apis = DataUtil.toIdSet(param.getApi()); } + //查询所有创建字段的关联类型 QueryWrapper wrapper = new QueryWrapper<>(); - wrapper.in("api", apis).eq("is_link",1); + 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)); + //邮件发送前三位编码不存在的对象 Set safeSet = ConcurrentHashMap.newKeySet(); for (String api : apis) { @@ -1929,31 +2119,49 @@ public class DataImportNewServiceImpl implements DataImportNewService { String sql1 = null; String sql2 = null; if (beginDateStr != null){ - if (api.endsWith("Share")){ - sql1 = "where lastModifiedDate >= '" + beginDateStr + "'"; - sql2 = "lastModifiedDate >= '"+ beginDateStr +"' order by Id limit " ; - }else { - sql1 = "where SystemModstamp >= '" + beginDateStr + "'"; - sql2 = "SystemModstamp >= '"+ beginDateStr +"' order by Id limit " ; - } + String updateDateField = dataFieldService.returnUpdateDateField(api); + sql1 = "where " + updateDateField + " >= '" + beginDateStr + "'"; + sql2 = updateDateField + " >= '"+ beginDateStr +"' order by Id limit " ; }else { - sql1 = null; sql2 = "1=1 order by Id limit " ; } + 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+ "条数据！", param.getApi()); if (count >0 ) { - int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1; + int page = count % 10000 == 0 ? count / 10000 : (count / 10000) + 1; for (int i = 0; i < page; i++) { - log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 2000); + log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 10000); - List> mapList = customMapper.list("*", api, sql2 + i * 2000 + ",2000" ); + List> mapList = customMapper.list(allFiled, api, sql2 + i * 10000 + ",10000" ); for (int j = 1; j <= mapList.size(); j++) { List> updateMapList = new ArrayList<>(); @@ -1969,9 +2177,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { 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"))); - } + } + 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"))); } } } @@ -1986,10 +2194,26 @@ public class DataImportNewServiceImpl implements DataImportNewService { @Override public ReturnT checkDeletedData(SalesforceParam param) throws Exception { + if (param.getType() == 1){ + return munalCheckDeletedData(param); + }else { + return autoCheckDeletedData(param); + } + + } + + private ReturnT munalCheckDeletedData(SalesforceParam param) throws Exception { + return ReturnT.SUCCESS; } + private ReturnT autoCheckDeletedData(SalesforceParam param) throws Exception { + + return ReturnT.SUCCESS; + } + + @Override public ReturnT verifyDataJob(SalesforceParam param) throws Exception { @@ -2111,6 +2335,211 @@ public class DataImportNewServiceImpl implements DataImportNewService { return ReturnT.SUCCESS; } + @Override + public void getNewIdByField(String api, String queryFields) throws Exception { + List stringList = DataUtil.toIdList(queryFields); + // 检测表 不存在就生成 但不生成批次 + commonService.checkApi(api, true); + Map map = Maps.newHashMap(); + SalesforceParam salesforceParam = new SalesforceParam(); + salesforceParam.setApi(api); + salesforceParam.setIdField("Id"); + String maxId = null; + Field[] dsrFields; + + List fields = Lists.newArrayList(); + PartnerConnection connect = salesforceConnect.createConnect(); + + DescribeSObjectResult dsr = connect.describeSObject(api); + dsrFields = dsr.getFields(); + for (Field field : dsrFields) { + // 不查询文件 + if ("base64".equalsIgnoreCase(field.getType().toString())) { + continue; + } + fields.add(field.getName()); + } + if ("TaskRelation".equals(api) || "EventRelation".equals(api)){ + fields.add("Relation.type"); + } + if ("CollaborationGroupMember".equals(api) || "MemberId".equals(api)){ + fields.add("Member.type"); + } + if ("FeedAttachment".equals(api) || "FeedEntityId".equals(api)){ + fields.add("FeedEntity.type"); + } + if ("FeedAttachment".equals(api) || "RecordId".equals(api)){ + fields.add("Record.type"); + } + salesforceParam.setSelect(StringUtils.join(fields, ",")); + com.alibaba.fastjson2.JSONArray objects = null; + int count = 0, failCount = 0; + // 遍历 + while (true) { + try { + // 判断是否存在要排除的id + salesforceParam.setMaxId(maxId); + map.put("param", salesforceParam); + String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map); + log.info("query sql: {}", sql); + XxlJobLogger.log("query sql: {}", sql); + QueryResult queryResult = connect.queryAll(sql); + if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) { + break; + } + SObject[] records = queryResult.getRecords(); + objects = DataUtil.toJsonArray(records, dsrFields); + maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID); + + // 更新new_id + for (int i = 0; i < objects.size(); i++) { + JSONObject jsonObject = objects.getJSONObject(i); + try { + Set keys = jsonObject.keySet(); + // update + String id = jsonObject.getString(Const.ID); + List> maps = Lists.newArrayList(); + for (String key : keys) { + if ("AccountContactRelation".equals(api) && "ContactId".equals(key)) { + Map m = customMapper.getByNewId("Id", "Contact", jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("AccountContactRelation".equals(api) && "AccountId".equals(key)) { + Map m = customMapper.getByNewId("Id", "Account", jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("TaskRelation".equals(api) && "TaskId".equals(key)) { + Map m = customMapper.getByNewId("Id", "Task", jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("EventRelation".equals(api) && "EventId".equals(key)) { + Map m = customMapper.getByNewId("Id", "Event", jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if (("EventRelation".equals(api) || "TaskRelation".equals(api)) && "RelationId".equals(key)) { + Map m = customMapper.getByNewId("Id", jsonObject.getString("Relation_type"), jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("CollaborationGroupMember".equals(api) && "CollaborationGroupId".equals(key)) { + Map m = customMapper.getByNewId("Id", "CollaborationGroup", jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("CollaborationGroupMember".equals(api) && "MemberId".equals(key)) { + Map m = customMapper.getByNewId("Id", jsonObject.getString("Member_type"), jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("FeedAttachment".equals(api) && "FeedEntityId".equals(key)) { + Map m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_type"), jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if ("FeedAttachment".equals(api) && "RecordId".equals(key)) { + Map m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_type"), jsonObject.getString(key)); + if (m == null){ + break; + } + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value",m.get("Id")); + maps.add(paramMap); + continue; + } + if (stringList.contains(key)) { + Map paramMap = Maps.newHashMap(); + paramMap.put("key", key); + paramMap.put("value", jsonObject.getString(key)); + maps.add(paramMap); + } + } + customMapper.updateByField(maps, api, id); + } catch (Throwable throwable) { + if (throwable.toString().contains("interrupt")) { + log.error("may interrupt error:", throwable); + throw new InterruptedException(); + } + log.info(JSON.toJSONString(throwable)); + } + } + count += records.length; + log.info("getNewId success count: {}", count); + failCount = 0; + objects = null; + } catch (InterruptedException e) { + throw e; + } catch (Throwable throwable) { + failCount++; + log.error("getNewId error api:{}, data:{}", api, com.alibaba.fastjson2.JSON.toJSONString(objects), throwable); + if (failCount > Const.MAX_FAIL_COUNT) { + log.info("getNewId error data:" + com.alibaba.fastjson2.JSON.toJSONString(objects)); + } + } + } + + // 3、拼接sql，更新本地数据库new_id + } + + @Override + public void getTaskNewId(SalesforceParam param) throws Exception { + List> mapsList = customMapper.list("*", "Task", "where What_Type = EmailMessage and new_id is null"); + + } + + /** * 组装【单表】 一次性Insert 参数 */ @@ -2145,12 +2574,10 @@ public class DataImportNewServiceImpl implements DataImportNewService { for (String api : apis) { DataObject update = dataObjectService.getById(api); try { - - if (api.endsWith("Share") || "GroupMember".equals(api)){ + if (!dataFieldService.hasCreatedDate(api)){ insertSingleShareData(api,partnerConnection,update); continue; } - List salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); @@ -2199,78 +2626,95 @@ public class DataImportNewServiceImpl implements DataImportNewService { return ReturnT.SUCCESS; } + /** - * 组装【单表】一次性Insert 参数 + * 组装【多】一次性Insert 参数 */ - public ReturnT autoInsertSingleData(SalesforceParam param, List> futures) throws Exception { - List apis = DataUtil.toIdList(param.getApi()); + 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"); // 全量的时候 检测是否有自动任务锁住的表 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()); + QueryWrapper dw = new QueryWrapper<>(); + qw.eq("data_lock", 1); + List dos = dataObjectService.list(dw); + if (CollectionUtils.isNotEmpty(dos)) { + String apiNames = dos.stream().map(DataObject::getName).collect(Collectors.joining()); + log.info("api:" + apiNames + " is locked"); return new ReturnT<>(500, "api:" + apiNames + " is locked"); } } PartnerConnection partnerConnection = salesforceTargetConnect.createConnect(); - for (String api : apis) { - QueryWrapper qw = new QueryWrapper<>(); - qw.eq("name", api); - DataObject update = dataObjectService.getOne(qw); - try { - if (api.endsWith("Share") || "GroupMember".equals(api)){ - insertSingleShareData(api,partnerConnection,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 -> { - if(t.getSyncEndDate().toInstant().isAfter(t.getLastModifiedDate().toInstant())){ + 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,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; - } - return null; - }).collect(Collectors.toList()); - } + }).collect(Collectors.toList()); + } - // 手动任务优先执行 - for (SalesforceParam salesforceParam : salesforceParams) { - 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[]{})); - - } catch (Exception e) { - throw e; - } finally { - if (isFull) { - update.setNeedUpdate(false); - update.setName(api); + // 手动任务优先执行 + for (SalesforceParam salesforceParam : salesforceParams) { + Future future = salesforceExecutor.execute(() -> { + try { + insertSingleExecute(salesforceParam, partnerConnection,update); + } 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) { + 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; } @@ -2282,7 +2726,6 @@ public class DataImportNewServiceImpl implements DataImportNewService { Map infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); String api = param.getApi(); - TimeUnit.MILLISECONDS.sleep(1); QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List list = dataFieldService.list(dbQw); @@ -2302,18 +2745,19 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (1 == param.getType()) { if (api.endsWith("Share")){ sql = "where RowCause = 'Manual' and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; - sql2 = "RowCause = 'Manual' and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; + sql2 = "RowCause = 'Manual' 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 "; + 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 = 'Manual' and new_id is null and LastModifiedDate >= '" + beginDateStr + "' "; - sql2 = "RowCause = 'Manual' and new_id is null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where RowCause = 'Manual' and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "RowCause = 'Manual' and new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000"; }else { - sql = "where new_id is null and LastModifiedDate >= '" + beginDateStr + "' "; - sql2 = "new_id is null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where new_id is null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "new_id is null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit 10000"; } } //表内数据总量 @@ -2338,98 +2782,117 @@ public class DataImportNewServiceImpl implements DataImportNewService { } int targetCount = 0; - //批量插入200一次 - int page = count%200 == 0 ? count/200 : (count/200) + 1; - log.info("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); - for (int i = 0; i < page; i++) { - try { - List> data = customMapper.list("*", api, sql2+ i * 200 + ",200"); - int size = data.size(); + 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++) { + 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(); log.info("执行api:{}, 执行批次:{}, 执行数据量:{}，开始时间：{}，结束时间：{}", api, i+1, size,beginDateStr,endDateStr); - SObject[] accounts = new SObject[size]; - String[] ids = new String[size]; - for (int j = 1; j <= size; j++) { - SObject account = new SObject(); - account.setType(api); - //给对象赋值 - for (DataField dataField : list) { - String field = dataField.getField(); - String reference_to = dataField.getReferenceTo(); + 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); + //给对象赋值 + 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) { + 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)) { - //引用类型字段 - String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield)){ - reference_to = data.get(j - 1).get(linkfield)!=null?data.get(j - 1).get(linkfield).toString():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) && data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + } + if (reference_to == null){ + continue; + } + Map m = customMapper.getById("new_id", reference_to, value); + if (m != null && !m.isEmpty()) { + account.setField(field, m.get("new_id")); + }else { + String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; + EmailUtil.send("DataDump ERROR", message); + log.info(message); + break; + } } - if (reference_to == null){ - continue; - } - //判断reference_to内是否包含User字符串 - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } - Map m = customMapper.getById("new_id", reference_to, value); - if (m != null && !m.isEmpty()) { - account.setField(field, m.get("new_id")); + } else { + if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) { + account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value)); }else { - String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; - EmailUtil.send("DataDump ERROR", message); - log.info(message); - return; + account.setField(field, data.get(j - 1).get(field)); } } - } 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 (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")); + 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; + } } - ids[j-1] = data.get(j-1).get("Id").toString(); - accounts[j-1] = account; - if (i*200+j == count){ - break; + if (infoFlag != null && "1".equals(infoFlag.get("value"))){ + printlnAccountsDetails(accounts, list); } - } - - if (infoFlag != null && "1".equals(infoFlag.get("value"))){ - printlnAccountsDetails(accounts, list); - } - SaveResult[] saveResults = partnerConnection.create(accounts); - log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults)); - 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]); - }else{ - log.error("-------------saveResults: {}", JSON.toJSONString(saveResult)); + SaveResult[] saveResults = partnerConnection.create(accounts); + 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{ + log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult)); + } + index++; } - index++; + } catch (Exception e) { + log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e)); } - TimeUnit.MILLISECONDS.sleep(1); - } catch (Exception e) { - log.error("insertSingle error api:{},错误信息：{}", api, e.getMessage()); } } + SalesforceParam countParam = new SalesforceParam(); countParam.setApi(api); countParam.setBeginCreateDate(beginDate); @@ -2441,14 +2904,16 @@ public class DataImportNewServiceImpl implements DataImportNewService { // 不存在 过滤 countParam.setIsDeleted(null); } - // sf count - Integer sfNum = commonService.countSfNum(partnerConnection, countParam); + 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", sfNum); + .set("target_sf_num", count); dataBatchHistoryService.update(updateQw); UpdateWrapper updateQw2 = new UpdateWrapper<>(); @@ -2462,7 +2927,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { updateQw3.eq("name", api) .eq("sync_start_date", beginDate) .eq("sync_end_date", DateUtils.addSeconds(endDate, -1)) - .set("target_update_num", targetCount); + .set("target_update_num", count); dataBatchHistoryService.update(updateQw3); UpdateWrapper updateQw4 = new UpdateWrapper<>(); @@ -2489,10 +2954,10 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (api.endsWith("Share")){ sql = "where RowCause = 'Manual' and new_id is null "; - sql2 = "RowCause = 'Manual' and new_id is null order by Id asc limit "; + sql2 = "RowCause = 'Manual' and new_id is null limit 10000"; }else { sql = "where new_id is null "; - sql2 = "new_id is null order by Id asc limit "; + sql2 = "new_id is null limit 10000"; } //表内数据总量 @@ -2501,10 +2966,6 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (count == 0) { return; } - //批量插入200一次 - int page = count%200 == 0 ? count/200 : (count/200) + 1; - - log.error("总Insert数据 count:{}；总批次：{}; -api:{}；", count,page, api); //查询当前对象多态字段映射 QueryWrapper queryWrapper = new QueryWrapper<>(); @@ -2520,92 +2981,111 @@ public class DataImportNewServiceImpl implements DataImportNewService { )); } - for (int i = 0; i < page; i++) { - try { - List> data = customMapper.list("*", api, sql2+ i * 200 + ",200"); - int size = data.size(); - log.info("执行api:{}, 执行批次:{}, 执行数据量:{}", api, i+1, size); - SObject[] accounts = new SObject[size]; - String[] ids = new String[size]; - for (int j = 1; j <= size; j++) { - SObject account = new SObject(); - account.setType(api); - //给对象赋值 - for (DataField dataField : list) { - String field = dataField.getField(); - String reference_to = dataField.getReferenceTo(); + int num = count%10000 == 0 ? count/10000 : (count/10000) + 1; - 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) { + log.info("总Insert数据 count:{}；总批次：{}; -api:{}；", count,num, api); - if (!"null".equals(value) && StringUtils.isNotEmpty(value)) { - //引用类型字段 - String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield)){ - reference_to = data.get(j - 1).get(linkfield)!=null?data.get(j - 1).get(linkfield).toString():null; - } - if (reference_to == null){ - continue; - } - //判断reference_to内是否包含User字符串 - if (reference_to.contains(",User") || reference_to.contains("User,")) { - reference_to = "User"; - } - Map m = customMapper.getById("new_id", reference_to, value); - if (m != null && !m.isEmpty()) { - account.setField(field, m.get("new_id")); - }else { - String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; - EmailUtil.send("DataDump ERROR", message); - log.info(message); - return; - } - } - } 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 (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")); - } + for (int z = 0; z < num; z++) { - ids[j-1] = data.get(j-1).get("Id").toString(); - accounts[j-1] = account; - if (i*200+j == count){ + 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 { + 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(); - if (infoFlag != null && "1".equals(infoFlag.get("value"))){ - printlnAccountsDetails(accounts, list); - } - SaveResult[] saveResults = partnerConnection.create(accounts); - log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults)); - 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]); - }else{ - log.error("-------------saveResults: {}", JSON.toJSONString(saveResult)); + SObject[] accounts = new SObject[sized]; + String[] ids = new String[sized]; + for (int j = 1; j <= sized; j++) { + SObject account = new SObject(); + account.setType(api); + //给对象赋值 + 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) && data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + } + if (reference_to == null){ + continue; + } + Map m = customMapper.getById("new_id", reference_to, value); + if (m != null && !m.isEmpty()) { + account.setField(field, m.get("new_id")); + }else { + String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！"; + EmailUtil.send("DataDump ERROR", message); + 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 (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; + } } - index++; + + if (infoFlag != null && "1".equals(infoFlag.get("value"))){ + printlnAccountsDetails(accounts, list); + } + SaveResult[] saveResults = partnerConnection.create(accounts); + 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]); + }else{ + log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult)); + } + index++; + } + TimeUnit.MILLISECONDS.sleep(1); + } catch (Exception e) { + log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e)); } - TimeUnit.MILLISECONDS.sleep(1); - } catch (Exception e) { - log.error("insertSingle error api:{},错误信息：{}", api, e.getMessage()); } } } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java index 957ab82..756b71d 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java @@ -255,6 +255,7 @@ public class DataImportServiceImpl implements DataImportService { } private void manualCreatedNewId(SalesforceParam param, PartnerConnection partnerConnection) throws Exception { + Map infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); String api = param.getApi(); QueryWrapper dbQw = new QueryWrapper<>(); @@ -272,7 +273,6 @@ public class DataImportServiceImpl implements DataImportService { } //表内数据总量 Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"); - log.error("总Insert数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api); if (count == 0) { return; } @@ -291,121 +291,139 @@ public class DataImportServiceImpl implements DataImportService { )); } - //批量插入200一次 - int page = count%200 == 0 ? count/200 : (count/200) + 1; - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"); DataObject update = new DataObject(); update.setName(api); - for (int i = 0; i < page; i++) { - try { - List> data = customMapper.list("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 200"); - int size = data.size(); - log.info("执行api:{}, 执行page:{}, 执行size:{}", api, i+1, size); - SObject[] accounts = new SObject[size]; - String[] ids = new String[size]; - for (int j = 1; j <= size; j++) { - SObject account = new SObject(); - account.setType(api); - //找出sf对象必填字段，并且给默认值 - for (DataField dataField : list) { - if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField()) - || "Id".equals(dataField.getField())){ - continue; - } - if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){ - //object类型转Date类型 - Date date; - try { - date = sdf.parse(String.valueOf(data.get(j - 1).get("CreatedDate"))); - }catch (ParseException e){ - //解决当时间秒为0时，转换秒精度丢失问题 - date = sdf.parse(data.get(j - 1).get("CreatedDate")+":00"); - } - Calendar calendar = Calendar.getInstance(); - calendar.setTime(date); - account.setField("CreatedDate", calendar); - continue; - } - if ("CreatedById".equals(dataField.getField()) && dataField.getIsCreateable()){ - Map CreatedByIdMap = customMapper.getById("new_id", "User", data.get(j-1).get("CreatedById").toString()); - if(CreatedByIdMap != null ){ - account.setField("CreatedById", CreatedByIdMap.get("new_id")); - } - continue; - } - if (dataField.getIsCreateable() != null && dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) { + int num = count%10000 == 0 ? count/10000 : (count/10000) + 1; - if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){ - //引用类型 - String reference = dataField.getReferenceTo(); - //引用类型字段 - String linkfield = fieldMap.get(dataField.getField()); + log.info("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, num,beginDateStr, endDateStr, api); - if (StringUtils.isNotBlank(linkfield)){ - reference = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null; - } + for (int z = 0; z < num; z++) { - if (reference == null){ - continue; - } + List> mapsList = customMapper.list("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 10000"); - List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); - if (referenceMap.isEmpty()){ - QueryWrapper maxIndex = new QueryWrapper<>(); - maxIndex.select("IFNULL(max(data_index),0) as data_index"); - maxIndex.ne("name", api); - Map map = dataObjectService.getMap(maxIndex); - //如果必填lookup字段没有值，跳过 - update.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1); - dataObjectService.updateById(update); - return; - }else{ - account.setField(dataField.getField(), referenceMap.get(0).get("new_id")); - continue; - } - } - if ("picklist".equals(dataField.getSfType())){ - List> pickList = customMapper.list("value", "data_picklist", "api = '"+api+"' and field = '"+dataField.getField()+"' limit 1"); - account.setField(dataField.getField(), pickList.get(0).get("value")); + log.info("总Insert数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, z,beginDateStr, endDateStr, api); + + int size = mapsList.size(); + + int page = size%200 == 0 ? size/200 : (size/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); + int sized = mapList.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); + //找出sf对象必填字段，并且给默认值 + for (DataField dataField : list) { + if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField()) + || "Id".equals(dataField.getField())){ continue; } - account.setField(dataField.getField(), DataUtil.fieldTypeToSf(dataField)); + if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){ + //object类型转Date类型 + Date date; + try { + date = sdf.parse(String.valueOf(mapList.get(j - 1).get("CreatedDate"))); + }catch (ParseException e){ + //解决当时间秒为0时，转换秒精度丢失问题 + date = sdf.parse(mapList.get(j - 1).get("CreatedDate")+":00"); + } + Calendar calendar = Calendar.getInstance(); + calendar.setTime(date); + account.setField("CreatedDate", calendar); + continue; + } + if ("CreatedById".equals(dataField.getField()) && dataField.getIsCreateable()){ + Map CreatedByIdMap = customMapper.getById("new_id", "User", mapList.get(j-1).get("CreatedById").toString()); + if(CreatedByIdMap != null ){ + account.setField("CreatedById", CreatedByIdMap.get("new_id")); + } + continue; + } + if (dataField.getIsCreateable() != null && dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) { + + if ("reference".equals(dataField.getSfType()) && mapList.get(j - 1).get(dataField.getField()) != null){ + //引用类型 + String reference = dataField.getReferenceTo(); + //引用类型字段 + String linkfield = fieldMap.get(dataField.getField()); + + if (StringUtils.isNotBlank(linkfield)){ + reference = mapList.get(j-1).get(linkfield)!=null?mapList.get(j-1).get(linkfield).toString():null; + } + + if (reference == null){ + continue; + } + + List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); + if (referenceMap.isEmpty()){ + QueryWrapper maxIndex = new QueryWrapper<>(); + maxIndex.select("IFNULL(max(data_index),0) as data_index"); + maxIndex.ne("name", api); + Map map = dataObjectService.getMap(maxIndex); + //如果必填lookup字段没有值，跳过 + update.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1); + dataObjectService.updateById(update); + return; + }else{ + account.setField(dataField.getField(), referenceMap.get(0).get("new_id")); + continue; + } + } + if ("picklist".equals(dataField.getSfType())){ + List> pickList = customMapper.list("value", "data_picklist", "api = '"+api+"' and field = '"+dataField.getField()+"' limit 1"); + account.setField(dataField.getField(), pickList.get(0).get("value")); + continue; + } + account.setField(dataField.getField(), DataUtil.fieldTypeToSf(dataField)); + } + } + + ids[j-1] = mapList.get(j-1).get("Id").toString(); + accounts[j-1] = account; + if (i*200+j == size){ + break; } } - ids[j-1] = data.get(j-1).get("Id").toString(); - accounts[j-1] = account; - if (i*200+j == count){ - break; + if (infoFlag != null && "1".equals(infoFlag.get("value"))){ + printAccountsDetails(accounts, list); } - } - - if (infoFlag != null && "1".equals(infoFlag.get("value"))){ - printAccountsDetails(accounts, list); - } - SaveResult[] saveResults = partnerConnection.create(accounts); - log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults)); - 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]); - }else{ - log.error("-------------saveResults: {}", JSON.toJSONString(saveResult)); + SaveResult[] saveResults = partnerConnection.create(accounts); + log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults)); + 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]); + }else{ + log.error("-------------saveResults: {}", JSON.toJSONString(saveResult)); + } + index++; } - index++; + TimeUnit.MILLISECONDS.sleep(1); + } catch (Exception e) { + log.info(JSON.toJSONString(e)); } - TimeUnit.MILLISECONDS.sleep(1); - } catch (Exception e) { - log.info(JSON.toJSONString(e)); } } + SalesforceParam countParam = new SalesforceParam(); countParam.setApi(api); countParam.setBeginCreateDate(beginDate); @@ -643,7 +661,7 @@ public class DataImportServiceImpl implements DataImportService { } int targetCount = 0; - //批量插入200一次 + int page = count%200 == 0 ? count/200 : (count/200) + 1; for (int i = 0; i < page; i++) { List> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 200 + ",200"); @@ -816,7 +834,7 @@ public class DataImportServiceImpl implements DataImportService { int targetCount = 0; try { - //批量插入200一次 + int page = count % 200 == 0 ? count / 200 : (count / 200) + 1; for (int i = 0; i < page; i++) { TimeUnit.MILLISECONDS.sleep(1); diff --git a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java index 2578770..7db3900 100644 --- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java @@ -93,7 +93,7 @@ public class FileServiceImpl implements FileService { extraSqlTmp += "AND Id > '" + maxId + "'"; } // 获取未存储的附件id - List> list = customMapper.list("Id, url", api, extraSqlTmp + " AND is_dump = 1 order by id asc limit 10"); + List> list = customMapper.list("Id, url", api, extraSqlTmp + " AND is_dump = 1 order by id asc limit 200"); if (CollectionUtils.isEmpty(list)) { break; } @@ -112,7 +112,7 @@ public class FileServiceImpl implements FileService { List> maps = Lists.newArrayList(); Map paramMap = Maps.newHashMap(); paramMap.put("key", "is_dump"); - paramMap.put("value", false); + paramMap.put("value", 0); maps.add(paramMap); customMapper.updateById(api, maps, id); } @@ -124,7 +124,7 @@ public class FileServiceImpl implements FileService { successNum += list.size(); } log.info("dump file verify success, api:{}, success num:{}, retry dump num:{}", api, successNum, num); - XxlJobLogger.log("dump file verify success, api:{}, success num:{}, retry dump num:{}", api, successNum, num); + EmailUtil.send("DumpFile Verify Success", "api: " + api + ", success num: " + successNum + ", retry dump num: " + num); // 存在需重下的文件 执行一遍 if (num > 0) { dumpFile(api, field, true); @@ -694,9 +694,17 @@ public class FileServiceImpl implements FileService { break; } - Map lMap = customMapper.getById("new_id",parentType, parentId); + Map lMap = customMapper.getById("new_id","User", String.valueOf(map.get("OwnerId"))); if(lMap == null) { - String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！"; + String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + map.get("OwnerId") + "数据不存在！！！！"; + log.info(format); + EmailUtil.send("DataDump ERROR", format); + break; + } + + Map uMap = customMapper.getById("new_id",parentType, parentId); + if(uMap == null) { + String format = "文件ID：" + id + ",对应用户ID：" + parentId + "数据不存在！！！！"; log.info(format); EmailUtil.send("DataDump ERROR", format); break; @@ -711,6 +719,10 @@ public class FileServiceImpl implements FileService { JSONObject credentialsJsonParam = new JSONObject(); credentialsJsonParam.put("parentId", lMap.get("new_id")); credentialsJsonParam.put("Name", fileName); + credentialsJsonParam.put("Description", map.get("Description")); + credentialsJsonParam.put("OwnerId", uMap.get("new_id")); + credentialsJsonParam.put("CreatedDate", map.get("CreatedDate")); + credentialsJsonParam.put("CreatedById", map.get("CreatedById")); MultipartEntityBuilder builder = MultipartEntityBuilder.create(); builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON); @@ -731,6 +743,7 @@ public class FileServiceImpl implements FileService { maps.add(paramMap); } } else { + log.info("文件ID：" + id + "上传失败，返回信息：" + response); throw new RuntimeException(); } } diff --git a/src/main/java/com/celnet/datadump/util/DataUtil.java b/src/main/java/com/celnet/datadump/util/DataUtil.java index 2db2ea8..52801d5 100644 --- a/src/main/java/com/celnet/datadump/util/DataUtil.java +++ b/src/main/java/com/celnet/datadump/util/DataUtil.java @@ -127,6 +127,21 @@ public class DataUtil { return Arrays.asList(ids.replaceAll(StringUtils.SPACE, StringUtils.EMPTY).split(",")); } + + public static Set toIdSet(String ids) { + if (StringUtils.isBlank(ids)) { + return Collections.emptySet(); + } + Set set = new HashSet<>(); + String[] split = ids.replaceAll(StringUtils.SPACE, StringUtils.EMPTY).split(","); + for (String id : split) { + if (StringUtils.isNotBlank(id)) { + set.add(id); + } + } + return set; + } + private static final int MINUTE = 60; private static final int HOUR = 24; diff --git a/src/main/resources/mapper/CustomMapper.xml b/src/main/resources/mapper/CustomMapper.xml index 70468de..d05fb75 100644 --- a/src/main/resources/mapper/CustomMapper.xml +++ b/src/main/resources/mapper/CustomMapper.xml @@ -70,6 +70,16 @@ + + UPDATE `${tableName}` + SET new_id = #{newId} + + + ${map.key} = #{map.value} + + + + insert into `${tableName}` @@ -174,6 +184,18 @@ + + + +