From 4f0a941ab66beecc854012edd37551e4002d23e9 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 7 Jan 2026 13:52:00 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20=E5=88=9D=E6=AD=A5?= =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E6=8B=89?= =?UTF-8?q?=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/resources/application-plugins.yml | 4 +- .../salesforce/common/utils/SoqlBuilder.java | 31 +++ .../DataiIntegrationObjectController.java | 43 +++ .../model/domain/DataiIntegrationBatch.java | 19 ++ .../model/param/DataiSyncParam.java | 7 +- .../IDataiIntegrationObjectService.java | 8 + .../DataiIntegrationBatchServiceImpl.java | 260 +++++++++++++----- .../DataiIntegrationObjectServiceImpl.java | 125 +++++++-- 8 files changed, 399 insertions(+), 98 deletions(-) diff --git a/datai-admin/src/main/resources/application-plugins.yml b/datai-admin/src/main/resources/application-plugins.yml index 9d660d57..b8d871ba 100644 --- a/datai-admin/src/main/resources/application-plugins.yml +++ b/datai-admin/src/main/resources/application-plugins.yml @@ -3,13 +3,13 @@ spring: # redis 配置 redis: # 地址 - host: 134.175.52.143 + host: 127.0.0.1 # 端口，默认为6379 port: 6379 # 数据库索引 database: 0 # 密码 - password: czsj@2024 + password: # 连接超时时间 timeout: 10s lettuce: diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/utils/SoqlBuilder.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/utils/SoqlBuilder.java index 330cd8f2..5235cf5f 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/utils/SoqlBuilder.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/utils/SoqlBuilder.java @@ -611,6 +611,37 @@ public class SoqlBuilder { return soql.toString(); } + /** + * 构建COUNT查询语句 + * 生成类似 "SELECT count(Id) FROM Object WHERE ..." 的查询 + * + * 注意： + * - COUNT查询不使用SELECT子句中指定的字段，而是使用count(Id) + * - COUNT查询不支持GROUP BY、HAVING、ORDER BY、LIMIT、OFFSET + * - 只使用FROM和WHERE条件 + * + * @return COUNT查询语句 + * @throws IllegalStateException 如果FROM子句为空 + */ + public String buildCountQuery() { + if (fromObject == null || fromObject.trim().isEmpty()) { + throw new IllegalStateException("FROM子句不能为空，必须指定查询对象"); + } + + StringBuilder soql = new StringBuilder(); + + // 构建SELECT count(Id)子句 + soql.append("SELECT count(Id)"); + + // 构建FROM子句 + soql.append(" FROM ").append(fromObject); + + // 构建WHERE子句 + buildWhereClause(soql); + + return soql.toString(); + } + /** * 验证构建器配置 * diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/controller/DataiIntegrationObjectController.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/controller/DataiIntegrationObjectController.java index e06836d2..f6db3177 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/controller/DataiIntegrationObjectController.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/controller/DataiIntegrationObjectController.java @@ -258,4 +258,47 @@ public class DataiIntegrationObjectController extends BaseController return error("同步对象数据时发生异常: " + e.getMessage()); } } + + /** + * 同步多个对象数据到本地数据库 + * + * 该方法用于触发多个对象的批量数据同步操作，将多个Salesforce对象的数据同步到本地数据库 + * 同步操作包括全量同步和增量同步两种模式，具体模式由每个对象的配置决定 + * 该方法会依次同步每个对象，即使某个对象同步失败，也会继续同步其他对象 + * + * @param ids 对象ID数组，用于标识需要同步的多个Salesforce对象 + * @return AjaxResult 同步结果，包含成功/失败状态、每个对象的同步结果、总耗时等信息 + * - 成功时返回：success=true, message="多对象数据同步完成", 以及详细的同步信息 + * - 失败时返回：success=false, message=错误信息 + */ + @Operation(summary = "同步多个对象数据到本地数据库") + @PreAuthorize("@ss.hasPermi('integration:object:syncData')") + @Log(title = "对象同步控制", businessType = BusinessType.UPDATE) + @PostMapping("/syncMultipleData") + public AjaxResult syncMultipleObjectData(@RequestBody Integer[] ids) + { + if (ids == null || ids.length == 0) { + log.error("对象ID数组为空，无法同步数据"); + return error("对象ID不能为空"); + } + + try { + log.info("开始同步多个对象数据，对象ID数量: {}", ids.length); + + Map result = dataiIntegrationObjectService.syncMultipleObjectData(ids); + + if ((Boolean) result.get("success")) { + log.info("多对象数据同步完成，成功数量: {}, 失败数量: {}", + result.get("successCount"), result.get("failureCount")); + return success(result); + } else { + log.error("多对象数据同步失败，错误信息: {}", result.get("message")); + return error((String) result.get("message")); + } + + } catch (Exception e) { + log.error("同步多个对象数据时发生异常", e); + return error("同步多个对象数据时发生异常: " + e.getMessage()); + } + } } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationBatch.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationBatch.java index 789a6e5d..755c21d6 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationBatch.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationBatch.java @@ -11,6 +11,25 @@ import com.datai.common.core.domain.BaseEntity; /** * 数据批次对象 datai_integration_batch * + * 字段更新规则： + * 1. 创建批次时设置的字段（后续不允许修改）： + * - id: 批次ID + * - deptId: 部门ID + * - api: 对象API + * - label: 对象名称 + * - syncType: 同步类型（FULL/INCREMENTAL） + * - batchField: 批次字段 + * - syncStartDate: 开始同步时间（批次的时间范围起点） + * - syncEndDate: 结束同步时间（批次的时间范围终点） + * + * 2. 同步时更新的字段： + * - sfNum: SF数据量（每次同步时更新） + * - dbNum: 本地数据量（每次同步时更新） + * - syncStatus: 同步状态（true=成功，false=失败） + * - firstSyncTime: 首次同步时间（仅在首次同步时设置） + * - lastSyncTime: 最后同步时间（每次同步时更新） + * - updateTime: 更新时间（每次更新时自动设置） + * * @author datai * @date 2026-01-01 */ 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 f13ded1d..6baf8339 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 @@ -44,6 +44,11 @@ public class DataiSyncParam { * 开始修改时间 */ private Date beginModifyDate; + + /** + * 数据拉取截断时间 + */ + private Date lastDate; /** * 结束修改时间 */ @@ -69,7 +74,7 @@ public class DataiSyncParam { */ private String select; /** - * 是否查询删除记录 + * 是否存在IsDeleted删除字段 */ private Boolean isDeleted; diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/IDataiIntegrationObjectService.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/IDataiIntegrationObjectService.java index 8806f10c..18d81693 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/IDataiIntegrationObjectService.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/IDataiIntegrationObjectService.java @@ -116,4 +116,12 @@ public interface IDataiIntegrationObjectService * @return 同步结果 */ public Map syncSingleObjectData(Integer id); + + /** + * 同步多个对象数据到本地数据库 + * + * @param ids 对象ID集合 + * @return 同步结果 + */ + public Map syncMultipleObjectData(Integer[] ids); } 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 e099151e..2c96c952 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 @@ -175,7 +175,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS log.info("批次 {} 共有 {} 条失败的同步记录，准备重试", id, failedHistories.size()); - boolean retryResult = syncObjectDataByBatch(batch.getApi(), id.toString()); + boolean retryResult = syncObjectDataByBatch(batch.getApi(), id); if (retryResult) { log.info("批次 {} 重试成功", id); @@ -299,12 +299,13 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS } String objectApi = batch.getApi(); - String batchId = id.toString(); - log.info("准备同步对象 {} 的批次 {} 数据", objectApi, batchId); + log.info("准备同步对象 {} 的批次 {} 数据", objectApi, id); - boolean syncResult = syncObjectDataByBatch(objectApi, batchId); + boolean syncResult = syncObjectDataByBatch(objectApi, id); + batch = selectDataiIntegrationBatchById(id); + if (syncResult) { log.info("批次 {} 数据同步成功", id); result.put("success", true); @@ -312,12 +313,14 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS result.put("batchId", id); result.put("api", objectApi); result.put("label", batch.getLabel()); + result.put("syncNum", batch.getDbNum()); } else { log.error("批次 {} 数据同步失败", id); result.put("success", false); result.put("message", "批次数据同步失败"); result.put("batchId", id); result.put("api", objectApi); + result.put("syncNum", batch.getDbNum() != null ? batch.getDbNum() : 0); } } catch (Exception e) { @@ -336,10 +339,10 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS * @param batchId 批次ID * @return 同步结果 */ - private boolean syncObjectDataByBatch(String objectApi, String batchId) { + private boolean syncObjectDataByBatch(String objectApi, Integer batchId) { log.info("准备同步Salesforce对象的指定批次数据，对象API: {}, 批次ID: {}", objectApi, batchId); - if (objectApi == null || objectApi.trim().isEmpty() || batchId == null || batchId.trim().isEmpty()) { + if (objectApi == null || objectApi.trim().isEmpty() || batchId == null ) { log.error("对象API或批次ID为空，无法同步指定批次数据"); return false; } @@ -350,7 +353,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS LocalDateTime syncStartTime = LocalDateTime.now(); try { - batch = selectDataiIntegrationBatchById(Integer.parseInt(batchId)); + batch = selectDataiIntegrationBatchById(batchId); if (batch == null) { log.error("批次不存在，批次ID: {}", batchId); return false; @@ -370,8 +373,18 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS param.setIsDeleted(true); } - int totalCount = executeQueryAndProcessData(connection, param, fieldList); - log.info("对象 {} 批次 {} 数据同步完成，共处理 {} 条记录", objectApi, batchId, totalCount); + if (batch.getSyncStartDate() != null) { + param.setBeginDate(java.sql.Timestamp.valueOf(batch.getSyncStartDate())); + } + if (batch.getSyncEndDate() != null) { + param.setEndDate(java.sql.Timestamp.valueOf(batch.getSyncEndDate())); + } + + int sfTotalCount = querySalesforceDataCount(connection, param); + log.info("对象 {} 批次 {} Salesforce中共有 {} 条记录需要同步", objectApi, batchId, sfTotalCount); + + int dbProcessedCount = executeQueryAndProcessData(connection, param, fieldList); + log.info("对象 {} 批次 {} 数据同步完成，共处理 {} 条记录", objectApi, batchId, dbProcessedCount); long endTime = System.currentTimeMillis(); long duration = endTime - startTime; @@ -379,9 +392,9 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS batchHistory.setApi(objectApi); batchHistory.setLabel(batch.getLabel()); - batchHistory.setBatchId(Integer.parseInt(batchId)); + batchHistory.setBatchId(batchId); batchHistory.setBatchField(batch.getBatchField()); - batchHistory.setSyncNum(totalCount); + batchHistory.setSyncNum(dbProcessedCount); batchHistory.setSyncType(batch.getSyncType()); batchHistory.setSyncStatus(true); batchHistory.setStartTime(syncStartTime); @@ -390,9 +403,21 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS batchHistory.setSyncStartTime(syncStartTime); batchHistory.setSyncEndTime(syncEndTime); batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory); - log.info("批次 {} 历史记录已保存，同步数据量: {}, 耗时: {}ms", batchId, totalCount, duration); - - insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), Integer.parseInt(batchId), true, null, duration); + log.info("批次 {} 历史记录已保存，同步数据量: {}, 耗时: {}ms", batchId, dbProcessedCount, duration); + + batch.setSfNum(sfTotalCount); + batch.setDbNum(dbProcessedCount); + batch.setSyncStatus(true); + if (batch.getFirstSyncTime() == null) { + batch.setFirstSyncTime(syncStartTime); + } + batch.setLastSyncTime(syncEndTime); + batch.setUpdateTime(DateUtils.getNowDate()); + updateDataiIntegrationBatch(batch); + log.info("批次 {} 信息已更新，SF数据量: {}, 本地数据量: {}, 同步状态: {}, 首次同步时间: {}, 最后同步时间: {}", + batchId, sfTotalCount, dbProcessedCount, true, batch.getFirstSyncTime(), syncEndTime); + + insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), batchId, true, null, duration); return true; } catch (Exception e) { @@ -404,7 +429,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS batchHistory.setApi(objectApi); batchHistory.setLabel(batch != null ? batch.getLabel() : ""); - batchHistory.setBatchId(Integer.parseInt(batchId)); + batchHistory.setBatchId(batchId); batchHistory.setBatchField(batch != null ? batch.getBatchField() : ""); batchHistory.setSyncNum(0); batchHistory.setSyncType(batch != null ? batch.getSyncType() : ""); @@ -417,7 +442,15 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory); log.info("批次 {} 失败历史记录已保存，耗时: {}ms", batchId, duration); - insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", Integer.parseInt(batchId), false, e.getMessage(), duration); + if (batch != null) { + batch.setSyncStatus(false); + batch.setLastSyncTime(syncEndTime); + batch.setUpdateTime(DateUtils.getNowDate()); + updateDataiIntegrationBatch(batch); + log.info("批次 {} 信息已更新，同步状态: {}, 结束时间: {}", batchId, false, syncEndTime); + } + + insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", batchId, false, e.getMessage(), duration); return false; } @@ -514,79 +547,48 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS } /** - * 执行查询并处理结果 + * 查询Salesforce数据量 * - * @param connection SOAP连接 - * @param param 查询参数 - * @param fieldList 字段列表 - * @return 处理的数据条数 + * @param connection Salesforce连接 + * @param param 查询参数 + * @return 符合条件的总记录数 */ - private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List fieldList) { + private int querySalesforceDataCount(PartnerConnection connection, DataiSyncParam param) { int totalCount = 0; - JSONArray objects = null; - String maxId = null; - Date lastDate = null; - + 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) { - 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); + String countQuery = buildCountQuery(param); + log.info("查询数据量SQL: {}", countQuery); + + QueryResult result = connection.queryAll(countQuery); + + if (result != null && result.getRecords() != null && result.getRecords().length > 0) { + SObject record = result.getRecords()[0]; + Object countValue = record.getField("expr0"); + if (countValue != null) { + totalCount = Integer.parseInt(countValue.toString()); } - - if (result.isDone()) { - break; - } - - result = connection.queryMore(result.getQueryLocator()); - - TimeUnit.MILLISECONDS.sleep(100); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - log.error("查询数据时线程被中断，对象API: {}", param.getApi(), e); + + log.info("对象 {} 符合条件的总记录数: {}", param.getApi(), totalCount); } catch (Exception e) { - log.error("查询处理数据时发生异常，对象API: {}", param.getApi(), e); + log.error("查询数据量时发生异常，对象API: {}", param.getApi(), e); } + return totalCount; } /** - * 构建动态查询语句 + * 构建COUNT查询语句 * * @param param 查询参数 - * @return SOQL查询语句 + * @return SOQL COUNT查询语句 */ - private String buildDynamicQuery(DataiSyncParam param) { + private String buildCountQuery(DataiSyncParam param) { SoqlBuilder builder = new SoqlBuilder(); + builder.from(param.getApi()); - String[] selectFields = param.getSelect().split(","); - builder.select(selectFields) - .from(param.getApi()); - - if (param.getIsDeleted() != null && !param.getIsDeleted()) { + if (param.getIsDeleted() != null && param.getIsDeleted()) { builder.where("IsDeleted = false"); } @@ -617,6 +619,117 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS builder.whereLe(param.getBatchField(), param.getEndModifyDate()); } + return builder.buildCountQuery(); + } + + /** + * 执行查询并处理结果 + * + * @param connection SOAP连接 + * @param param 查询参数 + * @param fieldList 字段列表 + * @return 处理的数据条数 + */ + private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List fieldList) { + int totalCount = 0; + JSONArray objects = null; + String maxId = null; + Date lastDate = null; + + try { + DescribeSObjectResult describeSObject = connection.describeSObject(param.getApi()); + Field[] dsrFields = describeSObject.getFields(); + + while (true) { + DataiSyncParam queryParam = new DataiSyncParam(); + queryParam.setApi(param.getApi()); + queryParam.setSelect(param.getSelect()); + queryParam.setBatchField(param.getBatchField()); + queryParam.setIdField(param.getIdField()); + queryParam.setIsDeleted(param.getIsDeleted()); + + queryParam.setBeginDate(param.getBeginDate()); + queryParam.setEndDate(param.getEndDate()); + queryParam.setLastDate(lastDate); + queryParam.setMaxId(maxId); + + String query = buildDynamicQuery(queryParam); + log.info("执行查询SQL: {}", query); + QueryResult result = connection.queryAll(query); + + if (result.getRecords() != null && result.getRecords().length > 0) { + objects = ConvertUtil.toJsonArray(result.getRecords(), dsrFields); + Date maxDate = objects.getJSONObject(objects.size() - 1).getDate(param.getBatchField()); + 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); + } + + if (result.isDone() || result.getRecords() == null || result.getRecords().length == 0) { + break; + } + } + } catch (Exception e) { + log.error("查询处理数据时发生异常，对象API: {}", param.getApi(), e); + } + return totalCount; + } + + /** + * 构建动态查询语句 + * + * @param param 查询参数 + * @return SOQL查询语句 + */ + 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()) { + builder.where("IsDeleted = false"); + } + + 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.getLastDate() != null) { + builder.whereLe(param.getBatchField(), param.getLastDate()); + } + + 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()); } @@ -624,6 +737,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS if (param.getLimit() != null && param.getLimit() > 0) { builder.limit(param.getLimit()); } + return builder.build(); } @@ -771,4 +885,6 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS return partitionName; } + + } 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 36ce98e4..e00b8652 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 @@ -17,7 +17,6 @@ import com.datai.integration.service.IDataiIntegrationBatchService; import com.datai.integration.service.IDataiIntegrationFieldService; import com.datai.integration.service.IDataiIntegrationObjectService; 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; @@ -710,6 +709,86 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec return result; } + @Override + public Map syncMultipleObjectData(Integer[] ids) { + Map result = new HashMap<>(); + long startTime = System.currentTimeMillis(); + + try { + if (ids == null || ids.length == 0) { + log.error("对象ID数组不能为空"); + result.put("success", false); + result.put("message", "对象ID数组不能为空"); + return result; + } + + log.info("开始同步多个对象数据，对象数量: {}", ids.length); + + List> syncResults = new ArrayList<>(); + int successCount = 0; + int failureCount = 0; + int totalSyncRecords = 0; + List> failedObjects = new ArrayList<>(); + + for (int i = 0; i < ids.length; i++) { + Integer id = ids[i]; + log.info("开始同步第 {}/{} 个对象，对象ID: {}", i + 1, ids.length, id); + + try { + Map singleResult = syncSingleObjectData(id); + syncResults.add(singleResult); + + if ((Boolean) singleResult.get("success")) { + successCount++; + Integer totalCount = (Integer) singleResult.get("totalCount"); + if (totalCount != null) { + totalSyncRecords += totalCount; + } + log.info("对象 {} 同步成功，同步记录数: {}", id, totalCount); + } else { + failureCount++; + Map failedObject = new HashMap<>(); + failedObject.put("objectId", id); + failedObject.put("message", singleResult.get("message")); + failedObjects.add(failedObject); + log.error("对象 {} 同步失败: {}", id, singleResult.get("message")); + } + + } catch (Exception e) { + failureCount++; + Map failedObject = new HashMap<>(); + failedObject.put("objectId", id); + failedObject.put("message", "同步异常: " + e.getMessage()); + failedObjects.add(failedObject); + log.error("同步对象 {} 时发生异常", id, e); + } + } + + long endTime = System.currentTimeMillis(); + long duration = endTime - startTime; + + result.put("success", true); + result.put("message", "多对象数据同步完成"); + result.put("totalObjects", ids.length); + result.put("successCount", successCount); + result.put("failureCount", failureCount); + result.put("totalSyncRecords", totalSyncRecords); + result.put("duration", duration); + result.put("syncResults", syncResults); + result.put("failedObjects", failedObjects); + + log.info("多对象数据同步完成，总对象数: {}, 成功: {}, 失败: {}, 总同步记录数: {}, 耗时: {}ms", + ids.length, successCount, failureCount, totalSyncRecords, duration); + + } catch (Exception e) { + log.error("同步多个对象数据时发生异常", e); + result.put("success", false); + result.put("message", "同步多个对象数据时发生异常: " + e.getMessage()); + } + + return result; + } + /** * 全量数据拉取 * 查询所有批次并使用多线程并行拉取 @@ -739,7 +818,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec List> futures = new ArrayList<>(); List> batchResults = new CopyOnWriteArrayList<>(); - int totalSyncCount = 0; int successBatchCount = 0; int failedBatchCount = 0; @@ -787,9 +865,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec for (Map batchResult : batchResults) { if ((Boolean) batchResult.get("success")) { successBatchCount++; - if (batchResult.get("syncNum") != null) { - totalSyncCount += (Integer) batchResult.get("syncNum"); - } } else { failedBatchCount++; } @@ -800,7 +875,17 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec LocalDateTime now = LocalDateTime.now(); object.setLastSyncDate(now); - object.setTotalRows(totalSyncCount); + + DataiIntegrationBatch queryAllBatches = new DataiIntegrationBatch(); + queryAllBatches.setApi(objectApi); + List allBatches = dataiIntegrationBatchService.selectDataiIntegrationBatchList(queryAllBatches); + + int totalDbNum = allBatches.stream() + .filter(batch -> batch.getDbNum() != null) + .mapToInt(DataiIntegrationBatch::getDbNum) + .sum(); + + object.setTotalRows(totalDbNum); object.setSyncStatus(failedBatchCount == 0); object.setUpdateTime(DateUtils.getNowDate()); updateDataiIntegrationObject(object); @@ -809,7 +894,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec result.put("message", "全量数据拉取完成"); result.put("objectId", object.getId()); result.put("objectApi", objectApi); - result.put("totalCount", totalSyncCount); + result.put("totalCount", totalDbNum); result.put("duration", duration); result.put("syncType", "full"); result.put("lastBatchDate", object.getLastBatchDate()); @@ -818,7 +903,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec result.put("failedBatchCount", failedBatchCount); log.info("对象 {} 全量数据拉取完成，共处理 {} 个批次，成功 {} 个，失败 {} 个，总记录数: {}，耗时: {}ms", - objectApi, batches.size(), successBatchCount, failedBatchCount, totalSyncCount, duration); + objectApi, batches.size(), successBatchCount, failedBatchCount, totalDbNum, duration); } catch (Exception e) { log.error("全量数据拉取失败，对象API: {}", objectApi, e); @@ -857,8 +942,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec result.put("message", "无法获取Salesforce连接"); return result; } - - List fieldList = getSalesforceObjectFields(connection, objectApi); String batchField = object.getBatchField(); if (batchField == null || batchField.trim().isEmpty()) { @@ -876,7 +959,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec DataiIntegrationBatch incrementalBatch = new DataiIntegrationBatch(); incrementalBatch.setApi(objectApi); incrementalBatch.setLabel(object.getLabel()); - incrementalBatch.setBatchField(batchField); + incrementalBatch.setBatchField(dataiIntegrationFieldService.getUpdateField(objectApi)); incrementalBatch.setSyncType("INCREMENTAL"); incrementalBatch.setSyncStatus(false); incrementalBatch.setSyncStartDate(lastBatchDate); @@ -884,26 +967,22 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec incrementalBatch.setCreateTime(DateUtils.getNowDate()); incrementalBatch.setUpdateTime(DateUtils.getNowDate()); - int batchId = dataiIntegrationBatchService.insertDataiIntegrationBatch(incrementalBatch); + dataiIntegrationBatchService.insertDataiIntegrationBatch(incrementalBatch); + Integer batchId = incrementalBatch.getId(); log.info("为对象 {} 创建增量批次，批次ID: {}", objectApi, batchId); object.setLastBatchDate(incrementalBatch.getSyncEndDate()); - updateDataiIntegrationObject(object); - + Map batchResult = dataiIntegrationBatchService.syncBatchData(batchId); - + + Integer totalDbNum = customMapper.countBySQL(objectApi, null); long endTime = System.currentTimeMillis(); long duration = endTime - startTime; LocalDateTime now = LocalDateTime.now(); if ((Boolean) batchResult.get("success")) { object.setLastSyncDate(now); - Integer syncNum = (Integer) batchResult.get("syncNum"); - if (object.getTotalRows() != null) { - object.setTotalRows(object.getTotalRows() + syncNum); - } else { - object.setTotalRows(syncNum); - } + object.setTotalRows(totalDbNum); object.setSyncStatus(true); object.setUpdateTime(DateUtils.getNowDate()); updateDataiIntegrationObject(object); @@ -912,13 +991,13 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec result.put("message", "增量数据拉取成功"); result.put("objectId", object.getId()); result.put("objectApi", objectApi); - result.put("totalCount", syncNum); + result.put("totalCount", batchResult.get("")); result.put("duration", duration); result.put("syncType", "incremental"); result.put("lastBatchDate", object.getLastBatchDate()); result.put("lastSyncDate", object.getLastSyncDate()); - log.info("对象 {} 增量数据拉取成功，同步数据量: {}，耗时: {}ms", objectApi, syncNum, duration); + log.info("对象 {} 增量数据拉取成功，总数据量: {}，耗时: {}ms", objectApi, totalDbNum, duration); } else { object.setSyncStatus(false); object.setUpdateTime(DateUtils.getNowDate());