kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】增量数据质量校验

Browse Source
This commit is contained in:
Kris 2025-09-29 10:09:41 +08:00
parent 6a2492cea7
commit 6849133092
5 changed files with 649 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/main/java/com/celnet/datadump/job/DataVerifyJob.java
View File

@ -96,5 +96,31 @@ public class DataVerifyJob {
// 由于当前dataVerifyService中没有直接对应的方法，需要根据实际业务逻辑进行实现
return dataVerifyService.verifyQuality(param);
}
/**
* 增量数据质量校验
* 通过参数传入开始时间，查询修改时间大于该开始时间的数据，进行数据校验
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataVerifyIncrementalQualityJob")
public ReturnT<String> 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);
}
}

9
src/main/java/com/celnet/datadump/service/DataVerifyService.java
View File

@ -20,5 +20,12 @@ public interface DataVerifyService {
ReturnT<String> verifyCount(DataVerifyParam param) throws Exception;
ReturnT<String> verifyQuality(DataVerifyParam param) throws Exception;
/**
* 增量数据质量校验
* @param param 参数
* @return ReturnT
*/
ReturnT<String> verifyIncrementalQuality(DataVerifyParam param) throws Exception;
}
}

201
src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
View File

@ -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;
}
}

477
src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
View File

@ -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<DataField> 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<String, Field> targetFieldMap = fieldMap.get("target");
while (true) {
List<Map<String, Object>> recordIds = queryRecordIds(objectApi, batch, page);
List<Map<String, Object>> 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<String> 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<DataObject> 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<DataObject> 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<Future<?>> 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<DataField> fields = this.getVerifiableFields(objectApi);
if (fields.isEmpty()) {
log.info("对象: {} 没有需要校验的字段！！！", objectApi);
return;
}
// 2. 处理字段相关信息
Map<String, Map<String, Field>> 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<DataField> fields,
ExcelWriter excelWriter, ExcelWriter objectWriter, WriteSheet summarySheet, WriteSheet detailSheet,
PartnerConnection sourceConn, PartnerConnection targetConn,
Map<String, Map<String, Field>> 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<String, Field> sourceFieldMap = fieldMap.get("source");
Map<String, Field> targetFieldMap = fieldMap.get("target");
while (true) {
List<Map<String, Object>> 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<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size();
List<String> ids = Lists.newArrayList();
List<String> newIds = Lists.newArrayList();
HashMap<String, String> idLinks = new HashMap<>();
for (Map<String, Object> 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<String, SObject> sourceRecords = queryRecords(sourceConn, objectApi, ids);
sourceRecordNum += sourceRecords.size();
Map<String, SObject> 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<String> 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<String> 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<String, Object> objectMap = customMapper.getById("new_id", referenceTo, String.valueOf(sourceValue));
if (objectMap == null || !String.valueOf(objectMap.get("new_id")).equals(String.valueOf(targetValue))) {
List<String> 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<String> 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<String> 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<String> 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<String, SObject> queryRecords(PartnerConnection connect, String objectApi ,List<String> 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<Map<String, Object>> queryRecordIds(String objectApi, DataBatch batch, int page){
private List<Map<String, Object>> 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;
}

2
src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java
View File

@ -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) {
{