feat: 清洗任务增加重试次数记录和日志展示 (#280)

* feat: 清洗任务增加重试次数记录和日志展示

* feat: 数据清洗改为数据处理

Author: hhhhsc701

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

@@ -142,7 +142,7 @@ public CleaningTaskDto createTask(CreateCleaningTaskRequest request) {
 
         prepareTask(task, request.getInstance(), executorType);
         scanDataset(taskId, request.getSrcDatasetId());
-        taskScheduler.executeTask(taskId);
+
         return task;
     }
 
@@ -157,9 +157,12 @@ public List<CleaningResultDto> getTaskResults(String taskId) {
         return cleaningResultRepo.findByInstanceId(taskId);
     }
 
-    public List<CleaningTaskLog> getTaskLog(String taskId) {
+    public List<CleaningTaskLog> getTaskLog(String taskId, int retryCount) {
         cleanTaskValidator.checkTaskId(taskId);
         String logPath = FLOW_PATH + "/" + taskId + "/output.log";
+        if (retryCount > 0) {
+            logPath += "." + retryCount;
+        }
         try (Stream<String> lines = Files.lines(Paths.get(logPath))) {
             List<CleaningTaskLog> logs = new ArrayList<>();
             AtomicReference<String> lastLevel = new AtomicReference<>("INFO");
@@ -215,7 +218,7 @@ public void executeTask(String taskId) {
         CleaningTaskDto task = cleaningTaskRepo.findTaskById(taskId);
         scanDataset(taskId, task.getSrcDatasetId(), succeedSet);
         cleaningResultRepo.deleteByInstanceId(taskId, "FAILED");
-        taskScheduler.executeTask(taskId);
+        taskScheduler.executeTask(taskId, task.getRetryCount() + 1);
     }
 
     private void prepareTask(CleaningTaskDto task, List<OperatorInstanceDto> instances, ExecutorType executorType) {

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

@@ -20,15 +20,16 @@ public class CleaningTaskScheduler {
 
     private final ExecutorService taskExecutor = Executors.newFixedThreadPool(5);
 
-    public void executeTask(String taskId) {
-        taskExecutor.submit(() -> submitTask(taskId));
+    public void executeTask(String taskId, int retryCount) {
+        taskExecutor.submit(() -> submitTask(taskId, retryCount));
     }
 
-    private void submitTask(String taskId) {
+    private void submitTask(String taskId, int retryCount) {
         CleaningTaskDto task = new CleaningTaskDto();
         task.setId(taskId);
         task.setStatus(CleaningTaskStatusEnum.RUNNING);
         task.setStartedAt(LocalDateTime.now());
+        task.setRetryCount(retryCount);
         cleaningTaskRepo.updateTask(task);
         runtimeClient.submitTask(taskId);
     }

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

@@ -36,6 +36,8 @@ public class CleaningTask extends BaseEntity<String> {
 
     private Integer fileCount;
 
+    private Integer retryCount;
+
     private LocalDateTime startedAt;
 
     private LocalDateTime finishedAt;

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/interfaces/dto/CleaningTaskDto.java

@@ -38,6 +38,8 @@ public class CleaningTaskDto {
 
     private Integer fileCount;
 
+    private Integer retryCount;
+
     private CleaningTaskStatusEnum status;
 
     private String templateId;

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/interfaces/rest/CleaningTaskController.java

@@ -1,6 +1,7 @@
 package com.datamate.cleaning.interfaces.rest;
 
 import com.datamate.cleaning.application.CleaningTaskService;
+import com.datamate.cleaning.application.scheduler.CleaningTaskScheduler;
 import com.datamate.cleaning.interfaces.dto.*;
 import com.datamate.common.interfaces.PagedResponse;
 import lombok.RequiredArgsConstructor;
@@ -18,6 +19,8 @@
 public class CleaningTaskController {
     private final CleaningTaskService cleaningTaskService;
 
+    private final CleaningTaskScheduler taskScheduler;
+
     @GetMapping
     public PagedResponse<CleaningTaskDto> cleaningTasksGet(
             @RequestParam("page") Integer page,
@@ -36,7 +39,9 @@ public CleaningTaskDto cleaningTasksPost(@McpToolParam(description = "创建任
         if (request.getInstance().isEmpty() && StringUtils.isNotBlank(request.getTemplateId())) {
             request.setInstance(cleaningTaskService.getInstanceByTemplateId(request.getTemplateId()));
         }
-        return cleaningTaskService.createTask(request);
+        CleaningTaskDto task = cleaningTaskService.createTask(request);
+        taskScheduler.executeTask(task.getId(), 0);
+        return task;
     }
 
     @PostMapping("/{taskId}/stop")
@@ -74,8 +79,9 @@ public List<CleaningResultDto> cleaningTasksTaskIdGetResult(@PathVariable("taskI
         return cleaningTaskService.getTaskResults(taskId);
     }
 
-    @GetMapping("/{taskId}/log")
-    public List<CleaningTaskLog> cleaningTasksTaskIdGetLog(@PathVariable("taskId") String taskId) {
-        return cleaningTaskService.getTaskLog(taskId);
+    @GetMapping("/{taskId}/log/{retryCount}")
+    public List<CleaningTaskLog> cleaningTasksTaskIdGetLog(@PathVariable("taskId") String taskId,
+                                                           @PathVariable("retryCount") int retryCount) {
+        return cleaningTaskService.getTaskLog(taskId, retryCount);
     }
 }

frontend/src/pages/DataCleansing/Detail/TaskDetail.tsx

@@ -76,10 +76,10 @@ export default function CleansingTaskDetail() {
 
   const [taskLog, setTaskLog] = useState();
 
-  const fetchTaskLog = async () => {
+  const fetchTaskLog = async (retryCount: number) => {
     if (!id) return;
     try {
-      const { data } = await queryCleaningTaskLogByIdUsingGet(id);
+      const { data } = await queryCleaningTaskLogByIdUsingGet(id, retryCount);
       setTaskLog(data);
     } catch (error) {
       message.error("获取清洗日志失败");
@@ -90,7 +90,7 @@ export default function CleansingTaskDetail() {
   const handleRefresh = async () => {
     fetchTaskDetail();
     {activeTab === "files" && await fetchTaskResult()}
-    {activeTab === "logs" && await fetchTaskLog()}
+    {activeTab === "logs" && await fetchTaskLog(task.retryCount)}
   };
 
   useEffect(() => {
@@ -190,7 +190,7 @@ export default function CleansingTaskDetail() {
 
   const breadItems = [
     {
-      title: <Link to="/data/cleansing">数据清洗</Link>,
+      title: <Link to="/data/cleansing">数据处理</Link>,
     },
     {
       title: "清洗任务详情",
@@ -215,7 +215,7 @@ export default function CleansingTaskDetail() {
           )}
           {activeTab === "operators" && <OperatorTable task={task} />}
           {activeTab === "files" && <FileTable result={result} fetchTaskResult={fetchTaskResult} />}
-          {activeTab === "logs" && <LogsTable taskLog={taskLog} fetchTaskLog={fetchTaskLog} />}
+          {activeTab === "logs" && <LogsTable taskLog={taskLog} fetchTaskLog={fetchTaskLog} retryCount={task.retryCount} />}
         </div>
       </div>
     </>

frontend/src/pages/DataCleansing/Detail/TemplateDetail.tsx

@@ -94,7 +94,7 @@ export default function CleansingTemplateDetail() {
 
   const breadItems = [
     {
-      title: <Link to="/data/cleansing">数据清洗</Link>,
+      title: <Link to="/data/cleansing">数据处理</Link>,
     },
     {
       title: "模板详情",

frontend/src/pages/DataCleansing/Detail/components/BasicInfo.tsx

@@ -56,6 +56,8 @@ export default function BasicInfo({ task }: { task: CleansingTask }) {
       ),
       span: 2,
     },
+    { key: "finishedTime", label: "结束时间", children: task?.finishedAt },
+    { key: "name", label: "重试次数", children: task?.retryCount },
   ];
 
   return (

frontend/src/pages/DataCleansing/Detail/components/LogsTable.tsx

@@ -1,16 +1,35 @@
-import {useEffect} from "react";
+import {useEffect, useState} from "react";
 import {useParams} from "react-router";
 import {FileClock} from "lucide-react";
 
-export default function LogsTable({taskLog, fetchTaskLog} : {taskLog: any[], fetchTaskLog: () => Promise<any>}) {
+export default function LogsTable({taskLog, fetchTaskLog, retryCount} : {taskLog: any[], fetchTaskLog: () => Promise<any>, retryCount: number}) {
   const { id = "" } = useParams();
+  const [selectedLog, setSelectedLog] = useState(retryCount + 1);
 
   useEffect(() => {
-    fetchTaskLog();
-  }, [id]);
+    fetchTaskLog(selectedLog - 1);
+  }, [id, selectedLog]);
 
   return taskLog?.length > 0 ? (
     <>
+      {/* --- 新增区域：左上角 Select 组件 --- */}
+      <div className="flex items-center justify-between pb-3">
+        <div className="flex items-center gap-3">
+          <label className="text-sm font-medium text-gray-500">选择运行轮次:</label>
+          <select
+            value={selectedLog}
+            onChange={(e) => setSelectedLog(Number(e.target.value))}
+            className="bg-gray-700 border border-gray-600 !text-white text-sm rounded-md focus:ring-blue-500 focus:border-blue-500 block px-2.5 py-1.5 min-w-[120px]"
+          >
+            {Array.from({ length: retryCount + 1 }, (_, i) => retryCount + 1 - i).map((num) => (
+              <option key={num} value={num}>
+                第 {num} 次
+              </option>
+            ))}
+          </select>
+        </div>
+        <span className="text-s text-gray-500 px-2">当前展示: 第 {selectedLog} 次</span>
+      </div>
       <div className="text-gray-300 p-4 border border-gray-700 bg-gray-800 rounded-lg">
         <div className="font-mono text-sm">
           {taskLog?.map?.((log, index) => (

frontend/src/pages/DataCleansing/Home/DataCleansing.tsx

@@ -22,7 +22,7 @@ export default function DataProcessingPage() {
     <div className="h-full flex flex-col gap-4">
       {/* Header */}
       <div className="flex justify-between items-center">
-        <h1 className="text-xl font-bold">数据清洗</h1>
+        <h1 className="text-xl font-bold">数据处理</h1>
         <div className="flex gap-2">
           <Button
             icon={<PlusOutlined />}