Merge remote-tracking branch 'origin/develop_930' into develop_930

Author: Dallas98

backend/services/data-collection-service/pom.xml

@@ -164,6 +164,37 @@
                     <classifier>exec</classifier>
                 </configuration>
             </plugin>
+            <plugin>
+              <groupId>org.apache.maven.plugins</groupId>
+              <artifactId>maven-compiler-plugin</artifactId>
+              <version>3.11.0</version>
+              <configuration>
+                <source>${maven.compiler.source}</source>
+                <target>${maven.compiler.target}</target>
+                <annotationProcessorPaths>
+                  <!-- 顺序很重要 -->
+                  <path>
+                    <groupId>org.projectlombok</groupId>
+                    <artifactId>lombok</artifactId>
+                    <version>${lombok.version}</version>
+                  </path>
+                  <path>
+                    <groupId>org.projectlombok</groupId>
+                    <artifactId>lombok-mapstruct-binding</artifactId>
+                    <version>${lombok-mapstruct-binding.version}</version>
+                  </path>
+                  <path>
+                    <groupId>org.mapstruct</groupId>
+                    <artifactId>mapstruct-processor</artifactId>
+                    <version>${mapstruct.version}</version>
+                  </path>
+                </annotationProcessorPaths>
+                <compilerArgs>
+                  <arg>-parameters</arg>
+                  <arg>-Amapstruct.defaultComponentModel=spring</arg>
+                </compilerArgs>
+              </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/CollectionTaskService.java

@@ -13,7 +13,10 @@
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 @Slf4j
 @Service
@@ -29,13 +32,17 @@ public CollectionTask create(CollectionTask task) {
         task.setCreatedAt(LocalDateTime.now());
         task.setUpdatedAt(LocalDateTime.now());
         taskMapper.insert(task);
+        executeTaskNow(task);
+        return task;
+    }
+
+    private void executeTaskNow(CollectionTask task) {
         if (Objects.equals(task.getSyncMode(), SyncMode.ONCE.getValue())) {
             TaskExecution exec = dataxExecutionService.createExecution(task);
             int timeout = task.getTimeoutSeconds() == null ? 3600 : task.getTimeoutSeconds();
             dataxExecutionService.runAsync(task, exec.getId(), timeout);
             log.info("Triggered DataX execution for task {} at {}, execId={}", task.getId(), LocalDateTime.now(), exec.getId());
         }
-        return task;
     }
 
     @Transactional
