kris/datai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 初步实现数据库同步到本地

Browse Source
This commit is contained in:
Kris 2026-01-06 18:54:13 +08:00
parent a78da492c7
commit cb09602386
14 changed files with 1161 additions and 746 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/constant/SalesforceConstants.java
View File

@ -330,4 +330,11 @@ public final class SalesforceConstants {
* 格式示例: 2024-01-01
*/
public static final String SF_DATE_ONLY_FORMAT = "yyyy-MM-dd";
public static final String SF_DATE_FORMAT_BULK = "yyyy-MM-dd'T'HH:mm:ss.SSSX";
/**
* sf 时间格式
*/
public static final String SF_TIME_FORMAT = "HH:mm:ss.SSS'Z'";
}

35
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationObject.java
View File

@ -118,6 +118,11 @@ public class DataiIntegrationObject extends BaseEntity
@Excel(name = "批次字段")
private String batchField;
/** 是否分区 */
@Schema(title = "是否分区")
@Excel(name = "是否分区")
private Boolean isPartitioned;
/** 本地记录数 */
@Schema(title = "本地记录数")
@Excel(name = "本地记录数")
@ -128,10 +133,10 @@ public class DataiIntegrationObject extends BaseEntity
@Excel(name = "最近同步时间")
private LocalDateTime lastSyncDate;
/** 全量同步时间 */
@Schema(title = "全量同步时间")
@Excel(name = "全量同步时间")
private LocalDateTime lastFullSyncDate;
/** 最后批次时间 */
@Schema(title = "最后批次时间")
@Excel(name = "最后批次时间")
private LocalDateTime lastBatchDate;
/** 状态 */
@Schema(title = "状态")
@ -362,6 +367,17 @@ public class DataiIntegrationObject extends BaseEntity
}
public void setIsPartitioned(Boolean isPartitioned)
{
this.isPartitioned = isPartitioned;
}
public Boolean getIsPartitioned()
{
return isPartitioned;
}
public void setTotalRows(Integer totalRows)
{
this.totalRows = totalRows;
@ -384,14 +400,14 @@ public class DataiIntegrationObject extends BaseEntity
}
public void setLastFullSyncDate(LocalDateTime lastFullSyncDate)
public void setLastBatchDate(LocalDateTime lastBatchDate)
{
this.lastFullSyncDate = lastFullSyncDate;
this.lastBatchDate = lastBatchDate;
}
public LocalDateTime getLastFullSyncDate()
public LocalDateTime getLastBatchDate()
{
return lastFullSyncDate;
return lastBatchDate;
}
@ -441,9 +457,10 @@ public class DataiIntegrationObject extends BaseEntity
.append("isIncremental", getIsIncremental())
.append("objectIndex", getObjectIndex())
.append("batchField", getBatchField())
.append("isPartitioned", getIsPartitioned())
.append("totalRows", getTotalRows())
.append("lastSyncDate", getLastSyncDate())
.append("lastFullSyncDate", getLastFullSyncDate())
.append("lastBatchDate", getLastBatchDate())
.append("syncStatus", getSyncStatus())
.append("errorMessage", getErrorMessage())
.append("remark", getRemark())

7
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/dto/DataiIntegrationObjectDto.java
View File

@ -84,14 +84,17 @@ public class DataiIntegrationObjectDto implements Serializable
/** 批次字段 */
private String batchField;
/** 是否分区 */
private Boolean isPartitioned;
/** 本地记录数 */
private Integer totalRows;
/** 最近同步时间 */
private LocalDateTime lastSyncDate;
/** 全量同步时间 */
private LocalDateTime lastFullSyncDate;
/** 最后批次时间 */
private LocalDateTime lastBatchDate;
/** 状态 */
private Boolean syncStatus;

47
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java
View File

@ -9,7 +9,7 @@ import java.util.Date;
import java.util.Set;
/**
* @author Radian
* @author Kris
* @description
* @date 2022/11/30
*/
@ -33,16 +33,45 @@ public class DataiSyncParam {
*/
private Date endDate;
/**
* 同步类型 1：存量 2：增量
* 开始创建时间
*/
private Integer type;
private Date beginCreateDate;
/**
* fastjson 过滤器 只打印当前类所有属性
* @return SimplePropertyPreFilter
* 结束创建时间
*/
public static SimplePropertyPreFilter getFilter() {
return new SimplePropertyPreFilter(DataiSyncParam.class, Arrays.stream(DataiSyncParam.class.getDeclaredFields()).map(Field::getName).toArray(String[]::new));
}
private Date endCreateDate;
/**
* 开始修改时间
*/
private Date beginModifyDate;
/**
* 结束修改时间
*/
private Date endModifyDate;
/**
* 最大Id
*/
private String maxId;
/**
* id字段
*/
private String idField = "Id";
/**
* 时间字段
*/
private String batchField = "CreatedDate";
/**
* 查询限制数量
*/
private Integer limit;
/**
* 查询字段 多个英文逗号分割
*/
private String select;
/**
* 是否查询删除记录
*/
private Boolean isDeleted;
}

7
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/vo/DataiIntegrationObjectVo.java
View File

