diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index b382f4c..0f2054f 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -138,6 +138,30 @@ public class DataDumpNewJob { return commonBatchService.dumpBigObject(param); } + /** + * 源系统BulkV1查询 + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("sourceSystemBulkV1QueryJob") + public ReturnT sourceSystemBulkV1QueryJob(String paramStr) throws Exception { + log.info("sourceSystemBulkV1QueryJob execute start .................."); + SalesforceParam param = new SalesforceParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, SalesforceParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + + return commonBatchService.dumpBigObjectSourceV1(param); + } + /** * 存量任务（BigObject_V2） * diff --git a/src/main/java/com/celnet/datadump/service/CommonBatchService.java b/src/main/java/com/celnet/datadump/service/CommonBatchService.java index 0bbdd0b..34775ee 100644 --- a/src/main/java/com/celnet/datadump/service/CommonBatchService.java +++ b/src/main/java/com/celnet/datadump/service/CommonBatchService.java @@ -11,7 +11,8 @@ public interface CommonBatchService { ReturnT dumpBigObjectV2(SalesforceParam param) throws Exception; - ReturnT importBigObject(SalesforceParam param) throws Exception; + + ReturnT dumpBigObjectSourceV1(SalesforceParam param) throws Exception; } \ No newline at end of file diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index 9f7119f..0936fa7 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -182,6 +182,21 @@ public class CommonBatchServiceImpl implements CommonBatchService { } } + @Override + public ReturnT dumpBigObjectSourceV1(SalesforceParam param) throws Exception { + try { + // 手动任务 + ReturnT result = manualBigObjectSourceV1Dump(param); + if (result != null) { + return result; + } + } catch (Throwable throwable) { + log.error("dump error", throwable); + throw throwable; + } + return ReturnT.SUCCESS; + } + private ReturnT manualBigObjectDump(SalesforceParam param) throws Exception { String api = param.getApi(); @@ -199,7 +214,7 @@ public class CommonBatchServiceImpl implements CommonBatchService { .filter(field -> !"SystemModstamp".equals(field) && !"LastModifiedDate".equals(field)) .collect(Collectors.joining(", ")); - String sql = "select Id__c,OppId__c from " + param.getApi(); + String sql = "select " + fieldStr + " from " + param.getApi() ; log.info("构建查询SQL: {}", sql); job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll); @@ -326,6 +341,67 @@ public class CommonBatchServiceImpl implements CommonBatchService { return ReturnT.SUCCESS; } + /** + * 源系统BulkV1查询实现 + * + * @param param 参数 + * @return 执行结果 + * @throws Exception 异常 + */ + private ReturnT manualBigObjectSourceV1Dump(SalesforceParam param) throws Exception { + + String api = param.getApi(); + + String sql = param.getSql(); + + Integer type = param.getType(); + + BulkConnection bulkConnect = salesforceConnect.createBulkConnect(); + + JobInfo job = null; + + try { + + log.info("开始源系统BulkV1查询BigObject数据, API: {}, SQL: {}", api,sql); + + // 创建Bulk作业 + job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll); + log.info("创建Bulk V1查询作业成功, 作业ID: {}", job.getId()); + + // 提交查询批次 + BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8))); + log.info("创建查询批次成功, 批次ID: {}", batchInfo.getId()); + + if (type == 0){ + return ReturnT.SUCCESS; + } + + // 等待作业完成 + int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo); + log.info("查询批次处理完成, 处理记录数: {}", completionOne); + + // 如果没有数据，直接结束 + if (completionOne == 0){ + log.info("无更多数据, 结束处理"); + BulkUtil.closeJob(bulkConnect, job.getId()); + return ReturnT.SUCCESS; + } + + // 获取查询结果列表 + QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId()); + log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length); + + BulkUtil.closeJob(bulkConnect, job.getId()); + log.info("关闭作业成功, 作业ID: {}", job.getId()); + + }catch (Exception e) { + log.error("源系统BulkV1查询BigObject数据失败, API: {}", api, e); + throw e; + } + log.info("源系统BulkV1查询BigObject数据任务完成, API: {}", api); + return ReturnT.SUCCESS; + } + private ReturnT manualBigObjectDumpV2(SalesforceParam param) throws Exception { String api = param.getApi(); diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java index 8b867eb..554912e 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -1265,7 +1265,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { //批量插入200一次 int page = count % 200 == 0 ? count / 200 : (count / 200) + 1; for (int i = 0; i < page; i++) { - List> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit " + i * 200 + ",200"); + List> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit 200"); SObject[] accounts = new SObject[linkList.size()]; String[] ids = new String[linkList.size()]; int index = 0; @@ -1310,18 +1310,24 @@ public class DataImportNewServiceImpl implements DataImportNewService { printlnAccountsDetails(accounts,dataFields); } SaveResult[] saveResults = connection.create(accounts); + ; for (int j = 0; j < saveResults.length; j++) { - if (!saveResults[j].getSuccess()) { - String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", api, com.alibaba.fastjson2.JSON.toJSONString(DataDumpParam.getFilter()), JSON.toJSONString(saveResults[j])); - log.error(format); - } else { - List> dList = new ArrayList<>(); - Map linkMap = new HashMap<>(); - linkMap.put("key", "new_id"); - linkMap.put("value", saveResults[j].getId()); - dList.add(linkMap); + if (saveResults[j].getSuccess()) { + List> maps = new ArrayList<>(); + Map m = new HashMap<>(); + m.put("key", "new_id"); + m.put("value", saveResults[j].getId()); + maps.add(m); + customMapper.updateById(api, maps, ids[j]); log.info("ContentDocumentLink Id: {}，对应的new_id: {} 更新成功！ " , ids[j], saveResults[j].getId()); - customMapper.updateById("ContentDocumentLink", dList, ids[j]); + }else{ + List> maps = new ArrayList<>(); + Map linkMap1 = new HashMap<>(); + linkMap1.put("key", "error_message"); + linkMap1.put("value", JSON.toJSONString(saveResults[j].getErrors())); + maps.add(linkMap1); + customMapper.updateById(api, maps, ids[j]); + log.error("Id:{},saveResults: {}",ids[j], JSON.toJSONString(saveResults[j])); } } } catch (Exception e) {