@@ -63,18 +70,7 @@ public List<CollectionTask> list(Integer page, Integer size, String status, Stri
 
     @Transactional
     public TaskExecution startExecution(CollectionTask task) {
-        TaskExecution exec = new TaskExecution();
-        exec.setId(UUID.randomUUID().toString());
-        exec.setTaskId(task.getId());
-        exec.setTaskName(task.getName());
-        exec.setStatus(TaskStatus.RUNNING);
-        exec.setProgress(0.0);
-        exec.setStartedAt(LocalDateTime.now());
-        exec.setCreatedAt(LocalDateTime.now());
-        executionMapper.insert(exec);
-        taskMapper.updateLastExecution(task.getId(), exec.getId());
-        taskMapper.updateStatus(task.getId(), TaskStatus.RUNNING.name());
-        return exec;
+        return dataxExecutionService.createExecution(task);
     }
 
     // ---- Template related merged methods ----

backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/DataxExecutionService.java

@@ -28,22 +28,15 @@ public class DataxExecutionService {
 
     private final DataxJobBuilder jobBuilder;
     private final DataxProcessRunner processRunner;
-    private final DataxProperties props;
     private final TaskExecutionMapper executionMapper;
     private final CollectionTaskMapper taskMapper;
 
 
     @Transactional
     public TaskExecution createExecution(CollectionTask task) {
-
-        TaskExecution exec = new TaskExecution();
-        exec.setId(UUID.randomUUID().toString());
+        TaskExecution exec = TaskExecution.initTaskExecution();
         exec.setTaskId(task.getId());
         exec.setTaskName(task.getName());
-        exec.setStatus(TaskStatus.RUNNING);
-        exec.setProgress(0.0);
-        exec.setStartedAt(LocalDateTime.now());
-        exec.setCreatedAt(LocalDateTime.now());
         executionMapper.insert(exec);
         taskMapper.updateLastExecution(task.getId(), exec.getId());
         taskMapper.updateStatus(task.getId(), TaskStatus.RUNNING.name());

backend/services/data-collection-service/src/main/java/com/dataengine/collection/domain/model/CollectionTask.java

@@ -1,8 +1,13 @@
 package com.dataengine.collection.domain.model;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Data;
 
 import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.Map;
 
 @Data
 public class CollectionTask {
@@ -22,4 +27,19 @@ public class CollectionTask {
     private LocalDateTime updatedAt;
     private String createdBy;
     private String updatedBy;
+
+    public void addPath() {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            Map<String, Object> parameter = objectMapper.readValue(
+                config,
+                new TypeReference<>() {}
+            );
+            parameter.put("destPath", "/dataset/local/" + id);
+            parameter.put("filePaths", Collections.singletonList(parameter.get("destPath")));
+            config = objectMapper.writeValueAsString(parameter);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

backend/services/data-collection-service/src/main/java/com/dataengine/collection/domain/model/TaskExecution.java

@@ -3,6 +3,7 @@
 import lombok.Data;
 
 import java.time.LocalDateTime;
+import java.util.UUID;
 
 @Data
 public class TaskExecution {
@@ -25,4 +26,14 @@ public class TaskExecution {
     private String config;
     private String result;
     private LocalDateTime createdAt;
+
+    public static TaskExecution initTaskExecution() {
+        TaskExecution exec = new TaskExecution();
+        exec.setId(UUID.randomUUID().toString());
+        exec.setStatus(TaskStatus.RUNNING);
+        exec.setProgress(0.0);
+        exec.setStartedAt(LocalDateTime.now());
+        exec.setCreatedAt(LocalDateTime.now());
+        return exec;
+    }
 }

backend/services/data-collection-service/src/main/java/com/dataengine/collection/infrastructure/runtime/datax/DataxJobBuilder.java

@@ -13,6 +13,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

backend/services/data-collection-service/src/main/java/com/dataengine/collection/interfaces/converter/CollectionTaskConverter.java

@@ -0,0 +1,53 @@
+package com.dataengine.collection.interfaces.converter;
+
+import com.dataengine.collection.domain.model.CollectionTask;
+import com.dataengine.collection.domain.model.DataxTemplate;
+import com.dataengine.collection.interfaces.dto.*;
+import com.dataengine.common.infrastructure.exception.BusinessException;
+import com.dataengine.common.infrastructure.exception.SystemErrorCode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.Named;
+import org.mapstruct.factory.Mappers;
+
+import java.util.Map;
+
+@Mapper
+public interface CollectionTaskConverter {
+    CollectionTaskConverter INSTANCE = Mappers.getMapper(CollectionTaskConverter.class);
+
+    @Mapping(source = "config", target = "config", qualifiedByName = "parseJsonToMap")
+    CollectionTaskResponse toResponse(CollectionTask task);
+
+    CollectionTaskSummary toSummary(CollectionTask task);
+
+    DataxTemplateSummary toTemplateSummary(DataxTemplate template);
+
+    @Mapping(source = "config", target = "config", qualifiedByName = "mapToJsonString")
+    CollectionTask toCollectionTask(CreateCollectionTaskRequest request);
+
+    @Mapping(source = "config", target = "config", qualifiedByName = "mapToJsonString")
+    CollectionTask toCollectionTask(UpdateCollectionTaskRequest request);
+
+    @Named("parseJsonToMap")
+    default Map<String, Object> parseJsonToMap(String json) {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            return objectMapper.readValue(json, Map.class);
+        } catch (Exception e) {
+            throw BusinessException.of(SystemErrorCode.INVALID_PARAMETER);
+        }
+    }
+
+    @Named("mapToJsonString")
+    default String mapToJsonString(Map<String, Object> map) {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            return objectMapper.writeValueAsString(map != null ? map : Map.of());
+        } catch (Exception e) {
+            throw BusinessException.of(SystemErrorCode.INVALID_PARAMETER);
+        }
+    }
+}