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

【feat】 20250822版本提交

Browse Source
This commit is contained in:
Kris 2025-08-22 16:35:20 +08:00
parent f951e3627e
commit 389bfea6bc
12 changed files with 559 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
View File

@ -380,25 +380,6 @@ public class DataDumpNewJob {
return dataImportNewService.checkDeletedData(param);
}
/**
* 数据统计核对
* @param paramStr 参数json
* @return result
*/
@XxlJob("verifyDataJob")
public ReturnT<String> verifyDataJob(String paramStr) throws Exception {
log.info("verifyDataJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
return dataImportNewService.verifyDataJob(param);
}
/**
* ContentFolderMember 拉取

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

@ -47,4 +47,54 @@ public class DataVerifyJob {
return dataVerifyService.verify(param);
}
/**
* 校验数据总量
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataVerifyCountJob")
public ReturnT<String> dataVerifyCountJob(String paramStr) throws Exception {
log.info("dataVerifyCountJob execute start ..................");
DataVerifyParam param = new DataVerifyParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, DataVerifyParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
if (param.getType() == null) {
return new ReturnT<>(500, "参数缺失！");
}
return dataVerifyService.verifyCount(param);
}
/**
* 校验数据质量
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataVerifyQualityJob")
public ReturnT<String> dataVerifyQualityJob(String paramStr) throws Exception {
log.info("dataVerifyQualityJob execute start ..................");
DataVerifyParam param = new DataVerifyParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, DataVerifyParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
if (param.getType() == null) {
return new ReturnT<>(500, "参数缺失！");
}
// 假设这里调用一个专门用于数据质量校验的服务方法
// 由于当前dataVerifyService中没有直接对应的方法，需要根据实际业务逻辑进行实现
return dataVerifyService.VerifyQuality(param);
}
}

6
src/main/java/com/celnet/datadump/param/DataVerifyParam.java
View File

@ -4,6 +4,8 @@ import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.Date;
/**
* @author Radian
* @description 数据校验参数
@ -47,4 +49,8 @@ public class DataVerifyParam {
@ApiModelProperty(hidden = true)
private Integer type = 1;
@ApiModelProperty(value = "开始时间", hidden = true)
private Date beginDate;
}

3
src/main/java/com/celnet/datadump/param/SalesforceParam.java
View File

@ -89,6 +89,9 @@ public class SalesforceParam extends DataDumpParam implements Cloneable {
@ApiModelProperty(value = "是否单线程", hidden = true)
private Boolean isSingleThread = false;
@ApiModelProperty(value = "是否有new_id", hidden = true)
private Boolean hasNewId = false;
@Override
public SalesforceParam clone() {
try {

1
src/main/java/com/celnet/datadump/service/DataImportNewService.java
View File

@ -22,7 +22,6 @@ public interface DataImportNewService {
ReturnT<String> checkDeletedData(SalesforceParam param) throws Exception;
ReturnT<String> verifyDataJob(SalesforceParam param) throws Exception;
ReturnT<String> insertSingle(SalesforceParam param) throws Exception;

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

@ -16,4 +16,9 @@ public interface DataVerifyService {
* @return ReturnT
*/
ReturnT<String> verify(DataVerifyParam param) throws Exception;
ReturnT<String> verifyCount(DataVerifyParam param) throws Exception;
ReturnT<String> VerifyQuality(DataVerifyParam param) throws Exception;
}

31
src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
View File

