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

【feat】 20250826版本

Browse Source
This commit is contained in:
Kris 2025-08-26 15:08:41 +08:00
parent 389bfea6bc
commit 912efda685
9 changed files with 116 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.run/data-dump（hitpoint）.run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（hitpoint）" type="docker-deploy" factoryName="dockerfile" server-name="hitpoint">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

14
.run/data-dump（百思）.run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（百思）" type="docker-deploy" factoryName="dockerfile" server-name="雨花石阿里云docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

14
.run/data-dump（稳健医疗).run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（稳健医疗)" type="docker-deploy" factoryName="dockerfile" server-name="稳健医疗docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

14
.run/data-dump（鸿利达） .run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（鸿利达） " type="docker-deploy" factoryName="dockerfile" server-name="鸿利达docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

6
src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
View File

@ -865,15 +865,17 @@ public class CommonServiceImpl implements CommonService {
// 根据id查询数据库 取出已存在数据的id
List<String> ids = Arrays.stream(records).map(SObject::getId).collect(Collectors.toList());
DataObject one = dataObjectService.getById(api);
QueryWrapper<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<String> list = dataFieldService.list(dbQw).stream().map(DataField::getField).collect(Collectors.toList());
List<String> existsIds = customMapper.getIds(api, ids);
for (int i = 0; i < objects.size(); i++) {
JSONObject jsonObject = objects.getJSONObject(i);
try {
Set<String> keys = jsonObject.keySet();
// update
String id = jsonObject.getString(Const.ID);
List<Map<String, Object>> maps = Lists.newArrayList();
for (String key : keys) {
for (String key : list) {
if (fields.stream().anyMatch(key::equalsIgnoreCase)) {
// ContentVersion表，插入更新时把title中的/替换为-
if ("ContentVersion".equals(one.getName()) && "Title".equals(key)){

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

@ -1998,7 +1998,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
return salesforceParam;
}).collect(Collectors.toList());
}
Set<String> safeSet = ConcurrentHashMap.newKeySet();
Set<String> safeSet = new HashSet<>();
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
@ -2122,12 +2122,13 @@ public class DataImportNewServiceImpl implements DataImportNewService {
DataObject::getName,
(existing, replacement) -> replacement));
//邮件发送前三位编码不存在的对象
Set<String> safeSet = ConcurrentHashMap.newKeySet();
for (String api : apis) {
//邮件发送前三位编码不存在的对象
Set<String> safeSet = new HashSet<>();
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
dbQw.eq("api", api).eq("is_link",1).eq("is_create",1);
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
if (linkConfigs.isEmpty()){
continue;
@ -2201,7 +2202,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
// 表内数据总量
Integer count = customMapper.countBySQL(api, sql1);
log.info("表api：{} 存在" +count+ "条数据！", param.getApi());
log.info("表api：{} 存在" +count+ "条数据！", api);
if (count >0 ) {
int page = count % 10000 == 0 ? count / 10000 : (count / 10000) + 1;
@ -2789,19 +2790,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
.orderByAsc("data_index")
.last(" limit 10");
// 全量的时候 检测是否有自动任务锁住的表
boolean isFull = CollectionUtils.isEmpty(param.getIds());
if (isFull) {
QueryWrapper<DataObject> dw = new QueryWrapper<>();
qw.eq("data_lock", 1);
List<DataObject> dos = dataObjectService.list(dw);
if (CollectionUtils.isNotEmpty(dos)) {
String apiNames = dos.stream().map(DataObject::getName).collect(Collectors.joining());
log.info("api:" + apiNames + " is locked");
return new ReturnT<>(500, "api:" + apiNames + " is locked");
}
}
PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
while (true) {
List<DataObject> dataObjects = dataObjectService.list(qw);
@ -2819,7 +2807,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
dataObjectService.updateById(update);
if (!dataFieldService.hasCreatedDate(api)){
insertSingleShareData(api,partnerConnection,update);
insertSingleShareData(api,partnerConnection,dataObject);
update.setNeedUpdate(false);
dataObjectService.updateById(update);
continue;
}
@ -2857,7 +2847,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
update.setDataWork(0);
update.setNeedUpdate(false);
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
@ -3066,6 +3056,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e));
}
}
if (z*10000+size >= count){
break;
}
}
SalesforceParam countParam = new SalesforceParam();
@ -3278,9 +3271,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}catch (InterruptedException e){
return;
}catch (Exception e) {
log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e));
log.error("insertSingle error api:{}", api, e);
}
}
if (z*10000+size >= count){
break;
}
}
}
}

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

@ -438,6 +438,10 @@ public class DataImportServiceImpl implements DataImportService {
log.info(JSON.toJSONString(e));
}
}
if (z*10000+size >= count){
break;
}
}
SalesforceParam countParam = new SalesforceParam();

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

