Merge branch 'develop_930' of github.com:ModelEngine-Group/data-platform into develop_930

Author: chenghh-9609

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTaskService.java

@@ -29,6 +29,7 @@
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -50,7 +51,11 @@ public class CleaningTaskService {
 
     public List<CleaningTask> getTasks(String status, String keywords, Integer page, Integer size) {
         Integer offset = page * size;
-        return cleaningTaskMapper.findTasksByStatus(status, keywords, size, offset);
+        return cleaningTaskMapper.findTasks(status, keywords, size, offset);
+    }
+
+    public int countTasks(String status, String keywords) {
+        return cleaningTaskMapper.findTasks(status, keywords, null, null).size();
     }
 
     @Transactional
@@ -68,16 +73,16 @@ public CleaningTask createTask(CreateCleaningTaskRequest request) {
         task.setSrcDatasetName(request.getSrcDatasetName());
         task.setDestDatasetId(datasetResponse.getId());
         task.setDestDatasetName(datasetResponse.getName());
+        task.setBeforeSize(datasetResponse.getTotalSize());
         cleaningTaskMapper.insertTask(task);
 
         List<OperatorInstancePo> instancePos = request.getInstance().stream()
                 .map(OperatorInstanceConverter.INSTANCE::operatorToDo).toList();
         operatorInstanceMapper.insertInstance(taskId, instancePos);
 
         prepareTask(task, request.getInstance());
-        scanDataset(task.getId(), request.getSrcDatasetId());
-
-        taskExecutor.submit(() -> executeTask(task));
+        scanDataset(taskId, request.getSrcDatasetId());
+        executeTask(taskId);
         return task;
     }
 
@@ -90,15 +95,14 @@ public void deleteTask(String taskId) {
         cleaningTaskMapper.deleteTask(taskId);
     }
 
-    public void executeTask(CleaningTask task) {
-        task.setStatus(CleaningTask.StatusEnum.RUNNING);
-        cleaningTaskMapper.updateTaskStatus(task);
-        submitTask(task.getId());
+    public void executeTask(String taskId) {
+        taskExecutor.submit(() -> submitTask(taskId));
     }
 
     private void prepareTask(CleaningTask task, List<OperatorInstance> instances) {
         TaskProcess process = new TaskProcess();
         process.setInstanceId(task.getId());
+        process.setDatasetId(task.getDestDatasetId());
         process.setDatasetPath(FLOW_PATH + "/" + task.getId() + "/dataset.jsonl");
         process.setExportPath(DATASET_PATH + "/" + task.getDestDatasetId());
         process.setExecutorType(ExecutorType.DATA_PLATFORM.getValue());
@@ -137,7 +141,7 @@ private void scanDataset(String taskId, String srcDatasetId) {
             }
             List<Map<String, Object>> files = datasetFile.getContent().stream()
                     .map(content -> Map.of("fileName", (Object) content.getFileName(),
-                            "fileSize", content.getSize() + "B",
+                            "fileSize", content.getSize(),
                             "filePath", content.getFilePath(),
                             "fileType", content.getFileType(),
                             "fileId", content.getId()))
@@ -148,6 +152,11 @@ private void scanDataset(String taskId, String srcDatasetId) {
     }
 
     private void submitTask(String taskId) {
+        CleaningTask task = new CleaningTask();
+        task.setId(taskId);
+        task.setStatus(CleaningTask.StatusEnum.RUNNING);
+        task.setStartedAt(LocalDateTime.now());
+        cleaningTaskMapper.updateTask(task);
         RuntimeClient.submitTask(taskId);
     }
 
@@ -175,6 +184,6 @@ public void stopTask(String taskId) {
         CleaningTask task = new CleaningTask();
         task.setId(taskId);
         task.setStatus(CleaningTask.StatusEnum.STOPPED);
-        cleaningTaskMapper.updateTaskStatus(task);
+        cleaningTaskMapper.updateTask(task);
     }
 }

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTemplateService.java

@@ -1,35 +1,45 @@
 package com.dataengine.cleaning.application.service;
 
 
+import com.dataengine.cleaning.domain.converter.OperatorInstanceConverter;
+import com.dataengine.cleaning.domain.model.OperatorInstancePo;
 import com.dataengine.cleaning.domain.model.TemplateWithInstance;
 import com.dataengine.cleaning.infrastructure.persistence.mapper.CleaningTemplateMapper;
+import com.dataengine.cleaning.infrastructure.persistence.mapper.OperatorInstanceMapper;
 import com.dataengine.cleaning.interfaces.dto.CleaningTemplate;
 import com.dataengine.cleaning.interfaces.dto.CreateCleaningTemplateRequest;
 import com.dataengine.cleaning.interfaces.dto.OperatorResponse;
 import com.dataengine.cleaning.interfaces.dto.UpdateCleaningTemplateRequest;
+import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.apache.ibatis.annotations.Param;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Service
+@RequiredArgsConstructor
 public class CleaningTemplateService {
+    private final CleaningTemplateMapper cleaningTemplateMapper;
 
-    @Autowired
-    private CleaningTemplateMapper cleaningTemplateMapper;
+    private final OperatorInstanceMapper operatorInstanceMapper;
 
-    public List<CleaningTemplate> getTemplates() {
+    public List<CleaningTemplate> getTemplates(String keywords, Integer page, Integer size) {
         List<OperatorResponse> allOperators = cleaningTemplateMapper.findAllOperators();
         Map<String, OperatorResponse> operatorsMap = allOperators.stream()
                 .collect(Collectors.toMap(OperatorResponse::getId, Function.identity()));
 
-        List<TemplateWithInstance> allTemplates = cleaningTemplateMapper.findAllTemplates();
+        Integer offset = null;
+        if (page != null && size != null) {
+            offset = page * size;
+        }
+        List<TemplateWithInstance> allTemplates = cleaningTemplateMapper.findAllTemplates(keywords, size, offset);
         Map<String, List<TemplateWithInstance>> templatesMap = allTemplates.stream()
                 .collect(Collectors.groupingBy(TemplateWithInstance::getId));
         return templatesMap.entrySet().stream().map(twi -> {
@@ -53,12 +63,22 @@ public List<CleaningTemplate> getTemplates() {
         }).toList();
     }
 
+    public int countTemplates(String keywords) {
+        return cleaningTemplateMapper.findAllTemplates(keywords, null, null).size();
+    }
+
     @Transactional
     public CleaningTemplate createTemplate(CreateCleaningTemplateRequest request) {
         CleaningTemplate template = new CleaningTemplate();
+        String templateId = UUID.randomUUID().toString();
+        template.setId(templateId);
         template.setName(request.getName());
         template.setDescription(request.getDescription());
         cleaningTemplateMapper.insertTemplate(template);
+
+        List<OperatorInstancePo> instancePos = request.getInstance().stream()
+                .map(OperatorInstanceConverter.INSTANCE::operatorToDo).toList();
+        operatorInstanceMapper.insertInstance(templateId, instancePos);
         return template;
     }
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/domain/model/TaskProcess.java

@@ -12,6 +12,8 @@
 public class TaskProcess {
     private String instanceId;
 
+    private String datasetId;
+
     private String datasetPath;
 
     private String exportPath;

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/domain/model/TemplateWithInstance.java

@@ -4,7 +4,7 @@
 import lombok.Setter;
 import org.springframework.format.annotation.DateTimeFormat;
 
-import java.time.OffsetDateTime;
+import java.time.LocalDateTime;
 
 
 @Getter
@@ -17,10 +17,10 @@ public class TemplateWithInstance {
     private String description;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime createdAt;
+    private LocalDateTime createdAt;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime updatedAt;
+    private LocalDateTime updatedAt;
 
     private String operatorId;
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTaskMapper.java

@@ -8,14 +8,14 @@
 
 @Mapper
 public interface CleaningTaskMapper {
-    List<CleaningTask> findTasksByStatus(@Param("status") String status, @Param("keywords") String keywords,
+    List<CleaningTask> findTasks(@Param("status") String status, @Param("keywords") String keywords,
                                          @Param("size") Integer size, @Param("offset") Integer offset);
 
     CleaningTask findTaskById(@Param("taskId") String taskId);
 
     void insertTask(CleaningTask task);
 
-    void updateTaskStatus(CleaningTask task);
+    void updateTask(CleaningTask task);
 
     void deleteTask(@Param("taskId") String taskId);
 }

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTemplateMapper.java

@@ -11,7 +11,8 @@
 @Mapper
 public interface CleaningTemplateMapper {
 
-    List<TemplateWithInstance> findAllTemplates();
+    List<TemplateWithInstance> findAllTemplates(@Param("keywords") String keywords,
+                                                @Param("size") Integer size, @Param("offset") Integer offset);
 
     List<OperatorResponse> findAllOperators();
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTaskController.java

@@ -1,21 +1,15 @@
 package com.dataengine.cleaning.interfaces.api;
 
+import com.dataengine.cleaning.application.service.CleaningTaskService;
 import com.dataengine.cleaning.interfaces.dto.CleaningTask;
 import com.dataengine.cleaning.interfaces.dto.CreateCleaningTaskRequest;
-import com.dataengine.cleaning.application.service.CleaningTaskService;
-
+import com.dataengine.common.infrastructure.common.Response;
 import com.dataengine.common.interfaces.PagedResponse;
-import com.dataengine.common.interfaces.Response;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.ResponseEntity;
-import org.springframework.web.bind.annotation.DeleteMapping;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
 
 
 @RestController
@@ -29,8 +23,10 @@ public ResponseEntity<Response<PagedResponse<CleaningTask>>> cleaningTasksGet(
             @RequestParam("page") Integer page,
             @RequestParam("size") Integer size, @RequestParam(value = "status", required = false) String status,
             @RequestParam(value = "keywords", required = false) String keywords) {
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(cleaningTaskService.getTasks(status, keywords, page,
-                size))));
+        List<CleaningTask> tasks = cleaningTaskService.getTasks(status, keywords, page, size);
+        int count = cleaningTaskService.countTasks(status, keywords);
+        int totalPages = (count + size + 1) / size;
+        return ResponseEntity.ok(Response.ok(PagedResponse.of(tasks, page, count, totalPages)));
     }
 
     @PostMapping
@@ -44,11 +40,9 @@ public ResponseEntity<Response<Object>> cleaningTasksStop(@PathVariable("taskId"
         return ResponseEntity.ok(Response.ok(null));
     }
 
-    @PostMapping("/{taskId}/start")
+    @PostMapping("/{taskId}/execute")
     public ResponseEntity<Response<Object>> cleaningTasksStart(@PathVariable("taskId") String taskId) {
-        CleaningTask task = new CleaningTask();
-        task.setId(taskId);
-        cleaningTaskService.executeTask(task);
+        cleaningTaskService.executeTask(taskId);
         return ResponseEntity.ok(Response.ok(null));
     }
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTemplateController.java

@@ -1,12 +1,11 @@
 package com.dataengine.cleaning.interfaces.api;
 
 import com.dataengine.cleaning.application.service.CleaningTemplateService;
-
 import com.dataengine.cleaning.interfaces.dto.CleaningTemplate;
 import com.dataengine.cleaning.interfaces.dto.CreateCleaningTemplateRequest;
 import com.dataengine.cleaning.interfaces.dto.UpdateCleaningTemplateRequest;
+import com.dataengine.common.infrastructure.common.Response;
 import com.dataengine.common.interfaces.PagedResponse;
-import com.dataengine.common.interfaces.Response;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.DeleteMapping;
@@ -16,8 +15,11 @@
 import org.springframework.web.bind.annotation.PutMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.List;
+
 
 @RestController
 @RequestMapping("/cleaning/templates")
@@ -26,8 +28,18 @@ public class CleaningTemplateController {
     private final CleaningTemplateService cleaningTemplateService;
 
     @GetMapping
-    public ResponseEntity<Response<PagedResponse<CleaningTemplate>>> cleaningTemplatesGet() {
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(cleaningTemplateService.getTemplates())));
+    public ResponseEntity<Response<PagedResponse<CleaningTemplate>>> cleaningTemplatesGet(
+            @RequestParam(value = "page", required = false) Integer page,
+            @RequestParam(value = "size", required = false) Integer size,
+            @RequestParam(value = "keywords", required = false) String keyword) {
+        List<CleaningTemplate> templates = cleaningTemplateService.getTemplates(keyword, page, size);
+        if (page != null && size != null) {
+            int count  = cleaningTemplateService.countTemplates(keyword);
+            int totalPages = (count + size + 1) / size;
+            return ResponseEntity.ok(Response.ok(PagedResponse.of(templates, page, count, totalPages)));
+        } else {
+            return ResponseEntity.ok(Response.ok(PagedResponse.of(templates)));
+        }
     }
 
     @PostMapping

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/dto/CleaningTask.java

@@ -3,7 +3,7 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonValue;
 
-import java.time.OffsetDateTime;
+import java.time.LocalDateTime;
 import java.util.List;
 
 import lombok.Getter;
@@ -32,19 +32,23 @@ public class CleaningTask {
 
     private String destDatasetName;
 
+    private long beforeSize;
+
+    private long afterSize;
+
     /**
      * 任务当前状态
      */
     public enum StatusEnum {
-        PENDING("pending"),
+        PENDING("PENDING"),
 
-        RUNNING("running"),
+        RUNNING("RUNNING"),
 
-        COMPLETED("completed"),
+        COMPLETED("COMPLETED"),
 
-        STOPPED("stopped"),
+        STOPPED("STOPPED"),
 
-        FAILED("failed");
+        FAILED("FAILED");
 
         private final String value;
 
@@ -77,12 +81,12 @@ public static StatusEnum fromValue(String value) {
     private CleaningProcess progress;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime createdAt;
+    private LocalDateTime createdAt;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime startedAt;
+    private LocalDateTime startedAt;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime finishedAt;
+    private LocalDateTime finishedAt;
 }
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/dto/CleaningTemplate.java

@@ -1,6 +1,6 @@
 package com.dataengine.cleaning.interfaces.dto;
 
-import java.time.OffsetDateTime;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -25,9 +25,9 @@ public class CleaningTemplate {
     private List<OperatorResponse> instance = new ArrayList<>();
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime createdAt;
+    private LocalDateTime createdAt;
 
     @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private OffsetDateTime updatedAt;
+    private LocalDateTime updatedAt;
 }