From 684913309296e1abf93e3be924ecb97411b9899a Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Mon, 29 Sep 2025 10:09:41 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=E5=A2=9E=E9=87=8F?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E8=B4=A8=E9=87=8F=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../celnet/datadump/job/DataVerifyJob.java | 28 +- .../datadump/service/DataVerifyService.java | 9 +- .../service/impl/CommonBatchServiceImpl.java | 201 ++++++-- .../service/impl/DataVerifyServiceImpl.java | 477 +++++++++++++++++- .../service/impl/FileServiceImpl.java | 2 +- 5 files changed, 649 insertions(+), 68 deletions(-) diff --git a/src/main/java/com/celnet/datadump/job/DataVerifyJob.java b/src/main/java/com/celnet/datadump/job/DataVerifyJob.java index a28de9f..ce998da 100644 --- a/src/main/java/com/celnet/datadump/job/DataVerifyJob.java +++ b/src/main/java/com/celnet/datadump/job/DataVerifyJob.java @@ -96,5 +96,31 @@ public class DataVerifyJob { // 由于当前dataVerifyService中没有直接对应的方法，需要根据实际业务逻辑进行实现 return dataVerifyService.verifyQuality(param); } + + /** + * 增量数据质量校验 + * 通过参数传入开始时间，查询修改时间大于该开始时间的数据，进行数据校验 + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataVerifyIncrementalQualityJob") + public ReturnT dataVerifyIncrementalQualityJob(String paramStr) throws Exception { + log.info("dataVerifyIncrementalQualityJob 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.getBeginDate() == null) { + return new ReturnT<>(500, "开始时间参数缺失！"); + } -} + // 调用增量数据质量校验服务方法 + return dataVerifyService.verifyIncrementalQuality(param); + } + +} \ No newline at end of file diff --git a/src/main/java/com/celnet/datadump/service/DataVerifyService.java b/src/main/java/com/celnet/datadump/service/DataVerifyService.java index 58f0c24..25a5e69 100644 --- a/src/main/java/com/celnet/datadump/service/DataVerifyService.java +++ b/src/main/java/com/celnet/datadump/service/DataVerifyService.java @@ -20,5 +20,12 @@ public interface DataVerifyService { ReturnT verifyCount(DataVerifyParam param) throws Exception; ReturnT verifyQuality(DataVerifyParam param) throws Exception; + + /** + * 增量数据质量校验 + * @param param 参数 + * @return ReturnT + */ + ReturnT verifyIncrementalQuality(DataVerifyParam param) throws Exception; -} +} \ No newline at end of file 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 709774e..327b815 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -336,60 +336,91 @@ public class CommonBatchServiceImpl implements CommonBatchService { // 使用Bulk V2 API导入所有文件 for (File csvFile : files) { - try { - String filePath = csvFile.getAbsolutePath(); - 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; + // 添加重试机制 + 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 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); + + // 等待作业完成 + 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; } } - - // 创建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); - 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) { - log.error("处理文件 {} 时发生错误: {}", csvFile.getName(), e.getMessage(), e); } } @@ -1244,4 +1275,74 @@ public class CommonBatchServiceImpl implements CommonBatchService { log.info("dataDumpJob done api:{}, count:{}", dataBatchHistory.getName(), num); } + /** + * 检查作业状态 + */ + private String checkJobState(String url, String jobId, BulkConnection connection) throws Exception { + CloseableHttpClient httpClient = HttpClients.createDefault(); + String jobUrl = url + "services/data/v56.0/jobs/ingest/" + jobId; + HttpGet httpGet = new HttpGet(jobUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + httpGet.setHeader("Content-Type", "application/json"); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + String responseBody = EntityUtils.toString(response.getEntity()); + JSONObject jobInfo = JSON.parseObject(responseBody); + return jobInfo.getString("state"); + } else { + throw new RuntimeException("获取作业状态失败，状态码: " + statusCode); + } + } finally { + response.close(); + httpClient.close(); + } + } + + /** + * 获取作业错误信息 + */ + private String getJobErrorMessage(String url, String jobId, BulkConnection connection) throws Exception { + CloseableHttpClient httpClient = HttpClients.createDefault(); + String jobUrl = url + "services/data/v56.0/jobs/ingest/" + jobId; + HttpGet httpGet = new HttpGet(jobUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + httpGet.setHeader("Content-Type", "application/json"); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + String responseBody = EntityUtils.toString(response.getEntity()); + JSONObject jobInfo = JSON.parseObject(responseBody); + return jobInfo.getString("errorMessage"); + } else { + return "无法获取错误信息，状态码: " + statusCode; + } + } finally { + response.close(); + httpClient.close(); + } + } + + /** + * 判断是否应该重试 + */ + private boolean shouldRetry(Exception e) { + String message = e.getMessage(); + if (message == null) return false; + + // 针对并发修改异常进行重试 + if (message.contains("ConcurrentTableMutationException") || + message.contains("Concurrent modification") || + message.contains("InternalServerError")) { + return true; + } + + // 其他可能的临时性错误也可以加入这里 + + return false; + } } 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 e0deb3e..44b8583 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java @@ -110,7 +110,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { } while (startPage <= page.getPages()); } catch (Throwable throwable) { - log.error("verify error", throwable); + log.info("verify error", throwable); throw new RuntimeException(throwable); } // 等待线程执行完毕 @@ -360,7 +360,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { }catch (InterruptedException e){ return ReturnT.FAIL; }catch (Exception e){ - log.error("数据量统计失败，错误信息：{}", e.getMessage()); + log.info("数据量统计失败，错误信息：{}", e.getMessage()); } String head = "校验数据总量"; @@ -418,7 +418,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { log.info("数据质量校验执行完成，耗时: {} ms", endTime - startTime); } catch (Exception e) { long errorTime = System.currentTimeMillis(); - log.error("数据质量校验执行失败，耗时: {} ms", errorTime - startTime, e); + log.info("数据质量校验执行失败，耗时: {} ms", errorTime - startTime, e); return new ReturnT<>(ReturnT.FAIL.getCode(), "执行数据质量校验失败: " + e.getMessage()); } @@ -433,7 +433,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { sendQualityCheckEmail(param, filePath); log.info("邮件发送完成"); } catch (Exception e) { - log.error("邮件发送失败", e); + log.info("邮件发送失败", e); // 邮件发送失败不影响主流程，仅记录错误日志 } } else { @@ -443,7 +443,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { log.info(">>> 数据质量校验任务执行完成 <<<"); return ReturnT.SUCCESS; } catch (Exception e) { - log.error("数据质量校验过程中发生未预期的错误", e); + log.info("数据质量校验过程中发生未预期的错误", e); return new ReturnT<>(ReturnT.FAIL.getCode(), "数据质量校验失败: " + e.getMessage()); } finally { log.info("数据质量校验任务结束"); @@ -511,7 +511,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { try { salesforceExecutor.waitForFutures(futures.toArray(new Future[0])); } catch (InterruptedException e) { - log.error("线程执行被中断", e); + log.info("线程执行被中断", e); Thread.currentThread().interrupt(); } @@ -556,7 +556,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { String objectApi = dataObject.getName(); List fields = getVerifiableFields(objectApi); if (fields.isEmpty()) { - log.error("对象: {} 没有需要校验的字段！！！", objectApi); + log.info("对象: {} 没有需要校验的字段！！！", objectApi); return; } @@ -659,7 +659,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { } } } catch (Exception e) { - log.error("处理字段信息时发生异常: ", e); + log.info("处理字段信息时发生异常: ", e); } return fieldMap; } @@ -687,7 +687,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { Map targetFieldMap = fieldMap.get("target"); while (true) { - List> recordIds = queryRecordIds(objectApi, batch, page); + List> recordIds = queryRecordIds(objectApi, startDate, endDate, page); if (recordIds.isEmpty()) break; int i = 0; while (true){ @@ -873,8 +873,455 @@ public class DataVerifyServiceImpl implements DataVerifyService { log.info("数据批次处理完成: 批次ID={} 批次开始时间={} 批次结束时间={} 对象API={}", batch.getId(),startDate,endDate, objectApi); } + + /** + * 增量数据质量校验 + * 通过参数传入开始时间，查询修改时间大于该开始时间的数据，进行数据校验 + * + * @param param 参数 + * @return ReturnT + */ + @Override + public ReturnT verifyIncrementalQuality(DataVerifyParam param) throws Exception { + log.info(">>> 开始执行增量数据质量校验任务 <<<"); + log.info("输入参数: {}", param != null ? param.toString() : "null"); + + // 检查必要参数 + if (param.getBeginDate() == null) { + log.info("增量数据质量校验缺少开始时间参数"); + return new ReturnT<>(ReturnT.FAIL.getCode(), "缺少开始时间参数"); + } + + try { + // 获取需要校验的数据对象列表 + log.info("步骤1: 获取需要校验的数据对象列表"); + List dataObjects = this.getDataObjectsForQualityCheck(param); + log.info("成功获取数据对象列表，共 {} 个对象", dataObjects.size()); + + // 如果没有需要校验的对象，直接返回失败 + if (dataObjects.isEmpty()) { + log.warn("没有需要校验的数据对象，终止执行"); + return ReturnT.FAIL; + } + log.info("校验对象列表: {}", dataObjects.stream().map(DataObject::getName).collect(Collectors.joining(", "))); + // 创建Salesforce连接 + log.info("步骤2: 创建Salesforce连接"); + log.debug("正在创建源系统Salesforce连接..."); + PartnerConnection sourceConnect = salesforceConnect.createConnect(); + log.debug("源系统Salesforce连接创建成功"); + + log.debug("正在创建目标系统Salesforce连接..."); + PartnerConnection targetConnect = salesforceTargetConnect.createConnect(); + log.debug("目标系统Salesforce连接创建成功"); + log.info("Salesforce连接创建完成"); + // 生成文件路径 + log.info("步骤3: 生成增量质量检查报告文件路径"); + String filePath = this.generateIncrementalQualityCheckFilePath(param.getBeginDate()); + log.info("增量质量检查报告文件路径: {}", filePath); + + // 执行增量数据质量校验并写入Excel文件 + log.info("步骤4: 执行增量数据质量校验并写入Excel文件"); + long startTime = System.currentTimeMillis(); + log.info("开始时间: {}", new Date(startTime)); + + try { + this.performIncrementalQualityCheckAndWriteToExcel(dataObjects, sourceConnect, targetConnect, filePath, param.getBeginDate(), param.getSendEmail()); + long endTime = System.currentTimeMillis(); + log.info("增量数据质量校验执行完成，耗时: {} ms", endTime - startTime); + } catch (Exception e) { + long errorTime = System.currentTimeMillis(); + log.info("增量数据质量校验执行失败，耗时: {} ms", errorTime - startTime, e); + return new ReturnT<>(ReturnT.FAIL.getCode(), "执行增量数据质量校验失败: " + e.getMessage()); + } + + // 记录日志 + log.info("增量数据质量校验完成，报告文件路径: {}", filePath); + + // 发送邮件（如果需要） + log.info("步骤5: 检查并发送增量质量检查邮件"); + if (param.getSendEmail()) { + log.debug("需要发送邮件，开始执行邮件发送逻辑"); + try { + this.sendIncrementalQualityCheckEmail(param, filePath); + log.info("邮件发送完成"); + } catch (Exception e) { + log.info("邮件发送失败", e); + // 邮件发送失败不影响主流程，仅记录错误日志 + } + } else { + log.info("无需发送邮件，跳过邮件发送步骤"); + } + + log.info(">>> 增量数据质量校验任务执行完成 <<<"); + return ReturnT.SUCCESS; + } catch (Exception e) { + log.info("增量数据质量校验过程中发生未预期的错误", e); + return new ReturnT<>(ReturnT.FAIL.getCode(), "增量数据质量校验失败: " + e.getMessage()); + } finally { + log.info("增量数据质量校验任务结束"); + } + } + + /** + * 生成增量质量检查报告文件路径 + * + * @param beginDate 开始时间 + * @return 文件路径 + */ + private String generateIncrementalQualityCheckFilePath(Date beginDate) { + return TEMP_FILE_PATH + "增量数据质量校验_" + DateFormatUtils.format(beginDate, "yyyyMMdd") + "_" + + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx"; + } + + /** + * 执行增量质量检查并将结果写入Excel文件 + * + * @param dataObjects 数据对象列表 + * @param sourceConn 源系统连接 + * @param targetConn 目标系统连接 + * @param filePath 文件路径 + * @param beginDate 开始时间 + * @param sendEmail 是否发送邮件 + * @throws Exception 异常 + */ + private void performIncrementalQualityCheckAndWriteToExcel(List dataObjects, + PartnerConnection sourceConn, + PartnerConnection targetConn, + String filePath, + Date beginDate, + boolean sendEmail) throws Exception { + try (ExcelWriter excelWriter = EasyExcel.write(filePath).build()) { + // 1. 创建汇总 Sheet + WriteSheet summarySheet = this.createSummarySheet(excelWriter); + // 2. 创建字段类型 Sheet + WriteSheet fieldTypeSheet = this.createFieldTypeSheet(excelWriter); + + List> futures = Lists.newArrayList(); + dataObjects.forEach(dataObject -> { + Future future = salesforceExecutor.execute(() -> { + this.processIncrementalDataObject(dataObject, excelWriter, summarySheet, fieldTypeSheet, + sourceConn, targetConn, beginDate, sendEmail); + }, 1, 1); + futures.add(future); + }); + + // 等待所有线程执行完毕 + try { + salesforceExecutor.waitForFutures(futures.toArray(new Future[0])); + } catch (InterruptedException e) { + log.info("线程执行被中断", e); + Thread.currentThread().interrupt(); + } + } + } + + /** + * 发送增量质量检查邮件 + * + * @param param 参数 + * @param filePath 文件路径 + */ + private void sendIncrementalQualityCheckEmail(DataVerifyParam param, String filePath) { + String head = "增量数据质量校验报告"; + String text = head + "生成时间：" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + + "，开始时间：" + DateFormatUtils.format(param.getBeginDate(), "yyyy-MM-dd HH:mm:ss"); + + // 发送邮件（如果需要） + if (param.getSendEmail()) { + if (StringUtils.isNotBlank(param.getSendTo())) { + String[] sendTo = DataUtil.toIdList(param.getSendTo()).toArray(new String[]{}); + EmailUtil.sendMineWithFile(head, text, false, Lists.newArrayList(filePath), sendTo); + } else { + EmailUtil.sendMineWithFile(head, text, false, Lists.newArrayList(filePath)); + } + } + } + + /** + * 处理单个数据对象（增量） + */ + private void processIncrementalDataObject(DataObject dataObject, ExcelWriter excelWriter, WriteSheet summarySheet, + WriteSheet fieldTypeSheet, PartnerConnection sourceConn, + PartnerConnection targetConn, Date beginDate, boolean sendEmail) { + // 1. 准备对象相关信息 + String objectApi = dataObject.getName(); + List fields = this.getVerifiableFields(objectApi); + if (fields.isEmpty()) { + log.info("对象: {} 没有需要校验的字段！！！", objectApi); + return; + } + + // 2. 处理字段相关信息 + Map> fieldMap = this.handlerFieldInfo(fields, fieldTypeSheet, targetConn, sourceConn, dataObject, excelWriter); + + String qualityCheckFilePath = this.generateObjectIncrementalQualityCheckFilePath(objectApi, beginDate); + + try (ExcelWriter objectWriter = EasyExcel.write(qualityCheckFilePath).build()) { + WriteSheet detailSheet = this.createDetailSheet(excelWriter, dataObject); + + // 3. 处理增量数据（不通过batch） + this.processIncrementalData(objectApi, fields, excelWriter, objectWriter, summarySheet, detailSheet, + sourceConn, targetConn, fieldMap, beginDate); + } catch (Exception e) { + log.info("处理对象 {} 增量数据时发生异常", objectApi, e); + } + + if (sendEmail) { + this.sendObjectQualityCheckEmail(objectApi + "_增量", qualityCheckFilePath); + } + } + + /** + * 生成对象增量质量检查报告文件路径 + * + * @param objectApi 对象API + * @param beginDate 开始时间 + * @return 文件路径 + */ + private String generateObjectIncrementalQualityCheckFilePath(String objectApi, Date beginDate) { + return TEMP_FILE_PATH + objectApi + "_增量_" + DateFormatUtils.format(beginDate, "yyyyMMdd") + "_" + + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx"; + } + + /** + * 处理增量数据（不通过batch） + */ + private void processIncrementalData(String objectApi, List fields, + ExcelWriter excelWriter, ExcelWriter objectWriter, WriteSheet summarySheet, WriteSheet detailSheet, + PartnerConnection sourceConn, PartnerConnection targetConn, + Map> fieldMap, Date beginDate) { + String startDate = DateFormatUtils.format(beginDate, "yyyy-MM-dd HH:mm:ss"); + String endDate = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"); + + log.info("开始处理增量数据: 对象API={} 开始时间={} 结束时间={}", objectApi, startDate, endDate); + + // 分页处理记录 + int page = 0; + int totalRecordNum = 0; + int targetRecordNum = 0; + int sourceRecordNum = 0; + int sorceOnlyNum = 0; + int targetOnlyNum = 0; + boolean isError = false; + + 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){ + 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) { + ids.add("'" + idMap.get("Id").toString() + "'"); + newIds.add("'" + idMap.get("new_id").toString() + "'"); + idLinks.put(idMap.get("Id").toString(), idMap.get("new_id").toString()); + } + + // 批量查询源和目标数据 + Map sourceRecords = queryRecords(sourceConn, objectApi, ids); + sourceRecordNum += sourceRecords.size(); + + Map targetRecords = queryRecords(targetConn, objectApi, newIds); + targetRecordNum += targetRecords.size(); + + // 比对记录 + for (String id : idLinks.keySet()) { + 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(" "); + 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); + isError = true; + continue; + }else if (sourceRecord != null && targetRecord == null) { + sorceOnlyNum ++; + List row = Lists.newArrayList(); + row.add(" "); + 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); + isError = true; + continue; + } + + // 比对字段值 + for (DataField field : fields) { + String fieldName = field.getField(); + + if (Const.EXCLUDE_FIELDS.contains(fieldName)) + continue; // 跳过字段比较 + + Object sourceValue = sourceRecord.getField(fieldName); + Object targetValue = targetRecord.getField(fieldName); + + Field sorceField = sourceFieldMap.get(fieldName); + Field targetField = targetFieldMap.get(fieldName); + + if (sourceValue == null && targetValue == null) { + continue; + } + + if (field.getSfType().equals("reference") && !field.getReferenceTo().equals("data_picklist")) { + //判断reference_to内是否包含User字符串 + String referenceTo = field.getReferenceTo(); + if (field.getReferenceTo().contains(",User") || field.getReferenceTo().contains("User,")) { + referenceTo = "User"; + } + 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(" "); + 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); + isError = true; + } + } else if (!Objects.equals(sourceValue, targetValue)){ + List row = Lists.newArrayList(); + row.add(" "); + 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); + isError = true; + } + } + } + i++; + } + + page++; + + if (page * MAX_PAGE_RECORDS >= MAX_TOTAL_RECORDS) { + log.info("达到最大处理记录数限制，停止处理"); + break; // 防止无限循环 + } + } + + // 写入批次汇总结果 + List row = Lists.newArrayList(); + row.add(" "); + 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(isError?"不通过":"通过"); + + // 写入Excel的summarySheet + excelWriter.write(Collections.singletonList(row), summarySheet); + } + + /** + * 写入详细记录 + */ + private void writeDetailRecord(ExcelWriter objectWriter, WriteSheet detailSheet, String batchId, String sourceId, + String targetId, String objectApi, String status, String fieldApi, String fieldLabel, + String sourceValue, String targetValue, String sourceFieldType, String targetFieldType, + Object sourceCreatedDate, Object sourceLastModifiedDate, Object targetLastModifiedDate) { + List row = Lists.newArrayList(); + row.add(batchId != null ? batchId : ""); + row.add(sourceId != null ? sourceId : ""); + row.add(targetId != null ? targetId : ""); + row.add(objectApi != null ? objectApi : ""); + row.add(fieldApi != null ? fieldApi : ""); + row.add(fieldLabel != null ? fieldLabel : ""); + row.add(sourceValue != null ? sourceValue : ""); + row.add(targetValue != null ? targetValue : ""); + row.add(status != null ? status : ""); + row.add(sourceFieldType != null ? sourceFieldType : ""); + row.add(targetFieldType != null ? targetFieldType : ""); + row.add(""); // 源系统是否可写入（暂时留空） + row.add(""); // 目标系统是否可写入（暂时留空） + row.add(sourceCreatedDate != null ? String.valueOf(sourceCreatedDate) : ""); + row.add(sourceLastModifiedDate != null ? String.valueOf(sourceLastModifiedDate) : ""); + row.add(targetLastModifiedDate != null ? String.valueOf(targetLastModifiedDate) : ""); + + objectWriter.write(Collections.singletonList(row), detailSheet); + } private Map queryRecords(PartnerConnection connect, String objectApi ,List ids){ String idCondition = "Id IN (" + String.join(",", ids) + ")"; @@ -891,16 +1338,16 @@ public class DataVerifyServiceImpl implements DataVerifyService { } } } catch (Exception e) { - log.error("查询ORG数据时发生异常: {}", e.getMessage()); + log.info("查询ORG数据时发生异常: {}", e.getMessage()); } return queryMap; } - private List> queryRecordIds(String objectApi, DataBatch batch, int page){ + private List> queryRecordIds(String objectApi, String startDate, String endDate, int page){ // 查询本地数据库获取ID列表 - String idSql = " CreatedDate >= '" + DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss") + "'" + - " AND CreatedDate < '" + DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss") + "'" + + String idSql = " CreatedDate >= '" + startDate + "'" + + " AND CreatedDate < '" + endDate + "'" + " ORDER BY Id ASC LIMIT " + (page * MAX_PAGE_RECORDS) + "," + MAX_PAGE_RECORDS; return customMapper.list("Id,new_id", objectApi, idSql); @@ -1205,7 +1652,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { FileUtils.delete(new File(filePath)); } catch (Throwable throwable) { - log.error("send verify email error", throwable); + log.info("send verify email error", throwable); throw new RuntimeException(throwable); } } @@ -1269,7 +1716,7 @@ public class DataVerifyServiceImpl implements DataVerifyService { throw e; } catch (Throwable throwable) { failCount++; - log.error("verify error", throwable); + log.info("verify error", throwable); if (failCount > MAX_FAIL_COUNT) { throw throwable; } 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 1dec4c0..3f8cfa1 100644 --- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java @@ -570,7 +570,7 @@ public class FileServiceImpl implements FileService { TimeUnit.MILLISECONDS.sleep(1); break; } catch (Exception throwable) { - log.error("upload file error, id: {}", id, throwable); + log.error("upload file error, id: {}, 返回实体信息：{}", id,respContent, throwable); failCount++; if (Const.MAX_FAIL_COUNT < failCount) { {