【feat】 源系统BulkV1查询

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<String> 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）
      *

src/main/java/com/celnet/datadump/service/CommonBatchService.java

@@ -11,7 +11,8 @@ public interface CommonBatchService {
 
     ReturnT<String> dumpBigObjectV2(SalesforceParam param) throws Exception;
 
-
     ReturnT<String> importBigObject(SalesforceParam param) throws Exception;
+    
+    ReturnT<String> dumpBigObjectSourceV1(SalesforceParam param) throws Exception;
 
 }

src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java

@@ -182,6 +182,21 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         }
     }
     
+    @Override
+    public ReturnT<String> dumpBigObjectSourceV1(SalesforceParam param) throws Exception {
+        try {
+            // 手动任务
+            ReturnT<String> result = manualBigObjectSourceV1Dump(param);
+            if (result != null) {
+                return result;
+            }
+        } catch (Throwable throwable) {
+            log.error("dump error", throwable);
+            throw throwable;
+        }
+        return ReturnT.SUCCESS;
+    }
+    
     private ReturnT<String> 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<String> 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<String> manualBigObjectDumpV2(SalesforceParam param) throws Exception {
 
         String api = param.getApi();

src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java

@@ -1258,7 +1258,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<Map<String, Object>> 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<Map<String, Object>> 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;
@@ -1303,18 +1303,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()) {
+                            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]));
+                                List<Map<String, Object>> maps = new ArrayList<>();
-                                log.error(format);
+                                Map<String, Object> m = new HashMap<>();
-                            } else {
+                                m.put("key", "new_id");
-                                List<Map<String, Object>> dList = new ArrayList<>();
+                                m.put("value", saveResults[j].getId());
-                                Map<String, Object> linkMap = new HashMap<>();
+                                maps.add(m);
-                                linkMap.put("key", "new_id");
+                                customMapper.updateById(api, maps, ids[j]);
-                                linkMap.put("value", saveResults[j].getId());
-                                dList.add(linkMap);
                                 log.info("ContentDocumentLink Id: {}，对应的new_id: {} 更新成功！ " , ids[j], saveResults[j].getId());
-                                customMapper.updateById("ContentDocumentLink", dList, ids[j]);
+                            }else{
+                                List<Map<String, Object>> maps = new ArrayList<>();
+                                Map<String, Object> 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) {