@ -79,14 +79,17 @@ public class DataiIntegrationObjectVo implements Serializable {
/** 批次字段 */
private String batchField;
/** 是否分区 */
private Boolean isPartitioned;
/** 本地记录数 */
private Integer totalRows;
/** 最近同步时间 */
private LocalDateTime lastSyncDate;
/** 全量同步时间 */
private LocalDateTime lastFullSyncDate;
/** 最后批次时间 */
private LocalDateTime lastBatchDate;
/** 状态 */
private Boolean syncStatus;

397
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java
View File

@ -1,17 +1,18 @@
package com.datai.integration.service.impl;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.*;
import java.util.function.Supplier;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.datai.common.utils.DateUtils;
import com.datai.common.utils.SecurityUtils;
import com.datai.integration.model.param.DataiSyncParam;
import com.datai.salesforce.common.utils.SoqlBuilder;
import com.datai.integration.util.ConvertUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@ -356,14 +357,14 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
}
PartnerConnection connection = retryOperation(() -> soapConnectionFactory.getConnection(), 3, 1000);
log.info("成功获取Salesforce SOAP连接");
List<String> fieldList = getSalesforceObjectFields(connection, objectApi);
log.info("获取到对象 {} 的字段列表: {}", objectApi, fieldList);
SalesforceParam param = new SalesforceParam();
DataiSyncParam param = new DataiSyncParam();
param.setApi(objectApi);
param.setSelect(String.join(",", fieldList));
param.setBatchField(batch.getBatchField());
if (integrationFieldService.isDeletedFieldExists(objectApi)) {
param.setIsDeleted(true);
@ -391,7 +392,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory);
log.info("批次 {} 历史记录已保存，同步数据量: {}, 耗时: {}ms", batchId, totalCount, duration);
insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), true, null, duration);
insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), Integer.parseInt(batchId), true, null, duration);
return true;
} catch (Exception e) {
@ -416,16 +417,17 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory);
log.info("批次 {} 失败历史记录已保存，耗时: {}ms", batchId, duration);
insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", false, e.getMessage(), duration);
insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", Integer.parseInt(batchId), false, e.getMessage(), duration);
return false;
}
}
private void insertSyncLog(String objectApi, String label, String syncType, boolean success, String errorMessage, long duration) {
private void insertSyncLog(String objectApi, String label, String syncType, Integer batchId, boolean success, String errorMessage, long duration) {
try {
com.datai.integration.model.domain.DataiIntegrationSyncLog syncLog = new com.datai.integration.model.domain.DataiIntegrationSyncLog();
syncLog.setObjectApi(objectApi);
syncLog.setBatchId(batchId);
syncLog.setOperationType("FULL".equals(syncType) ? "全量同步" : "增量同步");
syncLog.setOperationStatus(success ? "成功" : "失败");
syncLog.setErrorMessage(errorMessage);
@ -519,17 +521,38 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
* @param fieldList 字段列表
* @return 处理的数据条数
*/
private int executeQueryAndProcessData(PartnerConnection connection, SalesforceParam param, List<String> fieldList) {
private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List<String> fieldList) {
int totalCount = 0;
try {
String query = buildDynamicQuery(param);
log.info("执行查询SQL: {}", query);
JSONArray objects = null;
String maxId = null;
Date lastDate = null;
QueryResult result = connection.queryAll(query);
try {
DescribeSObjectResult describeSObject = connection.describeSObject(param.getApi());
Field[] dsrFields = describeSObject.getFields();
while (true) {
// 获取创建时间
param.setBeginDate(lastDate);
// 判断是否存在要排除的id
param.setMaxId(maxId);
String query = buildDynamicQuery(param);
log.info("执行查询SQL: {}", query);
QueryResult result = connection.queryAll(query);
if (result.getRecords() != null && result.getRecords().length > 0) {
totalCount += processQueryResult(param.getApi(), result, fieldList);
objects = ConvertUtil.toJsonArray(result.getRecords(), dsrFields);
Date maxDate = objects.getJSONObject(objects.size() - 1).getDate(param.getBatchField());
// 在当前批次中找出所有具有相同时间戳的记录，并获取其中ID最大的那条记录的ID
// 这个ID将作为下一批次查询的起始位置，避免重复处理相同时间戳的数据
Optional<String> maxIdOptional = objects.stream()
.map(t -> (JSONObject) t)
.filter(t -> maxDate.equals(t.getDate(param.getBatchField())))
.map(t -> t.getString("Id"))
.max(String::compareTo);
maxId = maxIdOptional.orElse(null);
lastDate = maxDate;
totalCount += processQueryResult(param.getApi(), result.getRecords(), objects);
log.info("已处理 {} 条记录", totalCount);
}
@ -556,35 +579,65 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
* @param param 查询参数
* @return SOQL查询语句
*/
private String buildDynamicQuery(SalesforceParam param) {
StringBuilder queryBuilder = new StringBuilder();
queryBuilder.append("SELECT ").append(param.getSelect())
.append(" FROM ").append(param.getApi());
private String buildDynamicQuery(DataiSyncParam param) {
SoqlBuilder builder = new SoqlBuilder();
if (param.getIsDeleted() != null && param.getIsDeleted()) {
queryBuilder.append(" WHERE IsDeleted = true");
String[] selectFields = param.getSelect().split(",");
builder.select(selectFields)
.from(param.getApi());
if (param.getIsDeleted() != null && !param.getIsDeleted()) {
builder.where("IsDeleted = false");
}
return queryBuilder.toString();
if (param.getBeginDate() != null && param.getEndDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate())
.whereLe(param.getBatchField(), param.getEndDate());
} else if (param.getBeginDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate());
} else if (param.getEndDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndDate());
}
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginCreateDate())
.whereLe(param.getBatchField(), param.getEndCreateDate());
} else if (param.getBeginCreateDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginCreateDate());
} else if (param.getEndCreateDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndCreateDate());
}
if (param.getBeginModifyDate() != null && param.getEndModifyDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginModifyDate())
.whereLe(param.getBatchField(), param.getEndModifyDate());
} else if (param.getBeginModifyDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginModifyDate());
} else if (param.getEndModifyDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndModifyDate());
}
if (StringUtils.isNotEmpty(param.getMaxId())) {
builder.whereGt(param.getIdField(), param.getMaxId());
}
if (param.getLimit() != null && param.getLimit() > 0) {
builder.limit(param.getLimit());
}
return builder.build();
}
/**
* 处理查询结果
*
* @param api API名称
* @param result 查询结果
* @param fieldList 字段列表
* @return 处理的数据条数
*/
private int processQueryResult(String api, QueryResult result, List<String> fieldList) {
private int processQueryResult(String api, SObject[] records, JSONArray objects) {
int count = 0;
if (result == null || result.getRecords() == null) {
log.info("处理API {} 的查询结果，共 0 条记录", api);
return count;
}
SObject[] records = result.getRecords();
log.info("处理API {} 的查询结果，共 {} 条记录", api, records.length);
boolean isPartitioned = checkIfTablePartitioned(api);
@ -593,135 +646,95 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchField = integrationFieldService.getDateField(api);
}
Map<String, List<Map<String, Object>>> partitionedData = new HashMap<>();
List<Map<String, Object>> normalData = new ArrayList<>();
for (SObject record : records) {
if (record == null) {
continue;
}
try {
Map<String, Object> recordMap = convertSObjectToMap(record, fieldList);
if (isPartitioned) {
String partitionName = "p_default";
if (StringUtils.isNotEmpty(batchField) && recordMap.containsKey(batchField.toLowerCase())) {
Object dateValue = recordMap.get(batchField.toLowerCase());
if (dateValue instanceof Date) {
Calendar calendar = Calendar.getInstance();
calendar.setTime((Date) dateValue);
int year = calendar.get(Calendar.YEAR);
partitionName = "p" + year;
}
}
partitionedData.computeIfAbsent(partitionName, k -> new ArrayList<>()).add(recordMap);
} else {
normalData.add(recordMap);
}
count++;
} catch (Exception e) {
String recordId = record.getId() != null ? record.getId() : "未知";
log.error("处理记录时发生异常，记录ID: {}", recordId, e);
}
if (objects == null || objects.isEmpty()) {
log.info("API {} 没有需要处理的数据", api);
return 0;
}
try {
if (isPartitioned) {
for (Map.Entry<String, List<Map<String, Object>>> entry : partitionedData.entrySet()) {
String partitionName = entry.getKey();
List<Map<String, Object>> dataList = entry.getValue();
if (!dataList.isEmpty()) {
if (!dataList.isEmpty()) {
Collection<String> keys = dataList.get(0).keySet();
Collection<Collection<Object>> values = new ArrayList<>();
for (Map<String, Object> recordMap : dataList) {
Collection<Object> recordValues = new ArrayList<>();
for (String key : keys) {
recordValues.add(recordMap.get(key));
}
values.add(recordValues);
}
customMapper.saveBatchToPartition(api.toLowerCase(), partitionName, keys, values);
log.info("批量插入分区 {} 数据 {} 条", partitionName, dataList.size());
}
}
List<String> ids = Arrays.stream(records).map(SObject::getId).collect(Collectors.toList());
List<String> existsIds = customMapper.getIds(api, ids);
String partitionName = null;
if (isPartitioned && batchField != null) {
partitionName = buildPartitionName(batchField, objects);
}
List<Map<String, Object>> insertMaps = new ArrayList<>();
List<Map<String, Object>> updateMaps = new ArrayList<>();
List<String> updateIds = new ArrayList<>();
for (int i = 0; i < objects.size(); i++) {
JSONObject jsonObject = objects.getJSONObject(i);
String id = jsonObject.getString("Id");
Map<String, Object> recordMap = new HashMap<>();
for (String key : jsonObject.keySet()) {
recordMap.put(key, jsonObject.get(key));
}
} else {
if (!normalData.isEmpty()) {
Collection<String> keys = normalData.get(0).keySet();
Collection<Collection<Object>> values = new ArrayList<>();
for (Map<String, Object> recordMap : normalData) {
Collection<Object> recordValues = new ArrayList<>();
for (String key : keys) {
recordValues.add(recordMap.get(key));
}
values.add(recordValues);
}
customMapper.saveBatch(api.toLowerCase(), keys, values);
log.info("批量插入普通表数据 {} 条", normalData.size());
if (existsIds.contains(id)) {
updateMaps.add(recordMap);
updateIds.add(id);
} else {
insertMaps.add(recordMap);
}
}
if (!insertMaps.isEmpty()) {
List<String> keys = new ArrayList<>(insertMaps.get(0).keySet());
List<Collection<Object>> values = new ArrayList<>();
for (Map<String, Object> recordMap : insertMaps) {
List<Object> rowValues = new ArrayList<>();
for (String key : keys) {
rowValues.add(recordMap.get(key));
}
values.add(rowValues);
}
if (isPartitioned && partitionName != null) {
customMapper.saveBatchToPartition(api, partitionName, keys, values);
log.info("成功插入 {} 条记录到分区表 {} 的分区 {}", values.size(), api, partitionName);
} else {
customMapper.saveBatch(api, keys, values);
log.info("成功插入 {} 条记录到表 {}", values.size(), api);
}
count += values.size();
}
if (!updateMaps.isEmpty()) {
for (int i = 0; i < updateMaps.size(); i++) {
Map<String, Object> recordMap = updateMaps.get(i);
String id = updateIds.get(i);
List<Map<String, Object>> maps = new ArrayList<>();
for (Map.Entry<String, Object> entry : recordMap.entrySet()) {
Map<String, Object> paramMap = new HashMap<>();
paramMap.put("key", entry.getKey());
paramMap.put("value", entry.getValue());
maps.add(paramMap);
}
if (isPartitioned && partitionName != null) {
customMapper.updateByIdFromPartition(api, partitionName, maps, id);
} else {
customMapper.updateById(api, maps, id);
}
}
log.info("成功更新 {} 条记录到表 {}", updateMaps.size(), api);
count += updateMaps.size();
}
} catch (Exception e) {
log.error("批量处理数据时发生异常，API: {}", api, e);
fallbackToSingleRecordProcessing(api, records, fieldList, isPartitioned, batchField);
log.error("处理API {} 的查询结果时发生异常", api, e);
throw new RuntimeException("处理查询结果失败: " + e.getMessage(), e);
}
return count;
}
/**
* 批量处理失败时的回退方案，使用单条记录处理
*
* @param api API名称
* @param records 记录列表
* @param fieldList 字段列表
* @param isPartitioned 是否为分区表
* @param batchField 批次字段
*/
private void fallbackToSingleRecordProcessing(String api, SObject[] records, List<String> fieldList, boolean isPartitioned, String batchField) {
log.warn("批量处理数据失败，将回退到单条记录处理，API: {}", api);
if (records == null || records.length == 0) {
log.info("回退处理API {} 的记录，共 0 条记录", api);
return;
}
for (SObject record : records) {
if (record == null) {
continue;
}
try {
Map<String, Object> recordMap = convertSObjectToMap(record, fieldList);
if (isPartitioned) {
String partitionName = "p_default";
if (StringUtils.isNotEmpty(batchField) && recordMap.containsKey(batchField.toLowerCase())) {
Object dateValue = recordMap.get(batchField.toLowerCase());
if (dateValue instanceof Date) {
Calendar calendar = Calendar.getInstance();
calendar.setTime((Date) dateValue);
int year = calendar.get(Calendar.YEAR);
partitionName = "p" + year;
}
}
customMapper.upsertToPartition(api.toLowerCase(), partitionName, recordMap);
} else {
customMapper.upsert(api.toLowerCase(), recordMap);
}
} catch (Exception e) {
String recordId = record.getId() != null ? record.getId() : "未知";
log.error("单条处理记录时发生异常，记录ID: {}", recordId, e);
}
}
}
/**
* 检查表是否已分区
@ -735,73 +748,27 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
}
/**
* 将SObject转换为Map
* 根据批次字段和数据构建分区名
*
* @param record SObject记录
* @param fieldList 字段列表
* @return 转换后的Map
* @param batchField 批次字段名
* @param objects 数据对象数组
* @return 分区名（如p2025）
*/
private Map<String, Object> convertSObjectToMap(SObject record, List<String> fieldList) {
Map<String, Object> recordMap = new HashMap<>();
if (record.getId() != null) {
recordMap.put("Id", record.getId());
}
for (String field : fieldList) {
if ("Id".equalsIgnoreCase(field)) {
continue;
}
try {
Object value = record.getField(field);
if (value != null) {
if (value instanceof java.util.Calendar) {
recordMap.put(field.toLowerCase(), ((Calendar) value).getTime());
} else {
recordMap.put(field.toLowerCase(), value);
}
} else {
recordMap.put(field.toLowerCase(), null);
}
} catch (Exception e) {
log.warn("获取字段 {} 的值时发生异常", field, e);
private String buildPartitionName(String batchField, JSONArray objects) {
String partitionName = "p_default";
if (StringUtils.isNotEmpty(batchField) && !objects.isEmpty()) {
JSONObject firstRecord = objects.getJSONObject(0);
Object dateValue = firstRecord.get(batchField);
if (dateValue instanceof Date) {
Calendar calendar = Calendar.getInstance();
calendar.setTime((Date) dateValue);
int year = calendar.get(Calendar.YEAR);
partitionName = "p" + year;
}
}
return recordMap;
}
/**
* Salesforce查询参数内部类
*/
private static class SalesforceParam {
private String api;
private String select;
private Boolean isDeleted;
public String getApi() {
return api;
}
public void setApi(String api) {
this.api = api;
}
public String getSelect() {
return select;
}
public void setSelect(String select) {
this.select = select;
}
public Boolean getIsDeleted() {
return isDeleted;
}
public void setIsDeleted(Boolean isDeleted) {
this.isDeleted = isDeleted;
}
return partitionName;
}
}

75
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationMetadataChangeServiceImpl.java
View File

@ -11,6 +11,7 @@ import com.datai.common.utils.CacheUtils;
import com.datai.integration.factory.impl.SOAPConnectionFactory;
import com.datai.integration.mapper.CustomMapper;
import com.datai.integration.model.domain.DataiIntegrationBatch;
import com.datai.integration.util.ConvertUtil;
import com.datai.salesforce.common.utils.SoqlBuilder;
import com.sforce.soap.partner.*;
import com.sforce.soap.partner.sobject.SObject;
@ -588,7 +589,7 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
for (com.sforce.soap.partner.Field field : objDetail.getFields()) {
Map<String, Object> fieldDef = new HashMap<>();
fieldDef.put("name", field.getName());
fieldDef.put("type", convertSalesforceTypeToMySQL(field.getType() != null ? field.getType().toString() : null));
fieldDef.put("type", ConvertUtil.fieldTypeToMysql(field));
fieldDef.put("comment", field.getLabel() != null ? field.getLabel().replaceAll("'", "\\\\'") : "");
fieldDefinitions.add(fieldDef);
fields.add(field.getName());
@ -819,7 +820,7 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
for (com.sforce.soap.partner.Field field : objDetail.getFields()) {
if (fieldApi.equals(field.getName())) {
String mysqlType = convertSalesforceTypeToMySQL(field.getType() != null ? field.getType().toString() : null);
String mysqlType = ConvertUtil.fieldTypeToMysql(field);
customMapper.addField(objectApi, fieldApi, mysqlType, field.isNillable());
log.info("成功添加字段: {}.{} 类型: {}", objectApi, fieldApi, mysqlType);
@ -855,7 +856,7 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
for (com.sforce.soap.partner.Field field : objDetail.getFields()) {
if (fieldApi.equals(field.getName())) {
String mysqlType = convertSalesforceTypeToMySQL(field.getType() != null ? field.getType().toString() : null);
String mysqlType = ConvertUtil.fieldTypeToMysql(field);
customMapper.modifyField(objectApi, fieldApi, mysqlType, field.isNillable());
log.info("成功修改字段: {}.{} 类型: {}", objectApi, fieldApi, mysqlType);
@ -888,59 +889,6 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
}
}
private String convertSalesforceTypeToMySQL(String typeStr) {
if (typeStr == null || typeStr.isEmpty()) {
return "VARCHAR(255)";
}
switch (typeStr) {
case "id":
return "VARCHAR(18)";
case "string":
case "email":
case "url":
case "phone":
return "VARCHAR(255)";
case "textarea":
return "TEXT";
case "boolean":
return "TINYINT(1)";
case "int":
return "INT";
case "double":
return "DOUBLE";
case "currency":
return "DECIMAL(18,4)";
case "date":
return "DATE";
case "datetime":
return "DATETIME";
case "time":
return "TIME";
case "percent":
return "DECIMAL(10,2)";
case "reference":
return "VARCHAR(18)";
case "picklist":
case "multipicklist":
return "VARCHAR(255)";
case "combobox":
return "VARCHAR(255)";
case "base64":
return "LONGBLOB";
case "anyType":
return "VARCHAR(255)";
case "address":
return "TEXT";
case "location":
return "VARCHAR(255)";
case "encryptedstring":
return "VARCHAR(255)";
default:
return "VARCHAR(255)";
}
}
@Override
public Map<String, Object> pullAllMetadataChanges() {
// 初始化返回结果
@ -1961,18 +1909,20 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
}
String systemDataStartTimeStr = CacheUtils.get(salesforceConfigCacheManager.getEnvironmentCacheKey(), "salesforce.data.start.time", String.class);
java.time.LocalDate startDate = java.time.LocalDate.now().minusYears(5);
java.time.LocalDateTime startDate = java.time.LocalDateTime.now().minusYears(5);
startDate = startDate.withHour(0).withMinute(0).withSecond(0).withNano(0);
if (systemDataStartTimeStr != null && !systemDataStartTimeStr.trim().isEmpty()) {
try {
startDate = java.time.LocalDate.parse(systemDataStartTimeStr.trim());
java.time.LocalDate startDateLocalDate = java.time.LocalDate.parse(systemDataStartTimeStr.trim());
startDate = startDateLocalDate.atStartOfDay();
log.info("从缓存获取系统数据开始时间: {}", systemDataStartTimeStr);
} catch (Exception e) {
log.warn("解析系统数据开始时间配置失败: {}, 使用默认值", systemDataStartTimeStr, e);
}
}
java.time.LocalDate endDate = java.time.LocalDate.now();
java.time.LocalDateTime endDate = java.time.LocalDateTime.now();
DataiIntegrationBatch batchTemplate = new DataiIntegrationBatch();
batchTemplate.setApi(objectApi);
@ -1984,14 +1934,17 @@ public class DataiIntegrationMetadataChangeServiceImpl implements IDataiIntegrat
batchTemplate.setUpdateTime(DateUtils.getNowDate());
saveBatch(SalesforceConfigConstants.BATCH_TYPE_YEAR,
java.sql.Date.valueOf(startDate),
java.sql.Date.valueOf(endDate),
java.sql.Timestamp.valueOf(startDate),
java.sql.Timestamp.valueOf(endDate),
objectApi,
objectLabel,
batchField,
connection,
batchTemplate);
object.setLastBatchDate(endDate);
dataiIntegrationObjectService.updateDataiIntegrationObject(object);
log.info("对象 {} 批次创建完成", objectApi);
} catch (Exception e) {
log.error("为对象 {} 创建批次时出错: {}", objectApi, e.getMessage(), e);

422
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java
View File

@ -1,6 +1,5 @@
package com.datai.integration.service.impl;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
@ -13,10 +12,12 @@ import com.datai.integration.mapper.CustomMapper;
import com.datai.integration.mapper.DataiIntegrationObjectMapper;
import com.datai.integration.model.domain.DataiIntegrationBatch;
import com.datai.integration.model.domain.DataiIntegrationObject;
import com.datai.integration.model.param.DataiSyncParam;
import com.datai.integration.service.IDataiIntegrationBatchService;
import com.datai.integration.service.IDataiIntegrationFieldService;
import com.datai.integration.service.IDataiIntegrationObjectService;
import com.datai.salesforce.common.param.SalesforceParam;
import com.datai.salesforce.common.utils.SoqlBuilder;
import com.datai.integration.util.ConvertUtil;
import com.datai.setting.future.SalesforceExecutor;
import com.sforce.soap.partner.DescribeSObjectResult;
import com.sforce.soap.partner.Field;
@ -197,7 +198,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
data.put("lastSyncEndTime", syncStats.get("lastSyncEndTime"));
data.put("fullSyncCount", syncStats.get("fullSyncCount"));
data.put("incrementalSyncCount", syncStats.get("incrementalSyncCount"));
data.put("lastFullSyncDate", object.getLastFullSyncDate());
data.put("lastBatchDate", object.getLastBatchDate());
data.put("lastSyncDate", object.getLastSyncDate());
data.put("syncStatus", object.getSyncStatus());
data.put("errorMessage", object.getErrorMessage());
@ -585,9 +586,9 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
}
}
if (object.getLastFullSyncDate() != null) {
if (latestFullSyncDate == null || object.getLastFullSyncDate().isAfter(latestFullSyncDate)) {
latestFullSyncDate = object.getLastFullSyncDate();
if (object.getLastBatchDate() != null) {
if (latestFullSyncDate == null || object.getLastBatchDate().isAfter(latestFullSyncDate)) {
latestFullSyncDate = object.getLastBatchDate();
}
}
}
@ -634,7 +635,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
* 同步单个Salesforce对象的数据到本地数据库
*
* @param id 对象ID，用于标识需要同步的Salesforce对象
* @return Map<String, Object> 同步结果，包含success、message、objectId、objectApi、totalCount、duration、syncType、lastFullSyncDate等字段
* @return Map<String, Object> 同步结果，包含success、message、objectId、objectApi、totalCount、duration、syncType、lastBatchDate等字段
* @throws ConnectionException 当获取Salesforce连接失败时抛出
* @throws RuntimeException 当同步过程中发生其他异常时抛出
*/
@ -681,14 +682,14 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
log.info("准备同步对象数据，对象API: {}", objectApi);
LocalDateTime lastFullSyncDate = object.getLastFullSyncDate();
LocalDateTime lastSyncDate = object.getLastSyncDate();
Map<String, Object> syncResult;
if (lastFullSyncDate == null) {
log.info("对象 {} 的全量拉取时间为空，执行全量数据拉取", objectApi);
if (lastSyncDate == null) {
log.info("对象 {} 的最近同步时间为空，执行全量数据拉取", objectApi);
syncResult = syncFullData(object, startTime);
} else {
log.info("对象 {} 的全量拉取时间不为空（{}），执行增量数据拉取", objectApi, lastFullSyncDate);
log.info("对象 {} 的最近同步时间不为空（{}），执行增量数据拉取", objectApi, lastSyncDate);
syncResult = syncIncrementalData(object, startTime);
}
@ -798,7 +799,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
long duration = endTime - startTime;
LocalDateTime now = LocalDateTime.now();
object.setLastFullSyncDate(now);
object.setLastSyncDate(now);
object.setTotalRows(totalSyncCount);
object.setSyncStatus(failedBatchCount == 0);
@ -812,7 +812,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("totalCount", totalSyncCount);
result.put("duration", duration);
result.put("syncType", "full");
result.put("lastFullSyncDate", object.getLastFullSyncDate());
result.put("lastBatchDate", object.getLastBatchDate());
result.put("totalBatchCount", batches.size());
result.put("successBatchCount", successBatchCount);
result.put("failedBatchCount", failedBatchCount);
@ -840,7 +840,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
private Map<String, Object> syncIncrementalData(DataiIntegrationObject object, long startTime) {
Map<String, Object> result = new HashMap<>();
String objectApi = object.getApi().trim();
LocalDateTime lastFullSyncDate = object.getLastFullSyncDate();
LocalDateTime lastBatchDate = object.getLastBatchDate();
try {
if (!Boolean.TRUE.equals(object.getIsIncremental())) {
@ -879,7 +879,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
incrementalBatch.setBatchField(batchField);
incrementalBatch.setSyncType("INCREMENTAL");
incrementalBatch.setSyncStatus(false);
incrementalBatch.setSyncStartDate(lastFullSyncDate);
incrementalBatch.setSyncStartDate(lastBatchDate);
incrementalBatch.setSyncEndDate(LocalDateTime.now());
incrementalBatch.setCreateTime(DateUtils.getNowDate());
incrementalBatch.setUpdateTime(DateUtils.getNowDate());
@ -887,6 +887,9 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
int batchId = dataiIntegrationBatchService.insertDataiIntegrationBatch(incrementalBatch);
log.info("为对象 {} 创建增量批次，批次ID: {}", objectApi, batchId);
object.setLastBatchDate(incrementalBatch.getSyncEndDate());
updateDataiIntegrationObject(object);
Map<String, Object> batchResult = dataiIntegrationBatchService.syncBatchData(batchId);
long endTime = System.currentTimeMillis();
@ -912,7 +915,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("totalCount", syncNum);
result.put("duration", duration);
result.put("syncType", "incremental");
result.put("lastFullSyncDate", object.getLastFullSyncDate());
result.put("lastBatchDate", object.getLastBatchDate());
result.put("lastSyncDate", object.getLastSyncDate());
log.info("对象 {} 增量数据拉取成功，同步数据量: {}，耗时: {}ms", objectApi, syncNum, duration);
@ -960,149 +963,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
return fields;
}
private int executeQueryAndProcessData(PartnerConnection connection, SalesforceParam param, List<String> fieldList)
{
int totalCount = 0;
int failCount = 0;
int batch = 0;
Date lastCreatedDate = null;
String maxId = null;
final int MAX_FAIL_COUNT = 3;
boolean isFirstQuery = true;
try {
while (true) {
try {
String query = buildDynamicQuery(param, lastCreatedDate, maxId, isFirstQuery);
log.info("执行查询，批次: {}, SQL: {}", batch, query);
QueryResult queryResult = connection.query(query);
if (queryResult == null || queryResult.getRecords() == null || queryResult.getRecords().length == 0) {
log.info("批次 {} 查询完成，无更多数据", batch);
break;
}
int currentBatchCount = processQueryResult(param.getApi(), queryResult, fieldList);
totalCount += currentBatchCount;
SObject[] records = queryResult.getRecords();
lastCreatedDate = getLastCreatedDate(records, param.getDateField());
maxId = getMaxId(records, param.getDateField(), lastCreatedDate);
log.info("批次 {} 处理完成，本批次记录数: {}, 总记录数: {}, 最后时间: {}, 最大ID: {}",
batch, currentBatchCount, totalCount, lastCreatedDate, maxId);
failCount = 0;
batch++;
isFirstQuery = false;
}catch (Exception e) {
failCount++;
log.error("批次 {} 查询失败，失败次数: {}/{}，错误信息: {}",
batch, failCount, MAX_FAIL_COUNT, e.getMessage(), e);
if (failCount > MAX_FAIL_COUNT) {
log.error("达到最大失败次数 {}，停止查询", MAX_FAIL_COUNT);
throw new RuntimeException("查询失败次数过多: " + e.getMessage(), e);
}
}
}
log.info("查询完成，总批次: {}, 总记录数: {}", batch, totalCount);
} catch (Exception e) {
log.error("执行查询并处理数据失败: {}", e.getMessage(), e);
}
return totalCount;
}
/**
* 从查询结果中获取最后一条记录的日期字段值
*
* @param records Salesforce查询返回的记录数组
* @param dateField 日期字段名称，用于提取时间戳
* @return 最后一条记录的日期字段值，如果获取失败则返回null
*/
private Date getLastCreatedDate(SObject[] records, String dateField)
{
// 检查记录数组是否为空
if (records == null || records.length == 0) {
return null;
}
// 获取最后一条记录
SObject lastRecord = records[records.length - 1];
if (lastRecord == null) {
return null;
}
try {
// 提取指定的日期字段值
Object dateValue = lastRecord.getField(dateField);
if (dateValue instanceof Calendar) {
// 将Calendar类型转换为Date类型
return ((Calendar) dateValue).getTime();
}
} catch (Exception e) {
log.warn("获取最后创建时间失败: {}", e.getMessage());
}
return null;
}
/**
* 从查询结果中获取指定日期对应的最大ID值
*
* @param records Salesforce查询返回的记录数组
* @param dateField 日期字段名称，用于比较日期值
* @param lastCreatedDate 日期参考值，只查找该日期的记录
* @return 指定日期对应的最大ID值，如果没有匹配记录则返回null
*/
private String getMaxId(SObject[] records, String dateField, Date lastCreatedDate)
{
// 检查记录数组和日期值是否为空，避免空指针异常
if (records == null || records.length == 0 || lastCreatedDate == null) {
return null;
}
// 初始化最大ID变量，用于存储遍历过程中找到的最大ID值
String maxId = null;
// 遍历所有记录，查找日期字段值等于lastCreatedDate的记录
for (SObject record : records) {
// 跳过空记录，避免空指针异常
if (record == null) {
continue;
}
try {
// 提取记录的日期字段值
Object dateValue = record.getField(dateField);
if (dateValue instanceof Calendar) {
// 将Calendar类型转换为Date类型，便于比较
Date recordDate = ((Calendar) dateValue).getTime();
// 只处理与参考日期相同的记录
if (recordDate.equals(lastCreatedDate)) {
// 获取记录的ID
String recordId = record.getId();
// 更新最大ID：如果当前记录ID大于已知的最大ID，则更新
if (recordId != null && (maxId == null || recordId.compareTo(maxId) > 0)) {
maxId = recordId;
}
}
}
} catch (Exception e) {
// 字段获取失败时记录警告日志，继续处理其他记录
log.warn("获取记录ID失败: {}", e.getMessage());
}
}
// 返回指定日期对应的最大ID值，该值将作为下一批次查询的起始ID
return maxId;
}
/**
* 构建动态SOQL查询语句
@ -1113,247 +973,47 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
* @param isFirstQuery 是否为首次查询
* @return 构建好的SOQL查询语句
*/
private String buildDynamicQuery(SalesforceParam param, Date lastCreatedDate, String maxId, boolean isFirstQuery)
private String buildDynamicQuery(DataiSyncParam param, Date lastCreatedDate, String maxId, boolean isFirstQuery)
{
// 使用StringBuilder构建查询语句，提高字符串拼接效率
StringBuilder queryBuilder = new StringBuilder();
// 构建SELECT和FROM子句
queryBuilder.append("SELECT ").append(param.getSelect())
.append(" FROM ").append(param.getApi());
SoqlBuilder builder = new SoqlBuilder();
// 创建条件列表，用于存储所有WHERE条件
List<String> conditions = new ArrayList<>();
String[] selectFields = param.getSelect().split(",");
builder.select(selectFields)
.from(param.getApi());
// 如果配置了查询已删除记录，添加IsDeleted条件
if (param.getIsDeleted() != null && param.getIsDeleted()) {
conditions.add("IsDeleted = true");
builder.where("IsDeleted = true");
}
// 首次查询时，如果配置了开始修改日期，添加日期范围条件
if (isFirstQuery && param.getBeginModifyDate() != null) {
String dateField = param.getDateField();
String dateField = param.getBatchField();
if (StringUtils.isNotEmpty(dateField)) {
// 创建UTC时区的日期格式化器
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'.000Z'");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
// 格式化日期为Salesforce支持的格式
String dateStr = sdf.format(param.getBeginModifyDate());
// 添加日期大于等于条件
conditions.add(dateField + " >= " + dateStr);
builder.whereGe(dateField, param.getBeginModifyDate());
}
}
// 非首次查询时，使用最后创建日期作为查询条件
if (!isFirstQuery && lastCreatedDate != null) {
String dateField = param.getDateField();
String dateField = param.getBatchField();
if (StringUtils.isNotEmpty(dateField)) {
// 创建UTC时区的日期格式化器
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'.000Z'");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
// 格式化日期为Salesforce支持的格式
String dateStr = sdf.format(lastCreatedDate);
// 添加日期大于等于条件
conditions.add(dateField + " >= " + dateStr);
builder.whereGe(dateField, lastCreatedDate);
}
}
// 如果配置了最大ID，添加ID大于条件，实现精确分页
if (StringUtils.isNotEmpty(maxId)) {
conditions.add("Id > '" + maxId + "'");
builder.whereGt("Id", maxId);
}
// 如果有条件，构建WHERE子句
if (!conditions.isEmpty()) {
queryBuilder.append(" WHERE ").append(String.join(" AND ", conditions));
}
builder.orderBy(param.getBatchField(), SoqlBuilder.SortOrder.ASC)
.orderBy("Id", SoqlBuilder.SortOrder.ASC);
// 添加ORDER BY子句，按日期字段和ID排序，确保数据顺序一致
queryBuilder.append(" ORDER BY ").append(param.getDateField()).append(", Id");
// 如果配置了查询限制，添加LIMIT子句
if (param.getLimit() != null && param.getLimit() > 0) {
queryBuilder.append(" LIMIT ").append(param.getLimit());
builder.limit(param.getLimit());
}
// 返回构建好的SOQL查询语句
return queryBuilder.toString();
return builder.build();
}
/**
* 处理查询结果并将数据保存到数据库
*
* @param api Salesforce对象的API名称，用于标识目标表
* @param result Salesforce查询返回的结果集，包含所有需要同步的记录
* @param fieldList 需要同步的字段列表，用于控制数据转换和存储的字段范围
* @return 成功处理的记录数量
*/
private int processQueryResult(String api, QueryResult result, List<String> fieldList)
{
// 初始化计数器，用于统计成功处理的记录数量
int count = 0;
// 检查查询结果是否为空，避免空指针异常
if (result == null || result.getRecords() == null) {
log.info("处理API {} 的查询结果，共 0 条记录", api);
return count;
}
// 获取查询结果中的所有记录
SObject[] records = result.getRecords();
log.info("处理API {} 的查询结果，共 {} 条记录", api, records.length);
// 检查目标表是否为分区表，决定数据保存策略
boolean isPartitioned = checkIfTablePartitioned(api);
// 初始化批次字段变量，用于分区表的数据分配
String batchField = null;
// 如果是分区表，获取批次字段配置
if (isPartitioned) {
// 根据API名称查询对象配置
DataiIntegrationObject object = dataiIntegrationObjectMapper.selectDataiIntegrationObjectByApi(api);
if (object != null && StringUtils.isNotEmpty(object.getBatchField())) {
// 使用对象配置的批次字段
batchField = object.getBatchField();
log.info("对象 {} 配置了批次字段: {}", api, batchField);
} else {
// 如果对象未配置批次字段，使用日期字段作为默认批次字段
batchField = dataiIntegrationFieldService.getDateField(api);
log.info("对象 {} 未配置批次字段，使用日期字段: {}", api, batchField);
}
}
// 创建分区数据Map，用于存储按分区名分组的数据
Map<String, List<Map<String, Object>>> partitionedData = new HashMap<>();
// 创建普通数据列表，用于存储非分区表的数据
List<Map<String, Object>> normalData = new ArrayList<>();
// 遍历所有记录，进行数据转换和分区分配
for (SObject record : records) {
// 跳过空记录，避免空指针异常
if (record == null) {
continue;
}
try {
// 将SObject对象转换为Map格式，便于后续处理
Map<String, Object> recordMap = convertSObjectToMap(record, fieldList);
// 如果是分区表且配置了批次字段，进行分区分配
if (isPartitioned && StringUtils.isNotEmpty(batchField)) {
// 初始化分区名称为默认分区
String partitionName = "p_default";
// 检查记录中是否包含批次字段
if (recordMap.containsKey(batchField.toLowerCase())) {
// 获取批次字段的值
Object batchValue = recordMap.get(batchField.toLowerCase());
if (batchValue instanceof Date) {
// 如果批次字段值为Date类型，使用年份作为分区标识（如p2025）
Calendar calendar = Calendar.getInstance();
calendar.setTime((Date) batchValue);
int year = calendar.get(Calendar.YEAR);
partitionName = "p" + year;
log.debug("记录根据批次字段 {} 的值 {} 分配到分区 {}", batchField, batchValue, partitionName);
} else if (batchValue != null) {
// 如果批次字段值为其他类型，使用哈希值作为分区标识
partitionName = "p_" + String.valueOf(batchValue).hashCode();
log.debug("记录根据批次字段 {} 的值 {} 分配到分区 {}", batchField, batchValue, partitionName);
}
}
// 将记录添加到对应分区的数据列表中
partitionedData.computeIfAbsent(partitionName, k -> new ArrayList<>()).add(recordMap);
} else {
// 非分区表，直接添加到普通数据列表
normalData.add(recordMap);
}
// 增加处理计数
count++;
} catch (Exception e) {
// 记录处理异常，但继续处理其他记录
String recordId = record.getId() != null ? record.getId() : "未知";
log.error("处理记录时发生异常，记录ID: {}", recordId, e);
}
}
try {
// 如果是分区表且有分区数据，批量保存到各个分区
if (isPartitioned && !partitionedData.isEmpty()) {
log.info("开始批量保存分区数据，共 {} 个分区", partitionedData.size());
// 遍历所有分区，保存数据
for (Map.Entry<String, List<Map<String, Object>>> entry : partitionedData.entrySet()) {
String partitionName = entry.getKey();
List<Map<String, Object>> dataList = entry.getValue();
// 保存数据到指定分区
saveDataToPartition(api, partitionName, dataList);
log.info("成功保存 {} 条记录到分区 {}", dataList.size(), partitionName);
}
} else if (!normalData.isEmpty()) {
// 非分区表，直接保存数据到目标表
saveDataToTable(api, normalData);
log.info("成功保存 {} 条记录到表 {}", normalData.size(), api);
}
} catch (Exception e) {
// 记录批量保存失败的异常
log.error("批量保存数据到数据库失败: {}", e.getMessage(), e);
}
// 返回成功处理的记录数量
return count;
}
/**
* 将Salesforce的SObject对象转换为Map格式
*
* @param record Salesforce的SObject对象，包含从Salesforce查询返回的单条记录数据
* @param fieldList 需要提取的字段列表，控制转换过程中包含哪些字段
* @return 包含所有指定字段值的Map对象，字段名统一为小写格式
*/
private Map<String, Object> convertSObjectToMap(SObject record, List<String> fieldList)
{
// 创建Map对象用于存储转换后的记录数据
Map<String, Object> recordMap = new HashMap<>();
// 添加记录ID字段，Id字段总是被包含在结果中
if (record.getId() != null) {
recordMap.put("Id", record.getId());
}
// 遍历所有需要提取的字段
for (String field : fieldList) {
// 跳过Id字段，因为已经单独处理过
if ("Id".equalsIgnoreCase(field)) {
continue;
}
try {
// 从SObject中获取字段值
Object value = record.getField(field);
if (value != null) {
// 检查字段值是否为Calendar类型（日期类型）
if (value instanceof java.util.Calendar) {
// 将Calendar类型转换为Date类型，便于数据库存储
recordMap.put(field.toLowerCase(), ((Calendar) value).getTime());
} else {
// 其他类型直接存储，字段名转换为小写以适应数据库命名规范
recordMap.put(field.toLowerCase(), value);
}
} else {
// 空值字段也保留在结果中，值为null
recordMap.put(field.toLowerCase(), null);
}
} catch (Exception e) {
// 字段获取异常时记录警告日志，但不中断处理流程
log.warn("获取字段 {} 的值时发生异常", field, e);
}
}
// 返回转换后的Map对象
return recordMap;
}
/**
* 检查指定表是否为分区表
*
@ -1363,11 +1023,13 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
private boolean checkIfTablePartitioned(String tableName)
{
try {
// 调用CustomMapper查询数据库元数据，判断表是否为分区表
return customMapper.isPartitioned(tableName);
DataiIntegrationObject object = dataiIntegrationObjectMapper.selectDataiIntegrationObjectByApi(tableName);
if (object != null && object.getIsPartitioned() != null) {
return object.getIsPartitioned();
}
return false;
} catch (Exception e) {
// 查询失败时记录警告日志，默认按非分区表处理
log.warn("检查表 {} 是否分区失败: {}", tableName, e.getMessage());
log.warn("检查对象 {} 分区状态失败: {}，默认按非分区表处理", tableName, e.getMessage());
return false;
}
}

100
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/SalesforceDataPullServiceImpl.java
View File

@ -5,6 +5,7 @@ import com.datai.integration.model.domain.DataiIntegrationFilterLookup;
import com.datai.integration.model.domain.DataiIntegrationMetadataChange;
import com.datai.integration.model.domain.DataiIntegrationObject;
import com.datai.integration.model.domain.DataiIntegrationPicklist;
import com.datai.integration.model.param.DataiSyncParam;
import com.datai.integration.factory.impl.SOAPConnectionFactory;
import com.datai.integration.mapper.CustomMapper;
import com.datai.integration.service.IDataiIntegrationFieldService;
@ -13,7 +14,8 @@ import com.datai.integration.service.IDataiIntegrationMetadataChangeService;
import com.datai.integration.service.IDataiIntegrationObjectService;
import com.datai.integration.service.IDataiIntegrationPicklistService;
import com.datai.integration.service.ISalesforceDataPullService;
import com.datai.salesforce.common.param.SalesforceParam;
import com.datai.salesforce.common.utils.SoqlBuilder;
import com.datai.integration.util.ConvertUtil;
import com.datai.setting.future.SalesforceExecutor;
import com.sforce.soap.partner.DescribeGlobalResult;
import com.sforce.soap.partner.DescribeGlobalSObjectResult;
@ -661,7 +663,7 @@ public class SalesforceDataPullServiceImpl implements ISalesforceDataPullService
log.info("获取到对象 {} 的字段列表，共 {} 个字段", objectApi, fieldList.size());
// 构建查询参数
SalesforceParam param = new SalesforceParam();
DataiSyncParam param = new DataiSyncParam();
param.setApi(objectApi);
param.setSelect(String.join(",", fieldList));
@ -737,7 +739,7 @@ public class SalesforceDataPullServiceImpl implements ISalesforceDataPullService
* @param fieldList 字段列表
* @return 处理的数据条数
*/
private int executeQueryAndProcessData(PartnerConnection connection, SalesforceParam param, List<String> fieldList) {
private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List<String> fieldList) {
int totalCount = 0;
try {
// 构建查询语句
@ -780,18 +782,55 @@ public class SalesforceDataPullServiceImpl implements ISalesforceDataPullService
* @param param 查询参数
* @return SOQL查询语句
*/
private String buildDynamicQuery(SalesforceParam param) {
// 构建简单的查询语句（实际实现需要根据实际情况调整）
StringBuilder queryBuilder = new StringBuilder();
queryBuilder.append("SELECT ").append(param.getSelect())
.append(" FROM ").append(param.getApi());
private String buildDynamicQuery(DataiSyncParam param) {
SoqlBuilder builder = new SoqlBuilder();
String[] selectFields = param.getSelect().split(",");
builder.select(selectFields)
.from(param.getApi());
// 添加删除记录查询条件
if (param.getIsDeleted() != null && param.getIsDeleted()) {
queryBuilder.append(" WHERE IsDeleted = true");
builder.where("IsDeleted = true");
} else if (param.getIsDeleted() != null && !param.getIsDeleted()) {
builder.where("IsDeleted = false");
}
return queryBuilder.toString();
if (param.getBeginDate() != null && param.getEndDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate())
.whereLe(param.getBatchField(), param.getEndDate());
} else if (param.getBeginDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate());
} else if (param.getEndDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndDate());
}
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
builder.whereGe("CreatedDate", param.getBeginCreateDate())
.whereLe("CreatedDate", param.getEndCreateDate());
} else if (param.getBeginCreateDate() != null) {
builder.whereGe("CreatedDate", param.getBeginCreateDate());
} else if (param.getEndCreateDate() != null) {
builder.whereLe("CreatedDate", param.getEndCreateDate());
}
if (param.getBeginModifyDate() != null && param.getEndModifyDate() != null) {
builder.whereGe("LastModifiedDate", param.getBeginModifyDate())
.whereLe("LastModifiedDate", param.getEndModifyDate());
} else if (param.getBeginModifyDate() != null) {
builder.whereGe("LastModifiedDate", param.getBeginModifyDate());
} else if (param.getEndModifyDate() != null) {
builder.whereLe("LastModifiedDate", param.getEndModifyDate());
}
if (StringUtils.isNotEmpty(param.getMaxId())) {
builder.whereGt(param.getIdField(), param.getMaxId());
}
if (param.getLimit() != null && param.getLimit() > 0) {
builder.limit(param.getLimit());
}
return builder.build();
}
/**
@ -991,42 +1030,7 @@ public class SalesforceDataPullServiceImpl implements ISalesforceDataPullService
* @return 转换后的Map
*/
private Map<String, Object> convertSObjectToMap(SObject record, List<String> fieldList) {
Map<String, Object> recordMap = new HashMap<>();
// 添加记录ID
if (record.getId() != null) {
recordMap.put("Id", record.getId());
}
// 添加其他字段
for (String field : fieldList) {
// 跳过ID字段，因为已经处理过了
if ("Id".equalsIgnoreCase(field)) {
continue;
}
try {
Object value = record.getField(field);
// 处理不同类型的字段值
if (value != null) {
// 处理日期类型
if (value instanceof java.util.Calendar) {
recordMap.put(field.toLowerCase(), ((Calendar) value).getTime());
}
// 处理其他类型
else {
recordMap.put(field.toLowerCase(), value);
}
} else {
// 对于null值，可以选择不放入map或者放入null
recordMap.put(field.toLowerCase(), null);
}
} catch (Exception e) {
log.warn("获取字段 {} 的值时发生异常", field, e);
}
}
return recordMap;
return null;
}
@ -1158,7 +1162,7 @@ public class SalesforceDataPullServiceImpl implements ISalesforceDataPullService
log.info("获取到对象 {} 的字段列表: {}", objectApi, fieldList);
// 构建查询参数
SalesforceParam param = new SalesforceParam();
DataiSyncParam param = new DataiSyncParam();
param.setApi(objectApi);
param.setSelect(String.join(",", fieldList));

756
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/util/ConvertUtil.java Normal file
View File

@ -0,0 +1,756 @@
package com.datai.integration.util;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.datai.salesforce.common.constant.SalesforceConstants;
import com.sforce.soap.partner.Field;
import com.sforce.soap.partner.sobject.SObject;
import com.sforce.ws.bind.XmlObject;
import com.sforce.ws.types.Time;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.map.CaseInsensitiveMap;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.commons.lang3.time.FastTimeZone;
import org.springframework.stereotype.Component;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;
/**
* Salesforce到MySQL数据转换工具类
*
* 主要功能：
* 1. 将Salesforce的SObject对象转换为Map
* 2. 处理各种数据类型的转换，确保与MySQL数据库兼容
* 3. 过滤和转换复杂类型字段
* 4. 字段类型映射（Salesforce -> MySQL）
* 5. 日期时间处理和转换
* 6. 时间范围计算
* 7. 任务分割
*
* 支持的类型转换：
* - Calendar -> Date: 日期类型转换
* - Boolean -> Integer(0/1): 布尔值转换为整数，兼容MySQL的TINYINT
* - XmlObjectWrapper -> String: 复杂字段转换为字符串
* - 其他类型 -> 保持原样
*/
@Component
@Slf4j
public class ConvertUtil {
private static final String XML_OBJECT_WRAPPER_CLASS = "com.sforce.ws.bind.XmlObjectWrapper";
private static final FastDateFormat SF_DATE_FORMAT = FastDateFormat.getInstance(SalesforceConstants.SF_DATE_FORMAT, FastTimeZone.getGmtTimeZone());
private static final FastDateFormat SF_TIME_FORMAT = FastDateFormat.getInstance(SalesforceConstants.SF_TIME_FORMAT, FastTimeZone.getGmtTimeZone());
private static final FastDateFormat SF_DATE_FORMAT_BULK = FastDateFormat.getInstance(SalesforceConstants.SF_DATE_FORMAT_BULK, FastTimeZone.getGmtTimeZone());
private static final String MYSQL_DATE_FORMAT = "yyyy-MM-dd";
private static final String MYSQL_DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
private static final int MINUTE = 60;
private static final int HOUR = 24;
private ConvertUtil() {
}
/**
* 添加ID字段到Map
*
* @param record SObject对象
* @param recordMap 目标Map
*/
private static void addIdField(SObject record, Map<String, Object> recordMap) {
if (record.getId() != null) {
recordMap.put("Id", record.getId());
}
}
/**
* sObject 数组转 JSONArray
*
* @param records sObject 数组
* @return JSONArray
*/
public static JSONArray toJsonArray(SObject[] records, Field[] fields) throws ParseException {
JSONArray jsonArray = new JSONArray();
for (SObject sObject : records) {
jsonArray.add(toJsonObject(sObject, fields));
}
return jsonArray;
}
/**
* sObject 数组转 JSONObject
*
* @param sObject sObject
* @param fields 字段
* @return JSONObject
*/
public static JSONObject toJsonObject(SObject sObject, Field[] fields) throws ParseException {
Iterator<XmlObject> children = sObject.getChildren();
JSONObject jsonObject = new JSONObject();
CaseInsensitiveMap map = new CaseInsensitiveMap();
for (Field field : fields) {
map.put(field.getName(), field.getType().toString());
}
while (children.hasNext()) {
XmlObject next = children.next();
String name = next.getName().getLocalPart();
Object value = next.getValue();
// 全部当string处理
if (next instanceof SObject) {
SObject sObjectChild = (SObject) next;
Iterator<XmlObject> children1 = sObjectChild.getChildren();
while (children1.hasNext()) {
XmlObject childNext = children1.next();
String childName = childNext.getName().getLocalPart();
Object childValue = childNext.getValue();
// partner会多一个type 值为对象名称 这里过滤掉
if ("type".equals(childName)) {
continue;
}
if (childValue == null) {
continue;
}
jsonObject.put(name+"_"+childName, childValue);
}
}
// partner会多一个type 值为对象名称 这里过滤掉
if ("type".equals(name)) {
continue;
}
if (value == null) {
continue;
}
String type = (String) map.get(name);
if (type == null) {
continue;
}
switch (type) {
case "date":
jsonObject.put(name, DateUtils.parseDate((String) value, "yyyy-MM-dd"));
break;
case "datetime":
jsonObject.put(name, SF_DATE_FORMAT.parse((String) value));
break;
case "time":
jsonObject.put(name, SF_TIME_FORMAT.parse((String) value));
break;
case "boolean":
jsonObject.put(name, BooleanUtils.toBoolean((String) value));
break;
case "long":
case "int":
jsonObject.put(name, Integer.parseInt((String) value));
break;
case "double":
case "currency":
case "percent":
jsonObject.put(name, Double.parseDouble((String) value));
break;
case "base64":
// 文件不处理 统一存oss或者server
break;
default:
if (value instanceof GregorianCalendar) {
jsonObject.put(name, DateFormatUtils.format(((GregorianCalendar) value), "yyyy-MM-dd HH:mm:ss"));
} else if (value instanceof Time){
jsonObject.put(name, DateFormatUtils.format((((Time) value).getTimeInMillis()), "HH:mm:ss"));
}else {
jsonObject.put(name, String.valueOf(value));
}
break;
}
}
return jsonObject;
}
/**
* 转换Calendar为Date
*
* @param calendar Calendar对象
* @return Date对象
*/
private static Date convertCalendarToDate(Calendar calendar) {
return calendar.getTime();
}
/**
* 转换Boolean为Integer
*
* @param bool 布尔值
* @return 整数（true=1, false=0）
*/
private static Integer convertBooleanToInteger(Boolean bool) {
return bool ? 1 : 0;
}
/**
* 判断是否为XmlObjectWrapper类型
*
* @param value 对象
* @return true如果是XmlObjectWrapper类型
*/
private static boolean isXmlObjectWrapper(Object value) {
return XML_OBJECT_WRAPPER_CLASS.equals(value.getClass().getName());
}
/**
* 转换XmlObjectWrapper为String
*
* @param field 字段名
* @param value XmlObjectWrapper对象
* @return 字符串表示
*/
private static String convertXmlObjectWrapperToString(String field, Object value) {
log.warn("跳过复杂字段 {}，类型为 XmlObjectWrapper，转换为字符串", field);
return String.valueOf(value);
}
/**
* 检查字段是否为复杂类型
*
* @param value 字段值
* @return true如果是复杂类型
*/
public static boolean isComplexType(Object value) {
if (value == null) {
return false;
}
return isXmlObjectWrapper(value);
}
/**
* 检查字段是否需要特殊转换
*
* @param value 字段值
* @return true如果需要转换
*/
public static boolean needsConversion(Object value) {
if (value == null) {
return false;
}
return value instanceof java.util.Calendar
|| value instanceof Boolean
|| value instanceof Time
|| isComplexType(value);
}
/**
* 将Salesforce字段类型转换为MySQL字段类型
*
* @param field Salesforce字段对象
* @return MySQL字段类型
*/
public static String fieldTypeToMysql(Field field) {
String type = field.getType().toString();
int length = field.getLength();
String result = type;
switch (type) {
case "id":
case "reference":
result = "varchar(" + length + ")";
break;
case "string":
if ("Name".equalsIgnoreCase(field.getName())) {
result = "varchar(" + length + ")";
} else {
result = "text";
}
break;
case "url":
case "email":
case "phone":
case "picklist":
case "textarea":
case "multipicklist":
case "address":
case "combobox":
case "location":
case "encryptedstring":
case "anyType":
result = "text";
break;
case "date":
result = "date";
break;
case "datetime":
result = "datetime";
break;
case "boolean":
result = "tinyint(1)";
break;
case "time":
result = "time(3)";
break;
case "long":
case "int":
result = "int(" + length + ")";
break;
case "double":
case "currency":
case "percent":
if (length > 4) {
result = "double(" + length + ")";
} else {
result = "decimal(40,20)";
}
break;
case "base64":
result = "longblob";
break;
default:
result = "text";
break;
}
return result;
}
/**
* 计算两个时间差
*
* @param date1 日期1
* @param date2 日期2
* @return 返回日期格式 xxdxxhxxmin
*/
public static String calTime(Date date1, Date date2) {
if (date1 == null || date2 == null) {
return null;
}
long min = Math.abs(date2.getTime() - date1.getTime()) / 1000 / 60;
StringBuilder stringBuilder = new StringBuilder();
if (min >= MINUTE) {
long h = min / MINUTE;
long realMin = min % MINUTE;
if (h >= HOUR) {
stringBuilder.append(h / HOUR).append("d");
long realH = h % HOUR;
if (realH > 0) {
stringBuilder.append(realH).append("h");
}
} else {
stringBuilder.append(h).append("h");
}
if (realMin > 0) {
stringBuilder.append(realMin).append("min");
}
} else {
stringBuilder.append(min).append("min");
}
return stringBuilder.toString();
}
/**
* 返回本周最后一天 如果大于结束日期 返回结束日期
*
* @param endDate 结束日期
* @param startDate 开始日期
* @return 本周最后一天
*/
public static Date getWeekLastDay(Date endDate, Date startDate) {
Date weekLastDay;
Calendar cal = DateUtils.toCalendar(startDate);
cal.set(Calendar.DAY_OF_WEEK, cal.getActualMaximum(Calendar.DAY_OF_WEEK));
cal.add(Calendar.DAY_OF_WEEK, 1);
weekLastDay = cal.getTime();
if (weekLastDay.compareTo(endDate) > 0) {
weekLastDay = endDate;
}
return weekLastDay;
}
/**
* 返回本月最后一天 如果大于结束日期 返回结束日期
*
* @param endDate 结束日期
* @param startDate 开始日期
* @return 本月最后一天
*/
public static Date getMonthLastDay(Date endDate, Date startDate) {
Date monthLastDay;
Calendar cal = DateUtils.toCalendar(startDate);
cal.set(Calendar.DAY_OF_MONTH, cal.getActualMaximum(Calendar.DAY_OF_MONTH));
cal.add(Calendar.DAY_OF_MONTH, 1);
monthLastDay = cal.getTime();
if (monthLastDay.compareTo(endDate) > 0) {
monthLastDay = endDate;
}
return monthLastDay;
}
/**
* 返回本年最后一天 如果大于结束日期 返回结束日期
*
* @param endDate 结束日期
* @param startDate 开始日期
* @return 本年最后一天
*/
public static Date getYearLastDay(Date endDate, Date startDate) {
Date yearLastDay;
startDate = DateUtils.addYears(startDate, 1);
Calendar cal = DateUtils.toCalendar(startDate);
cal.set(Calendar.YEAR, cal.get(Calendar.YEAR));
yearLastDay = cal.getTime();
if (yearLastDay.compareTo(endDate) > 0) {
yearLastDay = endDate;
}
return yearLastDay;
}
/**
* 根据字段类型将值转换为MySQL类型并返回
*
* @param type 字段类型
* @param value 字段值
* @return 转换后的值
* @throws ParseException 日期解析异常
*/
public static Object convertMysqlValue(String type, Object value) throws ParseException {
return convertMysqlValueInternal(type, value, false);
}
/**
* 根据字段类型将值转换为MySQL类型并返回（Bulk API专用）
*
* @param type 字段类型
* @param value 字段值
* @return 转换后的值
* @throws ParseException 日期解析异常
*/
public static Object convertMysqlBulkValue(String type, Object value) throws ParseException {
return convertMysqlValueInternal(type, value, true);
}
private static Object convertMysqlValueInternal(String type, Object value, boolean isBulk) throws ParseException {
if (value == null) {
return null;
}
String strValue = value instanceof String ? (String) value : null;
switch (type) {
case "date":
if (StringUtils.isEmpty(strValue)) {
return null;
}
return DateUtils.parseDate(strValue, MYSQL_DATE_FORMAT);
case "datetime":
if (StringUtils.isEmpty(strValue)) {
return null;
}
if (isBulk) {
return SF_DATE_FORMAT_BULK.parse(strValue);
} else {
return SF_DATE_FORMAT.parse(strValue);
}
case "time":
if (StringUtils.isEmpty(strValue)) {
return null;
}
return SF_TIME_FORMAT.parse(strValue);
case "boolean":
if (StringUtils.isEmpty(strValue)) {
return false;
}
return Boolean.parseBoolean(strValue);
case "long":
case "int":
if (StringUtils.isEmpty(strValue)) {
return 0;
}
return Integer.parseInt(strValue);
case "double":
case "currency":
case "percent":
if (StringUtils.isEmpty(strValue)) {
return 0.0;
}
return Double.parseDouble(strValue);
case "base64":
return null;
default:
if (value instanceof GregorianCalendar) {
return DateFormatUtils.format(((GregorianCalendar) value), "yyyy-MM-dd HH:mm:ss");
} else if (value instanceof Time) {
return DateFormatUtils.format((((Time) value).getTimeInMillis()), "HH:mm:ss");
} else {
return String.valueOf(value);
}
}
}
/**
* 将本地数据转换为Salesforce数据（Partner API）
*
* @param fieldType 字段类型
* @param data 数据值
* @return 转换后的值
* @throws ParseException 解析异常
*/
public static Object localDataToSfData(String fieldType, String data) throws ParseException {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
Date date;
Calendar calendar = Calendar.getInstance();
switch (fieldType) {
case "int":
return Integer.parseInt(data);
case "double":
case "currency":
case "percent":
return new BigDecimal(data);
case "boolean":
return Boolean.parseBoolean(data);
case "date":
try {
date = sd.parse(data);
calendar.setTime(date);
return calendar;
} catch (ParseException e) {
log.error("日期解析失败: {}", data, e);
return null;
}
case "datetime":
try {
date = sd.parse(data);
calendar.setTime(date);
return calendar;
} catch (ParseException e) {
log.error("日期时间解析失败: {}", data, e);
return null;
}
case "time":
return data;
default:
return data;
}
}
/**
* 将本地数据转换为Salesforce数据（Bulk API）
*
* @param fieldType 字段类型
* @param data 数据值
* @return 转换后的值
* @throws ParseException 解析异常
*/
public static Object localBulkDataToSfData(String fieldType, String data) throws ParseException {
switch (fieldType) {
case "int":
return Integer.parseInt(data);
case "double":
case "currency":
case "percent":
return new BigDecimal(data);
case "boolean":
return Boolean.parseBoolean(data);
case "date":
return data + "T08:00:00Z";
case "datetime":
LocalDateTime localDateTime = LocalDateTime.parse(data);
LocalDateTime plusEightHours = localDateTime.plusHours(-8);
OffsetDateTime offsetDateTime = plusEightHours.atOffset(ZoneOffset.ofHours(0));
return offsetDateTime.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);
case "time":
return adjustHour(data);
default:
return data;
}
}
/**
* 调整时间格式
*
* @param time 时间字符串
* @return 调整后的时间字符串
*/
private static String adjustHour(String time) {
if (StringUtils.isBlank(time)) {
return time;
}
return time;
}
/**
* 格式化日期时间为MySQL格式
*
* @param date 日期对象
* @return MySQL格式日期时间字符串
*/
public static String formatDateToMysql(Date date) {
if (date == null) {
return null;
}
SimpleDateFormat sdf = new SimpleDateFormat(MYSQL_DATETIME_FORMAT);
return sdf.format(date);
}
/**
* 解析MySQL日期字符串
*
* @param dateStr MySQL日期字符串
* @return Date对象
* @throws ParseException 解析异常
*/
public static Date parseMysqlDate(String dateStr) throws ParseException {
if (StringUtils.isBlank(dateStr)) {
return null;
}
SimpleDateFormat sdf = new SimpleDateFormat(MYSQL_DATETIME_FORMAT);
return sdf.parse(dateStr);
}
/**
* 获取当天的开始时间（00:00:00）
*
* @param date 日期对象
* @return 当天开始时间
*/
public static Date getStartOfDay(Date date) {
if (date == null) {
return null;
}
Calendar calendar = Calendar.getInstance();
calendar.setTime(date);
calendar.set(Calendar.HOUR_OF_DAY, 0);
calendar.set(Calendar.MINUTE, 0);
calendar.set(Calendar.SECOND, 0);
calendar.set(Calendar.MILLISECOND, 0);
return calendar.getTime();
}
/**
* 获取当天的结束时间（23:59:59）
*
* @param date 日期对象
* @return 当天结束时间
*/
public static Date getEndOfDay(Date date) {
if (date == null) {
return null;
}
Calendar calendar = Calendar.getInstance();
calendar.setTime(date);
calendar.set(Calendar.HOUR_OF_DAY, 23);
calendar.set(Calendar.MINUTE, 59);
calendar.set(Calendar.SECOND, 59);
calendar.set(Calendar.MILLISECOND, 999);
return calendar.getTime();
}
/**
* 判断日期是否在指定范围内
*
* @param date 待判断的日期
* @param start 开始日期
* @param end 结束日期
* @return true如果在范围内
*/
public static boolean isDateInRange(Date date, Date start, Date end) {
if (date == null) {
return false;
}
if (start != null && date.before(start)) {
return false;
}
if (end != null && date.after(end)) {
return false;
}
return true;
}
/**
* 添加天数到日期
*
* @param date 日期对象
* @param days 天数
* @return 新日期
*/
public static Date addDays(Date date, int days) {
if (date == null) {
return null;
}
return DateUtils.addDays(date, days);
}
/**
* 添加小时到日期
*
* @param date 日期对象
* @param hours 小时数
* @return 新日期
*/
public static Date addHours(Date date, int hours) {
if (date == null) {
return null;
}
return DateUtils.addHours(date, hours);
}
/**
* 添加分钟到日期
*
* @param date 日期对象
* @param minutes 分钟数
* @return 新日期
*/
public static Date addMinutes(Date date, int minutes) {
if (date == null) {
return null;
}
return DateUtils.addMinutes(date, minutes);
}
/**
* 比较两个日期是否是同一天
*
* @param date1 日期1
* @param date2 日期2
* @return true如果是同一天
*/
public static boolean isSameDay(Date date1, Date date2) {
if (date1 == null || date2 == null) {
return false;
}
return DateUtils.isSameDay(date1, date2);
}
/**
* 截断日期到指定精度
*
* @param date 日期对象
* @param field 精度字段（Calendar.DAY_OF_MONTH, Calendar.HOUR_OF_DAY等）
* @return 截断后的日期
*/
public static Date truncate(Date date, int field) {
if (date == null) {
return null;
}
return DateUtils.truncate(date, field);
}
}

26
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/doc/同步单对象数据接口文档.md
View File

@ -7,8 +7,8 @@
### 1.2 接口描述
该接口用于触发指定对象的单次数据同步操作，将Salesforce对象的数据同步到本地数据库。同步操作包括全量同步和增量同步两种模式，具体模式由对象的配置决定：
- **全量同步**：当对象的全量拉取时间（lastFullSyncDate）为空时，执行全量数据拉取，查询所有全量同步批次并使用多线程并行拉取
- **增量同步**：当对象的全量拉取时间不为空时，执行增量数据拉取，创建增量批次并拉取增量数据
- **全量同步**：当对象的最后批次时间（lastBatchDate）为空时，执行全量数据拉取，查询所有全量同步批次并使用多线程并行拉取
- **增量同步**：当对象的最后批次时间不为空时，执行增量数据拉取，创建增量批次并拉取增量数据
### 1.3 接口地址
```
@ -88,7 +88,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
"totalCount": 12345,
"duration": 15234,
"syncType": "full",
"lastFullSyncDate": "2025-12-27T10:30:00",
"lastBatchDate": "2025-12-27T10:30:00",
"totalBatchCount": 5,
"successBatchCount": 5,
"failedBatchCount": 0
@ -117,7 +117,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
| totalCount | Integer | 同步的总记录数 |
| duration | Long | 同步耗时（毫秒） |
| syncType | String | 同步类型，"full"表示全量同步，"incremental"表示增量同步 |
| lastFullSyncDate | LocalDateTime | 最后全量同步时间 |
| lastBatchDate | LocalDateTime | 最后批次时间 |
| lastSyncDate | LocalDateTime | 最后同步时间（增量同步时返回） |
| totalBatchCount | Integer | 总批次数量（全量同步时返回） |
| successBatchCount | Integer | 成功批次数量（全量同步时返回） |
@ -138,7 +138,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
"totalCount": 12345,
"duration": 15234,
"syncType": "full",
"lastFullSyncDate": "2025-12-27T10:30:00",
"lastBatchDate": "2025-12-27T10:30:00",
"totalBatchCount": 5,
"successBatchCount": 5,
"failedBatchCount": 0
@ -159,7 +159,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
"totalCount": 234,
"duration": 3456,
"syncType": "incremental",
"lastFullSyncDate": "2025-12-26T10:30:00",
"lastBatchDate": "2025-12-26T10:30:00",
"lastSyncDate": "2025-12-27T10:30:00"
}
}
@ -178,7 +178,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
"totalCount": 10000,
"duration": 12000,
"syncType": "full",
"lastFullSyncDate": "2025-12-27T10:30:00",
"lastBatchDate": "2025-12-27T10:30:00",
"totalBatchCount": 5,
"successBatchCount": 4,
"failedBatchCount": 1
@ -280,14 +280,14 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
- 检查对象是否存在
- 检查对象API是否配置
- 检查对象是否启用同步
4. **同步模式判断**：判断对象的全量拉取时间（lastFullSyncDate）是否为空
4. **同步模式判断**：判断对象的最后批次时间（lastBatchDate）是否为空
5. **全量同步执行**：
- 查询所有 `syncType="FULL"` 的批次
- 使用 `SalesforceExecutor` 线程池多线程并行拉取多个批次
- 统计成功和失败的批次数量
- 统计总同步记录数
6. **对象状态更新**：
- 更新对象的 `lastFullSyncDate` 为当前时间
- 更新对象的 `lastBatchDate` 为当前时间
- 更新对象的 `lastSyncDate` 为当前时间
- 更新对象的 `totalRows` 为总记录数
- 更新对象的 `syncStatus`（全部成功为true，否则为false）
@ -301,7 +301,7 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
- 检查对象是否存在
- 检查对象API是否配置
- 检查对象是否启用同步
4. **同步模式判断**：判断对象的全量拉取时间（lastFullSyncDate）是否为空
4. **同步模式判断**：判断对象的最后批次时间（lastBatchDate）是否为空
5. **增量同步执行**：
- 检查对象是否启用增量更新（isIncremental）
- 获取Salesforce连接
@ -321,8 +321,8 @@ fetch('http://localhost:8080/integration/object/1/syncData', {
| 条件 | 同步模式 | 说明 |
|------|----------|------|
| lastFullSyncDate == null | 全量同步 | 对象首次同步，执行全量数据拉取 |
| lastFullSyncDate != null | 增量同步 | 对象已执行过全量同步，执行增量数据拉取 |
| lastBatchDate == null | 全量同步 | 对象首次同步，执行全量数据拉取 |
| lastBatchDate != null | 增量同步 | 对象已执行过全量同步，执行增量数据拉取 |
### 4.3 全量同步特性
@ -607,7 +607,7 @@ POST /integration/batch/{id}/syncData
| isWork | Boolean | 是否启用同步 |
| isIncremental | Boolean | 是否启用增量更新 |
| batchField | String | 批次字段 |
| lastFullSyncDate | LocalDateTime | 最后全量同步时间 |
| lastBatchDate | LocalDateTime | 最后批次时间 |
| lastSyncDate | LocalDateTime | 最后同步时间 |
| totalRows | Integer | 总记录数 |
| syncStatus | Boolean | 同步状态 |

18
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/mapper/integration/DataiIntegrationObjectMapper.xml
View File

@ -25,9 +25,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="isIncremental" column="is_incremental" />
<result property="objectIndex" column="object_index" />
<result property="batchField" column="batch_field" />
<result property="isPartitioned" column="is_partitioned" />
<result property="totalRows" column="total_rows" />
<result property="lastSyncDate" column="last_sync_date" />
<result property="lastFullSyncDate" column="last_full_sync_date" />
<result property="lastBatchDate" column="last_batch_date" />
<result property="syncStatus" column="sync_status" />
<result property="errorMessage" column="error_message" />
<result property="remark" column="remark" />
@ -59,9 +60,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
dio.is_incremental,
dio.object_index,
dio.batch_field,
dio.is_partitioned,
dio.total_rows,
dio.last_sync_date,
dio.last_full_sync_date,
dio.last_batch_date,
dio.sync_status,
dio.error_message,
dio.remark,
@ -94,9 +96,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="isIncremental != null "> and dio.is_incremental = #{isIncremental}</if>
<if test="objectIndex != null "> and dio.object_index = #{objectIndex}</if>
<if test="batchField != null and batchField != ''"> and dio.batch_field = #{batchField}</if>
<if test="isPartitioned != null "> and dio.is_partitioned = #{isPartitioned}</if>
<if test="totalRows != null "> and dio.total_rows = #{totalRows}</if>
<if test="lastSyncDate != null "> and dio.last_sync_date = #{lastSyncDate}</if>
<if test="lastFullSyncDate != null "> and dio.last_full_sync_date = #{lastFullSyncDate}</if>
<if test="lastBatchDate != null "> and dio.last_batch_date = #{lastBatchDate}</if>
<if test="syncStatus != null "> and dio.sync_status = #{syncStatus}</if>
<if test="errorMessage != null and errorMessage != ''"> and dio.error_message = #{errorMessage}</if>
</where>
@ -134,9 +137,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="isIncremental != null">is_incremental,</if>
<if test="objectIndex != null">object_index,</if>
<if test="batchField != null">batch_field,</if>
<if test="isPartitioned != null">is_partitioned,</if>
<if test="totalRows != null">total_rows,</if>
<if test="lastSyncDate != null">last_sync_date,</if>
<if test="lastFullSyncDate != null">last_full_sync_date,</if>
<if test="lastBatchDate != null">last_batch_date,</if>
<if test="syncStatus != null">sync_status,</if>
<if test="errorMessage != null">error_message,</if>
<if test="remark != null">remark,</if>
@ -165,9 +169,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="isIncremental != null">#{isIncremental},</if>
<if test="objectIndex != null">#{objectIndex},</if>
<if test="batchField != null">#{batchField},</if>
<if test="isPartitioned != null">#{isPartitioned},</if>
<if test="totalRows != null">#{totalRows},</if>
<if test="lastSyncDate != null">#{lastSyncDate},</if>
<if test="lastFullSyncDate != null">#{lastFullSyncDate},</if>
<if test="lastBatchDate != null">#{lastBatchDate},</if>
<if test="syncStatus != null">#{syncStatus},</if>
<if test="errorMessage != null">#{errorMessage},</if>
<if test="remark != null">#{remark},</if>
@ -200,9 +205,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="isIncremental != null">is_incremental = #{isIncremental},</if>
<if test="objectIndex != null">object_index = #{objectIndex},</if>
<if test="batchField != null">batch_field = #{batchField},</if>
<if test="isPartitioned != null">is_partitioned = #{isPartitioned},</if>
<if test="totalRows != null">total_rows = #{totalRows},</if>
<if test="lastSyncDate != null">last_sync_date = #{lastSyncDate},</if>
<if test="lastFullSyncDate != null">last_full_sync_date = #{lastFullSyncDate},</if>
<if test="lastBatchDate != null">last_batch_date = #{lastBatchDate},</if>
<if test="syncStatus != null">sync_status = #{syncStatus},</if>
<if test="errorMessage != null">error_message = #{errorMessage},</if>
<if test="remark != null">remark = #{remark},</if>

7
datai-scenes/datai-scene-salesforce/sql/add_is_partitioned_column.sql Normal file
View File

@ -0,0 +1,7 @@
-- 为 datai_integration_object 表添加 is_partitioned 字段
-- 执行日期: 2026-01-06
-- 说明: 添加是否分区字段，用于标识对象是否使用分区表
ALTER TABLE `datai_integration_object`
ADD COLUMN `is_partitioned` tinyint NULL DEFAULT 0 COMMENT '是否分区'
AFTER `batch_field`;

3
datai-scenes/datai-scene-salesforce/sql/datai_table.sql
View File

@ -346,9 +346,10 @@ CREATE TABLE `datai_integration_object` (
`is_incremental` tinyint NULL DEFAULT 1 COMMENT '增量更新',
`object_index` int NULL DEFAULT 0 COMMENT '排序权重',
`batch_field` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT 'LastModifiedDate' COMMENT '批次字段',
`is_partitioned` tinyint NULL DEFAULT 0 COMMENT '是否分区',
`total_rows` int NULL DEFAULT 0 COMMENT '本地记录数',
`last_sync_date` datetime NULL DEFAULT NULL COMMENT '最近同步时间',
`last_full_sync_date` datetime NULL DEFAULT NULL COMMENT '全量同步时间',
`last_batch_date` datetime NULL DEFAULT NULL COMMENT '最后批次时间',
`sync_status` tinyint NULL DEFAULT 0 COMMENT '状态',
`error_message` text CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL COMMENT '失败原因',
`remark` varchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '备注',