From 6a2492cea79e07a844ec059f44e5ec7a6952761f Mon Sep 17 00:00:00 2001
From: Kris <2893855659@qq.com>
Date: Sun, 28 Sep 2025 14:49:57 +0800
Subject: [PATCH] =?UTF-8?q?[feat]=20=E6=8F=90=E4=BA=A4=E5=88=9D=E7=89=88Bi?=
 =?UTF-8?q?gObject=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../service/impl/CommonBatchServiceImpl.java  | 193 +++++++++++++++---
 .../service/impl/DataVerifyServiceImpl.java   |   3 +-
 .../com/celnet/datadump/util/BulkUtil.java    |   2 +-
 3 files changed, 161 insertions(+), 37 deletions(-)

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 4098224..709774e 100644
--- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
@@ -158,18 +158,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
     
     @Override
     public ReturnT<String> importBigObject(SalesforceParam param) throws Exception {
-        List<Future<?>> futures = Lists.newArrayList();
         try {
-            if (StringUtils.isNotBlank(param.getApi())) {
-                // 手动任务
-                ReturnT<String> result = manualBigObjectImport(param, futures);
-                if (result != null) {
-                    return result;
-                }
-            }
-            return ReturnT.SUCCESS;
+            return manualBigObjectImport(param);
         } catch (Throwable throwable) {
-            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
             log.error("import error", throwable);
             throw throwable;
         }
@@ -186,9 +177,11 @@ public class CommonBatchServiceImpl implements CommonBatchService {
 
             log.info("开始导出BigObject数据, API: {}", api);
 
+
             List<DataField> fieldList = dataFieldService.list(new QueryWrapper<DataField>().eq("api", param.getApi()));
             String fieldStr = fieldList.stream()
                     .map(DataField::getField)
+                    .filter(field -> !"SystemModstamp".equals(field) && !"LastModifiedDate".equals(field))
                     .collect(Collectors.joining(", "));
 
             String sql = "select " + fieldStr + " from " + param.getApi() ;
@@ -242,6 +235,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 long currentFileSize = 0;
 
                 int resultIndex = 0;
+
+                String headerLine = null;
                 for (final String resultId : queryResultList.getResult()) {
                     log.info("处理结果ID: {}, 进度: {}/{}", resultId, ++resultIndex, queryResultList.getResult().length);
 
@@ -251,6 +246,13 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                         String line;
                         int lineCount = 0;
                         while ((line = bufferedReader.readLine()) != null) {
+                            // 保存第一行作为标题行，并替换CreatedDate和CreatedById为Original_Created_Date__c和Original_Created_By__c
+                            // 但要确保不替换CreatedDate__c和CreatedById__c这样的自定义字段
+                            line = line.replaceAll("(?<!__)CreatedDate(?!__)", "Original_Created_Date__c")
+                                    .replaceAll("(?<!__)CreatedById(?!__)", "Original_Created_By__c");
+                            if (headerLine == null) {
+                                headerLine = line;
+                            }
                             lineCount++;
                             byte[] lineBytes = (line + "\n").getBytes(StandardCharsets.UTF_8);
 
@@ -272,7 +274,13 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                                 fileWriter = new FileWriter(currentFilePath.toFile(), true);
                                 bufferedWriter = new BufferedWriter(fileWriter);
                                 currentFileSize = 0;
-                                lineCount = 0;
+                                lineCount = 1; // 重置行计数，新文件将写入标题行
+                                
+                                // 在新文件中写入标题行
+                                if (headerLine != null) {
+                                    bufferedWriter.write(headerLine + "\n");
+                                    currentFileSize += (headerLine + "\n").getBytes(StandardCharsets.UTF_8).length;
+                                }
                             }
 
                             // 写入数据
@@ -304,13 +312,14 @@ public class CommonBatchServiceImpl implements CommonBatchService {
     }
 
 
-private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future<?>> futures) throws Exception {
+    private ReturnT<String> manualBigObjectImport(SalesforceParam param) throws Exception {
         String api = param.getApi();
         
         // 创建目标ORG连接
         BulkConnection targetBulkConnect = salesforceTargetConnect.createBulkConnect();
 
-        File directory = new File(api);
+        // 修改目录路径，使用与导出相同的路径规范
+        File directory = new File(Const.SERVER_FILE_PATH + "/" + api);
         if (!directory.exists() || !directory.isDirectory()) {
             throw new FileNotFoundException(api + "目录不存在: " + directory.getAbsolutePath());
         }
@@ -327,26 +336,61 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
 
         // 使用Bulk V2 API导入所有文件
         for (File csvFile : files) {
-            String filePath = csvFile.getAbsolutePath();
-            log.info("处理CSV文件: {}", filePath);
-            
-            // 创建Bulk V2 API作业
-            String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint();
-            String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
-            
-            // 创建Bulk V2作业
-            String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect);
-            log.info("创建Bulk V2作业成功, 作业ID: {}", jobId);
-            
-            // 上传CSV文件数据
-            uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath);
-            
-            // 关闭作业
-            closeBulkV2Job(url, jobId, targetBulkConnect);
-            
-            // 等待作业完成
-            BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect);
-            log.info("文件 {} 数据导入完成", csvFile.getName());
+            try {
+                String filePath = csvFile.getAbsolutePath();
+                log.info("开始处理CSV文件: {}", filePath);
+                
+                // 检查CSV文件是否为空或只有一行标题
+                try (BufferedReader reader = new BufferedReader(new FileReader(csvFile))) {
+                    String firstLine = reader.readLine();
+                    if (firstLine == null || firstLine.trim().isEmpty()) {
+                        log.warn("CSV文件为空，跳过处理: {}", filePath);
+                        continue;
+                    }
+                    
+                    String secondLine = reader.readLine();
+                    if (secondLine == null || secondLine.trim().isEmpty()) {
+                        log.warn("CSV文件只包含标题行，跳过处理: {}", filePath);
+                        continue;
+                    }
+                }
+                
+                // 创建Bulk V2 API作业
+                String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint();
+                String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
+                log.debug("Salesforce实例URL: {}", url);
+                
+                // 创建Bulk V2作业
+                String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect);
+                log.info("创建Bulk V2作业成功, 作业ID: {}", jobId);
+                
+                // 上传CSV文件数据
+                log.info("开始上传CSV文件数据，文件路径: {}, 作业ID: {}", filePath, jobId);
+                uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath);
+                log.info("CSV文件数据上传完成，文件路径: {}, 作业ID: {}", filePath, jobId);
+
+                // 上传文件完成
+                log.info("开始关闭文件上传作业，作业ID: {}", jobId);
+                closeBulkV2Job(url, jobId, targetBulkConnect);
+                log.info("作业关闭请求已发送，作业ID: {}", jobId);
+
+                // 等待作业完成
+                log.info("等待作业完成，作业ID: {}", jobId);
+                BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect);
+                log.info("文件 {} 数据导入完成，作业ID: {}", csvFile.getName(), jobId);
+                
+                // 下载作业结果文件（成功、失败、未处理记录）
+                log.info("开始下载作业结果文件，作业ID: {}", jobId);
+                downloadJobResultFiles(url, jobId, targetBulkConnect, api, csvFile.getName());
+                log.info("作业结果文件下载完成，作业ID: {}", jobId);
+
+                TimeUnit.SECONDS.sleep(30);
+
+            } catch (InterruptedException interruptedExc ){
+                return ReturnT.FAIL;
+            }catch (Exception e) {
+                log.error("处理文件 {} 时发生错误: {}", csvFile.getName(), e.getMessage(), e);
+            }
         }
 
         log.info("所有BigObject数据导入完成: {}", api);
