数据清洗接口

hhhhsc701 · hhhhsc701 · commit a345d490d6f3 · 2025-09-26T11:24:16.000+08:00
diff --git a/backend/openapi/specs/data-cleaning.yaml b/backend/openapi/specs/data-cleaning.yaml
@@ -142,7 +142,6 @@ paths:
           example: ''
           schema:
             type: string
-            format: uuid
       responses:
         '200':
           description: 成功获取任务详情
@@ -183,7 +182,6 @@ paths:
           example: ''
           schema:
             type: string
-            format: uuid
       responses:
         '204':
           description: 任务删除成功
@@ -288,7 +286,6 @@ paths:
           example: ''
           schema:
             type: string
-            format: uuid
       responses:
         '200':
           description: 成功获取模板详情
@@ -329,7 +326,6 @@ paths:
           example: ''
           schema:
             type: string
-            format: uuid
       requestBody:
         content:
           application/json:
@@ -384,7 +380,6 @@ paths:
           example: ''
           schema:
             type: string
-            format: uuid
       responses:
         '204':
           description: 模板删除成功
@@ -513,7 +508,7 @@ components:
         instance:
           type: array
           items: &ref_3
-            $ref: '#/components/schemas/OperatorResponse'
+            $ref: '#/components/schemas/OperatorInstance'
           description: 模板定义的清洗规则和配置
     CreateCleaningTemplateRequest:
       type: object
@@ -527,8 +522,10 @@ components:
         description:
           type: string
           description: 模板描述
-        instance: &ref_4
-          $ref: '#/components/schemas/OperatorInstance'
+        instance:
+          type: array
+          items: *ref_3
+          description: 任务的具体配置（如果非模板创建，则直接定义）'
     CleaningTemplate:
       type: object
       required:
@@ -539,7 +536,6 @@ components:
       properties:
         id:
           type: string
-          format: uuid
           description: 模板唯一标识符
         name:
           type: string
@@ -549,7 +545,8 @@ components:
           description: 模板描述
         instance:
           type: array
-          items: *ref_3
+          items: &ref_4
+            $ref: '#/components/schemas/OperatorResponse'
           description: 模板定义的清洗规则和配置
         createdAt:
           type: string
@@ -574,7 +571,10 @@ components:
           description: 任务描述
         sourceType:
           type: string
-        instance: *ref_4
+        instance:
+          type: array
+          items: *ref_3
+          description: 任务的具体配置（如果非模板创建，则直接定义）
     ErrorResponse:
       type: object
       properties:
@@ -595,7 +595,6 @@ components:
       properties:
         id:
           type: string
-          format: uuid
           description: 任务唯一标识符
         name:
           type: string
@@ -613,11 +612,10 @@ components:
             - failed
         templateId:
           type: string
-          format: uuid
           description: 关联的模板ID（如果基于模板创建）
         instance:
           type: array
