From 656ca02cebacbfa3adfa81c99e44b00b623a7f83 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 30 Oct 2025 10:42:10 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../datadump/config/SalesforceConnect.java | 22 +- .../config/SalesforceTargetConnect.java | 18 +- .../service/impl/CommonBatchServiceImpl.java | 145 +++-- .../service/impl/DataVerifyServiceImpl.java | 505 ++++++++++-------- .../service/impl/FileServiceImpl.java | 2 + .../service/impl/OrgConfigServiceImpl.java | 71 ++- 6 files changed, 411 insertions(+), 352 deletions(-) diff --git a/src/main/java/com/celnet/datadump/config/SalesforceConnect.java b/src/main/java/com/celnet/datadump/config/SalesforceConnect.java index e9de93b..99409bd 100644 --- a/src/main/java/com/celnet/datadump/config/SalesforceConnect.java +++ b/src/main/java/com/celnet/datadump/config/SalesforceConnect.java @@ -10,6 +10,7 @@ import com.sforce.soap.partner.PartnerConnection; import com.sforce.ws.ConnectionException; import com.sforce.ws.ConnectorConfig; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @@ -54,16 +55,19 @@ public class SalesforceConnect { if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", String.valueOf(map1.get("value"))); } - if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) { - map.put("seesionId", (String) map1.get("value")); + if ("SOURCE_ORG_TOKEN".equals(map1.get("code"))) { + map.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); } } String username = map.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); - config.setPassword(map.get("password")); - config.setSessionId(map.get("seesionId")); + if (StringUtils.isNotEmpty(map.get("token"))){ + config.setPassword(map.get("password") + map.get("token")); + }else { + config.setPassword(map.get("password")); + } String url = map.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); @@ -110,11 +114,15 @@ public class SalesforceConnect { if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", String.valueOf(map1.get("value"))); } - if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) { - map.put("seesionId", (String) map1.get("value")); + if ("SOURCE_ORG_TOKEN".equals(map1.get("code"))) { + map.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); } } - return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url")); + if (StringUtils.isNotEmpty(map.get("token"))){ + return BulkUtil.getBulkConnection(map.get("username"),map.get("password") + map.get("token"),map.get("url")); + }else { + return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url")); + } } catch (Exception e) { failCount ++; log.error("源ORG连接异常！休眠一分钟再次发起重试~~", e); diff --git a/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java b/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java index 6a21504..6dc8e25 100644 --- a/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java +++ b/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java @@ -10,6 +10,7 @@ import com.sforce.soap.partner.PartnerConnection; import com.sforce.ws.ConnectionException; import com.sforce.ws.ConnectorConfig; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @@ -49,16 +50,19 @@ public class SalesforceTargetConnect { if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", (String) map1.get("value")); } - if ("TARGET_ORG_SEESION".equals(map1.get("code"))) { - map.put("seesionId", (String) map1.get("value")); + if ("TARGET_ORG_TOKEN".equals(map1.get("code"))) { + map.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); } } String username = map.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); - config.setPassword(map.get("password")); - config.setSessionId(map.get("seesionId")); + if (StringUtils.isNotEmpty(map.get("token"))){ + config.setPassword(map.get("password") + map.get("token")); + }else { + config.setPassword(map.get("password")); + } String url = map.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); @@ -109,11 +113,11 @@ public class SalesforceTargetConnect { if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", (String) map1.get("value")); } - if ("TARGET_ORG_SEESION".equals(map1.get("code"))) { - map.put("seesionId", (String) map1.get("value")); + if ("TARGET_ORG_TOKEN".equals(map1.get("code"))) { + map.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); } } - return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url")); + return BulkUtil.getBulkConnection(map.get("username"),map.get("password") + map.get("token"),map.get("url")); } catch (Exception e) { failCount++; log.error("目标ORG连接异常！休眠一分钟再次发起重试~~", e); diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index b58dd27..9f7119f 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -199,7 +199,7 @@ public class CommonBatchServiceImpl implements CommonBatchService { .filter(field -> !"SystemModstamp".equals(field) && !"LastModifiedDate".equals(field)) .collect(Collectors.joining(", ")); - String sql = "select " + fieldStr + " from " + param.getApi() ; + String sql = "select Id__c,OppId__c from " + param.getApi(); log.info("构建查询SQL: {}", sql); job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll); @@ -234,7 +234,7 @@ public class CommonBatchServiceImpl implements CommonBatchService { // 使用确定性命名策略，避免使用随机时间戳 String fileName = Const.SERVER_FILE_PATH + "/" + api+ "/" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss"); - final long maxFileSize = 50 * 1024 * 1024; // 50MB + final long maxFileSize = 20 * 1024 * 1024; // 50MB int fileIndex = 0; Path currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv"); @@ -502,91 +502,70 @@ public class CommonBatchServiceImpl implements CommonBatchService { // 使用Bulk V2 API导入所有文件 for (File csvFile : files) { - // 添加重试机制 - int maxRetries = 3; - int retryCount = 0; - boolean success = false; - - while (retryCount < maxRetries && !success) { - try { - String filePath = csvFile.getAbsolutePath(); - log.info("开始处理CSV文件: {}, 重试次数: {}", filePath, retryCount); - - // 检查CSV文件是否为空或只有一行标题 - try (BufferedReader reader = new BufferedReader(new FileReader(csvFile))) { - String firstLine = reader.readLine(); - if (firstLine == null || firstLine.trim().isEmpty()) { - log.warn("CSV文件为空，跳过处理: {}", filePath); - continue; - } - - String secondLine = reader.readLine(); - if (secondLine == null || secondLine.trim().isEmpty()) { - log.warn("CSV文件只包含标题行，跳过处理: {}", filePath); - continue; - } + String filePath = csvFile.getAbsolutePath(); + try { + log.info("开始处理CSV文件: {}", filePath); + + // 检查CSV文件是否为空或只有一行标题 + try (BufferedReader reader = new BufferedReader(new FileReader(csvFile))) { + String firstLine = reader.readLine(); + if (firstLine == null || firstLine.trim().isEmpty()) { + log.warn("CSV文件为空，跳过处理: {}", filePath); + continue; } - - // 创建Bulk V2 API作业 - String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint(); - String url = restEndpoint.substring(0, restEndpoint.indexOf("services/")); - log.debug("Salesforce实例URL: {}", url); - - // 创建Bulk V2作业 - String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect); - log.info("创建Bulk V2作业成功, 作业ID: {}", jobId); - - // 上传CSV文件数据 - log.info("开始上传CSV文件数据，文件路径: {}, 作业ID: {}", filePath, jobId); - uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath); - log.info("CSV文件数据上传完成，文件路径: {}, 作业ID: {}", filePath, jobId); - // 上传文件完成 - log.info("开始关闭文件上传作业，作业ID: {}", jobId); - closeBulkV2Job(url, jobId, targetBulkConnect); - log.info("作业关闭请求已发送，作业ID: {}", jobId); - - // 等待作业完成 - log.info("等待作业完成，作业ID: {}", jobId); - BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect); - - // 检查作业最终状态 - String jobState = checkJobState(url, jobId, targetBulkConnect); - if ("Failed".equals(jobState)) { - String errorMessage = getJobErrorMessage(url, jobId, targetBulkConnect); - throw new RuntimeException("作业失败，错误信息: " + errorMessage); - } - - log.info("文件 {} 数据导入完成，作业ID: {}", csvFile.getName(), jobId); - - // 下载作业结果文件（成功、失败、未处理记录） - log.info("开始下载作业结果文件，作业ID: {}", jobId); - downloadJobResultFiles(url, jobId, targetBulkConnect, api, csvFile.getName()); - log.info("作业结果文件下载完成，作业ID: {}", jobId); - - success = true; - TimeUnit.SECONDS.sleep(30); - - } catch (InterruptedException interruptedExc) { - return ReturnT.FAIL; - } catch (Exception e) { - retryCount++; - log.error("处理文件 {} 时发生错误，重试次数: {}/{}，错误: {}", - csvFile.getName(), retryCount, maxRetries, e.getMessage(), e); - - // 对于特定可重试的错误，如并发修改异常，进行重试 - if (shouldRetry(e) && retryCount < maxRetries) { - // 指数退避策略 - long backoffTime = (long) Math.pow(2, retryCount) * 1000; - log.info("等待 {} ms 后进行重试...", backoffTime); - Thread.sleep(backoffTime); - } else if (retryCount >= maxRetries) { - log.error("处理文件 {} 达到最大重试次数，放弃处理", csvFile.getName()); - throw e; - } else { - throw e; + String secondLine = reader.readLine(); + if (secondLine == null || secondLine.trim().isEmpty()) { + log.warn("CSV文件只包含标题行，跳过处理: {}", filePath); + continue; } } + + // 创建Bulk V2 API作业 + String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint(); + String url = restEndpoint.substring(0, restEndpoint.indexOf("services/")); + log.debug("Salesforce实例URL: {}", url); + + // 创建Bulk V2作业 + String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect); + log.info("创建Bulk V2作业成功, 作业ID: {}", jobId); + + // 上传CSV文件数据 + log.info("开始上传CSV文件数据，文件路径: {}, 作业ID: {}", filePath, jobId); + uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath); + log.info("CSV文件数据上传完成，文件路径: {}, 作业ID: {}", filePath, jobId); + + // 上传文件完成 + log.info("开始关闭文件上传作业，作业ID: {}", jobId); + closeBulkV2Job(url, jobId, targetBulkConnect); + log.info("作业关闭请求已发送，作业ID: {}", jobId); + + // �ab等待作业完成 + log.info("等待作业完成，作业ID: {}", jobId); + BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect); + + // 检查作业最终状态 + String jobState = checkJobState(url, jobId, targetBulkConnect); + if ("Failed".equals(jobState)) { + String errorMessage = getJobErrorMessage(url, jobId, targetBulkConnect); + throw new RuntimeException("作业失败，错误信息: " + errorMessage); + } + + log.info("文件 {} 数据导入完成，作业ID: {}", csvFile.getName(), jobId); + + // 下载作业结果文件（成功、失败、未处理记录） + log.info("开始下载作业结果文件，作业ID: {}", jobId); + downloadJobResultFiles(url, jobId, targetBulkConnect, api, csvFile.getName()); + log.info("作业结果文件下载完成，作业ID: {}", jobId); + + + TimeUnit.SECONDS.sleep(30); + + } catch (InterruptedException interruptedExc) { + return ReturnT.FAIL; + } catch (Exception e) { + EmailUtil.send("DataDump Import Failed", "BigObject数据导入失败，API: " + api +"，CSV文件：" + filePath + "错误信息: " + e.getMessage()); + throw new RuntimeException( e); } } 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 11a28c4..64ab2a7 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java @@ -609,6 +609,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { // 2. 处理字段相关信息 Map> fieldMap = handlerFieldInfo(fields, fieldTypeSheet, targetConn, sourceConn, dataObject, excelWriter); + HashSet objectHashSet = new HashSet<>(); // 3. 处理每个批次 List batches = getDataBatches(objectApi); @@ -621,7 +622,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { // 为每个批次单独创建校验文件 String batchQualityCheckFilePath = generateObjectQualityCheckFilePath(objectApi + "_" + batch.getId()); try (ExcelWriter objectWriter = EasyExcel.write(batchQualityCheckFilePath).build()) { - batchHasError = processDataBatch(batch, objectApi, fields, excelWriter, objectWriter, summarySheet, dataObject, sourceConn, targetConn, fieldMap, ignoreBeginDate) || batchHasError; + batchHasError = processDataBatch(batch, objectApi, fields, excelWriter, objectWriter, summarySheet, dataObject, sourceConn, targetConn, fieldMap, ignoreBeginDate,objectHashSet) || batchHasError; }catch (Exception e){ log.error("API={},批次ID={}处理异常：{}",objectApi,batch.getId(), e.getMessage(),e); } @@ -649,81 +650,32 @@ public class DataVerifyServiceImpl implements DataVerifyService { arrayList.add(currentFilePath); objectWriter = EasyExcel.write(currentFilePath).build(); - for (DataBatch batch : batches) { - String startDate = DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss"); - String endDate = DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss"); + try { + for (DataBatch batch : batches) { + String startDate = DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss"); + String endDate = DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss"); - log.info("开始处理数据批次: 批次ID={} 批次总量(大于十万只校验十万)={} 批次开始时间={} 批次结束时间={} 对象API={} 忽略开始时间={}", batch.getId(),batch.getFirstSfNum(),startDate,endDate, objectApi, ignoreBeginDate); + log.info("开始处理数据批次: 批次ID={} 批次总量(大于十万只校验十万)={} 批次开始时间={} 批次结束时间={} 对象API={} 忽略开始时间={}", batch.getId(),batch.getFirstSfNum(),startDate,endDate, objectApi, ignoreBeginDate); - try { - // 为每个批次创建新的detail sheet - detailSheet = createDetailSheet(objectWriter, dataObject); - // 分页处理记录 - int page = 0; - int totalRecordNum = 0; - int targetRecordNum = 0; - int sourceRecordNum = 0; - int sorceOnlyNum = 0; - int targetOnlyNum = 0; + try { + // 为每个批次创建新的detail sheet + detailSheet = createDetailSheet(objectWriter, dataObject); + // 分页处理记录 + int page = 0; + int totalRecordNum = 0; + int targetRecordNum = 0; + int sourceRecordNum = 0; + int sorceOnlyNum = 0; + int targetOnlyNum = 0; - Map sourceFieldMap = fieldMap.get("source"); - Map targetFieldMap = fieldMap.get("target"); + Map sourceFieldMap = fieldMap.get("source"); + Map targetFieldMap = fieldMap.get("target"); - while (true) { - List> recordIds = queryRecordIds(objectApi, startDate, endDate, page); - if (recordIds.isEmpty()) break; - int i = 0; - while (true){ - // 检查是否需要创建新文件 - if (writtenRows >= MAX_ROWS_PER_FILE) { - log.info("创建新文件，当前已写入行数: {}", writtenRows); - if (objectWriter != null) { - objectWriter.finish(); - } - fileIndex++; - currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex); - arrayList.add(currentFilePath); - objectWriter = EasyExcel.write(currentFilePath).build(); - detailSheet = createDetailSheet(objectWriter, dataObject); - writtenRows = 0; - } - - int startIndex = MAX_BATCH_RECORDS * i; - int endIndex = Math.min(startIndex + MAX_BATCH_RECORDS, recordIds.size()); - if (startIndex >= endIndex) { - break; - } - - List> idList = recordIds.subList(startIndex, endIndex); - totalRecordNum += idList.size(); - - List ids = Lists.newArrayList(); - List newIds = Lists.newArrayList(); - HashMap idLinks = new HashMap<>(); - for (Map idMap : idList) { - Object idObj = idMap.get("Id"); - Object newIdObj = idMap.get("new_id"); - - if (idObj != null) { - ids.add("'" + idObj.toString() + "'"); - } - if (newIdObj != null) { - newIds.add("'" + newIdObj.toString() + "'"); - } - if (idObj != null && newIdObj != null) { - idLinks.put(idObj.toString(), newIdObj.toString()); - } - } - - // 批量查询源和目标数据 - Map sourceRecords = queryRecords(sourceConn, objectApi, ids); - sourceRecordNum += sourceRecords.size(); - - Map targetRecords = queryRecords(targetConn, objectApi, newIds); - targetRecordNum += targetRecords.size(); - - // 比对记录 - for (String id : idLinks.keySet()) { + while (true) { + List> recordIds = queryRecordIds(objectApi, startDate, endDate, page); + if (recordIds.isEmpty()) break; + int i = 0; + while (true){ // 检查是否需要创建新文件 if (writtenRows >= MAX_ROWS_PER_FILE) { log.info("创建新文件，当前已写入行数: {}", writtenRows); @@ -738,129 +690,206 @@ public class DataVerifyServiceImpl implements DataVerifyService { writtenRows = 0; } - String newId = idLinks.get(id); - SObject sourceRecord = sourceRecords.get(id); - SObject targetRecord = targetRecords.get(newId); - - if (sourceRecord == null && targetRecord != null) { - targetOnlyNum ++; - List row = Lists.newArrayList(); - row.add(String.valueOf(batch.getId())); - row.add(id); - row.add(newId); - row.add(objectApi); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add("目标系统仅有"); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(String.valueOf(targetRecord.getField("LastModifiedDate"))); - // 写入到Excel的detailSheet，而不是添加到某个detailData列表 - objectWriter.write(Collections.singletonList(row), detailSheet); - writtenRows++; - hasError = true; - continue; - }else if (sourceRecord != null && targetRecord == null) { - sorceOnlyNum ++; - List row = Lists.newArrayList(); - row.add(String.valueOf(batch.getId())); - row.add(id); - row.add(newId); - row.add(objectApi); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add("源ORG仅有"); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(" "); - row.add(String.valueOf(sourceRecord.getField("CreatedDate"))); - row.add(String.valueOf(sourceRecord.getField("LastModifiedDate"))); - row.add(" "); - // 写入到Excel的detailSheet，而不是添加到某个detailData列表 - objectWriter.write(Collections.singletonList(row), detailSheet); - writtenRows++; - hasError = true; - continue; + int startIndex = MAX_BATCH_RECORDS * i; + int endIndex = Math.min(startIndex + MAX_BATCH_RECORDS, recordIds.size()); + if (startIndex >= endIndex) { + break; } - boolean isSame = true; - if (ignoreBeginDate != null) { - Object lastModifiedObj = sourceRecord.getField("LastModifiedDate"); - if (lastModifiedObj != null) { - try { - // 解析Salesforce返回的日期格式 2025-10-12T03:27:54.000Z - String lastModifiedStr = lastModifiedObj.toString(); - Date lastModifiedDate = org.apache.commons.lang3.time.DateUtils.parseDate(lastModifiedStr, "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"); + List> idList = recordIds.subList(startIndex, endIndex); + totalRecordNum += idList.size(); - if (lastModifiedDate.before(ignoreBeginDate)) { - isSame = false; - } - } catch (Exception e) { - log.warn("解析LastModifiedDate失败: {}", lastModifiedObj, e); - } + List ids = Lists.newArrayList(); + List newIds = Lists.newArrayList(); + HashMap idLinks = new HashMap<>(); + for (Map idMap : idList) { + Object idObj = idMap.get("Id"); + Object newIdObj = idMap.get("new_id"); + + if (idObj != null) { + ids.add("'" + idObj.toString() + "'"); + } + if (newIdObj != null) { + newIds.add("'" + newIdObj.toString() + "'"); + } + if (idObj != null && newIdObj != null) { + idLinks.put(idObj.toString(), newIdObj.toString()); } } - if (isSame){ - // 比对字段值 - for (DataField field : fields) { - // 检查是否需要创建新文件 - if (writtenRows >= MAX_ROWS_PER_FILE) { - log.info("创建新文件，当前已写入行数: {}", writtenRows); - if (objectWriter != null) { - objectWriter.finish(); + // 批量查询源和目标数据 + Map sourceRecords = queryRecords(sourceConn, objectApi, ids); + sourceRecordNum += sourceRecords.size(); + + Map targetRecords = queryRecords(targetConn, objectApi, newIds); + targetRecordNum += targetRecords.size(); + + // 比对记录 + for (String id : idLinks.keySet()) { + // 检查是否需要创建新文件 + if (writtenRows >= MAX_ROWS_PER_FILE) { + log.info("创建新文件，当前已写入行数: {}", writtenRows); + if (objectWriter != null) { + objectWriter.finish(); + } + fileIndex++; + currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex); + arrayList.add(currentFilePath); + objectWriter = EasyExcel.write(currentFilePath).build(); + detailSheet = createDetailSheet(objectWriter, dataObject); + writtenRows = 0; + } + + String newId = idLinks.get(id); + SObject sourceRecord = sourceRecords.get(id); + SObject targetRecord = targetRecords.get(newId); + + if (sourceRecord == null && targetRecord != null) { + targetOnlyNum ++; + List row = Lists.newArrayList(); + row.add(String.valueOf(batch.getId())); + row.add(id); + row.add(newId); + row.add(objectApi); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add("目标系统仅有"); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(String.valueOf(targetRecord.getField("LastModifiedDate"))); + // 写入到Excel的detailSheet，而不是添加到某个detailData列表 + objectWriter.write(Collections.singletonList(row), detailSheet); + writtenRows++; + hasError = true; + continue; + }else if (sourceRecord != null && targetRecord == null) { + sorceOnlyNum ++; + List row = Lists.newArrayList(); + row.add(String.valueOf(batch.getId())); + row.add(id); + row.add(newId); + row.add(objectApi); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add("源ORG仅有"); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(" "); + row.add(String.valueOf(sourceRecord.getField("CreatedDate"))); + row.add(String.valueOf(sourceRecord.getField("LastModifiedDate"))); + row.add(" "); + // 写入到Excel的detailSheet，而不是添加到某个detailData列表 + objectWriter.write(Collections.singletonList(row), detailSheet); + writtenRows++; + hasError = true; + continue; + } + + boolean isSame = true; + if (ignoreBeginDate != null) { + Object lastModifiedObj = sourceRecord.getField("LastModifiedDate"); + if (lastModifiedObj != null) { + try { + // 解析Salesforce返回的日期格式 2025-10-12T03:27:54.000Z + String lastModifiedStr = lastModifiedObj.toString(); + Date lastModifiedDate = org.apache.commons.lang3.time.DateUtils.parseDate(lastModifiedStr, "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"); + + if (lastModifiedDate.before(ignoreBeginDate)) { + isSame = false; + } + } catch (Exception e) { + log.warn("解析LastModifiedDate失败: {}", lastModifiedObj, e); } - fileIndex++; - currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex); - arrayList.add(currentFilePath); - objectWriter = EasyExcel.write(currentFilePath).build(); - detailSheet = createDetailSheet(objectWriter, dataObject); - writtenRows = 0; } + } - String fieldName = field.getField(); + if (isSame){ + // 比对字段值 + for (DataField field : fields) { + // 检查是否需要创建新文件 + if (writtenRows >= MAX_ROWS_PER_FILE) { + log.info("创建新文件，当前已写入行数: {}", writtenRows); + if (objectWriter != null) { + objectWriter.finish(); + } + fileIndex++; + currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex); + arrayList.add(currentFilePath); + objectWriter = EasyExcel.write(currentFilePath).build(); + detailSheet = createDetailSheet(objectWriter, dataObject); + writtenRows = 0; + } - if (Const.EXCLUDE_FIELDS.contains(fieldName)) - continue; // 跳过字段比较 + String fieldName = field.getField(); - Object sourceValue = sourceRecord.getField(fieldName); - Object targetValue = targetRecord.getField(fieldName); + if (Const.EXCLUDE_FIELDS.contains(fieldName)) + continue; // 跳过字段比较 - Field sorceField = sourceFieldMap.get(fieldName); - Field targetField = targetFieldMap.get(fieldName); + Object sourceValue = sourceRecord.getField(fieldName); + Object targetValue = targetRecord.getField(fieldName); - if (sourceValue == null && targetValue == null) { - continue; - } + Field sorceField = sourceFieldMap.get(fieldName); + Field targetField = targetFieldMap.get(fieldName); - //判断reference_to内是否包含User字符串 - String referenceTo = field.getReferenceTo(); - if (referenceTo !=null && (referenceTo.contains(",User") || referenceTo.contains("User,"))) { - referenceTo = "User"; - } - - if (referenceTo != null &&field.getSfType().equals("reference") && !referenceTo.equals("data_picklist")) { - - String[] split = referenceTo.split(","); - if (split.length > 1 && !"User".equals(referenceTo)){ + if (sourceValue == null && targetValue == null) { continue; } - if (StringUtils.isBlank(referenceTo)){ + if (targetField == null){ + objectHashSet.add(fieldName); continue; } - Map objectMap = customMapper.getById("new_id", referenceTo, String.valueOf(sourceValue)); + //判断reference_to内是否包含User字符串 + String referenceTo = field.getReferenceTo(); + if (referenceTo !=null && (referenceTo.contains(",User") || referenceTo.contains("User,"))) { + referenceTo = "User"; + } - if (objectMap == null || !String.valueOf(objectMap.get("new_id")).equals(String.valueOf(targetValue))) { + if (referenceTo != null &&field.getSfType().equals("reference") && !referenceTo.equals("data_picklist")) { + + String[] split = referenceTo.split(","); + if (split.length > 1 && !"User".equals(referenceTo)){ + continue; + } + if (StringUtils.isBlank(referenceTo)){ + continue; + } + + Map objectMap = customMapper.getById("new_id", referenceTo, String.valueOf(sourceValue)); + + if (objectMap == null || !String.valueOf(objectMap.get("new_id")).equals(String.valueOf(targetValue))) { + List row = Lists.newArrayList(); + row.add(String.valueOf(batch.getId())); + row.add(id); + row.add(newId); + row.add(objectApi); + row.add(field.getField()); + row.add(field.getName()); + row.add(String.valueOf(sourceValue)); + row.add(String.valueOf(targetValue)); + row.add("字段值不一致"); + row.add(String.valueOf(sorceField.getType())); + row.add(String.valueOf(targetField.getType())); + row.add(String.valueOf(sorceField.isCreateable())); + row.add(String.valueOf(targetField.isCreateable())); + row.add(String.valueOf(sourceRecord.getField("CreatedDate"))); + row.add(String.valueOf(sourceRecord.getField("LastModifiedDate"))); + row.add(String.valueOf(targetRecord.getField("LastModifiedDate"))); + // 写入到Excel的detailSheet，而不是添加到某个detailData列表 + objectWriter.write(Collections.singletonList(row), detailSheet); + writtenRows++; + hasError = true; + } + } else if (!Objects.equals(sourceValue, targetValue)){ List row = Lists.newArrayList(); row.add(String.valueOf(batch.getId())); row.add(id); @@ -883,73 +912,55 @@ public class DataVerifyServiceImpl implements DataVerifyService { writtenRows++; hasError = true; } - } else if (!Objects.equals(sourceValue, targetValue)){ - List row = Lists.newArrayList(); - row.add(String.valueOf(batch.getId())); - row.add(id); - row.add(newId); - row.add(objectApi); - row.add(field.getField()); - row.add(field.getName()); - row.add(String.valueOf(sourceValue)); - row.add(String.valueOf(targetValue)); - row.add("字段值不一致"); - row.add(String.valueOf(sorceField.getType())); - row.add(String.valueOf(targetField.getType())); - row.add(String.valueOf(sorceField.isCreateable())); - row.add(String.valueOf(targetField.isCreateable())); - row.add(String.valueOf(sourceRecord.getField("CreatedDate"))); - row.add(String.valueOf(sourceRecord.getField("LastModifiedDate"))); - row.add(String.valueOf(targetRecord.getField("LastModifiedDate"))); - // 写入到Excel的detailSheet，而不是添加到某个detailData列表 - objectWriter.write(Collections.singletonList(row), detailSheet); - writtenRows++; - hasError = true; } } } + i++; + } + + page++; + + if (page * MAX_PAGE_RECORDS >= MAX_TOTAL_RECORDS) { + log.info("达到最大处理记录数限制，停止处理"); + break; // 防止无限循环 } - i++; } - page++; + // 写入批次汇总结果 + List row = Lists.newArrayList(); + row.add(String.valueOf(batch.getId())); + row.add(objectApi); + row.add(startDate); + row.add(endDate); + row.add(String.valueOf(sourceRecordNum)); + row.add(String.valueOf(targetRecordNum)); + row.add(String.valueOf(totalRecordNum)); + row.add(String.valueOf(sorceOnlyNum)); + row.add(String.valueOf(targetOnlyNum)); + row.add(hasError?"不通过":"通过"); - if (page * MAX_PAGE_RECORDS >= MAX_TOTAL_RECORDS) { - log.info("达到最大处理记录数限制，停止处理"); - break; // 防止无限循环 + // 写入Excel的summarySheet + synchronized (this) { + excelWriter.write(Collections.singletonList(row), summarySheet); } + } catch (Exception e){ + // 写入批次汇总结果 + List row = Lists.newArrayList(); + row.add(String.valueOf(batch.getId())); + row.add(objectApi); + row.add(startDate); + row.add(endDate); + row.add("校验异常"); + log.error("API={},处理异常：{}",objectApi, e.getMessage(),e); } - - // 写入批次汇总结果 - List row = Lists.newArrayList(); - row.add(String.valueOf(batch.getId())); - row.add(objectApi); - row.add(startDate); - row.add(endDate); - row.add(String.valueOf(sourceRecordNum)); - row.add(String.valueOf(targetRecordNum)); - row.add(String.valueOf(totalRecordNum)); - row.add(String.valueOf(sorceOnlyNum)); - row.add(String.valueOf(targetOnlyNum)); - row.add(hasError?"不通过":"通过"); - - // 写入Excel的summarySheet - synchronized (this) { - excelWriter.write(Collections.singletonList(row), summarySheet); - } - } catch (Exception e){ - // 写入批次汇总结果 - List row = Lists.newArrayList(); - row.add(String.valueOf(batch.getId())); - row.add(objectApi); - row.add(startDate); - row.add(endDate); - row.add("校验异常"); - log.error("API={},处理异常：{}",objectApi, e.getMessage(),e); + } + } finally { + // 确保ExcelWriter被正确关闭，这样数据才会被写入文件 + if (objectWriter != null) { + objectWriter.finish(); } } - if (sendEmail && hasError){ sendObjectQualityCheckEmailList(objectApi, arrayList); }else { @@ -957,6 +968,12 @@ public class DataVerifyServiceImpl implements DataVerifyService { } } + if (!objectHashSet.isEmpty()){ + String missingFields = "对象 " +objectApi + "在目标ORG中不存在以下字段:" + String.join(", ", objectHashSet); + log.info(missingFields); + EmailUtil.send("数据质量校验 ERROR", missingFields); + } + } @@ -1052,7 +1069,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { */ private boolean processDataBatch(DataBatch batch, String objectApi, List fields, ExcelWriter excelWriter,ExcelWriter objectWriter, WriteSheet summarySheet, DataObject dataObject, - PartnerConnection sourceConn, PartnerConnection targetConn,Map> fieldMap, Date ignoreBeginDate) { + PartnerConnection sourceConn, PartnerConnection targetConn,Map> fieldMap, Date ignoreBeginDate,Set< String> objectHashSet) { WriteSheet detailSheet = createDetailSheet(excelWriter, dataObject); @@ -1073,6 +1090,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { Map sourceFieldMap = fieldMap.get("source"); Map targetFieldMap = fieldMap.get("target"); + while (true) { List> recordIds = queryRecordIds(objectApi, startDate, endDate, page); if (recordIds.isEmpty()) break; @@ -1203,6 +1221,11 @@ public class DataVerifyServiceImpl implements DataVerifyService { continue; } + if (targetField == null){ + objectHashSet.add(fieldName); + continue; + } + //判断reference_to内是否包含User字符串 String referenceTo = field.getReferenceTo(); if (referenceTo !=null && (referenceTo.contains(",User") || referenceTo.contains("User,"))) { @@ -1491,17 +1514,25 @@ private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String file String qualityCheckFilePath = this.generateObjectIncrementalQualityCheckFilePath(objectApi, beginDate); + HashSet objectHashSet = new HashSet<>(); + boolean hasError = false; try (ExcelWriter objectWriter = EasyExcel.write(qualityCheckFilePath).build()) { WriteSheet detailSheet = this.createDetailSheet(objectWriter, dataObject); // 修复：使用正确的ExcelWriter // 3. 处理增量数据（不通过batch） hasError = this.processIncrementalData(objectApi, fields, excelWriter, objectWriter, summarySheet, detailSheet, - sourceConn, targetConn, fieldMap, beginDate, endDate); + sourceConn, targetConn, fieldMap, beginDate, endDate, objectHashSet); } catch (Exception e) { log.info("处理对象 {} 增量数据时发生异常", objectApi, e); } + if (!objectHashSet.isEmpty()){ + String missingFields = "对象 " +objectApi + "在目标ORG中不存在以下字段:" + String.join(", ", objectHashSet); + log.info(missingFields); + EmailUtil.send("数据质量校验 ERROR", missingFields); + } + if (sendEmail && hasError) { this.sendObjectQualityCheckEmail(objectApi + "_增量", qualityCheckFilePath); } @@ -1525,7 +1556,7 @@ private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String file private boolean processIncrementalData(String objectApi, List fields, ExcelWriter excelWriter, ExcelWriter objectWriter, WriteSheet summarySheet, WriteSheet detailSheet, PartnerConnection sourceConn, PartnerConnection targetConn, - Map> fieldMap, Date beginDate, Date endwDate) { + Map> fieldMap, Date beginDate, Date endwDate ,Set< String> objectHashSet) { String startDate = DateFormatUtils.format(beginDate, "yyyy-MM-dd HH:mm:ss"); String endDate = DateFormatUtils.format(endwDate, "yyyy-MM-dd HH:mm:ss"); @@ -1540,6 +1571,8 @@ private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String file Map sourceFieldMap = fieldMap.get("source"); Map targetFieldMap = fieldMap.get("target"); + + try { DescribeSObjectResult dsr = sourceConn.describeSObject(objectApi); Field[] dsrFields = dsr.getFields(); @@ -1676,6 +1709,11 @@ private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String file continue; } + if (targetField == null){ + objectHashSet.add(fieldName); + continue; + } + //判断reference_to内是否包含User字符串 String referenceTo = field.getReferenceTo(); if (referenceTo !=null && (referenceTo.contains(",User") || referenceTo.contains("User,"))) { @@ -1747,6 +1785,7 @@ private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String file log.error("verify error", throwable); } + // 写入批次汇总结果 List row = Lists.newArrayList(); row.add(" "); 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 3f8cfa1..8960f6b 100644 --- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java @@ -561,6 +561,8 @@ public class FileServiceImpl implements FileService { log.info("文件上传错误，返回：" + JSON.toJSONString(response)); throw new RuntimeException(); } + }else { + log.info("文件路径URL为空，数据库执行SQL！！！"); } Map paramMap = Maps.newHashMap(); paramMap.put("key", "is_upload"); diff --git a/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java index d9e76a4..56f70af 100644 --- a/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java @@ -14,6 +14,7 @@ import com.celnet.datadump.util.EmailUtil; import com.sforce.soap.partner.PartnerConnection; import com.sforce.ws.ConnectionException; import com.sforce.ws.ConnectorConfig; +import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -23,11 +24,10 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import static com.celnet.datadump.global.SystemConfigCode.EXECUTOR_SIZE; - /** * org配置表 服务实现类 */ +@Slf4j @Service public class OrgConfigServiceImpl extends ServiceImpl implements OrgConfigService { @@ -43,31 +43,40 @@ public class OrgConfigServiceImpl extends ServiceImpl sourceOrgMap = new HashMap<>(); try { List> poll = customerMapper.list("code,value","org_config",null); //遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值 - Map map = new HashMap<>(); for (Map map1 : poll) { if ("SOURCE_ORG_URL".equals(map1.get("code"))) { - map.put("url", (String) map1.get("value")); + sourceOrgMap.put("url", (String) map1.get("value")); sourceOrgUrl = map1.get("value").toString(); } if ("SOURCE_ORG_USERNAME".equals(map1.get("code"))) { - map.put("username", (String) map1.get("value")); + sourceOrgMap.put("username", (String) map1.get("value")); } if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) { - map.put("password", String.valueOf(map1.get("value"))); + sourceOrgMap.put("password", String.valueOf(map1.get("value"))); } if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) { dumpUrl = (String) map1.get("value"); } + if ("SOURCE_ORG_TOKEN".equals(map1.get("code"))) { + sourceOrgMap.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); + } } - String username = map.get("username"); + String username = sourceOrgMap.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); - config.setPassword(map.get("password")); - String url = map.get("url"); + if (StringUtils.isNotEmpty(sourceOrgMap.get("token"))){ + config.setPassword(sourceOrgMap.get("password") + sourceOrgMap.get("token")); + }else { + config.setPassword(sourceOrgMap.get("password")); + } + log.info("username:{}", username); + log.info("password:{}", sourceOrgMap.get("password") + sourceOrgMap.get("token")); + String url = sourceOrgMap.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); config.setConnectionTimeout(60 * 60 * 1000); @@ -75,38 +84,52 @@ public class OrgConfigServiceImpl extends ServiceImpl targetOrgMap = new HashMap<>(); try { List> poll = customerMapper.list("code,value","org_config",null); //遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值 - Map map = new HashMap<>(); for (Map map1 : poll) { if ("TARGET_ORG_URL".equals(map1.get("code"))) { - map.put("url", (String) map1.get("value")); + targetOrgMap.put("url", (String) map1.get("value")); targetOrgUrl = map1.get("value").toString(); } if ("TARGET_ORG_USERNAME".equals(map1.get("code"))) { - map.put("username", (String) map1.get("value")); + targetOrgMap.put("username", (String) map1.get("value")); } if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) { - map.put("password", (String) map1.get("value")); + targetOrgMap.put("password", (String) map1.get("value")); } if ("FILE_UPLOAD_URL".equals(map1.get("code"))) { uploadUrl = (String) map1.get("value"); } + if ("TARGET_ORG_TOKEN".equals(map1.get("code"))) { + targetOrgMap.put("token", map1.get("value") == null ? "" :(String) map1.get("value")); + } } - String username = map.get("username"); + String username = targetOrgMap.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); - config.setPassword(map.get("password")); - String url = map.get("url"); + if (StringUtils.isNotEmpty(targetOrgMap.get("token"))){ + config.setPassword(targetOrgMap.get("password") + targetOrgMap.get("token")); + }else { + config.setPassword(targetOrgMap.get("password")); + } + log.info("username:{}", username); + log.info("password:{}", targetOrgMap.get("password") + targetOrgMap.get("token")); + String url = targetOrgMap.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); config.setConnectionTimeout(60 * 60 * 1000); @@ -114,10 +137,14 @@ public class OrgConfigServiceImpl extends ServiceImpl