@@ -365,16 +409,25 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
     private void uploadCsvDataToJob(String url, String jobId, BulkConnection connection, String csvFilePath) throws Exception {
         CloseableHttpClient httpClient = HttpClients.createDefault();
         String uploadUrl = url + "services/data/v56.0/jobs/ingest/" + jobId + "/batches/";
+
+        // 创建HttpPut请求，用于上传CSV文件数据到Bulk API 2.0作业
         HttpPut httpPut = new HttpPut(uploadUrl);
+        // 设置请求内容类型为CSV格式
         httpPut.setHeader("Content-Type", "text/csv");
+        // 设置认证信息，使用Bearer Token方式认证
         httpPut.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
+        // 设置接受的响应内容类型为JSON格式
+        httpPut.setHeader("Accept", "application/json");
+        // 设置美化打印输出，便于调试和日志查看
+        httpPut.setHeader("X-PrettyPrint", "1");
         
         FileEntity fileEntity = new FileEntity(new File(csvFilePath));
         httpPut.setEntity(fileEntity);
-        
+
         CloseableHttpResponse response = httpClient.execute(httpPut);
         try {
             int statusCode = response.getStatusLine().getStatusCode();
+
             if (statusCode != 201) { // 201 Created
                 String errorBody = EntityUtils.toString(response.getEntity());
                 throw new RuntimeException("Failed to upload data to Bulk API 2.0 job, status code: " + statusCode + ", error: " + errorBody);
@@ -420,6 +473,78 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
         log.info("作业关闭成功，作业ID: {}", jobId);
     }
     
+    /**
+     * 下载Bulk API V2作业的结果文件（成功、失败、未处理记录）
+     * 
+     * @param url Salesforce实例URL
+     * @param jobId 作业ID
+     * @param connection BulkConnection连接对象
+     * @param api API名称
+     * @param sourceFileName 源文件名
+     * @throws Exception 下载过程中发生错误时抛出异常
+     */
+    private void downloadJobResultFiles(String url, String jobId, BulkConnection connection, String api, String sourceFileName) throws Exception {
+        // 确保结果目录存在
+        String resultDirPath = Const.SERVER_FILE_PATH + "/" + api + "/results";
+        Path resultDir = Paths.get(resultDirPath);
+        if (!Files.exists(resultDir)) {
+            Files.createDirectories(resultDir);
+            log.info("创建结果目录: {}", resultDir.toAbsolutePath());
+        }
+        
+        // 构造基础文件名（不含扩展名）
+        String baseFileName = sourceFileName;
+        if (sourceFileName.endsWith(".csv")) {
+            baseFileName = sourceFileName.substring(0, sourceFileName.length() - 4);
+        }
+        
+        // 下载成功记录
+        downloadJobResultFile(url, jobId, connection, "successfulResults", resultDirPath, baseFileName + "_successful.csv");
+        
+        // 下载失败记录
+        downloadJobResultFile(url, jobId, connection, "failedResults", resultDirPath, baseFileName + "_failed.csv");
+        
+        // 下载未处理记录
+        downloadJobResultFile(url, jobId, connection, "unprocessedrecords", resultDirPath, baseFileName + "_unprocessed.csv");
+    }
+    
+    /**
+     * 下载单个作业结果文件
+     * 
+     * @param url Salesforce实例URL
+     * @param jobId 作业ID
+     * @param connection BulkConnection连接对象
+     * @param resultType 结果类型 (successfulResults, failedResults, unprocessedrecords)
+     * @param resultDirPath 结果目录路径
+     * @param fileName 文件名
+     * @throws Exception 下载过程中发生错误时抛出异常
+     */
+    private void downloadJobResultFile(String url, String jobId, BulkConnection connection, 
+                                      String resultType, String resultDirPath, String fileName) throws Exception {
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        String resultUrl = url + "services/data/v56.0/jobs/ingest/" + jobId + "/" + resultType;
+        HttpGet httpGet = new HttpGet(resultUrl);
+        httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
+        httpGet.setHeader("Content-Type", "application/json; charset=UTF-8");
+        
+        CloseableHttpResponse response = httpClient.execute(httpGet);
+        try {
+            int statusCode = response.getStatusLine().getStatusCode();
+            if (statusCode == 200) {
+                String responseBody = EntityUtils.toString(response.getEntity());
+                Path resultFilePath = Paths.get(resultDirPath, fileName);
+                Files.write(resultFilePath, responseBody.getBytes(StandardCharsets.UTF_8));
+                log.info("下载{}文件成功: {}", resultType, resultFilePath.toAbsolutePath());
+            } else {
+                String errorBody = EntityUtils.toString(response.getEntity());
+                log.warn("下载{}文件失败，状态码: {}，错误: {}", resultType, statusCode, errorBody);
+            }
+        } finally {
+            response.close();
+            httpClient.close();
+        }
+    }
+    
     private ReturnT<String> manualBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
         List<String> apis;
         apis = DataUtil.toIdList(param.getApi());
diff --git a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
index 3aaa99d..e0deb3e 100644
--- a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
@@ -652,7 +652,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
                     targetRow.add(targetField.isNillable());
                     targetRow.add(targetField.isUnique());
                     targetRow.add(targetField.isPolymorphicForeignKey() ? "是" : "否");
-                    targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", sourceField.getReferenceTo()));
+                    targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", targetField.getReferenceTo()));
 
                     // 写入Excel
                     excelWriter.write(Collections.singletonList(targetRow), fieldTypeSheet);
@@ -866,7 +866,6 @@ public class DataVerifyServiceImpl implements DataVerifyService {
         row.add(String.valueOf(totalRecordNum));
         row.add(String.valueOf(sorceOnlyNum));
         row.add(String.valueOf(targetOnlyNum));
-        row.add(String.valueOf(totalRecordNum * fields.size()));
         row.add(isError?"不通过":"通过");
 
         // 写入Excel的summarySheet
diff --git a/src/main/java/com/celnet/datadump/util/BulkUtil.java b/src/main/java/com/celnet/datadump/util/BulkUtil.java
index 5a9541b..4f50f40 100644
--- a/src/main/java/com/celnet/datadump/util/BulkUtil.java
+++ b/src/main/java/com/celnet/datadump/util/BulkUtil.java
@@ -214,7 +214,7 @@ public class BulkUtil {
      */
     public static void waitForBulkV2JobCompletion(String url, String jobId, BulkConnection connection) throws Exception {
         CloseableHttpClient httpClient = HttpClients.createDefault();
-        String jobStatusUrl = url + "services/data/v56.0/jobs/query/" + jobId;
+        String jobStatusUrl = url + "services/data/v56.0/jobs/ingest/" + jobId;
         boolean jobCompleted = false;
 
         while (!jobCompleted) {