@ -138,7 +138,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
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"));
head.add(Lists.newArrayList("对象API","对象Label","源ORG数据量","本地数据量","本地存在new_id数据量","目标ORG数据量","目标存在old_id数据量","本地重复new_id","目标重复old_id","源VS本地是否一致","目标VS本地是否一致","源VS目标是否一致","比对结果"));
excelWriter.write(head, writeSheet);
WriteSheet writeSheet1 = EasyExcel.writerSheet("本地new_id重复").build();
@ -177,14 +177,14 @@ public class DataVerifyServiceImpl implements DataVerifyService {
}
QueryResult queryResult = null;
Integer sourceNum = 0;
int sourceNum = 0;
int retryCount = 0;
while (retryCount < 3) {
try {
// 查询源ORG数据量
queryResult = connect.queryAll(sql);
SObject record = queryResult.getRecords()[0];
sourceNum = (Integer) record.getField("num");
sourceNum = (int) record.getField("num");
log.info("api :{} 源ORG num: {}，执行sql：{}", api, sourceNum, sql);
break;
} catch (Exception e) {
@ -195,8 +195,8 @@ public class DataVerifyServiceImpl implements DataVerifyService {
}
}
}
Integer localNum = -1;
Integer localNewIdNum = -1;
int localNum = -1;
int localNewIdNum = -1;
List<Map<String, Object>> mapList = new ArrayList<>();
@ -232,13 +232,13 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 查询目标ORG数据量
QueryResult targetQueryResult = null;
Integer targetNum = 0;
int targetNum = 0;
int targetRetryCount = 0;
while (targetRetryCount < 3) {
try {
targetQueryResult = targetConnect.queryAll(sql);
SObject targetRecord = targetQueryResult.getRecords()[0];
targetNum = (Integer) targetRecord.getField("num");
targetNum = (int) targetRecord.getField("num");
log.info("api :{} 目标ORG num: {}，执行sql：{}", api,sql, targetNum);
break;
} catch (Exception e) {
@ -252,14 +252,13 @@ public class DataVerifyServiceImpl implements DataVerifyService {
QueryResult targetQueryOldIdResult = null;
int oldIdRetryCount = 0;
Integer targetOldIdNum = 0;
int targetOldIdNum = 0;
if (dataObject.getIsEditable()){
while (oldIdRetryCount < 3) {
try {
targetQueryOldIdResult = targetConnect.queryAll(oldIdSql);
SObject targetOldIdRecord = targetQueryOldIdResult.getRecords()[0];
targetOldIdNum = (Integer) targetOldIdRecord.getField("num");
targetOldIdNum = (int) targetOldIdRecord.getField("num");
log.info("api :{} 目标ORG存在old_id数据量: {}，执行sql：{}", api, targetOldIdNum, oldIdSql);
break;
} catch (Exception e) {
@ -316,13 +315,8 @@ public class DataVerifyServiceImpl implements DataVerifyService {
} catch (Exception e) {
oldIdRepuRetryCount++;
log.info("查询目标ORG存在old_id重复数据失败，第{}次重试，执行sql：{}", oldIdRepuRetryCount, duplicateIdSql, e);
if (oldIdRepuRetryCount >= 3) {
targetOldIdNum = -1;
}
}
}
}else {
targetOldIdNum = -1;
}
// 写入一行数据
@ -335,9 +329,19 @@ public class DataVerifyServiceImpl implements DataVerifyService {
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()));
boolean equals = sourceNum == localNum;
boolean equals1 = targetNum == localNum;
boolean equals2 = sourceNum == targetNum;
rowData.add(equals?"是":"否");
rowData.add(equals1?"是":"否");
rowData.add(equals2?"是":"否");
rowData.add((equals && equals1 && equals2)?"一致":"");
data.add(rowData);
excelWriter.write(data, writeSheet);
}

19
src/main/java/com/celnet/datadump/util/DataUtil.java
View File

@ -494,7 +494,7 @@ public class DataUtil {
public static String adjustHour(String timeStr) {
// 提取小时部分并转换为整数
int hour = Integer.parseInt(timeStr.substring(0, 2));
int hour = Integer.parseInt(timeStr.substring(0, 2));
// 减去8小时并处理跨天逻辑
int adjustedHour = hour - 8;
@ -503,12 +503,23 @@ public class DataUtil {
}
// 格式化为两位字符串（自动补零）
String newHour = String.format("%02d", adjustedHour);
String newHour = String.format("%02d", adjustedHour);
// 拼接原始字符串的剩余部分（分钟+秒+毫秒）
return newHour + timeStr.substring(2);
// 拼接原始字符串的剩余部分（分钟+秒）
String timeWithoutMilliseconds = newHour + timeStr.substring(2);
// 添加毫秒部分和Z后缀
// 如果原始字符串不包含毫秒，则添加.000
if (timeStr.length() <= 8) { // HH:mm:ss 格式，长度为8
return timeWithoutMilliseconds + ".000Z";
} else if (timeStr.contains(".")) { // 已经包含毫秒
return timeWithoutMilliseconds + "Z";
} else {
return timeWithoutMilliseconds + ".000Z";
}
}
public static boolean isUpdate(String field){
switch (field) {
case "LastModifiedDate":