feature: 清洗任务详情页 (#73)

* feature: 清洗任务详情

* fix: 取消构建镜像，改为直接拉取

* fix: 增加清洗任务详情页

* fix: 增加清洗任务详情页

* fix: 算子列表可点击

* fix: 模板详情和更新

Author: hhhhsc701

.github/workflows/docker-image-save.yml

@@ -0,0 +1,27 @@
+name: docker-image-save.yml
+on:
+  workflow_call:
+    inputs:
+      service_name:
+        required: true
+        type: string
+
+jobs:
+  pull-and-save:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Pull Docker Image
+        run: |
+          LOWERCASE_REPO=$(echo "${{ github.repository_owner }}" | tr '[:upper:]' '[:lower:]')
+          docker pull ghcr.io/$LOWERCASE_REPO/datamate-${{ inputs.service_name }}:latest
+          docker tag ghcr.io/$LOWERCASE_REPO/datamate-${{ inputs.service_name }}:latest datamate-${{ inputs.service_name }}:latest
+
+      - name: Save Docker Image
+        run: |
+          docker save -o datamate-${{ inputs.service_name }}.tar datamate-${{ inputs.service_name }}:latest
+
+      - name: Upload Docker Image
+        uses: actions/upload-artifact@v4
+        with:
+          name: datamate-${{ inputs.service_name }}
+          path: datamate-${{ inputs.service_name }}.tar

.github/workflows/docker-images-reusable.yml

@@ -47,19 +47,7 @@ jobs:
           make build-${{ inputs.service_name }} VERSION=latest
 
       - name: Tag & Push Docker Image
-        if: github.event_name != 'pull_request' && !startsWith(github.workflow, 'Package')
+        if: github.event_name != 'pull_request'
         run: |
           docker tag datamate-${{ inputs.service_name }}:latest ${{ steps.set-tag.outputs.TAGS }}
-          docker push ${{ steps.set-tag.outputs.TAGS }}
-
-      - name: Save Docker Image
-        if: startsWith(github.workflow, 'Package')
-        run: |
-          docker save -o datamate-${{ inputs.service_name }}.tar datamate-${{ inputs.service_name }}:latest
-
-      - name: Upload Docker Image
-        if: startsWith(github.workflow, 'Package')
-        uses: actions/upload-artifact@v4
-        with:
-          name: datamate-${{ inputs.service_name }}
-          path: datamate-${{ inputs.service_name }}.tar
+          docker push ${{ steps.set-tag.outputs.TAGS }}

.github/workflows/package.yml

@@ -6,23 +6,33 @@ on:
 jobs:
   backend-docker-build:
     name: Build and Push Backend Docker Image
-    uses: ./.github/workflows/docker-image-backend.yml
+    uses: ./.github/workflows/docker-image-save.yml
+    with:
+      service_name: backend
 
   frontend-docker-build:
     name: Build and Push Frontend Docker Image
-    uses: ./.github/workflows/docker-image-frontend.yml
+    uses: ./.github/workflows/docker-image-save.yml
+    with:
+      service_name: frontend
 
   database-docker-build:
     name: Build and Push Database Docker Image
-    uses: ./.github/workflows/docker-image-database.yml
+    uses: ./.github/workflows/docker-image-save.yml
+    with:
+      service_name: database
 
   runtime-docker-build:
     name: Build and Push Runtime Docker Image
-    uses: ./.github/workflows/docker-image-runtime.yml
+    uses: ./.github/workflows/docker-image-save.yml
+    with:
+      service_name: runtime
 
   backend-python-docker-build:
     name: Build and Push Backend Python Docker Image
-    uses: ./.github/workflows/docker-image-backend-python.yml
+    uses: ./.github/workflows/docker-image-save.yml
+    with:
+      service_name: backend-python
 
   package-all:
     needs:
@@ -54,7 +64,7 @@ jobs:
       - name: Upload Package
         uses: actions/upload-artifact@v4
         with:
-          name: datamate
+          name: DataMate
           include-hidden-files: true
           path: |
             deployment/

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

@@ -11,10 +11,7 @@
 import com.datamate.cleaning.domain.repository.OperatorInstanceRepository;
 
 import com.datamate.cleaning.infrastructure.validator.CleanTaskValidator;
-import com.datamate.cleaning.interfaces.dto.CleaningProcess;
-import com.datamate.cleaning.interfaces.dto.CleaningTaskDto;
-import com.datamate.cleaning.interfaces.dto.CreateCleaningTaskRequest;
-import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
+import com.datamate.cleaning.interfaces.dto.*;
 import com.datamate.common.infrastructure.exception.BusinessException;
 import com.datamate.common.infrastructure.exception.SystemErrorCode;
 import com.datamate.datamanagement.application.DatasetApplicationService;
@@ -40,15 +37,19 @@
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
 
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class CleaningTaskService {
-    private final CleaningTaskRepository CleaningTaskRepo;
+    private final CleaningTaskRepository cleaningTaskRepo;
 
     private final OperatorInstanceRepository operatorInstanceRepo;
 
@@ -66,19 +67,24 @@ public class CleaningTaskService {
 
     private final String FLOW_PATH = "/flow";
 
+    private final Pattern LEVEL_PATTERN = Pattern.compile(
+            "\\b(TRACE|DEBUG|INFO|WARN|WARNING|ERROR|FATAL)\\b",
+            Pattern.CASE_INSENSITIVE
+    );
+
     public List<CleaningTaskDto> getTasks(String status, String keywords, Integer page, Integer size) {
-        List<CleaningTaskDto> tasks = CleaningTaskRepo.findTasks(status, keywords, page, size);
+        List<CleaningTaskDto> tasks = cleaningTaskRepo.findTasks(status, keywords, page, size);
         tasks.forEach(this::setProcess);
         return tasks;
     }
 
     private void setProcess(CleaningTaskDto task) {
-        int count = cleaningResultRepo.countByInstanceId(task.getId());
-        task.setProgress(CleaningProcess.of(task.getFileCount(), count));
+        int[] count = cleaningResultRepo.countByInstanceId(task.getId());
+        task.setProgress(CleaningProcess.of(task.getFileCount(), count[0], count[1]));
     }
 
     public int countTasks(String status, String keywords) {
-        return CleaningTaskRepo.findTasks(status, keywords, null, null).size();
+        return cleaningTaskRepo.findTasks(status, keywords, null, null).size();
     }
 
     @Transactional
@@ -105,7 +111,7 @@ public CleaningTaskDto createTask(CreateCleaningTaskRequest request) {
         task.setDestDatasetName(destDataset.getName());
         task.setBeforeSize(srcDataset.getSizeBytes());
         task.setFileCount(srcDataset.getFileCount().intValue());
-        CleaningTaskRepo.insertTask(task);
+        cleaningTaskRepo.insertTask(task);
 
         operatorInstanceRepo.insertInstance(taskId, request.getInstance());
 
@@ -116,14 +122,50 @@ public CleaningTaskDto createTask(CreateCleaningTaskRequest request) {
     }
 
     public CleaningTaskDto getTask(String taskId) {
-        CleaningTaskDto task = CleaningTaskRepo.findTaskById(taskId);
+        CleaningTaskDto task = cleaningTaskRepo.findTaskById(taskId);
         setProcess(task);
+        task.setInstance(operatorInstanceRepo.findOperatorByInstanceId(taskId));
         return task;
     }
 
+    public List<CleaningResultDto> getTaskResults(String taskId) {
+        return cleaningResultRepo.findByInstanceId(taskId);
+    }
+
+    public List<CleaningTaskLog> getTaskLog(String taskId) {
+        String logPath = FLOW_PATH + "/" + taskId + "/output.log";
+        try (Stream<String> lines = Files.lines(Paths.get(logPath))) {
+            List<CleaningTaskLog> logs = new ArrayList<>();
+            AtomicReference<String> lastLevel = new AtomicReference<>("INFO");
+            lines.forEach(line -> {
+                lastLevel.set(getLogLevel(line, lastLevel.get()));
+                CleaningTaskLog log = new CleaningTaskLog();
+                log.setLevel(lastLevel.get());
+                log.setMessage(line);
+                logs.add(log);
+            });
+            return logs;
+        } catch (IOException e) {
+            log.error("Fail to read log file {}", logPath, e);
+            return Collections.emptyList();
+        }
+    }
+
+    private String getLogLevel(String logLine, String defaultLevel) {
+        if (logLine == null || logLine.trim().isEmpty()) {
+            return defaultLevel;
+        }
+
+        Matcher matcher = LEVEL_PATTERN.matcher(logLine);
+        if (matcher.find()) {
+            return matcher.group(1).toUpperCase();
+        }
+        return defaultLevel;
+    }
+
     @Transactional
     public void deleteTask(String taskId) {
-        CleaningTaskRepo.deleteTaskById(taskId);
+        cleaningTaskRepo.deleteTaskById(taskId);
         operatorInstanceRepo.deleteByInstanceId(taskId);
         cleaningResultRepo.deleteByInstanceId(taskId);
     }
@@ -190,7 +232,7 @@ private void writeListMapToJsonlFile(List<Map<String, Object>> mapList, String f
 
         try (BufferedWriter writer = new BufferedWriter(new FileWriter(fileName))) {
             if (!mapList.isEmpty()) { // 检查列表是否为空，避免异常
-                String jsonString = objectMapper.writeValueAsString(mapList.get(0));
+                String jsonString = objectMapper.writeValueAsString(mapList.getFirst());
                 writer.write(jsonString);
 
                 for (int i = 1; i < mapList.size(); i++) {

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

@@ -5,7 +5,7 @@
 import com.datamate.cleaning.domain.repository.OperatorInstanceRepository;
 import com.datamate.cleaning.interfaces.dto.*;
 import com.datamate.cleaning.domain.model.entity.TemplateWithInstance;
-import com.datamate.operator.domain.repository.OperatorRepository;
+import com.datamate.operator.domain.repository.OperatorViewRepository;
 import com.datamate.operator.interfaces.dto.OperatorDto;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
@@ -26,10 +26,11 @@ public class CleaningTemplateService {
 
     private final OperatorInstanceRepository operatorInstanceRepo;
 
-    private final OperatorRepository operatorRepo;
+    private final OperatorViewRepository operatorViewRepo;
 
     public List<CleaningTemplateDto> getTemplates(String keywords) {
-        List<OperatorDto> allOperators = operatorRepo.findAllOperators();
+        List<OperatorDto> allOperators =
+                operatorViewRepo.findOperatorsByCriteria(null, null, null, null, null);
         Map<String, OperatorDto> operatorsMap = allOperators.stream()
                 .collect(Collectors.toMap(OperatorDto::getId, Function.identity()));
         List<TemplateWithInstance> allTemplates = cleaningTemplateRepo.findAllTemplates(keywords);
@@ -39,8 +40,8 @@ public List<CleaningTemplateDto> getTemplates(String keywords) {
             List<TemplateWithInstance> value = twi.getValue();
             CleaningTemplateDto template = new CleaningTemplateDto();
             template.setId(twi.getKey());
-            template.setName(value.get(0).getName());
-            template.setDescription(value.get(0).getDescription());
+            template.setName(value.getFirst().getName());
+            template.setDescription(value.getFirst().getDescription());
             template.setInstance(value.stream().filter(v -> StringUtils.isNotBlank(v.getOperatorId()))
                     .sorted(Comparator.comparingInt(TemplateWithInstance::getOpIndex))
                     .map(v -> {
@@ -50,8 +51,8 @@ public List<CleaningTemplateDto> getTemplates(String keywords) {
                         }
                         return operator;
                     }).toList());
-            template.setCreatedAt(value.get(0).getCreatedAt());
-            template.setUpdatedAt(value.get(0).getUpdatedAt());
+            template.setCreatedAt(value.getFirst().getCreatedAt());
+            template.setUpdatedAt(value.getFirst().getUpdatedAt());
             return template;
         }).toList();
     }
@@ -70,17 +71,22 @@ public CleaningTemplateDto createTemplate(CreateCleaningTemplateRequest request)
     }
 
     public CleaningTemplateDto getTemplate(String templateId) {
-        return cleaningTemplateRepo.findTemplateById(templateId);
+        CleaningTemplateDto template = cleaningTemplateRepo.findTemplateById(templateId);
+        template.setInstance(operatorInstanceRepo.findOperatorByInstanceId(templateId));
+        return template;
     }
 
     @Transactional
     public CleaningTemplateDto updateTemplate(String templateId, UpdateCleaningTemplateRequest request) {
         CleaningTemplateDto template = cleaningTemplateRepo.findTemplateById(templateId);
-        if (template != null) {
-            template.setName(request.getName());
-            template.setDescription(request.getDescription());
-            cleaningTemplateRepo.updateTemplate(template);
+        if (template == null) {
+            return null;
         }
+        template.setName(request.getName());
+        template.setDescription(request.getDescription());
+        cleaningTemplateRepo.updateTemplate(template);
+        operatorInstanceRepo.deleteByInstanceId(templateId);
+        operatorInstanceRepo.insertInstance(templateId, request.getInstance());
         return template;
     }
 

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/application/scheduler/CleaningTaskScheduler.java

@@ -16,6 +16,8 @@
 public class CleaningTaskScheduler {
     private final CleaningTaskRepository cleaningTaskRepo;
 
+    private final RuntimeClient runtimeClient;
+
     private final ExecutorService taskExecutor = Executors.newFixedThreadPool(5);
 
     public void executeTask(String taskId) {
@@ -28,11 +30,11 @@ private void submitTask(String taskId) {
         task.setStatus(CleaningTaskStatusEnum.RUNNING);
         task.setStartedAt(LocalDateTime.now());
         cleaningTaskRepo.updateTask(task);
-        RuntimeClient.submitTask(taskId);
+        runtimeClient.submitTask(taskId);
     }
 
     public void stopTask(String taskId) {
-        RuntimeClient.stopTask(taskId);
+        runtimeClient.stopTask(taskId);
         CleaningTaskDto task = new CleaningTaskDto();
         task.setId(taskId);
         task.setStatus(CleaningTaskStatusEnum.STOPPED);

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/domain/model/entity/Operator.java

@@ -3,9 +3,14 @@
 
 import com.baomidou.mybatisplus.extension.repository.IRepository;
 import com.datamate.cleaning.domain.model.entity.CleaningResult;
+import com.datamate.cleaning.interfaces.dto.CleaningResultDto;
+
+import java.util.List;
 
 public interface CleaningResultRepository extends IRepository<CleaningResult> {
     void deleteByInstanceId(String instanceId);
 
-    int countByInstanceId(String instanceId);
+    int[] countByInstanceId(String instanceId);
+
+    List<CleaningResultDto> findByInstanceId(String instanceId);
 }

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/domain/repository/CleaningResultRepository.java

@@ -3,11 +3,14 @@
 import com.baomidou.mybatisplus.extension.repository.IRepository;
 import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
 import com.datamate.cleaning.domain.model.entity.OperatorInstance;
+import com.datamate.operator.interfaces.dto.OperatorDto;
 
 import java.util.List;
 
 public interface OperatorInstanceRepository extends IRepository<OperatorInstance> {
     void insertInstance(String instanceId, List<OperatorInstanceDto> instances);
 
     void deleteByInstanceId(String instanceId);
+
+    List<OperatorDto> findOperatorByInstanceId(String instanceId);
 }