diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/constant/SalesforceConstants.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/constant/SalesforceConstants.java index 3ece102c..46122b5c 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/constant/SalesforceConstants.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/constant/SalesforceConstants.java @@ -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'"; } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationObject.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationObject.java index 4ddd2c45..5e80ccba 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationObject.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationObject.java @@ -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()) diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/dto/DataiIntegrationObjectDto.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/dto/DataiIntegrationObjectDto.java index 49756ece..b8465fec 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/dto/DataiIntegrationObjectDto.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/dto/DataiIntegrationObjectDto.java @@ -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; diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java index 9b86c94e..f13ded1d 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java @@ -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; + } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/vo/DataiIntegrationObjectVo.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/vo/DataiIntegrationObjectVo.java index 00846ffc..6e7cf0f4 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/vo/DataiIntegrationObjectVo.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/vo/DataiIntegrationObjectVo.java @@ -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; diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java index 119cb5aa..e099151e 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java @@ -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 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 fieldList) { + private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List 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 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 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>> partitionedData = new HashMap<>(); - List> normalData = new ArrayList<>(); - - for (SObject record : records) { - if (record == null) { - continue; - } - - try { - Map 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>> entry : partitionedData.entrySet()) { - String partitionName = entry.getKey(); - List> dataList = entry.getValue(); - if (!dataList.isEmpty()) { - if (!dataList.isEmpty()) { - Collection keys = dataList.get(0).keySet(); - Collection> values = new ArrayList<>(); - for (Map recordMap : dataList) { - Collection 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 ids = Arrays.stream(records).map(SObject::getId).collect(Collectors.toList()); + List existsIds = customMapper.getIds(api, ids); + + String partitionName = null; + if (isPartitioned && batchField != null) { + partitionName = buildPartitionName(batchField, objects); + } + + List> insertMaps = new ArrayList<>(); + List> updateMaps = new ArrayList<>(); + List updateIds = new ArrayList<>(); + + for (int i = 0; i < objects.size(); i++) { + JSONObject jsonObject = objects.getJSONObject(i); + String id = jsonObject.getString("Id"); + + Map recordMap = new HashMap<>(); + for (String key : jsonObject.keySet()) { + recordMap.put(key, jsonObject.get(key)); } - } else { - if (!normalData.isEmpty()) { - Collection keys = normalData.get(0).keySet(); - Collection> values = new ArrayList<>(); - for (Map recordMap : normalData) { - Collection 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 keys = new ArrayList<>(insertMaps.get(0).keySet()); + List> values = new ArrayList<>(); + + for (Map recordMap : insertMaps) { + List 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 recordMap = updateMaps.get(i); + String id = updateIds.get(i); + + List> maps = new ArrayList<>(); + for (Map.Entry entry : recordMap.entrySet()) { + Map 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 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 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 convertSObjectToMap(SObject record, List fieldList) { - Map 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; } } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationMetadataChangeServiceImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationMetadataChangeServiceImpl.java index d459c844..a475726d 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationMetadataChangeServiceImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationMetadataChangeServiceImpl.java @@ -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 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 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); diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java index 511d3c70..36ce98e4 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java @@ -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 同步结果，包含success、message、objectId、objectApi、totalCount、duration、syncType、lastFullSyncDate等字段 + * @return Map 同步结果，包含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 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 syncIncrementalData(DataiIntegrationObject object, long startTime) { Map 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 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 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 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 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>> partitionedData = new HashMap<>(); - // 创建普通数据列表，用于存储非分区表的数据 - List> normalData = new ArrayList<>(); - - // 遍历所有记录，进行数据转换和分区分配 - for (SObject record : records) { - // 跳过空记录，避免空指针异常 - if (record == null) { - continue; - } - - try { - // 将SObject对象转换为Map格式，便于后续处理 - Map 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>> entry : partitionedData.entrySet()) { - String partitionName = entry.getKey(); - List> 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 convertSObjectToMap(SObject record, List fieldList) - { - // 创建Map对象用于存储转换后的记录数据 - Map 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; } } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/SalesforceDataPullServiceImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/SalesforceDataPullServiceImpl.java index 3d01c177..9f3fae30 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/SalesforceDataPullServiceImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/SalesforceDataPullServiceImpl.java @@ -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 fieldList) { + private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List 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 convertSObjectToMap(SObject record, List fieldList) { - Map 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)); diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/util/ConvertUtil.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/util/ConvertUtil.java new file mode 100644 index 00000000..ca0c1539 --- /dev/null +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/util/ConvertUtil.java @@ -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 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 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 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); + } +} diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/doc/同步单对象数据接口文档.md b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/doc/同步单对象数据接口文档.md index 0e8e44a6..70d10441 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/doc/同步单对象数据接口文档.md +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/doc/同步单对象数据接口文档.md @@ -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 | 同步状态 | diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/mapper/integration/DataiIntegrationObjectMapper.xml b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/mapper/integration/DataiIntegrationObjectMapper.xml index 14b7153f..5b8bcfa4 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/mapper/integration/DataiIntegrationObjectMapper.xml +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/resources/mapper/integration/DataiIntegrationObjectMapper.xml @@ -25,9 +25,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + - + @@ -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" and dio.is_incremental = #{isIncremental} and dio.object_index = #{objectIndex} and dio.batch_field = #{batchField} + and dio.is_partitioned = #{isPartitioned} and dio.total_rows = #{totalRows} and dio.last_sync_date = #{lastSyncDate} - and dio.last_full_sync_date = #{lastFullSyncDate} + and dio.last_batch_date = #{lastBatchDate} and dio.sync_status = #{syncStatus} and dio.error_message = #{errorMessage} @@ -134,9 +137,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" is_incremental, object_index, batch_field, + is_partitioned, total_rows, last_sync_date, - last_full_sync_date, + last_batch_date, sync_status, error_message, remark, @@ -165,9 +169,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" #{isIncremental}, #{objectIndex}, #{batchField}, + #{isPartitioned}, #{totalRows}, #{lastSyncDate}, - #{lastFullSyncDate}, + #{lastBatchDate}, #{syncStatus}, #{errorMessage}, #{remark}, @@ -200,9 +205,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" is_incremental = #{isIncremental}, object_index = #{objectIndex}, batch_field = #{batchField}, + is_partitioned = #{isPartitioned}, total_rows = #{totalRows}, last_sync_date = #{lastSyncDate}, - last_full_sync_date = #{lastFullSyncDate}, + last_batch_date = #{lastBatchDate}, sync_status = #{syncStatus}, error_message = #{errorMessage}, remark = #{remark}, diff --git a/datai-scenes/datai-scene-salesforce/sql/add_is_partitioned_column.sql b/datai-scenes/datai-scene-salesforce/sql/add_is_partitioned_column.sql new file mode 100644 index 00000000..fce3bb16 --- /dev/null +++ b/datai-scenes/datai-scene-salesforce/sql/add_is_partitioned_column.sql @@ -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`; diff --git a/datai-scenes/datai-scene-salesforce/sql/datai_table.sql b/datai-scenes/datai-scene-salesforce/sql/datai_table.sql index b58c2f3a..6fb5b2b2 100644 --- a/datai-scenes/datai-scene-salesforce/sql/datai_table.sql +++ b/datai-scenes/datai-scene-salesforce/sql/datai_table.sql @@ -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 '备注',