-          items: *ref_3
+          items: *ref_4
           description: 任务的具体配置（如果非模板创建，则直接定义）
         progress:
           $ref: '#/components/schemas/CleaningProcess'
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/scheduler/TaskScheduler.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/scheduler/TaskScheduler.java
@@ -0,0 +1,8 @@
+package com.dataengine.cleaning.application.scheduler;
+
+public class TaskScheduler {
+    public void a() {
+
+
+    }
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTaskService.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTaskService.java
@@ -0,0 +1,66 @@
+package com.dataengine.cleaning.application.service;
+
+
+import com.dataengine.cleaning.infrastructure.persistence.mapper.CleaningTaskMapper;
+import com.dataengine.cleaning.interfaces.dto.CleaningTask;
+import com.dataengine.cleaning.interfaces.dto.CreateCleaningTaskRequest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Service
+public class CleaningTaskService {
+
+    @Autowired
+    private CleaningTaskMapper cleaningTaskMapper;
+
+    private final ExecutorService taskExecutor = Executors.newFixedThreadPool(5);
+
+    public List<CleaningTask> getTasks(String status) {
+        return cleaningTaskMapper.findTasksByStatus(status);
+    }
+
+    @Transactional
+    public CleaningTask createTask(CreateCleaningTaskRequest request) {
+        CleaningTask task = new CleaningTask();
+        task.setName(request.getName());
+        task.setDescription(request.getDescription());
+        task.setStatus(CleaningTask.StatusEnum.PENDING);
+        task.setId(UUID.randomUUID().toString());
+        cleaningTaskMapper.insertTask(task);
+
+        taskExecutor.submit(() -> executeTask(task));
+        return task;
+    }
+
+    public CleaningTask getTask(String taskId) {
+        return cleaningTaskMapper.findTaskById(taskId);
+    }
+
+    @Transactional
+    public void deleteTask(String taskId) {
+        cleaningTaskMapper.deleteTask(taskId);
+    }
+
+    private void executeTask(CleaningTask task) {
+        task.setStatus(CleaningTask.StatusEnum.RUNNING);
+        cleaningTaskMapper.updateTaskStatus(task);
+        String raySubmitId = submitTaskToRay(task);
+        trackTaskStatus(raySubmitId);
+        task.setStatus(CleaningTask.StatusEnum.COMPLETED);
+        cleaningTaskMapper.updateTaskStatus(task);
+    }
+
+    private String submitTaskToRay(CleaningTask task) {
+        return "";
+    }
+
+    private void trackTaskStatus (String submitId) {
+
+    }
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTemplateService.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTemplateService.java
@@ -0,0 +1,51 @@
+package com.dataengine.cleaning.application.service;
+
+
+import com.dataengine.cleaning.infrastructure.persistence.mapper.CleaningTemplateMapper;
+import com.dataengine.cleaning.interfaces.dto.CleaningTemplate;
+import com.dataengine.cleaning.interfaces.dto.CreateCleaningTemplateRequest;
+import com.dataengine.cleaning.interfaces.dto.UpdateCleaningTemplateRequest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import java.util.List;
+
+@Service
+public class CleaningTemplateService {
+
+    @Autowired
+    private CleaningTemplateMapper cleaningTemplateMapper;
+
+    public List<CleaningTemplate> getTemplates() {
+        return cleaningTemplateMapper.findAllTemplates();
+    }
+
+    @Transactional
+    public CleaningTemplate createTemplate(CreateCleaningTemplateRequest request) {
+        CleaningTemplate template = new CleaningTemplate();
+        template.setName(request.getName());
+        template.setDescription(request.getDescription());
+        cleaningTemplateMapper.insertTemplate(template);
+        return template;
+    }
+
+    public CleaningTemplate getTemplate(String templateId) {
+        return cleaningTemplateMapper.findTemplateById(templateId);
+    }
+
+    @Transactional
+    public CleaningTemplate updateTemplate(String templateId, UpdateCleaningTemplateRequest request) {
+        CleaningTemplate template = cleaningTemplateMapper.findTemplateById(templateId);
+        if (template != null) {
+            template.setName(request.getName());
+            template.setDescription(request.getDescription());
+            cleaningTemplateMapper.updateTemplate(template);
+        }
+        return template;
+    }
+
+    @Transactional
+    public void deleteTemplate(String templateId) {
+        cleaningTemplateMapper.deleteTemplate(templateId);
+    }
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTaskMapper.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTaskMapper.java
@@ -0,0 +1,20 @@
+package com.dataengine.cleaning.infrastructure.persistence.mapper;
+
+import com.dataengine.cleaning.interfaces.dto.CleaningTask;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface CleaningTaskMapper {
+    List<CleaningTask> findTasksByStatus(@Param("status") String status);
+
+    CleaningTask findTaskById(@Param("taskId") String taskId);
+
+    void insertTask(CleaningTask task);
+
+    void updateTaskStatus(CleaningTask task);
+
+    void deleteTask(@Param("taskId") String taskId);
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTemplateMapper.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTemplateMapper.java
@@ -0,0 +1,21 @@
+package com.dataengine.cleaning.infrastructure.persistence.mapper;
+
+import com.dataengine.cleaning.interfaces.dto.CleaningTemplate;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface CleaningTemplateMapper {
+
+    List<CleaningTemplate> findAllTemplates();
+
+    CleaningTemplate findTemplateById(@Param("templateId") String templateId);
+
+    void insertTemplate(CleaningTemplate template);
+
+    void updateTemplate(CleaningTemplate template);
+
+    void deleteTemplate(@Param("templateId") String templateId);
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTaskController.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTaskController.java
@@ -0,0 +1,39 @@
+package com.dataengine.cleaning.interfaces.api;
+
+import com.dataengine.cleaning.interfaces.dto.CleaningTask;
+import com.dataengine.cleaning.interfaces.dto.CreateCleaningTaskRequest;
+import com.dataengine.cleaning.application.service.CleaningTaskService;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+
+@RestController
+public class CleaningTaskController implements CleaningTaskApi {
+
+    @Autowired
+    private CleaningTaskService cleaningTaskService;
+
+    @Override
+    public ResponseEntity<List<CleaningTask>> cleaningTasksGet(String status) {
+        return ResponseEntity.ok(cleaningTaskService.getTasks(status));
+    }
+
+    @Override
+    public ResponseEntity<CleaningTask> cleaningTasksPost(CreateCleaningTaskRequest request) {
+        return ResponseEntity.ok(cleaningTaskService.createTask(request));
+    }
+
+    @Override
+    public ResponseEntity<CleaningTask> cleaningTasksTaskIdGet(String taskId) {
+        return ResponseEntity.ok(cleaningTaskService.getTask(taskId));
+    }
+
+    @Override
+    public ResponseEntity<Void> cleaningTasksTaskIdDelete(String taskId) {
+        cleaningTaskService.deleteTask(taskId);
+        return ResponseEntity.noContent().build();
+    }
+}
diff --git a/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTemplateController.java b/backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTemplateController.java
@@ -0,0 +1,46 @@
+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 org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+
+@RestController
+public class CleaningTemplateController implements CleaningTemplateApi {
+
+    @Autowired
+    private CleaningTemplateService cleaningTemplateService;
+
+    @Override
+    public ResponseEntity<List<CleaningTemplate>> cleaningTemplatesGet() {
+        return ResponseEntity.ok(cleaningTemplateService.getTemplates());
+    }
+
+    @Override
+    public ResponseEntity<CleaningTemplate> cleaningTemplatesPost(CreateCleaningTemplateRequest request) {
+        return ResponseEntity.ok(cleaningTemplateService.createTemplate(request));
+    }
+
+    @Override
+    public ResponseEntity<CleaningTemplate> cleaningTemplatesTemplateIdGet(String templateId) {
+        return ResponseEntity.ok(cleaningTemplateService.getTemplate(templateId));
+    }
+
+    @Override
+    public ResponseEntity<CleaningTemplate> cleaningTemplatesTemplateIdPut(String templateId,
+                                                                           UpdateCleaningTemplateRequest request) {
+        return ResponseEntity.ok(cleaningTemplateService.updateTemplate(templateId, request));
+    }
+
+    @Override
+    public ResponseEntity<Void> cleaningTemplatesTemplateIdDelete(String templateId) {
+        cleaningTemplateService.deleteTemplate(templateId);
+        return ResponseEntity.noContent().build();
+    }
+}
diff --git a/backend/services/data-cleaning-service/src/main/resources/mappers/CleaningTaskMapper.xml b/backend/services/data-cleaning-service/src/main/resources/mappers/CleaningTaskMapper.xml
@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.dataengine.cleaning.infrastructure.persistence.mapper.CleaningTaskMapper">
+
+    <select id="findTasksByStatus" resultType="com.dataengine.cleaning.interfaces.dto.CleaningTask">
+        SELECT * FROM t_clean_task WHERE status = #{status}
+    </select>
+
+    <select id="findTaskById" resultType="com.dataengine.cleaning.interfaces.dto.CleaningTask">
+        SELECT * FROM t_clean_task WHERE id = #{taskId}
+    </select>
+
+    <insert id="insertTask">
+        INSERT INTO t_clean_task (id, name, description, status, created_at)
+        VALUES (#{id}, #{name}, #{description}, #{status}, NOW())
+    </insert>
+
+    <update id="updateTaskStatus">
+        UPDATE t_clean_task SET status = #{status} WHERE id = #{id}
+    </update>
+
+    <delete id="deleteTask">
+        DELETE FROM t_clean_task WHERE id = #{taskId}
+    </delete>
+
+</mapper>
diff --git a/backend/services/data-cleaning-service/src/main/resources/mappers/CleaningTemplateMapper.xml b/backend/services/data-cleaning-service/src/main/resources/mappers/CleaningTemplateMapper.xml
diff --git a/backend/services/main-application/src/main/java/com/dataengine/main/DataEnginePlatformApplication.java b/backend/services/main-application/src/main/java/com/dataengine/main/DataEnginePlatformApplication.java
diff --git a/backend/services/operator-market-service/src/main/resources/db/operator-market.sql b/backend/services/operator-market-service/src/main/resources/db/operator-market.sql