@ -372,9 +372,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
String reference = dataField.getReferenceTo();
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield)){
reference = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null;
if (data.get(j-1).get(linkfield)!=null){
reference = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！");
}
}
if (reference == null){
continue;
@ -799,8 +802,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
@ -1186,8 +1193,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
@ -1325,8 +1336,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;

284
src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
View File

@ -805,6 +805,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
int j = 0;
try {
for (Map<String, Object> map : mapList) {
//需要置空的参数
List<String> fieldsToNull = new ArrayList<>();
SObject account = new SObject();
account.setType(api);
//给对象赋值
@ -827,8 +829,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
@ -844,13 +850,21 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
}
} else {
if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
account.setField(field, map.get(field));
if(StringUtils.isNotBlank(dataField.getSfType())){
if (map.get(field) != null){
account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
}else {
fieldsToNull.add(field);
}
}
}
}
if (!fieldsToNull.isEmpty()){
String[] fieldsToNullArray = fieldsToNull.toArray(new String[0]);
account.setField("fieldsToNull", fieldsToNullArray);
}
if (dataObject.getIsEditable()){
account.setField("old_owner_id__c", map.get("OwnerId"));
account.setField("old_sfdc_id__c", map.get("Id"));
@ -878,7 +892,11 @@ public class DataImportNewServiceImpl implements DataImportNewService {
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
} catch (Throwable e) {
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException interruptedException){
return;
}catch (Throwable e) {
log.info(JSON.toJSONString(e));
}
}
@ -1013,8 +1031,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
@ -1628,7 +1650,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
log.info("总文件数 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000");
List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_Type", api, " is_dump = 1 and is_upload = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ "limit " + i * 1000 + ",1000");
for (Map<String, Object> map : list) {
int failCount = 0;
@ -1641,7 +1663,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
id = (String) map.get("Id");
String url_fileName = (String) map.get("url");
String parentId = (String) map.get("ParentId");
String parentType = (String) map.get("Parent_type");
String parentType = (String) map.get("Parent_Type");
// 判断路径是否为空
if (StringUtils.isNotBlank(url_fileName)) {
String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
@ -2237,7 +2259,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if ((param.getEndCreateDate() == null || param.getBeginCreateDate() == null) && param.getIds().isEmpty()){
log.info("手动删除数据必须指定参数：开始与结束时间/Ids !!!!");
return ReturnT.SUCCESS;
return ReturnT.FAIL;
}
String sql = "";
@ -2256,108 +2278,136 @@ public class DataImportNewServiceImpl implements DataImportNewService {
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("删除数据，总批次：{}",page);
try{
for (int i = 0; i < page; i++) {
for (int i = 0; i < page; i++) {
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("*", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}else {
log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
}
}
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
List<Map<String, Object>> maps = new ArrayList<>();
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
customMapper.deleteOne(typeName, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
}
index++;
}
index++;
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return ReturnT.FAIL;
} catch (Exception e) {
log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
}
return ReturnT.SUCCESS;
}
private ReturnT<String> autoCheckDeletedData(SalesforceParam param) throws Exception {
String api = "DeleteEvent";
PartnerConnection connect = salesforceTargetConnect.createConnect();
return ReturnT.SUCCESS;
}
String sql = "SystemModstamp > " + DateUtil.format(DateUtils.addDays(new Date(), -15), "yyyy-MM-dd HH:mm:ss");
Integer count = customMapper.countBySQL(api, sql);
@Override
public ReturnT<String> verifyDataJob(SalesforceParam param) throws Exception {
int page = count%200 == 0 ? count/200 : (count/200) + 1;
QueryWrapper<DataBatch> qw = getDataBatchQueryWrapper(param);
log.info("删除数据，总批次：{}",page);
try{
for (int i = 0; i < page; i++) {
Map<String, List<DataBatch>> maps = Maps.newHashMap();
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
List<DataBatch> dataBatches = dataBatchService.list(qw);
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
String filePath = TEMP_FILE_PATH + "数据校验" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx";
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
try {
ExcelWriter excelWriter = EasyExcel.write(filePath, DataBatch.class).build();
Map<String, WriteSheet> sheetMap = Maps.newHashMap();
dataBatches.forEach(t -> {
List<DataBatch> list = maps.getOrDefault(t.getName(), Lists.newArrayList());
// 日期处理 减1s作为真实范围
t.setSyncEndDate(org.apache.commons.lang3.time.DateUtils.addSeconds(t.getSyncEndDate(), -1));
list.add(t);
maps.put(t.getName(), list);
});
maps.forEach((k, v) -> {
if (CollectionUtils.isNotEmpty(v)) {
// 如果同一个sheet只要创建一次
String label = v.get(0).getLabel();
// excel 不允许sheet名存在[\/?*:]等特殊字符
label = label.replaceAll("[\\\\/?*:]", " ");
WriteSheet writeSheet = sheetMap.getOrDefault(k, EasyExcel.writerSheet(label).build());
excelWriter.write(v, writeSheet);
sheetMap.put(k, writeSheet);
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("new_id,SobjectName", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}else {
log.info("删除数据，未找到对应Id：{}，对应类型：{}", list.get(z).get("Record"), list.get(z).get("SobjectName"));
}
}
});
}catch (Exception e){
log.error(JSON.toJSONString(e));
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
customMapper.deleteOne(typeName, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
}
index++;
}
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return ReturnT.FAIL;
} catch (Exception e) {
log.error("checkDeletedData error api:{},错误信息：{}", api, e.getMessage());
}
String head = "数据校验" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss");
EmailUtil.sendMineWithFile(head, null, false, Lists.newArrayList(filePath));
return ReturnT.SUCCESS;
}
/**
* 参数转换qw
*
@ -2542,7 +2592,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
continue;
}
if (("EventRelation".equals(api) || "TaskRelation".equals(api)) && "RelationId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Relation_type"), jsonObject.getString(key));
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Relation_Type"), jsonObject.getString(key));
if (m == null){
break;
}
@ -2564,7 +2614,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
continue;
}
if ("CollaborationGroupMember".equals(api) && "MemberId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Member_type"), jsonObject.getString(key));
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("Member_Type"), jsonObject.getString(key));
if (m == null){
break;
}
@ -2575,7 +2625,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
continue;
}
if ("FeedAttachment".equals(api) && "FeedEntityId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_type"), jsonObject.getString(key));
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
if (m == null){
break;
}
@ -2586,7 +2636,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
continue;
}
if ("FeedAttachment".equals(api) && "RecordId".equals(key)) {
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_type"), jsonObject.getString(key));
Map<String, Object> m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
if (m == null){
break;
}
@ -2697,15 +2747,19 @@ public class DataImportNewServiceImpl implements DataImportNewService {
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
if (salesforceParam.getIsSingleThread()){
insertSingleExecute(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
@ -2787,15 +2841,19 @@ public class DataImportNewServiceImpl implements DataImportNewService {
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
if (salesforceParam.getIsSingleThread()){
insertSingleExecute(salesforceParam, partnerConnection,update);
}else {
Future<?> future = salesforceExecutor.execute(() -> {
try {
insertSingleExecute(salesforceParam, partnerConnection,update);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 0);
futures.add(future);
}
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
@ -2866,7 +2924,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
//查询当前对象多态字段映射
QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",1);
List<LinkConfig> configs = linkConfigService.list(queryWrapper);
Map<String, String> fieldMap = new HashMap<>();
if (!configs.isEmpty()) {
@ -2931,8 +2989,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield) ){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;
@ -3143,8 +3205,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield) && data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
if (StringUtils.isNotBlank(linkfield)){
if (data.get(j-1).get(linkfield)!=null){
reference_to = data.get(j-1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ data.get(j-1).get("Id") +"的关联类型不存在！");
}
}
if (reference_to == null){
continue;

8
src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
View File

@ -360,11 +360,13 @@ public class DataImportServiceImpl implements DataImportService {
String reference = dataField.getReferenceTo();
//引用类型字段
String linkfield = fieldMap.get(dataField.getField());
if (StringUtils.isNotBlank(linkfield)){
reference = mapList.get(j-1).get(linkfield)!=null?mapList.get(j-1).get(linkfield).toString():null;
if (mapList.get(j - 1).get(linkfield)!=null){
reference = mapList.get(j - 1).get(linkfield).toString();
}else {
log.info("对象类型:" + api + "的数据:"+ mapList.get(j - 1).get("Id") +"的关联类型不存在！");
}
}
if (reference == null){
continue;
}

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

@ -8,33 +8,32 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.DataBatch;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.param.DataVerifyParam;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.service.CommonService;
import com.celnet.datadump.service.DataBatchService;
import com.celnet.datadump.service.DataFieldService;
import com.celnet.datadump.service.DataVerifyService;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.DataUtil;
import com.celnet.datadump.util.EmailUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.soap.partner.PartnerConnection;
import com.sforce.soap.partner.QueryResult;
import com.sforce.soap.partner.sobject.SObject;
import com.xxl.job.core.biz.model.ReturnT;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@ -60,6 +59,10 @@ public class DataVerifyServiceImpl implements DataVerifyService {
private SalesforceExecutor salesforceExecutor;
@Autowired
private SalesforceConnect salesforceConnect;
@Autowired
private SalesforceTargetConnect salesforceTargetConnect;
@Autowired
private DataObjectService dataObjectService;
private static final Integer MAX_FAIL_COUNT = 3;
@ -111,6 +114,260 @@ public class DataVerifyServiceImpl implements DataVerifyService {
}
}
@Override
public ReturnT<String> verifyCount(DataVerifyParam param) throws Exception {
QueryWrapper<DataObject> wrapper = new QueryWrapper<>();
wrapper.eq("data_work",1);
if (StringUtils.isNotBlank(param.getApi())){
wrapper.in("name",DataUtil.toIdList(param.getApi()));
}
List<DataObject> list = dataObjectService.list(wrapper);
if (list.isEmpty()){
log.info("没有对象存在！不进行数据量统计");
return ReturnT.FAIL;
}
PartnerConnection connect = salesforceConnect.createConnect();
PartnerConnection targetConnect = salesforceTargetConnect.createConnect();
String filePath = TEMP_FILE_PATH + "数据量统计_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + ".xlsx";
if (param.getBeginDate() != null) {
filePath = TEMP_FILE_PATH + "数据量统计_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss") + "_统计开始时间：" + DateFormatUtils.format(param.getBeginDate(), "yyyyMMdd") + ".xlsx";
}
try (ExcelWriter excelWriter = EasyExcel.write(filePath).build()) {
WriteSheet writeSheet = EasyExcel.writerSheet("数据量统计").build();
// 写入表头
List<List<String>> head = Lists.newArrayList();
head.add(Lists.newArrayList("对象API","对象Label","源ORG数据量","本地数据量","本地存在new_id数据量","目标ORG数据量","目标存在old_id数据量","HasDeleted","本地重复new_id","目标重复old_id"));
excelWriter.write(head, writeSheet);
WriteSheet writeSheet1 = EasyExcel.writerSheet("本地new_id重复").build();
// 写入表头
List<List<String>> head1 = Lists.newArrayList();
head1.add(Lists.newArrayList("对象API", "Id", "new_id"));
excelWriter.write(head1, writeSheet1);
WriteSheet writeSheet2 = EasyExcel.writerSheet("目标old_id重复").build();
// 写入表头
List<List<String>> head2 = Lists.newArrayList();
head2.add(Lists.newArrayList("对象API", "Id", "old_sfdc_id__c"));
excelWriter.write(head2, writeSheet2);
// 写入数据
for (DataObject dataObject : list) {
String api = dataObject.getName();
String extraSql = "";
SalesforceParam salesforceParam = new SalesforceParam();
salesforceParam.setApi(api);
Boolean hasDeleted = dataFieldService.hasDeleted(api);
if (hasDeleted) {
salesforceParam.setIsDeleted(false);
extraSql = " WHERE IsDeleted = false ";
}
String sql = "SELECT Count(Id) num FROM " + api + extraSql;
String oldIdSql = sql.contains("WHERE") ? sql + " AND old_sfdc_id__c != null":sql + " WHERE old_sfdc_id__c != null";
String duplicateIdSql = "SELECT old_sfdc_id__c FROM " + api + " WHERE old_sfdc_id__c != NULL GROUP BY old_sfdc_id__c HAVING COUNT(old_sfdc_id__c) > 1";
if (param.getBeginDate() != null){
String dateField = dataFieldService.returnDateField(api);
sql = sql.contains("WHERE") ? sql + " AND "+dateField + " >= '" + DateFormatUtils.format(param.getBeginDate(), "yyyy-MM-dd HH:mm:ss") + "'":sql + " WHERE "+dateField + " >= '" + DateFormatUtils.format(param.getBeginDate(), "yyyy-MM-dd HH:mm:ss") + "'";
}
QueryResult queryResult = null;
Integer sourceNum = 0;
int retryCount = 0;
while (retryCount < 3) {
try {
// 查询源ORG数据量
queryResult = connect.queryAll(sql);
SObject record = queryResult.getRecords()[0];
sourceNum = (Integer) record.getField("num");
log.info("api :{} 源ORG num: {}，执行sql：{}", api, sourceNum, sql);
break;
} catch (Exception e) {
retryCount++;
log.info("查询源ORG数据量失败，第{}次重试，执行sql：{}", retryCount, sql, e);
if (retryCount >= 3) {
sourceNum = -1;
}
}
}
Integer localNum = -1;
Integer localNewIdNum = -1;
List<Map<String, Object>> mapList = new ArrayList<>();
if (StringUtils.isNotBlank(customMapper.checkTable(api))){
// 查询本地数据量
localNum = customMapper.count(salesforceParam);
log.info("api :{} 本地 num: {}", api, localNum);
// 查询本地存在new_id数据量
salesforceParam.setHasNewId(true);
localNewIdNum = customMapper.count(salesforceParam);
log.info("api :{} 本地存在new_id num: {}", api, localNewIdNum);
// 查询本地重复new_id的数量
String newReIdSql = "new_id IN ( SELECT new_id FROM `" + api + "` WHERE new_id IS NOT NULL GROUP BY new_id HAVING COUNT(new_id) > 1 )";
mapList = customMapper.list("Id,new_id", api, newReIdSql);
log.info("api :{} 本地重复new_id num: {}", api, mapList.size());
if (!mapList.isEmpty()){
List<List<String>> duplicateData = Lists.newArrayList();
for (Map<String, Object> map : mapList) {
List<String> row = Lists.newArrayList();
row.add(api);
row.add((String) map.get("Id"));
row.add((String) map.get("new_id"));
duplicateData.add(row);
}
excelWriter.write(duplicateData, writeSheet1);
}
}
// 查询目标ORG数据量
QueryResult targetQueryResult = null;
Integer targetNum = 0;
int targetRetryCount = 0;
while (targetRetryCount < 3) {
try {
targetQueryResult = targetConnect.queryAll(sql);
SObject targetRecord = targetQueryResult.getRecords()[0];
targetNum = (Integer) targetRecord.getField("num");
log.info("api :{} 目标ORG num: {}，执行sql：{}", api,sql, targetNum);
break;
} catch (Exception e) {
targetRetryCount++;
log.info("查询目标ORG数据量失败，第{}次重试，执行sql：{}", targetRetryCount,sql, e);
if (targetRetryCount >= 3) {
targetNum = -1;
}
}
}
QueryResult targetQueryOldIdResult = null;
int oldIdRetryCount = 0;
Integer targetOldIdNum = 0;
if (dataObject.getIsEditable()){
while (oldIdRetryCount < 3) {
try {
targetQueryOldIdResult = targetConnect.queryAll(oldIdSql);
SObject targetOldIdRecord = targetQueryOldIdResult.getRecords()[0];
targetOldIdNum = (Integer) targetOldIdRecord.getField("num");
log.info("api :{} 目标ORG存在old_id数据量: {}，执行sql：{}", api, targetOldIdNum, oldIdSql);
break;
} catch (Exception e) {
oldIdRetryCount++;
log.info("查询目标ORG存在old_id数据量失败，第{}次重试，执行sql：{}", oldIdRetryCount, oldIdSql, e);
if (oldIdRetryCount >= 3) {
targetOldIdNum = -1;
}
}
}
}else {
targetOldIdNum = -1;
}
int oldIdRepuRetryCount = 0;
List<List<String>> duplicateOldData = Lists.newArrayList();
if (dataObject.getIsEditable()){
while (oldIdRepuRetryCount < 3) {
try {
List<String> oldIds = Lists.newArrayList();
QueryResult query = targetConnect.queryAll(duplicateIdSql);
if (query != null && query.getRecords() != null && query.getRecords().length > 0) {
for (SObject record : query.getRecords()) {
oldIds = Lists.newArrayList();
oldIds.add((String) record.getField("old_sfdc_id__c"));
}
if (oldIds.isEmpty()) {
break;
}
String idStr = "(";
for (String id : oldIds) {
idStr += "'" + id + "',"; // 拼接每个ID
}
if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束
idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号
}
idStr += ")";
String duplicateOldIdSql = "SELECT Id,old_sfdc_id__c FROM " + api + " WHERE old_sfdc_id__c IN " + idStr;
QueryResult queryResult1 = targetConnect.queryAll(duplicateOldIdSql);
if (queryResult1 != null && queryResult1.getRecords() != null && queryResult1.getRecords().length > 0) {
for (SObject record : queryResult1.getRecords()) {
List<String> row = Lists.newArrayList();
row.add(api);
row.add((String) record.getField("Id"));
row.add((String) record.getField("old_sfdc_id__c"));
duplicateOldData.add(row);
}
excelWriter.write(duplicateOldData, writeSheet2);
log.info("api :{} 查询目标ORG存在old_id重复数据，数据量: {}，执行sql：{}", api, queryResult1.getRecords().length, duplicateOldIdSql);
}
}
break;
} catch (Exception e) {
oldIdRepuRetryCount++;
log.info("查询目标ORG存在old_id重复数据失败，第{}次重试，执行sql：{}", oldIdRepuRetryCount, duplicateIdSql, e);
if (oldIdRepuRetryCount >= 3) {
targetOldIdNum = -1;
}
}
}
}else {
targetOldIdNum = -1;
}
// 写入一行数据
List<List<String>> data = Lists.newArrayList();
List<String> rowData = Lists.newArrayList();
rowData.add(api);
rowData.add(dataObject.getLabel());
rowData.add(sourceNum == -1 ? "错误" : String.valueOf(sourceNum));
rowData.add(localNum == -1 ? "错误" : String.valueOf(localNum));
rowData.add(localNewIdNum == -1 ? "错误" : String.valueOf(localNewIdNum));
rowData.add(targetNum == -1 ? "错误" : String.valueOf(targetNum));
rowData.add(targetOldIdNum == -1 ? "错误" : String.valueOf(targetOldIdNum));
rowData.add(String.valueOf(hasDeleted));
rowData.add(mapList.isEmpty() ? "无" : String.valueOf(mapList.size()));
rowData.add(duplicateOldData.isEmpty() ? "无" : String.valueOf(duplicateOldData.size()));
data.add(rowData);
excelWriter.write(data, writeSheet);
}
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return ReturnT.FAIL;
}catch (Exception e){
log.error("数据量统计失败，错误信息：{}", e.getMessage());
}
String head = "校验数据总量";
String text = head + "更新时间：" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss");
log.info("数据量统计完成，文件路径：{}", filePath);
EmailUtil.sendMineWithFile(head, text, false, Lists.newArrayList(filePath));
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> VerifyQuality(DataVerifyParam param) throws Exception {
return null;
}
/**
* 发送邮件
*

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

@ -654,7 +654,7 @@ public class FileServiceImpl implements FileService {
}
try {
// 获取未存储的附件id
List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_type", api, "is_upload = 0");
List<Map<String, Object>> list = customMapper.list("*", api, "is_upload = 0");
for (Map<String, Object> map : list) {
String finalUploadUrl = uploadUrl;
Future<?> future = salesforceExecutor.execute(() -> {
@ -670,7 +670,13 @@ public class FileServiceImpl implements FileService {
id = (String) map.get("Id");
String url_fileName = (String) map.get("url");
String parentId = (String) map.get("ParentId");
String parentType = (String) map.get("Parent_type");
String parentType = (String) map.get("Parent_Type");
String ownerId = (String) map.get("OwnerId");
String description = (String) map.get("Description");
String createdById = (String) map.get("CreatedById");
String fileName = (String) map.get("Name");
log.info("id: {}, url_fileName: {}, parentId: {}, parentType: {}, ownerId: {}, description: {}, createdById: {}",
id, url_fileName, parentId, parentType, ownerId, description, createdById);
// 判断路径是否为空
if (StringUtils.isNotBlank(url_fileName)) {
String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
@ -680,9 +686,6 @@ public class FileServiceImpl implements FileService {
log.info("文件不存在");
break;
}
String fileName = (String) map.get("Name");
log.info("文件名称：" + fileName);
// dataObject查询
QueryWrapper<DataObject> qw = new QueryWrapper<>();
@ -694,17 +697,25 @@ public class FileServiceImpl implements FileService {
break;
}
Map<String, Object> lMap = customMapper.getById("new_id","User", String.valueOf(map.get("OwnerId")));
Map<String, Object> lMap = customMapper.getById("new_id",parentType, String.valueOf(parentId));
if(lMap == null) {
String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + map.get("OwnerId") + "数据不存在！！！！";
String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
break;
}
Map<String, Object> uMap = customMapper.getById("new_id",parentType, parentId);
Map<String, Object> uMap = customMapper.getById("new_id","User", ownerId);
if(uMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + parentId + "数据不存在！！！！";
String format = "文件ID：" + id + ",对应用户ID：" + ownerId + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
break;
}
Map<String, Object> cMap = customMapper.getById("new_id","User", createdById);
if(cMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + createdById + "数据不存在！！！！";
log.info(format);
EmailUtil.send("DataDump ERROR", format);
break;
@ -719,10 +730,10 @@ public class FileServiceImpl implements FileService {
JSONObject credentialsJsonParam = new JSONObject();
credentialsJsonParam.put("parentId", lMap.get("new_id"));
credentialsJsonParam.put("Name", fileName);
credentialsJsonParam.put("Description", map.get("Description"));
credentialsJsonParam.put("Description", description);
credentialsJsonParam.put("OwnerId", uMap.get("new_id"));
credentialsJsonParam.put("CreatedDate", map.get("CreatedDate"));
credentialsJsonParam.put("CreatedById", map.get("CreatedById"));
// credentialsJsonParam.put("CreatedDate", map.get("CreatedDate"));
credentialsJsonParam.put("CreatedById", cMap.get("new_id"));
MultipartEntityBuilder builder = MultipartEntityBuilder.create();
builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
@ -743,7 +754,7 @@ public class FileServiceImpl implements FileService {
maps.add(paramMap);
}
} else {
log.info("文件ID：" + id + "上传失败，返回信息：" + response);
log.info("文件ID：" + id + "上传失败，返回信息：" + EntityUtils.toString(response.getEntity(), "UTF-8"));
throw new RuntimeException();
}
}

3
src/main/resources/mapper/CustomMapper.xml
View File

@ -132,6 +132,9 @@
<if test="param.isDeleted != null">
AND IsDeleted = #{param.isDeleted}
</if>
<if test="param.hasNewId">
AND new_id IS NOT NULL
</if>
</if>
</where>
</select>