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-10-30 10:42:10 +08:00
parent 713e545420
commit 656ca02ceb
6 changed files with 411 additions and 352 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/main/java/com/celnet/datadump/config/SalesforceConnect.java
View File

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

18
src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java
View File

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

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

@ -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);
// <EFBFBD>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);
}
}

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

@ -609,6 +609,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 2. 处理字段相关信息
Map<String, Map<String, Field>> fieldMap = handlerFieldInfo(fields, fieldTypeSheet, targetConn, sourceConn, dataObject, excelWriter);
HashSet<String> objectHashSet = new HashSet<>();
// 3. 处理每个批次
List<DataBatch> 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<String, Field> sourceFieldMap = fieldMap.get("source");
Map<String, Field> targetFieldMap = fieldMap.get("target");
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){
// 检查是否需要创建新文件
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<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) {
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<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()) {
while (true) {
List<Map<String, Object>> 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<String> 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<String> 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<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size();
if (lastModifiedDate.before(ignoreBeginDate)) {
isSame = false;
}
} catch (Exception e) {
log.warn("解析LastModifiedDate失败: {}", lastModifiedObj, e);
}
List<String> ids = Lists.newArrayList();
List<String> newIds = Lists.newArrayList();
HashMap<String, String> idLinks = new HashMap<>();
for (Map<String, Object> 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<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()) {
// 检查是否需要创建新文件
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<String> 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<String> 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<String, Object> 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<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(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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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<DataField> fields,
ExcelWriter excelWriter,ExcelWriter objectWriter, WriteSheet summarySheet, DataObject dataObject,
PartnerConnection sourceConn, PartnerConnection targetConn,Map<String, Map<String, Field>> fieldMap, Date ignoreBeginDate) {
PartnerConnection sourceConn, PartnerConnection targetConn,Map<String, Map<String, Field>> fieldMap, Date ignoreBeginDate,Set< String> objectHashSet) {
WriteSheet detailSheet = createDetailSheet(excelWriter, dataObject);
@ -1073,6 +1090,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
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;
@ -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<String> 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<DataField> fields,
ExcelWriter excelWriter, ExcelWriter objectWriter, WriteSheet summarySheet, WriteSheet detailSheet,
PartnerConnection sourceConn, PartnerConnection targetConn,
Map<String, Map<String, Field>> fieldMap, Date beginDate, Date endwDate) {
Map<String, Map<String, Field>> 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<String, Field> sourceFieldMap = fieldMap.get("source");
Map<String, Field> 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<String> row = Lists.newArrayList();
row.add(" ");

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

@ -561,6 +561,8 @@ public class FileServiceImpl implements FileService {
log.info("文件上传错误，返回：" + JSON.toJSONString(response));
throw new RuntimeException();
}
}else {
log.info("文件路径URL为空，数据库执行SQL！！！");
}
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_upload");

71
src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java
View File

@ -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<OrgConfigMapper, OrgConfig> implements OrgConfigService {
@ -43,31 +43,40 @@ public class OrgConfigServiceImpl extends ServiceImpl<OrgConfigMapper, OrgConfig
String uploadUrl = null;
String dumpUrl = null;
Map<String, String> sourceOrgMap = new HashMap<>();
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> 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<OrgConfigMapper, OrgConfig
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
} catch (ConnectionException e) {
String message = "源ORG连接配置错误！,\n地址：" + sourceOrgUrl + "\n错误信息：\n" + e.getMessage() ;;
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
log.info("exception message", e);
String message = String.format("源ORG连接配置错误！%n地址：%s%n用户名：%s%nToken：%s%n错误信息：%s",
sourceOrgUrl,
sourceOrgMap.get("username"),
sourceOrgMap.get("token"),
e.getMessage());
String format = String.format("ORG连接异常！%n详细信息：%n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
flag = false;
}
Map<String, String> targetOrgMap = new HashMap<>();
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> 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<OrgConfigMapper, OrgConfig
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
} catch (ConnectionException e) {
String message = "目标ORG连接配置错误！,\n地址：" + targetOrgUrl + "\n错误信息：\n" + e.getMessage() ;
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
log.info("exception message", e);
String message = String.format("目标ORG连接配置错误！%n地址：%s%n用户名：%s%nToken：%s%n错误信息：%s",
targetOrgUrl,
targetOrgMap.get("username"),
targetOrgMap.get("token"),
e.getMessage());
String format = String.format("ORG连接异常！%n详细信息：%n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
flag = false;
}