From 389bfea6bc58698cd4dfb9b94d43a713adb3f187 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Fri, 22 Aug 2025 16:35:20 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=2020250822=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../celnet/datadump/job/DataDumpNewJob.java | 19 -- .../celnet/datadump/job/DataVerifyJob.java | 50 +++ .../datadump/param/DataVerifyParam.java | 6 + .../datadump/param/SalesforceParam.java | 3 + .../service/DataImportNewService.java | 1 - .../datadump/service/DataVerifyService.java | 5 + .../impl/DataImportBatchServiceImpl.java | 31 +- .../impl/DataImportNewServiceImpl.java | 284 +++++++++++------- .../service/impl/DataImportServiceImpl.java | 8 +- .../service/impl/DataVerifyServiceImpl.java | 273 ++++++++++++++++- .../service/impl/FileServiceImpl.java | 37 ++- src/main/resources/mapper/CustomMapper.xml | 3 + 12 files changed, 559 insertions(+), 161 deletions(-) diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index b86092c..854bac5 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -380,25 +380,6 @@ public class DataDumpNewJob { return dataImportNewService.checkDeletedData(param); } - /** - * 数据统计核对 - * @param paramStr 参数json - * @return result - */ - @XxlJob("verifyDataJob") - public ReturnT verifyDataJob(String paramStr) throws Exception { - log.info("verifyDataJob execute start .................."); - SalesforceParam param = new SalesforceParam(); - try { - if (StringUtils.isNotBlank(paramStr)) { - param = JSON.parseObject(paramStr, SalesforceParam.class); - } - } catch (Throwable throwable) { - return new ReturnT<>(500, "参数解析失败！"); - } - - return dataImportNewService.verifyDataJob(param); - } /** * ContentFolderMember 拉取 diff --git a/src/main/java/com/celnet/datadump/job/DataVerifyJob.java b/src/main/java/com/celnet/datadump/job/DataVerifyJob.java index a358ce7..63e93ad 100644 --- a/src/main/java/com/celnet/datadump/job/DataVerifyJob.java +++ b/src/main/java/com/celnet/datadump/job/DataVerifyJob.java @@ -47,4 +47,54 @@ public class DataVerifyJob { return dataVerifyService.verify(param); } + /** + * 校验数据总量 + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataVerifyCountJob") + public ReturnT dataVerifyCountJob(String paramStr) throws Exception { + log.info("dataVerifyCountJob execute start .................."); + DataVerifyParam param = new DataVerifyParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, DataVerifyParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + if (param.getType() == null) { + return new ReturnT<>(500, "参数缺失！"); + } + + return dataVerifyService.verifyCount(param); + } + + /** + * 校验数据质量 + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataVerifyQualityJob") + public ReturnT dataVerifyQualityJob(String paramStr) throws Exception { + log.info("dataVerifyQualityJob execute start .................."); + DataVerifyParam param = new DataVerifyParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, DataVerifyParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + if (param.getType() == null) { + return new ReturnT<>(500, "参数缺失！"); + } + + // 假设这里调用一个专门用于数据质量校验的服务方法 + // 由于当前dataVerifyService中没有直接对应的方法，需要根据实际业务逻辑进行实现 + return dataVerifyService.VerifyQuality(param); + } + } diff --git a/src/main/java/com/celnet/datadump/param/DataVerifyParam.java b/src/main/java/com/celnet/datadump/param/DataVerifyParam.java index 066cb73..7beba0d 100644 --- a/src/main/java/com/celnet/datadump/param/DataVerifyParam.java +++ b/src/main/java/com/celnet/datadump/param/DataVerifyParam.java @@ -4,6 +4,8 @@ import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import lombok.Data; +import java.util.Date; + /** * @author Radian * @description 数据校验参数 @@ -47,4 +49,8 @@ public class DataVerifyParam { @ApiModelProperty(hidden = true) private Integer type = 1; + + @ApiModelProperty(value = "开始时间", hidden = true) + private Date beginDate; + } diff --git a/src/main/java/com/celnet/datadump/param/SalesforceParam.java b/src/main/java/com/celnet/datadump/param/SalesforceParam.java index dfa4b45..58ba701 100644 --- a/src/main/java/com/celnet/datadump/param/SalesforceParam.java +++ b/src/main/java/com/celnet/datadump/param/SalesforceParam.java @@ -89,6 +89,9 @@ public class SalesforceParam extends DataDumpParam implements Cloneable { @ApiModelProperty(value = "是否单线程", hidden = true) private Boolean isSingleThread = false; + + @ApiModelProperty(value = "是否有new_id", hidden = true) + private Boolean hasNewId = false; @Override public SalesforceParam clone() { try { diff --git a/src/main/java/com/celnet/datadump/service/DataImportNewService.java b/src/main/java/com/celnet/datadump/service/DataImportNewService.java index c9abf3c..52050e5 100644 --- a/src/main/java/com/celnet/datadump/service/DataImportNewService.java +++ b/src/main/java/com/celnet/datadump/service/DataImportNewService.java @@ -22,7 +22,6 @@ public interface DataImportNewService { ReturnT checkDeletedData(SalesforceParam param) throws Exception; - ReturnT verifyDataJob(SalesforceParam param) throws Exception; ReturnT insertSingle(SalesforceParam param) throws Exception; diff --git a/src/main/java/com/celnet/datadump/service/DataVerifyService.java b/src/main/java/com/celnet/datadump/service/DataVerifyService.java index dcad9b3..e7e8abc 100644 --- a/src/main/java/com/celnet/datadump/service/DataVerifyService.java +++ b/src/main/java/com/celnet/datadump/service/DataVerifyService.java @@ -16,4 +16,9 @@ public interface DataVerifyService { * @return ReturnT */ ReturnT verify(DataVerifyParam param) throws Exception; + + ReturnT verifyCount(DataVerifyParam param) throws Exception; + + ReturnT VerifyQuality(DataVerifyParam param) throws Exception; + } 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 a4daef4..c2462e2 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -372,9 +372,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { String reference = dataField.getReferenceTo(); //引用类型字段 String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield)){ - reference = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null; + if (data.get(j-1).get(linkfield)!=null){ + reference = data.get(j-1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！"); + } } if (reference == null){ continue; @@ -799,8 +802,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } //引用类型字段 String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){ - reference_to = map.get(linkfield).toString(); + if (StringUtils.isNotBlank(linkfield)){ + if (map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; @@ -1186,8 +1193,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } //引用类型字段 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 (StringUtils.isNotBlank(linkfield)){ + if (data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; @@ -1325,8 +1336,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } //引用类型字段 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 (StringUtils.isNotBlank(linkfield)){ + if (data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; 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 5c57078..f697896 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -805,6 +805,8 @@ public class DataImportNewServiceImpl implements DataImportNewService { int j = 0; try { for (Map map : mapList) { + //需要置空的参数 + List fieldsToNull = new ArrayList<>(); SObject account = new SObject(); account.setType(api); //给对象赋值 @@ -827,8 +829,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { } //引用类型字段 String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){ - reference_to = map.get(linkfield).toString(); + if (StringUtils.isNotBlank(linkfield)){ + if (map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; @@ -844,13 +850,21 @@ public class DataImportNewServiceImpl implements DataImportNewService { } } } 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(StringUtils.isNotBlank(dataField.getSfType())){ + if (map.get(field) != null){ + account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value)); + }else { + fieldsToNull.add(field); + } } } } + + if (!fieldsToNull.isEmpty()){ + String[] fieldsToNullArray = fieldsToNull.toArray(new String[0]); + account.setField("fieldsToNull", fieldsToNullArray); + } + if (dataObject.getIsEditable()){ account.setField("old_owner_id__c", map.get("OwnerId")); account.setField("old_sfdc_id__c", map.get("Id")); @@ -878,7 +892,11 @@ public class DataImportNewServiceImpl implements DataImportNewService { dataLog1.setEndTime(new Date()); dataLogService.save(dataLog1); - } catch (Throwable e) { + TimeUnit.MILLISECONDS.sleep(1); + + }catch (InterruptedException interruptedException){ + return; + }catch (Throwable e) { log.info(JSON.toJSONString(e)); } } @@ -1013,8 +1031,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { } //引用类型字段 String linkfield = fieldMap.get(dataField.getField()); - if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){ - reference_to = map.get(linkfield).toString(); + if (StringUtils.isNotBlank(linkfield)){ + if (map.get(linkfield)!=null){ + reference_to = map.get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; @@ -1628,7 +1650,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { log.info("总文件数 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); - List> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000"); + List> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_Type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000"); for (Map map : list) { int failCount = 0; @@ -1641,7 +1663,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { id = (String) map.get("Id"); String url_fileName = (String) map.get("url"); String parentId = (String) map.get("ParentId"); - String parentType = (String) map.get("Parent_type"); + String parentType = (String) map.get("Parent_Type"); // 判断路径是否为空 if (StringUtils.isNotBlank(url_fileName)) { String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName; @@ -2237,7 +2259,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { if ((param.getEndCreateDate() == null || param.getBeginCreateDate() == null) && param.getIds().isEmpty()){ log.info("手动删除数据必须指定参数：开始与结束时间/Ids !!!!"); - return ReturnT.SUCCESS; + return ReturnT.FAIL; } String sql = ""; @@ -2256,108 +2278,136 @@ public class DataImportNewServiceImpl implements DataImportNewService { int page = count%200 == 0 ? count/200 : (count/200) + 1; log.info("删除数据，总批次：{}",page); + try{ + for (int i = 0; i < page; i++) { - for (int i = 0; i < page; i++) { + List> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200"); - List> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200"); + log.info("删除数据，当前批次：{}，批次数：{}", i,list.size()); - log.info("删除数据，当前批次：{}，批次数：{}", i,list.size()); + Map idNameMap = new HashMap<>(); + String[] ids = new String[list.size()]; - Map idNameMap = new HashMap<>(); - String[] ids = new String[list.size()]; - for (int z = 0; z < list.size(); z++) { - Map objectMap = customMapper.getById("*", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString()); - if (objectMap != null && objectMap.get("new_id") != null) { - ids[z] = objectMap.get("new_id").toString(); + for (int z = 0; z < list.size(); z++) { + Map objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString()); + if (objectMap != null && objectMap.get("new_id") != null) { + ids[z] = objectMap.get("new_id").toString(); + idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString()); + }else { + log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName")); + } } - idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString()); - } - int index = 0; - DeleteResult[] deleteResults = connect.delete(ids); - for (DeleteResult deleteResult : deleteResults) { - if (deleteResult.isSuccess()){ - String typeName = idNameMap.get(ids[index]); - Map objectMap = customMapper.getById("*", typeName, ids[index]); + int index = 0; + DeleteResult[] deleteResults = connect.delete(ids); + for (DeleteResult deleteResult : deleteResults) { + if (deleteResult.isSuccess()){ + String typeName = idNameMap.get(ids[index]); + Map objectMap = customMapper.getById("*", typeName, ids[index]); - List> maps = new ArrayList<>(); + List> maps = new ArrayList<>(); - Map m1 = new HashMap<>(); - m1.put("key", "is_delete"); - m1.put("value", 1); + Map m1 = new HashMap<>(); + m1.put("key", "is_delete"); + m1.put("value", 1); - Map m2 = new HashMap<>(); - m2.put("key", "all_data"); - m2.put("value", JSON.toJSONString(objectMap)); + Map m2 = new HashMap<>(); + m2.put("key", "all_data"); + m2.put("value", JSON.toJSONString(objectMap)); - maps.add(m1); - maps.add(m2); - customMapper.updateById(api, maps, ids[index]); + maps.add(m1); + maps.add(m2); + customMapper.updateById(api, maps, ids[index]); - }else { - log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors()); + customMapper.deleteOne(typeName, ids[index]); + + }else { + log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors()); + } + index++; } - index++; } + TimeUnit.MILLISECONDS.sleep(1); + }catch (InterruptedException e){ + return ReturnT.FAIL; + } catch (Exception e) { + log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage()); } - return ReturnT.SUCCESS; } + private ReturnT autoCheckDeletedData(SalesforceParam param) throws Exception { + String api = "DeleteEvent"; + PartnerConnection connect = salesforceTargetConnect.createConnect(); - return ReturnT.SUCCESS; - } + String sql = "SystemModstamp > " + DateUtil.format(DateUtils.addDays(new Date(), -15), "yyyy-MM-dd HH:mm:ss"); + Integer count = customMapper.countBySQL(api, sql); - @Override - public ReturnT verifyDataJob(SalesforceParam param) throws Exception { + int page = count%200 == 0 ? count/200 : (count/200) + 1; - QueryWrapper qw = getDataBatchQueryWrapper(param); + log.info("删除数据，总批次：{}",page); + try{ + for (int i = 0; i < page; i++) { - Map> maps = Maps.newHashMap(); + List> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200"); - List dataBatches = dataBatchService.list(qw); + log.info("删除数据，当前批次：{}，批次数：{}", i,list.size()); - String filePath = TEMP_FILE_PATH + "数据校验" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx"; + Map idNameMap = new HashMap<>(); + String[] ids = new String[list.size()]; - try { - ExcelWriter excelWriter = EasyExcel.write(filePath, DataBatch.class).build(); - - Map sheetMap = Maps.newHashMap(); - - dataBatches.forEach(t -> { - List list = maps.getOrDefault(t.getName(), Lists.newArrayList()); - // 日期处理 减1s作为真实范围 - t.setSyncEndDate(org.apache.commons.lang3.time.DateUtils.addSeconds(t.getSyncEndDate(), -1)); - list.add(t); - maps.put(t.getName(), list); - }); - - maps.forEach((k, v) -> { - if (CollectionUtils.isNotEmpty(v)) { - // 如果同一个sheet只要创建一次 - String label = v.get(0).getLabel(); - // excel 不允许sheet名存在[\/?*:]等特殊字符 - label = label.replaceAll("[\\\\/?*:]", " "); - WriteSheet writeSheet = sheetMap.getOrDefault(k, EasyExcel.writerSheet(label).build()); - excelWriter.write(v, writeSheet); - sheetMap.put(k, writeSheet); + for (int z = 0; z < list.size(); z++) { + Map objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString()); + if (objectMap != null && objectMap.get("new_id") != null) { + ids[z] = objectMap.get("new_id").toString(); + idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString()); + }else { + log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName")); + } } - }); - }catch (Exception e){ - log.error(JSON.toJSONString(e)); + + int index = 0; + DeleteResult[] deleteResults = connect.delete(ids); + for (DeleteResult deleteResult : deleteResults) { + if (deleteResult.isSuccess()){ + String typeName = idNameMap.get(ids[index]); + Map objectMap = customMapper.getById("*", typeName, ids[index]); + + List> maps = new ArrayList<>(); + + Map m1 = new HashMap<>(); + m1.put("key", "is_delete"); + m1.put("value", 1); + + Map m2 = new HashMap<>(); + m2.put("key", "all_data"); + m2.put("value", JSON.toJSONString(objectMap)); + + maps.add(m1); + maps.add(m2); + customMapper.updateById(api, maps, ids[index]); + + customMapper.deleteOne(typeName, ids[index]); + + }else { + log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors()); + } + index++; + } + } + TimeUnit.MILLISECONDS.sleep(1); + }catch (InterruptedException e){ + return ReturnT.FAIL; + } catch (Exception e) { + log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage()); } - - String head = "数据校验" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss"); - - EmailUtil.sendMineWithFile(head, null, false, Lists.newArrayList(filePath)); - - return ReturnT.SUCCESS; } + /** * 参数转换qw * @@ -2542,7 +2592,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { continue; } if (("EventRelation".equals(api) || "TaskRelation".equals(api)) && "RelationId".equals(key)) { - Map m = customMapper.getByNewId("Id", jsonObject.getString("Relation_type"), jsonObject.getString(key)); + Map m = customMapper.getByNewId("Id", jsonObject.getString("Relation_Type"), jsonObject.getString(key)); if (m == null){ break; } @@ -2564,7 +2614,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { continue; } if ("CollaborationGroupMember".equals(api) && "MemberId".equals(key)) { - Map m = customMapper.getByNewId("Id", jsonObject.getString("Member_type"), jsonObject.getString(key)); + Map m = customMapper.getByNewId("Id", jsonObject.getString("Member_Type"), jsonObject.getString(key)); if (m == null){ break; } @@ -2575,7 +2625,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { continue; } if ("FeedAttachment".equals(api) && "FeedEntityId".equals(key)) { - Map m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_type"), jsonObject.getString(key)); + Map m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key)); if (m == null){ break; } @@ -2586,7 +2636,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { continue; } if ("FeedAttachment".equals(api) && "RecordId".equals(key)) { - Map m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_type"), jsonObject.getString(key)); + Map m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_Type"), jsonObject.getString(key)); if (m == null){ break; } @@ -2697,15 +2747,19 @@ public class DataImportNewServiceImpl implements DataImportNewService { // 手动任务优先执行 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); + if (salesforceParam.getIsSingleThread()){ + insertSingleExecute(salesforceParam, partnerConnection,update); + }else { + Future future = salesforceExecutor.execute(() -> { + try { + insertSingleExecute(salesforceParam, partnerConnection,update); + } catch (Throwable throwable) { + log.error("salesforceExecutor error", throwable); + throw new RuntimeException(throwable); + } + }, salesforceParam.getBatch(), 1); + futures.add(future); + } } // 等待当前所有线程执行完成 salesforceExecutor.waitForFutures(futures.toArray(new Future[]{})); @@ -2787,15 +2841,19 @@ public class DataImportNewServiceImpl implements DataImportNewService { // 手动任务优先执行 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); + if (salesforceParam.getIsSingleThread()){ + insertSingleExecute(salesforceParam, partnerConnection,update); + }else { + Future future = salesforceExecutor.execute(() -> { + try { + insertSingleExecute(salesforceParam, partnerConnection,update); + } catch (Throwable throwable) { + log.error("salesforceExecutor error", throwable); + throw new RuntimeException(throwable); + } + }, salesforceParam.getBatch(), 0); + futures.add(future); + } } // 等待当前所有线程执行完成 salesforceExecutor.waitForFutures(futures.toArray(new Future[]{})); @@ -2866,7 +2924,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { //查询当前对象多态字段映射 QueryWrapper queryWrapper = new QueryWrapper<>(); - queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); + queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",1); List configs = linkConfigService.list(queryWrapper); Map fieldMap = new HashMap<>(); if (!configs.isEmpty()) { @@ -2931,8 +2989,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { } //引用类型字段 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 (StringUtils.isNotBlank(linkfield) ){ + if (data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; @@ -3143,8 +3205,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { } //引用类型字段 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 (StringUtils.isNotBlank(linkfield)){ + if (data.get(j-1).get(linkfield)!=null){ + reference_to = data.get(j-1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！"); + } } if (reference_to == null){ continue; 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 b83f169..cace7be 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java @@ -360,11 +360,13 @@ public class DataImportServiceImpl implements DataImportService { 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 (mapList.get(j - 1).get(linkfield)!=null){ + reference = mapList.get(j - 1).get(linkfield).toString(); + }else { + log.info("对象类型:" + api + "的数据:"+ mapList.get(j - 1).get("Id") +"的关联类型不存在！"); + } } - if (reference == null){ continue; } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java index c7c966c..83fb863 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java @@ -8,33 +8,32 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; 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; import com.celnet.datadump.entity.DataBatch; +import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.param.DataVerifyParam; import com.celnet.datadump.param.SalesforceParam; import com.celnet.datadump.mapper.CustomMapper; -import com.celnet.datadump.service.CommonService; -import com.celnet.datadump.service.DataBatchService; -import com.celnet.datadump.service.DataFieldService; -import com.celnet.datadump.service.DataVerifyService; +import com.celnet.datadump.service.*; import com.celnet.datadump.util.DataUtil; import com.celnet.datadump.util.EmailUtil; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.sforce.soap.partner.PartnerConnection; +import com.sforce.soap.partner.QueryResult; +import com.sforce.soap.partner.sobject.SObject; import com.xxl.job.core.biz.model.ReturnT; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.time.DateFormatUtils; import org.apache.commons.lang3.time.DateUtils; +import org.checkerframework.checker.units.qual.A; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.File; -import java.util.Calendar; -import java.util.Date; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -60,6 +59,10 @@ public class DataVerifyServiceImpl implements DataVerifyService { private SalesforceExecutor salesforceExecutor; @Autowired private SalesforceConnect salesforceConnect; + @Autowired + private SalesforceTargetConnect salesforceTargetConnect; + @Autowired + private DataObjectService dataObjectService; private static final Integer MAX_FAIL_COUNT = 3; @@ -111,6 +114,260 @@ public class DataVerifyServiceImpl implements DataVerifyService { } } + @Override + public ReturnT verifyCount(DataVerifyParam param) throws Exception { + QueryWrapper wrapper = new QueryWrapper<>(); + wrapper.eq("data_work",1); + if (StringUtils.isNotBlank(param.getApi())){ + wrapper.in("name",DataUtil.toIdList(param.getApi())); + } + List list = dataObjectService.list(wrapper); + + if (list.isEmpty()){ + log.info("没有对象存在！不进行数据量统计"); + return ReturnT.FAIL; + } + PartnerConnection connect = salesforceConnect.createConnect(); + PartnerConnection targetConnect = salesforceTargetConnect.createConnect(); + String filePath = TEMP_FILE_PATH + "数据量统计_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx"; + if (param.getBeginDate() != null) { + filePath = TEMP_FILE_PATH + "数据量统计_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + "_统计开始时间：" + DateFormatUtils.format(param.getBeginDate(), "yyyyMMdd") + ".xlsx"; + } + + try (ExcelWriter excelWriter = EasyExcel.write(filePath).build()) { + WriteSheet writeSheet = EasyExcel.writerSheet("数据量统计").build(); + // 写入表头 + List> head = Lists.newArrayList(); + head.add(Lists.newArrayList("对象API","对象Label","源ORG数据量","本地数据量","本地存在new_id数据量","目标ORG数据量","目标存在old_id数据量","HasDeleted","本地重复new_id","目标重复old_id")); + excelWriter.write(head, writeSheet); + + WriteSheet writeSheet1 = EasyExcel.writerSheet("本地new_id重复").build(); + // 写入表头 + List> head1 = Lists.newArrayList(); + head1.add(Lists.newArrayList("对象API", "Id", "new_id")); + excelWriter.write(head1, writeSheet1); + + WriteSheet writeSheet2 = EasyExcel.writerSheet("目标old_id重复").build(); + // 写入表头 + List> head2 = Lists.newArrayList(); + head2.add(Lists.newArrayList("对象API", "Id", "old_sfdc_id__c")); + excelWriter.write(head2, writeSheet2); + + // 写入数据 + for (DataObject dataObject : list) { + String api = dataObject.getName(); + String extraSql = ""; + SalesforceParam salesforceParam = new SalesforceParam(); + salesforceParam.setApi(api); + Boolean hasDeleted = dataFieldService.hasDeleted(api); + if (hasDeleted) { + salesforceParam.setIsDeleted(false); + extraSql = " WHERE IsDeleted = false "; + } + + String sql = "SELECT Count(Id) num FROM " + api + extraSql; + + String oldIdSql = sql.contains("WHERE") ? sql + " AND old_sfdc_id__c != null":sql + " WHERE old_sfdc_id__c != null"; + + String duplicateIdSql = "SELECT old_sfdc_id__c FROM " + api + " WHERE old_sfdc_id__c != NULL GROUP BY old_sfdc_id__c HAVING COUNT(old_sfdc_id__c) > 1"; + + if (param.getBeginDate() != null){ + String dateField = dataFieldService.returnDateField(api); + sql = sql.contains("WHERE") ? sql + " AND "+dateField + " >= '" + DateFormatUtils.format(param.getBeginDate(), "yyyy-MM-dd HH:mm:ss") + "'":sql + " WHERE "+dateField + " >= '" + DateFormatUtils.format(param.getBeginDate(), "yyyy-MM-dd HH:mm:ss") + "'"; + } + + QueryResult queryResult = null; + Integer sourceNum = 0; + int retryCount = 0; + while (retryCount < 3) { + try { + // 查询源ORG数据量 + queryResult = connect.queryAll(sql); + SObject record = queryResult.getRecords()[0]; + sourceNum = (Integer) record.getField("num"); + log.info("api :{} 源ORG num: {}，执行sql：{}", api, sourceNum, sql); + break; + } catch (Exception e) { + retryCount++; + log.info("查询源ORG数据量失败，第{}次重试，执行sql：{}", retryCount, sql, e); + if (retryCount >= 3) { + sourceNum = -1; + } + } + } + Integer localNum = -1; + Integer localNewIdNum = -1; + + List> mapList = new ArrayList<>(); + + if (StringUtils.isNotBlank(customMapper.checkTable(api))){ + // 查询本地数据量 + localNum = customMapper.count(salesforceParam); + + log.info("api :{} 本地 num: {}", api, localNum); + + // 查询本地存在new_id数据量 + salesforceParam.setHasNewId(true); + localNewIdNum = customMapper.count(salesforceParam); + + log.info("api :{} 本地存在new_id num: {}", api, localNewIdNum); + + // 查询本地重复new_id的数量 + String newReIdSql = "new_id IN ( SELECT new_id FROM `" + api + "` WHERE new_id IS NOT NULL GROUP BY new_id HAVING COUNT(new_id) > 1 )"; + mapList = customMapper.list("Id,new_id", api, newReIdSql); + log.info("api :{} 本地重复new_id num: {}", api, mapList.size()); + + if (!mapList.isEmpty()){ + List> duplicateData = Lists.newArrayList(); + for (Map map : mapList) { + List row = Lists.newArrayList(); + row.add(api); + row.add((String) map.get("Id")); + row.add((String) map.get("new_id")); + duplicateData.add(row); + } + excelWriter.write(duplicateData, writeSheet1); + } + } + + // 查询目标ORG数据量 + QueryResult targetQueryResult = null; + Integer targetNum = 0; + int targetRetryCount = 0; + while (targetRetryCount < 3) { + try { + targetQueryResult = targetConnect.queryAll(sql); + SObject targetRecord = targetQueryResult.getRecords()[0]; + targetNum = (Integer) targetRecord.getField("num"); + log.info("api :{} 目标ORG num: {}，执行sql：{}", api,sql, targetNum); + break; + } catch (Exception e) { + targetRetryCount++; + log.info("查询目标ORG数据量失败，第{}次重试，执行sql：{}", targetRetryCount,sql, e); + if (targetRetryCount >= 3) { + targetNum = -1; + } + } + } + + QueryResult targetQueryOldIdResult = null; + int oldIdRetryCount = 0; + + Integer targetOldIdNum = 0; + if (dataObject.getIsEditable()){ + while (oldIdRetryCount < 3) { + try { + targetQueryOldIdResult = targetConnect.queryAll(oldIdSql); + SObject targetOldIdRecord = targetQueryOldIdResult.getRecords()[0]; + targetOldIdNum = (Integer) targetOldIdRecord.getField("num"); + log.info("api :{} 目标ORG存在old_id数据量: {}，执行sql：{}", api, targetOldIdNum, oldIdSql); + break; + } catch (Exception e) { + oldIdRetryCount++; + log.info("查询目标ORG存在old_id数据量失败，第{}次重试，执行sql：{}", oldIdRetryCount, oldIdSql, e); + if (oldIdRetryCount >= 3) { + targetOldIdNum = -1; + } + } + } + }else { + targetOldIdNum = -1; + } + + int oldIdRepuRetryCount = 0; + List> duplicateOldData = Lists.newArrayList(); + if (dataObject.getIsEditable()){ + while (oldIdRepuRetryCount < 3) { + try { + List oldIds = Lists.newArrayList(); + QueryResult query = targetConnect.queryAll(duplicateIdSql); + if (query != null && query.getRecords() != null && query.getRecords().length > 0) { + for (SObject record : query.getRecords()) { + oldIds = Lists.newArrayList(); + oldIds.add((String) record.getField("old_sfdc_id__c")); + } + if (oldIds.isEmpty()) { + break; + } + String idStr = "("; + for (String id : oldIds) { + idStr += "'" + id + "',"; // 拼接每个ID + } + if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束 + idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号 + } + idStr += ")"; + + String duplicateOldIdSql = "SELECT Id,old_sfdc_id__c FROM " + api + " WHERE old_sfdc_id__c IN " + idStr; + QueryResult queryResult1 = targetConnect.queryAll(duplicateOldIdSql); + if (queryResult1 != null && queryResult1.getRecords() != null && queryResult1.getRecords().length > 0) { + for (SObject record : queryResult1.getRecords()) { + List row = Lists.newArrayList(); + row.add(api); + row.add((String) record.getField("Id")); + row.add((String) record.getField("old_sfdc_id__c")); + duplicateOldData.add(row); + } + excelWriter.write(duplicateOldData, writeSheet2); + log.info("api :{} 查询目标ORG存在old_id重复数据，数据量: {}，执行sql：{}", api, queryResult1.getRecords().length, duplicateOldIdSql); + } + } + break; + } catch (Exception e) { + oldIdRepuRetryCount++; + log.info("查询目标ORG存在old_id重复数据失败，第{}次重试，执行sql：{}", oldIdRepuRetryCount, duplicateIdSql, e); + if (oldIdRepuRetryCount >= 3) { + targetOldIdNum = -1; + } + } + } + }else { + targetOldIdNum = -1; + } + + // 写入一行数据 + List> data = Lists.newArrayList(); + List rowData = Lists.newArrayList(); + rowData.add(api); + rowData.add(dataObject.getLabel()); + rowData.add(sourceNum == -1 ? "错误" : String.valueOf(sourceNum)); + rowData.add(localNum == -1 ? "错误" : String.valueOf(localNum)); + rowData.add(localNewIdNum == -1 ? "错误" : String.valueOf(localNewIdNum)); + rowData.add(targetNum == -1 ? "错误" : String.valueOf(targetNum)); + rowData.add(targetOldIdNum == -1 ? "错误" : String.valueOf(targetOldIdNum)); + rowData.add(String.valueOf(hasDeleted)); + rowData.add(mapList.isEmpty() ? "无" : String.valueOf(mapList.size())); + rowData.add(duplicateOldData.isEmpty() ? "无" : String.valueOf(duplicateOldData.size())); + data.add(rowData); + excelWriter.write(data, writeSheet); + } + TimeUnit.MILLISECONDS.sleep(1); + }catch (InterruptedException e){ + return ReturnT.FAIL; + }catch (Exception e){ + log.error("数据量统计失败，错误信息：{}", e.getMessage()); + } + + String head = "校验数据总量"; + String text = head + "更新时间：" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"); + + log.info("数据量统计完成，文件路径：{}", filePath); + + EmailUtil.sendMineWithFile(head, text, false, Lists.newArrayList(filePath)); + + return ReturnT.SUCCESS; + } + + @Override + public ReturnT VerifyQuality(DataVerifyParam param) throws Exception { + + + + + return null; + } + + /** * 发送邮件 * 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 7db3900..2bd3e02 100644 --- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java @@ -654,7 +654,7 @@ public class FileServiceImpl implements FileService { } try { // 获取未存储的附件id - List> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_type", api, "is_upload = 0"); + List> list = customMapper.list("*", api, "is_upload = 0"); for (Map map : list) { String finalUploadUrl = uploadUrl; Future future = salesforceExecutor.execute(() -> { @@ -670,7 +670,13 @@ public class FileServiceImpl implements FileService { id = (String) map.get("Id"); String url_fileName = (String) map.get("url"); String parentId = (String) map.get("ParentId"); - String parentType = (String) map.get("Parent_type"); + String parentType = (String) map.get("Parent_Type"); + String ownerId = (String) map.get("OwnerId"); + String description = (String) map.get("Description"); + String createdById = (String) map.get("CreatedById"); + String fileName = (String) map.get("Name"); + log.info("id: {}, url_fileName: {}, parentId: {}, parentType: {}, ownerId: {}, description: {}, createdById: {}", + id, url_fileName, parentId, parentType, ownerId, description, createdById); // 判断路径是否为空 if (StringUtils.isNotBlank(url_fileName)) { String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName; @@ -680,9 +686,6 @@ public class FileServiceImpl implements FileService { log.info("文件不存在"); break; } - String fileName = (String) map.get("Name"); - - log.info("文件名称：" + fileName); // dataObject查询 QueryWrapper qw = new QueryWrapper<>(); @@ -694,17 +697,25 @@ public class FileServiceImpl implements FileService { break; } - Map lMap = customMapper.getById("new_id","User", String.valueOf(map.get("OwnerId"))); + Map lMap = customMapper.getById("new_id",parentType, String.valueOf(parentId)); if(lMap == null) { - String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + map.get("OwnerId") + "数据不存在！！！！"; + String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！"; log.info(format); EmailUtil.send("DataDump ERROR", format); break; } - Map uMap = customMapper.getById("new_id",parentType, parentId); + Map uMap = customMapper.getById("new_id","User", ownerId); if(uMap == null) { - String format = "文件ID：" + id + ",对应用户ID：" + parentId + "数据不存在！！！！"; + String format = "文件ID：" + id + ",对应用户ID：" + ownerId + "数据不存在！！！！"; + log.info(format); + EmailUtil.send("DataDump ERROR", format); + break; + } + + Map cMap = customMapper.getById("new_id","User", createdById); + if(cMap == null) { + String format = "文件ID：" + id + ",对应用户ID：" + createdById + "数据不存在！！！！"; log.info(format); EmailUtil.send("DataDump ERROR", format); break; @@ -719,10 +730,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("Description", description); credentialsJsonParam.put("OwnerId", uMap.get("new_id")); - credentialsJsonParam.put("CreatedDate", map.get("CreatedDate")); - credentialsJsonParam.put("CreatedById", map.get("CreatedById")); +// credentialsJsonParam.put("CreatedDate", map.get("CreatedDate")); + credentialsJsonParam.put("CreatedById", cMap.get("new_id")); MultipartEntityBuilder builder = MultipartEntityBuilder.create(); builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON); @@ -743,7 +754,7 @@ public class FileServiceImpl implements FileService { maps.add(paramMap); } } else { - log.info("文件ID：" + id + "上传失败，返回信息：" + response); + log.info("文件ID：" + id + "上传失败，返回信息：" + EntityUtils.toString(response.getEntity(), "UTF-8")); throw new RuntimeException(); } } diff --git a/src/main/resources/mapper/CustomMapper.xml b/src/main/resources/mapper/CustomMapper.xml index 20ecc96..c17f2d0 100644 --- a/src/main/resources/mapper/CustomMapper.xml +++ b/src/main/resources/mapper/CustomMapper.xml @@ -132,6 +132,9 @@ AND IsDeleted = #{param.isDeleted} + + AND new_id IS NOT NULL +