feature：数据集导入数据集支持选择归集任务导入

backend/services/data-collection-service/src/main/java/com/datamate/collection/infrastructure/datax/DataxProcessRunner.java

@@ -6,6 +6,7 @@
 import com.datamate.collection.domain.process.ProcessRunner;
 import com.datamate.collection.infrastructure.datax.config.MysqlConfig;
 import com.datamate.collection.infrastructure.datax.config.NasConfig;
+import com.datamate.collection.infrastructure.datax.config.ObsConfig;
 import com.datamate.common.infrastructure.exception.BusinessException;
 import com.datamate.common.infrastructure.exception.SystemErrorCode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -92,18 +93,21 @@ private String getJobConfig(CollectionTask task) {
         try {
             ObjectMapper objectMapper = new ObjectMapper();
             TemplateType templateType = task.getTaskType();
-            switch (templateType) {
-                case NAS:
+            return switch (templateType) {
+                case NAS -> {
                     // NAS 特殊处理
                     NasConfig nasConfig = objectMapper.readValue(task.getConfig(), NasConfig.class);
-                    return nasConfig.toJobConfig(objectMapper, task);
-                case OBS:
-                case MYSQL:
+                    yield nasConfig.toJobConfig(objectMapper, task);
+                }
+                case OBS -> {
+                    ObsConfig obsConfig = objectMapper.readValue(task.getConfig(), ObsConfig.class);
+                    yield obsConfig.toJobConfig(objectMapper, task);
+                }
+                case MYSQL -> {
                     MysqlConfig mysqlConfig = objectMapper.readValue(task.getConfig(), MysqlConfig.class);
-                    return mysqlConfig.toJobConfig(objectMapper, task);
-                default:
-                    throw BusinessException.of(SystemErrorCode.UNKNOWN_ERROR, "Unsupported template type: " + templateType);
-            }
+                    yield mysqlConfig.toJobConfig(objectMapper, task);
+                }
+            };
         } catch (Exception e) {
             log.error("Failed to parse task config", e);
             throw new RuntimeException("Failed to parse task config", e);

backend/services/data-collection-service/src/main/java/com/datamate/collection/infrastructure/datax/config/ObsConfig.java

@@ -0,0 +1,61 @@
+package com.datamate.collection.infrastructure.datax.config;
+
+import com.datamate.collection.domain.model.entity.CollectionTask;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * OBS 归集配置类
+ *
+ * @since 2025/11/18
+ */
+@Getter
+@Setter
+public class ObsConfig implements  BaseConfig{
+    private String endpoint;
+    private String bucket;
+    private String accessKey;
+    private String secretKey;
+    private String prefix;
+
+    /**
+     * 将当前 OBS 配置构造成 DataX 所需的 job JSON 字符串。
+     */
+    public String toJobConfig(ObjectMapper objectMapper, CollectionTask task) throws Exception {
+        Map<String, Object> parameter = new HashMap<>();
+        if (endpoint != null) parameter.put("endpoint", endpoint);
+        if (bucket != null) parameter.put("bucket", bucket);
+        if (accessKey != null) parameter.put("accessKey", accessKey);
+        if (secretKey != null) parameter.put("secretKey", secretKey);
+        if (prefix != null) parameter.put("prefix", prefix);
+        parameter.put("destPath", task.getTargetPath());
+
+        Map<String, Object> job = new HashMap<>();
+        Map<String, Object> content = new HashMap<>();
+        Map<String, Object> reader = new HashMap<>();
+        reader.put("name", "obsreader");
+        reader.put("parameter", parameter);
+        content.put("reader", reader);
+
+        Map<String, Object> writer = new HashMap<>();
+        writer.put("name", "obswriter");
+        writer.put("parameter", parameter);
+        content.put("writer", writer);
+
+        job.put("content", List.of(content));
+        Map<String, Object> setting = new HashMap<>();
+        Map<String, Object> channel = new HashMap<>();
+        channel.put("channel", 2);
+        setting.put("speed", channel);
+        job.put("setting", setting);
+
+        Map<String, Object> jobConfig = new HashMap<>();
+        jobConfig.put("job", job);
+        return objectMapper.writeValueAsString(jobConfig);
+    }
+}

backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetApplicationService.java

@@ -11,14 +11,12 @@
 import com.datamate.datamanagement.domain.model.dataset.Tag;
 import com.datamate.datamanagement.infrastructure.client.CollectionTaskClient;
 import com.datamate.datamanagement.infrastructure.client.dto.CollectionTaskDetailResponse;
-import com.datamate.datamanagement.infrastructure.client.dto.LocalCollectionConfig;
 import com.datamate.datamanagement.infrastructure.exception.DataManagementErrorCode;
 import com.datamate.datamanagement.infrastructure.persistence.mapper.TagMapper;
 import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetFileRepository;
 import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetRepository;
 import com.datamate.datamanagement.interfaces.converter.DatasetConverter;
 import com.datamate.datamanagement.interfaces.dto.*;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
@@ -28,11 +26,13 @@
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.StringUtils;
 
+import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.*;
-import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * 数据集应用服务（对齐 DB schema，使用 UUID 字符串主键）
@@ -46,8 +46,7 @@ public class DatasetApplicationService {
     private final TagMapper tagMapper;
     private final DatasetFileRepository datasetFileRepository;
     private final CollectionTaskClient collectionTaskClient;
-    private final FileMetadataService fileMetadataService;
-    private final ObjectMapper objectMapper;
+    private final DatasetFileApplicationService datasetFileApplicationService;
 
     @Value("${datamate.data-management.base-path:/dataset}")
     private String datasetBasePath;
@@ -223,68 +222,38 @@ public AllDatasetStatisticsResponse getAllDatasetStatistics() {
     @Async
     public void processDataSourceAsync(String datasetId, String dataSourceId) {
         try {
-            log.info("开始处理数据源文件扫描，数据集ID: {}, 数据源ID: {}", datasetId, dataSourceId);
+            log.info("Initiating data source file scanning, dataset ID: {}, collection task ID: {}", datasetId, dataSourceId);
             List<String> filePaths = getFilePaths(dataSourceId);
             if (CollectionUtils.isEmpty(filePaths)) {
                 return;
             }
-            log.info("开始扫描文件，共 {} 个文件路径", filePaths.size());
-
-            List<DatasetFile> datasetFiles = fileMetadataService.scanFiles(filePaths, datasetId);
-            // 查询数据集中已存在的文件
-            List<DatasetFile> existDatasetFileList = datasetFileRepository.findAllByDatasetId(datasetId);
-            Map<String, DatasetFile> existDatasetFilePathMap = existDatasetFileList.stream().collect(Collectors.toMap(DatasetFile::getFilePath, Function.identity()));
-            Dataset dataset = datasetRepository.getById(datasetId);
-            dataset.setFiles(existDatasetFileList);
-
-            // 批量同步数据集文件表
-            asyncDatasetFile(datasetFiles, existDatasetFilePathMap, dataset, existDatasetFileList, filePaths);
-            datasetRepository.updateById(dataset);
+            log.info("Starting file scan, total files: {}", filePaths.size());
+            datasetFileApplicationService.copyFilesToDatasetDir(datasetId, new CopyFilesRequest(filePaths));
         } catch (Exception e) {
             log.error("处理数据源文件扫描失败，数据集ID: {}, 数据源ID: {}", datasetId, dataSourceId, e);
         }
     }
 
-    private void asyncDatasetFile(List<DatasetFile> datasetFiles, Map<String, DatasetFile> existDatasetFilePathMap, Dataset dataset, List<DatasetFile> existDatasetFileList, List<String> filePaths) {
-        if (CollectionUtils.isNotEmpty(datasetFiles)) {
-            for (DatasetFile datasetFile : datasetFiles) {
-                if (existDatasetFilePathMap.containsKey(datasetFile.getFilePath())) {
-                    DatasetFile existDatasetFile = existDatasetFilePathMap.get(datasetFile.getFilePath());
-                    dataset.removeFile(existDatasetFile);
-                    existDatasetFile.setFileSize(datasetFile.getFileSize());
-                    dataset.addFile(existDatasetFile);
-                    dataset.active();
-                    datasetFileRepository.updateById(existDatasetFile);
-                } else {
-                    dataset.addFile(datasetFile);
-                    dataset.active();
-                    datasetFileRepository.save(datasetFile);
-                }
-            }
-            log.info("文件元数据写入完成，共写入 {} 条记录", datasetFiles.size());
-        } else {
-            log.warn("未扫描到有效文件");
-        }
-        for (DatasetFile datasetFile : existDatasetFileList) {
-            String existFilePath = datasetFile.getFilePath();
-            for (String filePath : filePaths) {
-                if (existFilePath.equals(filePath) || existFilePath.startsWith(filePath)) {
-                    if (Files.notExists(Paths.get(existFilePath))) {
-                        dataset.removeFile(datasetFile);
-                        datasetFileRepository.removeById(datasetFile.getId());
-                    }
-                }
-            }
-        }
-    }
-
     private List<String> getFilePaths(String dataSourceId) {
         CollectionTaskDetailResponse taskDetail = collectionTaskClient.getTaskDetail(dataSourceId).getData();
         if (taskDetail == null) {
-            log.warn("获取归集任务详情失败，任务ID: {}", dataSourceId);
+            log.warn("Fail to get collection task detail, task ID: {}", dataSourceId);
+            return Collections.emptyList();
+        }
+        Path targetPath = Paths.get(taskDetail.getTargetPath());
+        if (!Files.exists(targetPath) || !Files.isDirectory(targetPath)) {
+            log.warn("Target path not exists or is not a directory: {}", taskDetail.getTargetPath());
+            return Collections.emptyList();
+        }
+
+        try (Stream<Path> paths = Files.walk(targetPath, 1)) {
+            return paths
+                .filter(Files::isRegularFile)  // 只保留文件，排除目录
+                .map(Path::toString)           // 转换为字符串路径
+                .collect(Collectors.toList());
+        } catch (IOException e) {
+            log.error("Fail to scan directory: {}", targetPath, e);
             return Collections.emptyList();
         }
-        log.info("获取到归集任务详情: {}", taskDetail);
-        return Collections.singletonList(taskDetail.getTargetPath());
     }
 }

backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetFileApplicationService.java

@@ -7,10 +7,12 @@
 import com.datamate.common.infrastructure.exception.BusinessAssert;
 import com.datamate.common.infrastructure.exception.BusinessException;
 import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.datamanagement.common.enums.DuplicateMethod;
 import com.datamate.datamanagement.domain.contants.DatasetConstant;
 import com.datamate.datamanagement.domain.model.dataset.Dataset;
 import com.datamate.datamanagement.domain.model.dataset.DatasetFile;
 import com.datamate.datamanagement.domain.model.dataset.DatasetFileUploadCheckInfo;
+import com.datamate.datamanagement.infrastructure.exception.DataManagementErrorCode;
 import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetFileRepository;
 import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetRepository;
 import com.datamate.datamanagement.interfaces.converter.DatasetConverter;
@@ -45,6 +47,8 @@
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipOutputStream;
 
@@ -63,6 +67,9 @@ public class DatasetFileApplicationService {
     @Value("${datamate.data-management.base-path:/dataset}")
     private String datasetBasePath;
 
+    @Value("${datamate.data-management.file.duplicate:COVER}")
+    private DuplicateMethod duplicateMethod;
+
     @Autowired
     public DatasetFileApplicationService(DatasetFileRepository datasetFileRepository,
                                          DatasetRepository datasetRepository, FileService fileService) {
@@ -254,13 +261,36 @@ private void saveFileInfoToDb(FileUploadResult fileUploadResult, UploadFileReque
                 .filePath(savedFile.getPath())
                 .fileType(AnalyzerUtils.getExtension(uploadFile.getFileName()))
                 .build();
-
-        datasetFileRepository.save(datasetFile);
+        dataset.setFiles(datasetFileRepository.findAllByDatasetId(datasetId));
+        setDatasetFileId(datasetFile, dataset);
+        datasetFileRepository.saveOrUpdate(datasetFile);
         dataset.addFile(datasetFile);
         dataset.active();
         datasetRepository.updateById(dataset);
     }
 
+    /**
+     * 为数据集文件设置文件id
+     *
+     * @param datasetFile 要设置id的文件
+     * @param dataset 数据集（包含文件列表）
+     */
+    private void setDatasetFileId(DatasetFile datasetFile, Dataset dataset) {
+        Map<String, DatasetFile> existDatasetFilMap = dataset.getFiles().stream().collect(Collectors.toMap(DatasetFile::getFilePath, Function.identity()));
+        DatasetFile existDatasetFile = existDatasetFilMap.get(datasetFile.getFilePath());
+        if (Objects.isNull(existDatasetFile)) {
+            return;
+        }
+        if (duplicateMethod == DuplicateMethod.ERROR) {
+            log.error("file {} already exists in dataset {}", datasetFile.getFileName(), datasetFile.getDatasetId());
+            throw BusinessException.of(DataManagementErrorCode.DATASET_FILE_ALREADY_EXISTS);
+        }
+        if (duplicateMethod == DuplicateMethod.COVER) {
+            dataset.removeFile(existDatasetFile);
+            datasetFile.setId(existDatasetFile.getId());
+        }
+    }
+
     /**
      * 复制文件到数据集目录
      *
@@ -273,6 +303,8 @@ public List<DatasetFile> copyFilesToDatasetDir(String datasetId, CopyFilesReques
         Dataset dataset = datasetRepository.getById(datasetId);
         BusinessAssert.notNull(dataset, SystemErrorCode.RESOURCE_NOT_FOUND);
         List<DatasetFile> copiedFiles = new ArrayList<>();
+        List<DatasetFile> existDatasetFiles = datasetFileRepository.findAllByDatasetId(datasetId);
+        dataset.setFiles(existDatasetFiles);
         for (String sourceFilePath : req.sourcePaths()) {
             Path sourcePath = Paths.get(sourceFilePath);
             if (!Files.exists(sourcePath) || !Files.isRegularFile(sourcePath)) {
@@ -292,10 +324,11 @@ public List<DatasetFile> copyFilesToDatasetDir(String datasetId, CopyFilesReques
                     .uploadTime(currentTime)
                     .lastAccessTime(currentTime)
                     .build();
+            setDatasetFileId(datasetFile, dataset);
             dataset.addFile(datasetFile);
             copiedFiles.add(datasetFile);
         }
-        datasetFileRepository.saveBatch(copiedFiles, 100);
+        datasetFileRepository.saveOrUpdateBatch(copiedFiles, 100);
         dataset.active();
         datasetRepository.updateById(dataset);
         CompletableFuture.runAsync(() -> copyFilesToDatasetDir(req.sourcePaths(), dataset));