Develop op (#35)

* refactor: enhance CleaningTaskService and related components with validation and repository updates
* feature: 支持算子上传创建

Makefile

@@ -82,8 +82,8 @@ deer-flow-docker-build:
 		git clone git@github.com:bytedance/deer-flow.git ../deer-flow; \
 	fi
 	sed -i "s/dark/light/g" "../deer-flow/web/src/components/deer-flow/theme-provider-wrapper.tsx"
-	cp deployment/docker/deer-flow/.env.example ../deer-flow/.env
+	cp -n deployment/docker/deer-flow/.env.example ../deer-flow/.env
-	cp deployment/docker/deer-flow/conf.yaml.example ../deer-flow/conf.yaml
+	cp -n deployment/docker/deer-flow/conf.yaml.example ../deer-flow/conf.yaml
 	cd ../deer-flow && docker compose build
 
 .PHONY: mineru-docker-build
@@ -131,16 +131,16 @@ mineru-k8s-uninstall:
 
 .PHONY: datamate-docker-install
 datamate-docker-install:
-	cd deployment/docker/datamate && cp .env.example .env && docker compose -f docker-compose.yml up -d
+	cd deployment/docker/datamate && cp -n .env.example .env && docker compose -f docker-compose.yml up -d
 
 .PHONY: datamate-docker-uninstall
 datamate-docker-uninstall:
-	cd deployment/docker/datamate && docker compose -f docker-compose.yml down
+	cd deployment/docker/datamate && docker compose -f docker-compose.yml down -v
 
 .PHONY: deer-flow-docker-install
 deer-flow-docker-install:
-	cd deployment/docker/datamate && cp .env.deer-flow.example .env && docker compose -f docker-compose.yml up -d
+	cd deployment/docker/datamate && cp -n .env.deer-flow.example .env && docker compose -f docker-compose.yml up -d
-	cd deployment/docker/deer-flow && cp .env.example .env && cp conf.yaml.example conf.yaml && docker compose -f docker-compose.yml up -d
+	cd deployment/docker/deer-flow && cp -n .env.example .env && cp -n conf.yaml.example conf.yaml && docker compose -f docker-compose.yml up -d
 
 .PHONY: deer-flow-docker-uninstall
 deer-flow-docker-uninstall:

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

@@ -27,6 +27,11 @@
           <artifactId>data-management-service</artifactId>
           <version>${project.version}</version>
         </dependency>
+        <dependency>
+          <groupId>com.datamate</groupId>
+          <artifactId>operator-market-service</artifactId>
+          <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>

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

@@ -1,6 +1,6 @@
 package com.datamate.cleaning;
 
-import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
@@ -8,9 +8,11 @@ import org.springframework.scheduling.annotation.EnableScheduling;
  * 数据归集服务配置类
  * 基于DataX的数据归集和同步服务，支持多种数据源的数据采集和归集
  */
-@SpringBootApplication
 @EnableAsync
 @EnableScheduling
+@ComponentScan(basePackages = {
+    "com.datamate.cleaning"
+})
 public class DataCleaningServiceConfiguration {
     // Configuration class for JAR packaging - no main method needed
 }

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

@@ -10,6 +10,7 @@ import com.datamate.cleaning.domain.repository.CleaningResultRepository;
 import com.datamate.cleaning.domain.repository.CleaningTaskRepository;
 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;
@@ -59,6 +60,8 @@ public class CleaningTaskService {
 
     private final DatasetFileApplicationService datasetFileService;
 
+    private final CleanTaskValidator cleanTaskValidator;
+
     private final String DATASET_PATH = "/dataset";
 
     private final String FLOW_PATH = "/flow";
@@ -80,6 +83,9 @@ public class CleaningTaskService {
 
     @Transactional
     public CleaningTaskDto createTask(CreateCleaningTaskRequest request) {
+        cleanTaskValidator.checkNameDuplication(request.getName());
+        cleanTaskValidator.checkInputAndOutput(request.getInstance());
+
         CreateDatasetRequest createDatasetRequest = new CreateDatasetRequest();
         createDatasetRequest.setName(request.getDestDatasetName());
         createDatasetRequest.setDatasetType(DatasetType.valueOf(request.getDestDatasetType()));

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

@@ -5,6 +5,8 @@ import com.datamate.cleaning.domain.repository.CleaningTemplateRepository;
 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.interfaces.dto.OperatorDto;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
@@ -24,8 +26,10 @@ public class CleaningTemplateService {
 
     private final OperatorInstanceRepository operatorInstanceRepo;
 
+    private final OperatorRepository operatorRepo;
+
     public List<CleaningTemplateDto> getTemplates(String keywords) {
-        List<OperatorDto> allOperators = operatorInstanceRepo.findAllOperators();
+        List<OperatorDto> allOperators = operatorRepo.findAllOperators();
         Map<String, OperatorDto> operatorsMap = allOperators.stream()
                 .collect(Collectors.toMap(OperatorDto::getId, Function.identity()));
         List<TemplateWithInstance> allTemplates = cleaningTemplateRepo.findAllTemplates(keywords);

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/common/exception/CleanErrorCode.java

@@ -12,7 +12,7 @@ public enum CleanErrorCode implements ErrorCode {
      */
     DUPLICATE_TASK_NAME("clean.0001", "清洗任务名称重复"),
 
-    CREATE_DATASET_FAILED("clean.0002", "创建数据集失败");
+    IN_AND_OUT_NOT_MATCH("clean.0002", "算子输入输出不匹配");
 
     private final String code;
     private final String message;

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

@@ -1,26 +0,0 @@
-package com.datamate.cleaning.domain.model;
-
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.List;
-
-
-@Getter
-@Setter
-@NoArgsConstructor
-public class CreateDatasetRequest {
-    /** 数据集名称 */
-    private String name;
-    /** 数据集描述 */
-    private String description;
-    /** 数据集类型 */
-    private String datasetType;
-    /** 标签列表 */
-    private List<String> tags;
-    /** 数据源 */
-    private String dataSource;
-    /** 目标位置 */
-    private String targetLocation;
-}

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

@@ -1,36 +0,0 @@
-package com.datamate.cleaning.domain.model;
-
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.time.LocalDateTime;
-
-
-@Getter
-@Setter
-@NoArgsConstructor
-public class DatasetFileResponse {
-    /** 文件ID */
-    private String id;
-    /** 文件名 */
-    private String fileName;
-    /** 原始文件名 */
-    private String originalName;
-    /** 文件类型 */
-    private String fileType;
-    /** 文件大小（字节） */
-    private Long fileSize;
-    /** 文件状态 */
-    private String status;
-    /** 文件描述 */
-    private String description;
-    /** 文件路径 */
-    private String filePath;
-    /** 上传时间 */
-    private LocalDateTime uploadTime;
-    /** 最后更新时间 */
-    private LocalDateTime lastAccessTime;
-    /** 上传者 */
-    private String uploadedBy;
-}

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

@@ -1,44 +0,0 @@
-package com.datamate.cleaning.domain.model;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.time.LocalDateTime;
-
-/**
- * 数据集实体（与数据库表 t_dm_datasets 对齐）
- */
-@Getter
-@Setter
-@NoArgsConstructor
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class DatasetResponse {
-    /** 数据集ID */
-    private String id;
-    /** 数据集名称 */
-    private String name;
-    /** 数据集描述 */
-    private String description;
-    /** 数据集类型 */
-    private String datasetType;
-    /** 数据集状态 */
-    private String status;
-    /** 数据源 */
-    private String dataSource;
-    /** 目标位置 */
-    private String targetLocation;
-    /** 文件数量 */
-    private Integer fileCount;
-    /** 总大小（字节） */
-    private Long totalSize;
-    /** 完成率（0-100） */
-    private Float completionRate;
-    /** 创建时间 */
-    private LocalDateTime createdAt;
-    /** 更新时间 */
-    private LocalDateTime updatedAt;
-    /** 创建者 */
-    private String createdBy;
-}

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

@@ -1,23 +0,0 @@
-package com.datamate.cleaning.domain.model;
-
-import lombok.Getter;
-import lombok.Setter;
-import java.util.List;
-
-/**
- * 数据集类型响应DTO
- */
-@Getter
-@Setter
-public class DatasetTypeResponse {
-    /** 类型编码 */
-    private String code;
-    /** 类型名称 */
-    private String name;
-    /** 类型描述 */
-    private String description;
-    /** 支持的文件格式 */
-    private List<String> supportedFormats;
-    /** 图标 */
-    private String icon;
-}

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

@@ -1,28 +0,0 @@
-package com.datamate.cleaning.domain.model;
-
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.List;
-
-
-@Getter
-@Setter
-@NoArgsConstructor
-public class PagedDatasetFileResponse {
-    /** 文件内容列表 */
-    private List<DatasetFileResponse> content;
-    /** 当前页码 */
-    private Integer page;
-    /** 每页大小 */
-    private Integer size;
-    /** 总元素数 */
-    private Integer totalElements;
-    /** 总页数 */
-    private Integer totalPages;
-    /** 是否为第一页 */
-    private Boolean first;
-    /** 是否为最后一页 */
-    private Boolean last;
-}

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

@@ -16,4 +16,6 @@ public interface CleaningTaskRepository extends IRepository<CleaningTask> {
     void updateTask(CleaningTaskDto task);
 
     void deleteTaskById(String taskId);
+
+    boolean isNameExist(String name);
 }

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

@@ -1,15 +1,12 @@
 package com.datamate.cleaning.domain.repository;
 
 import com.baomidou.mybatisplus.extension.repository.IRepository;
-import com.datamate.cleaning.interfaces.dto.OperatorDto;
 import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
 import com.datamate.cleaning.domain.model.entity.OperatorInstance;
 
 import java.util.List;
 
 public interface OperatorInstanceRepository extends IRepository<OperatorInstance> {
-    List<OperatorDto> findAllOperators();
-
     void insertInstance(String instanceId, List<OperatorInstanceDto> instances);
 
     void deleteByInstanceId(String instanceId);

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/infrastructure/converter/OperatorInstanceConverter.java

@@ -3,10 +3,10 @@ package com.datamate.cleaning.infrastructure.converter;
 
 import com.datamate.cleaning.domain.model.entity.OperatorInstance;
 import com.datamate.cleaning.domain.model.entity.Operator;
-import com.datamate.cleaning.interfaces.dto.OperatorDto;
 import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
 import com.datamate.common.infrastructure.exception.BusinessException;
 import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.operator.interfaces.dto.OperatorDto;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.mapstruct.Mapper;

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/infrastructure/persistence/Impl/CleaningTaskRepositoryImpl.java

@@ -51,4 +51,10 @@ public class CleaningTaskRepositoryImpl extends CrudRepository<CleaningTaskMappe
     public void deleteTaskById(String taskId) {
         mapper.deleteById(taskId);
     }
+
+    public boolean isNameExist(String name) {
+        LambdaQueryWrapper<CleaningTask> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(CleaningTask::getName, name);
+        return mapper.exists(queryWrapper);
+    }
 }

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/infrastructure/persistence/Impl/OperatorInstanceRepositoryImpl.java

@@ -3,7 +3,6 @@ package com.datamate.cleaning.infrastructure.persistence.Impl;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.repository.CrudRepository;
 import com.datamate.cleaning.infrastructure.converter.OperatorInstanceConverter;
-import com.datamate.cleaning.interfaces.dto.OperatorDto;
 import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
 import com.datamate.cleaning.domain.model.entity.OperatorInstance;
 import com.datamate.cleaning.domain.repository.OperatorInstanceRepository;
@@ -20,11 +19,6 @@ public class OperatorInstanceRepositoryImpl extends CrudRepository<OperatorInsta
     implements OperatorInstanceRepository {
     private final OperatorInstanceMapper mapper;
 
-    @Override
-    public List<OperatorDto> findAllOperators() {
-        return OperatorInstanceConverter.INSTANCE.fromEntityToDto(mapper.findAllOperators());
-    }
-
     @Override
     public void insertInstance(String instanceId, List<OperatorInstanceDto> instances) {
         List<OperatorInstance> operatorInstances = new ArrayList<>();

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/infrastructure/persistence/mapper/OperatorInstanceMapper.java

@@ -1,17 +1,10 @@
 package com.datamate.cleaning.infrastructure.persistence.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.datamate.cleaning.domain.model.entity.Operator;
 import com.datamate.cleaning.domain.model.entity.OperatorInstance;
 import org.apache.ibatis.annotations.Mapper;
-import org.apache.ibatis.annotations.Select;
-
-import java.util.List;
 
 
 @Mapper
 public interface OperatorInstanceMapper extends BaseMapper<OperatorInstance> {
-    @Select("SELECT id, name, description, version, inputs, outputs, runtime, settings, is_star, created_at, " +
-        "updated_at FROM t_operator")
-    List<Operator> findAllOperators();
 }

backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/infrastructure/validator/CleanTaskValidator.java

@@ -0,0 +1,40 @@
+package com.datamate.cleaning.infrastructure.validator;
+
+import com.datamate.cleaning.common.exception.CleanErrorCode;
+import com.datamate.cleaning.domain.repository.CleaningTaskRepository;
+import com.datamate.cleaning.interfaces.dto.OperatorInstanceDto;
+import com.datamate.common.infrastructure.exception.BusinessException;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Locale;
+
+
+@Component
+@RequiredArgsConstructor
+public class CleanTaskValidator {
+    private final CleaningTaskRepository cleaningTaskRepo;
+
+    public void checkNameDuplication (String name) {
+        if (cleaningTaskRepo.isNameExist(name)) {
+            throw BusinessException.of(CleanErrorCode.DUPLICATE_TASK_NAME);
+        }
+    }
+
+    public void checkInputAndOutput (List<OperatorInstanceDto> operators) {
+        if (operators == null || operators.size() <= 1) {
+            return;
+        }
+        for (int i = 1; i < operators.size(); i++) {
+            OperatorInstanceDto front = operators.get(i - 1);
+            OperatorInstanceDto back = operators.get(i);
+            if (!StringUtils.equals(front.getOutputs(), back.getInputs())) {
+                throw BusinessException.of(CleanErrorCode.IN_AND_OUT_NOT_MATCH,
+                    String.format(Locale.ROOT, "ops(name: [%s, %s]) inputs and outputs does not match",
+                        front.getName(), back.getName()));
+            }
+        }
+    }
+}

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

@@ -5,6 +5,7 @@ import com.datamate.cleaning.common.enums.CleaningTaskStatusEnum;
 import java.time.LocalDateTime;
 import java.util.List;
 
+import com.datamate.operator.interfaces.dto.OperatorDto;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.format.annotation.DateTimeFormat;

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

@@ -4,6 +4,7 @@ import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
 
+import com.datamate.operator.interfaces.dto.OperatorDto;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.format.annotation.DateTimeFormat;

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

@@ -1,41 +0,0 @@
-package com.datamate.cleaning.interfaces.dto;
-
-import java.time.LocalDateTime;
-
-import lombok.Getter;
-import lombok.Setter;
-import org.springframework.format.annotation.DateTimeFormat;
-
-/**
- * OperatorDto
- */
-
-@Getter
-@Setter
-public class OperatorDto {
-
-    private String id;
-
-    private String name;
-
-    private String description;
-
-    private String version;
-
-    private String inputs;
-
-    private String outputs;
-
-    private String runtime;
-
-    private String settings;
-
-    private Boolean isStar;
-
-    @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private LocalDateTime createdAt;
-
-    @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-    private LocalDateTime updatedAt;
-}
-

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

@@ -1,6 +1,7 @@
 package com.datamate.cleaning.interfaces.dto;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 
@@ -17,6 +18,14 @@ public class OperatorInstanceDto {
 
     private String id;
 
+    private String name;
+
+    private String inputs;
+
+    private String outputs;
+
+    private List<Integer> categories;
+
     private Map<String, Object> overrides = new HashMap<>();
 }
 

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

@@ -3,10 +3,8 @@ package com.datamate.cleaning.interfaces.rest;
 import com.datamate.cleaning.application.CleaningTaskService;
 import com.datamate.cleaning.interfaces.dto.CleaningTaskDto;
 import com.datamate.cleaning.interfaces.dto.CreateCleaningTaskRequest;
-import com.datamate.common.infrastructure.common.Response;
 import com.datamate.common.interfaces.PagedResponse;
 import lombok.RequiredArgsConstructor;
-import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 
 import java.util.List;
@@ -19,41 +17,41 @@ public class CleaningTaskController {
     private final CleaningTaskService cleaningTaskService;
 
     @GetMapping
-    public ResponseEntity<Response<PagedResponse<CleaningTaskDto>>> cleaningTasksGet(
+    public PagedResponse<CleaningTaskDto> cleaningTasksGet(
             @RequestParam("page") Integer page,
             @RequestParam("size") Integer size, @RequestParam(value = "status", required = false) String status,
             @RequestParam(value = "keywords", required = false) String keywords) {
         List<CleaningTaskDto> 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)));
+        return PagedResponse.of(tasks, page, count, totalPages);
     }
 
     @PostMapping
-    public ResponseEntity<Response<CleaningTaskDto>> cleaningTasksPost(@RequestBody CreateCleaningTaskRequest request) {
+    public CleaningTaskDto cleaningTasksPost(@RequestBody CreateCleaningTaskRequest request) {
-        return ResponseEntity.ok(Response.ok(cleaningTaskService.createTask(request)));
+        return cleaningTaskService.createTask(request);
     }
 
     @PostMapping("/{taskId}/stop")
-    public ResponseEntity<Response<Object>> cleaningTasksStop(@PathVariable("taskId") String taskId) {
+    public String cleaningTasksStop(@PathVariable("taskId") String taskId) {
         cleaningTaskService.stopTask(taskId);
-        return ResponseEntity.ok(Response.ok(null));
+        return taskId;
     }
 
     @PostMapping("/{taskId}/execute")
-    public ResponseEntity<Response<Object>> cleaningTasksStart(@PathVariable("taskId") String taskId) {
+    public String cleaningTasksStart(@PathVariable("taskId") String taskId) {
         cleaningTaskService.executeTask(taskId);
-        return ResponseEntity.ok(Response.ok(null));
+        return taskId;
     }
 
     @GetMapping("/{taskId}")
-    public ResponseEntity<Response<CleaningTaskDto>> cleaningTasksTaskIdGet(@PathVariable("taskId") String taskId) {
+    public CleaningTaskDto cleaningTasksTaskIdGet(@PathVariable("taskId") String taskId) {
-        return ResponseEntity.ok(Response.ok(cleaningTaskService.getTask(taskId)));
+        return cleaningTaskService.getTask(taskId);
     }
 
     @DeleteMapping("/{taskId}")
-    public ResponseEntity<Response<Object>> cleaningTasksTaskIdDelete(@PathVariable("taskId") String taskId) {
+    public String cleaningTasksTaskIdDelete(@PathVariable("taskId") String taskId) {
         cleaningTaskService.deleteTask(taskId);
-        return ResponseEntity.ok(Response.ok(null));
+        return taskId;
     }
 }

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

@@ -4,10 +4,8 @@ import com.datamate.cleaning.application.CleaningTemplateService;
 import com.datamate.cleaning.interfaces.dto.CleaningTemplateDto;
 import com.datamate.cleaning.interfaces.dto.CreateCleaningTemplateRequest;
 import com.datamate.cleaning.interfaces.dto.UpdateCleaningTemplateRequest;
-import com.datamate.common.infrastructure.common.Response;
 import com.datamate.common.interfaces.PagedResponse;
 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;
@@ -29,14 +27,14 @@ public class CleaningTemplateController {
     private final CleaningTemplateService cleaningTemplateService;
 
     @GetMapping
-    public ResponseEntity<Response<PagedResponse<CleaningTemplateDto>>> cleaningTemplatesGet(
+    public PagedResponse<CleaningTemplateDto> cleaningTemplatesGet(
             @RequestParam(value = "page", required = false) Integer page,
             @RequestParam(value = "size", required = false) Integer size,
             @RequestParam(value = "keywords", required = false) String keyword) {
         List<CleaningTemplateDto> templates = cleaningTemplateService.getTemplates(keyword);
         if (page == null || size == null) {
-            return ResponseEntity.ok(Response.ok(PagedResponse.of(templates.stream()
+            return PagedResponse.of(templates.stream()
-                    .sorted(Comparator.comparing(CleaningTemplateDto::getCreatedAt).reversed()).toList())));
+                    .sorted(Comparator.comparing(CleaningTemplateDto::getCreatedAt).reversed()).toList());
         }
         int count = templates.size();
         int totalPages = (count + size + 1) / size;
@@ -44,31 +42,31 @@ public class CleaningTemplateController {
                 .sorted(Comparator.comparing(CleaningTemplateDto::getCreatedAt).reversed())
                 .skip((long) page * size)
                 .limit(size).toList();
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(limitTemplates, page, count, totalPages)));
+        return PagedResponse.of(limitTemplates, page, count, totalPages);
     }
 
     @PostMapping
-    public ResponseEntity<Response<CleaningTemplateDto>> cleaningTemplatesPost(
+    public CleaningTemplateDto cleaningTemplatesPost(
             @RequestBody CreateCleaningTemplateRequest request) {
-        return ResponseEntity.ok(Response.ok(cleaningTemplateService.createTemplate(request)));
+        return cleaningTemplateService.createTemplate(request);
     }
 
     @GetMapping("/{templateId}")
-    public ResponseEntity<Response<CleaningTemplateDto>> cleaningTemplatesTemplateIdGet(
+    public CleaningTemplateDto cleaningTemplatesTemplateIdGet(
             @PathVariable("templateId") String templateId) {
-        return ResponseEntity.ok(Response.ok(cleaningTemplateService.getTemplate(templateId)));
+        return cleaningTemplateService.getTemplate(templateId);
     }
 
     @PutMapping("/{templateId}")
-    public ResponseEntity<Response<CleaningTemplateDto>> cleaningTemplatesTemplateIdPut(
+    public CleaningTemplateDto cleaningTemplatesTemplateIdPut(
             @PathVariable("templateId") String templateId, @RequestBody UpdateCleaningTemplateRequest request) {
-        return ResponseEntity.ok(Response.ok(cleaningTemplateService.updateTemplate(templateId, request)));
+        return cleaningTemplateService.updateTemplate(templateId, request);
     }
 
     @DeleteMapping("/{templateId}")
-    public ResponseEntity<Response<Object>> cleaningTemplatesTemplateIdDelete(
+    public String cleaningTemplatesTemplateIdDelete(
             @PathVariable("templateId") String templateId) {
         cleaningTemplateService.deleteTemplate(templateId);
-        return ResponseEntity.noContent().build();
+        return templateId;
     }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/application/CategoryService.java

@@ -6,13 +6,16 @@ import com.datamate.operator.domain.repository.CategoryRepository;
 import com.datamate.operator.interfaces.dto.CategoryDto;
 import com.datamate.operator.interfaces.dto.CategoryRelationDto;
 import com.datamate.operator.interfaces.dto.CategoryTreeResponse;
-import com.datamate.operator.interfaces.dto.SubCategory;
 import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 
+import java.time.LocalDateTime;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Service
@@ -26,37 +29,40 @@ public class CategoryService {
         List<CategoryDto> allCategories = categoryRepo.findAllCategories();
         List<CategoryRelationDto> allRelations = categoryRelationRepo.findAllRelation();
 
-        Map<Integer, Integer> relationMap = allRelations.stream()
+        Map<String, Integer> relationMap = allRelations.stream()
                 .collect(Collectors.groupingBy(
                         CategoryRelationDto::getCategoryId,
                         Collectors.collectingAndThen(Collectors.counting(), Math::toIntExact)));
 
-        Map<Integer, String> nameMap = allCategories.stream()
+        Map<String, CategoryDto> nameMap = allCategories.stream()
-                .collect(Collectors.toMap(CategoryDto::getId, CategoryDto::getName));
+                .collect(Collectors.toMap(CategoryDto::getId, Function.identity()));
-        Map<Integer, List<CategoryDto>> groupedByParentId = allCategories.stream()
+        Map<String, List<CategoryDto>> groupedByParentId = allCategories.stream()
-                .filter(relation -> relation.getParentId() > 0)
+                .filter(relation -> !StringUtils.equals(relation.getParentId(), "0"))
                 .collect(Collectors.groupingBy(CategoryDto::getParentId));
 
         return groupedByParentId.entrySet().stream()
-                .sorted(Map.Entry.comparingByKey())
+                .sorted(categoryComparator(nameMap))
                 .map(entry -> {
-                    Integer parentId = entry.getKey();
+                    String parentId = entry.getKey();
                     List<CategoryDto> group = entry.getValue();
                     CategoryTreeResponse response = new CategoryTreeResponse();
                     response.setId(parentId);
-                    response.setName(nameMap.get(parentId));
+                    response.setName(nameMap.get(parentId).getName());
                     AtomicInteger totalCount = new AtomicInteger();
-                    response.setCategories(group.stream().map(category -> {
+                    response.setCategories(group.stream().peek(category -> {
-                        SubCategory subCategory = new SubCategory();
+                        category.setCount(relationMap.getOrDefault(category.getId(), 0));
-                        subCategory.setId(category.getId());
-                        subCategory.setName(category.getName());
-                        subCategory.setCount(relationMap.getOrDefault(category.getId(), 0));
                         totalCount.getAndAdd(relationMap.getOrDefault(category.getId(), 0));
-                        subCategory.setParentId(parentId);
+                    }).sorted(Comparator.comparing(CategoryDto::getCreatedAt)).toList());
-                        return subCategory;
-                    }).toList());
                     response.setCount(totalCount.get());
                     return response;
                 }).toList();
     }
+
+    private Comparator<Map.Entry<String, List<CategoryDto>>> categoryComparator(Map<String, CategoryDto> categoryMap) {
+        return (entry1, entry2) -> {
+            LocalDateTime index1 = categoryMap.get(entry1.getKey()).getCreatedAt();
+            LocalDateTime index2 = categoryMap.get(entry2.getKey()).getCreatedAt();
+            return index1.compareTo(index2);
+        };
+    }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/application/LabelService.java

@@ -1,21 +0,0 @@
-package com.datamate.operator.application;
-
-import com.datamate.operator.interfaces.dto.LabelDto;
-import org.springframework.stereotype.Service;
-import java.util.List;
-import java.util.Collections;
-
-@Service
-public class LabelService {
-    public List<LabelDto> getLabels(Integer page, Integer size, String keyword) {
-        // TODO: 查询标签列表
-        return Collections.emptyList();
-    }
-    public void updateLabel(String id, List<LabelDto> updateLabelDtoRequest) {
-        // TODO: 更新标签
-    }
-    public void createLabels(LabelDto labelsPostRequest) {
-        // TODO: 批量创建标签
-    }
-}
-

backend/services/operator-market-service/src/main/java/com/datamate/operator/application/OperatorService.java

@@ -1,15 +1,22 @@
 package com.datamate.operator.application;
 
+import com.datamate.common.domain.model.ChunkUploadPreRequest;
+import com.datamate.common.domain.service.FileService;
+import com.datamate.operator.domain.contants.OperatorConstant;
 import com.datamate.operator.infrastructure.converter.OperatorConverter;
 import com.datamate.operator.domain.model.OperatorView;
 import com.datamate.operator.domain.repository.CategoryRelationRepository;
 import com.datamate.operator.domain.repository.OperatorRepository;
 import com.datamate.operator.domain.repository.OperatorViewRepository;
+import com.datamate.operator.infrastructure.parser.ParserHolder;
 import com.datamate.operator.interfaces.dto.OperatorDto;
+import com.datamate.operator.interfaces.dto.UploadOperatorRequest;
 import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
-import org.springframework.web.multipart.MultipartFile;
+import org.springframework.transaction.annotation.Transactional;
 
+import java.io.File;
 import java.util.List;
 
 @Service
@@ -21,14 +28,21 @@ public class OperatorService {
 
     private final CategoryRelationRepository relationRepo;
 
-    public List<OperatorDto> getOperators(Integer page, Integer size, List<Integer> categories,
+    private final ParserHolder parserHolder;
+
+    private final FileService fileService;
+
+    @Value("${operator.base.path:/operator}")
+    private String operatorBasePath;
+
+    public List<OperatorDto> getOperators(Integer page, Integer size, List<String> categories,
                                           String operatorName, Boolean isStar) {
         List<OperatorView> filteredOperators = operatorViewRepo.findOperatorsByCriteria(page, size, operatorName,
                 categories, isStar);
         return filteredOperators.stream().map(OperatorConverter.INSTANCE::fromEntityToDto).toList();
     }
 
-    public int getOperatorsCount(List<Integer> categories, String operatorName, Boolean isStar) {
+    public int getOperatorsCount(List<String> categories, String operatorName, Boolean isStar) {
         return operatorViewRepo.countOperatorsByCriteria(operatorName, categories, isStar);
     }
 
@@ -37,20 +51,60 @@ public class OperatorService {
         return OperatorConverter.INSTANCE.fromEntityToDto(operator);
     }
 
+    @Transactional
     public OperatorDto createOperator(OperatorDto req) {
         operatorRepo.insertOperator(req);
         relationRepo.batchInsert(req.getId(), req.getCategories());
+        parserHolder.extractTo(getFileType(req.getFileName()), getUploadPath(req.getFileName()),
+            getExtractPath(getFileNameWithoutExtension(req.getFileName())));
         return getOperatorById(req.getId());
     }
 
+    @Transactional
     public OperatorDto updateOperator(String id, OperatorDto req) {
         operatorRepo.updateOperator(req);
         relationRepo.batchInsert(id, req.getCategories());
+        parserHolder.extractTo(getFileType(req.getFileName()), getUploadPath(req.getFileName()),
+            getExtractPath(getFileNameWithoutExtension(req.getFileName())));
         return getOperatorById(id);
     }
 
-    public OperatorDto uploadOperator(MultipartFile file, String description) {
+    @Transactional
-        // TODO: 文件上传与解析
+    public void deleteOperator(String id) {
-        return new OperatorDto();
+        operatorRepo.deleteOperator(id);
+        relationRepo.deleteByOperatorId(id);
+    }
+
+    public OperatorDto uploadOperator(String fileName) {
+        return parserHolder.parseYamlFromArchive(getFileType(fileName), new File(getUploadPath(fileName)),
+            OperatorConstant.YAML_PATH);
+    }
+
+    public String preUpload() {
+        ChunkUploadPreRequest request = ChunkUploadPreRequest.builder().build();
+        request.setUploadPath(operatorBasePath + File.separator + "upload");
+        request.setTotalFileNum(1);
+        request.setServiceId(OperatorConstant.SERVICE_ID);
+        return fileService.preUpload(request);
+    }
+
+    public void chunkUpload(UploadOperatorRequest request) {
+        fileService.chunkUpload(OperatorConverter.INSTANCE.toChunkRequest(request));
+    }
+
+    private String getFileType(String fileName) {
+        return fileName.substring(fileName.lastIndexOf('.') + 1);
+    }
+
+    private String getFileNameWithoutExtension(String fileName) {
+        return fileName.substring(0, fileName.lastIndexOf('.'));
+    }
+
+    private String getUploadPath(String fileName) {
+        return operatorBasePath + File.separator + "upload" + File.separator + fileName;
+    }
+
+    private String getExtractPath(String fileName) {
+        return operatorBasePath + File.separator + "extract" + File.separator + fileName;
     }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/contants/OperatorConstant.java

@@ -0,0 +1,42 @@
+package com.datamate.operator.domain.contants;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class OperatorConstant {
+    public static String SERVICE_ID = "operator";
+
+    public static String YAML_PATH = "metadata.yml";
+
+    public static String CATEGORY_PYTHON = "python";
+
+    public static String CATEGORY_PYTHON_ID = "9eda9d5d-072b-499b-916c-797a0a8750e1";
+
+    public static String CATEGORY_JAVA = "java";
+
+    public static String CATEGORY_JAVA_ID = "b5bfc548-8ef6-417c-b8a6-a4197c078249";
+
+    public static String CATEGORY_CUSTOMIZED_ID = "ec2cdd17-8b93-4a81-88c4-ac9e98d10757";
+
+    public static String CATEGORY_TEXT_ID = "d8a5df7a-52a9-42c2-83c4-01062e60f597";
+
+    public static String CATEGORY_IMAGE_ID = "de36b61c-9e8a-4422-8c31-d30585c7100f";
+
+    public static String CATEGORY_AUDIO_ID = "42dd9392-73e4-458c-81ff-41751ada47b5";
+
+    public static String CATEGORY_VIDEO_ID = "a233d584-73c8-4188-ad5d-8f7c8dda9c27";
+
+    public static String CATEGORY_ALL_ID = "4d7dbd77-0a92-44f3-9056-2cd62d4a71e4";
+
+    public static Map<String, String> CATEGORY_MAP = new HashMap<>();
+
+    static {
+        CATEGORY_MAP.put(CATEGORY_PYTHON, CATEGORY_PYTHON_ID);
+        CATEGORY_MAP.put(CATEGORY_JAVA, CATEGORY_JAVA_ID);
+        CATEGORY_MAP.put("text", CATEGORY_TEXT_ID);
+        CATEGORY_MAP.put("image", CATEGORY_IMAGE_ID);
+        CATEGORY_MAP.put("audio", CATEGORY_AUDIO_ID);
+        CATEGORY_MAP.put("video", CATEGORY_VIDEO_ID);
+        CATEGORY_MAP.put("all", CATEGORY_ALL_ID);
+    }
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/model/Category.java

@@ -4,15 +4,21 @@ import com.baomidou.mybatisplus.annotation.TableName;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.time.LocalDateTime;
+
 @Setter
 @Getter
 @TableName(value = "t_operator_category", autoResultMap = true)
 public class Category {
-    private Integer id;
+    private String id;
 
     private String name;
 
+    private String value;
+
     private String type;
 
-    private Integer parentId;
+    private String parentId;
+
+    private LocalDateTime createdAt;
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/model/CategoryRelation.java

@@ -10,7 +10,7 @@ import lombok.Setter;
 @AllArgsConstructor
 @TableName(value = "t_operator_category_relation", autoResultMap = true)
 public class CategoryRelation {
-    private Integer categoryId;
+    private String categoryId;
 
     private String operatorId;
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/CategoryRelationRepository.java

@@ -3,7 +3,6 @@ package com.datamate.operator.domain.repository;
 import com.baomidou.mybatisplus.extension.repository.IRepository;
 import com.datamate.operator.domain.model.CategoryRelation;
 import com.datamate.operator.interfaces.dto.CategoryRelationDto;
-import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
@@ -11,5 +10,7 @@ public interface CategoryRelationRepository extends IRepository<CategoryRelation
 
     List<CategoryRelationDto> findAllRelation();
 
-    void batchInsert(@Param("operatorId") String operatorId, @Param("categories") List<Integer> categories);
+    void batchInsert(String operatorId, List<String> categories);
+
+    void deleteByOperatorId(String operatorId);
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/OperatorRepository.java

@@ -7,9 +7,11 @@ import com.datamate.operator.interfaces.dto.OperatorDto;
 import java.util.List;
 
 public interface OperatorRepository extends IRepository<Operator> {
-    List<Operator> findAllOperators();
+    List<OperatorDto> findAllOperators();
 
     void updateOperator(OperatorDto operator);
 
     void insertOperator(OperatorDto operator);
+
+    void deleteOperator(String id);
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/OperatorViewRepository.java

@@ -7,9 +7,9 @@ import java.util.List;
 
 public interface OperatorViewRepository extends IRepository<OperatorView> {
     List<OperatorView> findOperatorsByCriteria(Integer page, Integer size, String operatorName,
-                                               List<Integer> categories, Boolean isStar);
+                                               List<String> categories, Boolean isStar);
 
-    Integer countOperatorsByCriteria(String operatorName, List<Integer> categories, Boolean isStar);
+    Integer countOperatorsByCriteria(String operatorName, List<String> categories, Boolean isStar);
 
     OperatorView findOperatorById(String id);
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/converter/OperatorConverter.java

@@ -1,8 +1,10 @@
 package com.datamate.operator.infrastructure.converter;
 
+import com.datamate.common.domain.model.ChunkUploadRequest;
 import com.datamate.operator.domain.model.Operator;
 import com.datamate.operator.domain.model.OperatorView;
 import com.datamate.operator.interfaces.dto.OperatorDto;
+import com.datamate.operator.interfaces.dto.UploadOperatorRequest;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 import org.mapstruct.Named;
@@ -19,13 +21,17 @@ public interface OperatorConverter {
     @Mapping(target = "categories", source = "categories", qualifiedByName = "stringToList")
     OperatorDto fromEntityToDto(OperatorView operator);
 
+    List<OperatorDto> fromEntityToDto(List<Operator> operator);
+
     @Named("stringToList")
-    static List<Integer> stringToList(String input) {
+    static List<String> stringToList(String input) {
         if (input == null || input.isEmpty()) {
             return Collections.emptyList();
         }
-        return Arrays.stream(input.split(",")).map(Integer::valueOf).toList();
+        return Arrays.stream(input.split(",")).map(String::valueOf).toList();
     }
 
     Operator fromDtoToEntity(OperatorDto operator);
+
+    ChunkUploadRequest toChunkRequest(UploadOperatorRequest request);
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/exception/OperatorErrorCode.java

@@ -0,0 +1,21 @@
+package com.datamate.operator.infrastructure.exception;
+
+import com.datamate.common.infrastructure.exception.ErrorCode;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public enum OperatorErrorCode implements ErrorCode {
+    /**
+     * 不支持的文件类型
+     */
+    UNSUPPORTED_FILE_TYPE("op.0001", "不支持的文件类型"),
+
+    YAML_NOT_FOUND("op.0002", "算子中缺少元数据文件"),
+
+    FIELD_NOT_FOUND("op.0003", "缺少必要的字段");
+
+    private final String code;
+    private final String message;
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/parser/AbstractParser.java

@@ -0,0 +1,80 @@
+package com.datamate.operator.infrastructure.parser;
+
+import com.datamate.common.infrastructure.exception.BusinessException;
+import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.operator.domain.contants.OperatorConstant;
+import com.datamate.operator.infrastructure.exception.OperatorErrorCode;
+import com.datamate.operator.interfaces.dto.OperatorDto;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+public abstract class AbstractParser {
+    protected ObjectMapper objectMapper = new ObjectMapper();
+
+    protected OperatorDto parseYaml(InputStream yamlContent) {
+        Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
+        Map<String, Object> content = yaml.load(yamlContent);
+        OperatorDto operator = new OperatorDto();
+        operator.setId(toStringIfNotNull(content.get("raw_id")));
+        operator.setName(toStringIfNotNull(content.get("name")));
+        operator.setDescription(toStringIfNotNull(content.get("description")));
+        operator.setVersion(toStringIfNotNull(content.get("version")));
+        operator.setInputs(toStringIfNotNull(content.get("inputs")));
+        operator.setOutputs(toStringIfNotNull(content.get("outputs")));
+        operator.setRuntime(toJsonIfNotNull(content.get("runtime")));
+        operator.setSettings(toJsonIfNotNull(content.get("settings")));
+        List<String> categories = new ArrayList<>();
+        categories.add(OperatorConstant.CATEGORY_MAP.get(toLowerCaseIfNotNull(content.get("language"))));
+        categories.add(OperatorConstant.CATEGORY_MAP.get(toLowerCaseIfNotNull(content.get("modal"))));
+        categories.add(OperatorConstant.CATEGORY_CUSTOMIZED_ID);
+        operator.setCategories(categories);
+        return operator;
+    };
+
+    /**
+     * 从压缩包内读取指定路径的 yaml 文件并解析为指定类型
+     * @param archive 压缩包路径（zip 或 tar）
+     * @param entryPath 压缩包内部的文件路径，例如 "config/app.yaml" 或 "./config/app.yaml"
+     * @return 解析后的对象
+     */
+    public abstract OperatorDto parseYamlFromArchive(File archive, String entryPath);
+
+    /**
+     * 将压缩包解压到目标目录（保持相对路径）
+     * @param archive 压缩包路径
+     * @param targetDir 目标目录
+     */
+    public abstract void extractTo(File archive, String targetDir);
+
+    private String toStringIfNotNull(Object obj) {
+        if (obj == null) {
+            throw BusinessException.of(OperatorErrorCode.FIELD_NOT_FOUND);
+        }
+        return obj.toString();
+    }
+
+    private String toLowerCaseIfNotNull(Object obj) {
+        if (obj == null) {
+            throw BusinessException.of(OperatorErrorCode.FIELD_NOT_FOUND);
+        }
+        return obj.toString().toLowerCase(Locale.ROOT);
+    }
+
+    private String toJsonIfNotNull(Object obj) {
+        try {
+            return obj == null ? null : objectMapper.writeValueAsString(obj);
+        } catch (JsonProcessingException e) {
+            throw BusinessException.of(SystemErrorCode.UNKNOWN_ERROR, e.getMessage());
+        }
+    }
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/parser/ParserHolder.java

@@ -0,0 +1,62 @@
+package com.datamate.operator.infrastructure.parser;
+
+import com.datamate.common.infrastructure.exception.BusinessException;
+import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.operator.infrastructure.exception.OperatorErrorCode;
+import com.datamate.operator.interfaces.dto.OperatorDto;
+import jakarta.annotation.PostConstruct;
+import org.springframework.stereotype.Component;
+
+import java.io.File;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class ParserHolder {
+    // 存放 parser：key 为 parser 类型标识（例如 "zip" 或 "tar"），value 为 parser 实例
+    private final Map<String, AbstractParser> parserMap = new ConcurrentHashMap<>();
+
+    // 注册 parser（可在启动时调用）
+    public void registerParser(String type, AbstractParser parser) {
+        if (type == null || parser == null) {
+            throw BusinessException.of(SystemErrorCode.UNKNOWN_ERROR);
+        }
+        parserMap.put(type, parser);
+    }
+
+    // 根据类型获取 parser（可能为 null）
+    public AbstractParser getParser(String type) {
+        return parserMap.get(type);
+    }
+
+    // 便捷代理：从指定类型的压缩包中读取 entry 并解析为 clazz
+    public OperatorDto parseYamlFromArchive(String type, File archive, String entryPath) {
+        AbstractParser parser = getParser(type);
+        if (parser == null) {
+            throw BusinessException.of(OperatorErrorCode.UNSUPPORTED_FILE_TYPE,
+                "No parser registered for type: " + type);
+        }
+        return parser.parseYamlFromArchive(archive, entryPath);
+    }
+
+    // 便捷代理：将指定类型的压缩包解压到目标目录
+    public void extractTo(String type, File archive, String targetDir) {
+        AbstractParser parser = getParser(type);
+        if (parser == null) {
+            throw BusinessException.of(OperatorErrorCode.UNSUPPORTED_FILE_TYPE,
+                "No parser registered for type: " + type);
+        }
+        parser.extractTo(archive, targetDir);
+    }
+
+    public void extractTo(String type, String sourceDir, String targetDir) {
+        extractTo(type, new File(sourceDir), targetDir);
+    }
+
+    @PostConstruct
+    public void init() {
+        // 注册 zip 和 tar parser，key 可根据需要调整（例如 "zip"/"tar"）
+        registerParser("zip", new ZipParser());
+        registerParser("tar", new TarParser());
+    }
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/parser/TarParser.java

@@ -0,0 +1,77 @@
+package com.datamate.operator.infrastructure.parser;
+
+import com.datamate.common.infrastructure.exception.BusinessException;
+import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.operator.infrastructure.exception.OperatorErrorCode;
+import com.datamate.operator.interfaces.dto.OperatorDto;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Objects;
+
+public class TarParser extends AbstractParser {
+
+    @Override
+    public OperatorDto parseYamlFromArchive(File archive, String entryPath) {
+        // 允许带或不带前导 "./"
+        String normalized = entryPath.startsWith("./") ? entryPath.substring(2) : entryPath;
+        try (InputStream fis = Files.newInputStream(archive.toPath());
+             TarArchiveInputStream tis = new TarArchiveInputStream(fis)) {
+            TarArchiveEntry entry;
+            while ((entry = tis.getNextEntry()) != null) {
+                String name = entry.getName();
+                if (Objects.equals(name, entryPath) || Objects.equals(name, normalized)) {
+                    // 使用 SnakeYAML 解析当前 entry 的内容到目标类型
+                    return parseYaml(tis);
+                }
+            }
+        } catch (IOException e) {
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, e.getMessage());
+        }
+        throw BusinessException.of(OperatorErrorCode.YAML_NOT_FOUND, "Entry not found in tar: " + entryPath);
+    }
+
+    @Override
+    public void extractTo(File archive, String targetDir) {
+        Path targetPath = Paths.get(targetDir);
+        try (InputStream fis = Files.newInputStream(archive.toPath());
+             TarArchiveInputStream tis = new TarArchiveInputStream(fis)) {
+            Files.createDirectories(targetPath);
+            TarArchiveEntry entry;
+            while ((entry = tis.getNextEntry()) != null) {
+                String entryName = entry.getName();
+                // 去掉可能的前导 "./"
+                if (entryName.startsWith("./")) {
+                    entryName = entryName.substring(2);
+                }
+
+                Path resolved = targetPath.resolve(entryName).toAbsolutePath().normalize();
+                if (!resolved.startsWith(targetPath.toAbsolutePath().normalize())) {
+                    throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, "Bad tar entry: " + entryName);
+                }
+
+                if (entry.isDirectory()) {
+                    Files.createDirectories(resolved);
+                } else {
+                    Files.createDirectories(resolved.getParent());
+                    try (OutputStream os = Files.newOutputStream(resolved)) {
+                        byte[] buffer = new byte[8192];
+                        int len;
+                        while ((len = tis.read(buffer)) != -1) {
+                            os.write(buffer, 0, len);
+                        }
+                    }
+                }
+            }
+        } catch (IOException e) {
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, e.getMessage());
+        }
+    }
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/parser/ZipParser.java

@@ -0,0 +1,77 @@
+package com.datamate.operator.infrastructure.parser;
+
+import com.datamate.common.infrastructure.exception.BusinessException;
+import com.datamate.common.infrastructure.exception.SystemErrorCode;
+import com.datamate.operator.infrastructure.exception.OperatorErrorCode;
+import com.datamate.operator.interfaces.dto.OperatorDto;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Enumeration;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+public class ZipParser extends AbstractParser {
+
+    @Override
+    public OperatorDto parseYamlFromArchive(File archive, String entryPath) {
+        try (ZipFile zipFile = new ZipFile(archive)) {
+            // 允许带或不带前导 "./"
+            String normalized = entryPath.startsWith("./") ? entryPath.substring(2) : entryPath;
+            ZipEntry entry = zipFile.getEntry(entryPath);
+            if (entry == null) {
+                entry = zipFile.getEntry(normalized);
+            }
+            if (entry == null) {
+                throw BusinessException.of(OperatorErrorCode.YAML_NOT_FOUND, "Entry not found in zip: " + entryPath);
+            }
+            try (InputStream is = zipFile.getInputStream(entry)) {
+                // 使用 SnakeYAML 解析为目标类型
+                return parseYaml(is);
+            }
+        } catch (IOException e) {
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, e.getMessage());
+        }
+    }
+
+    @Override
+    public void extractTo(File archive, String targetDir) {
+        Path targetPath = Paths.get(targetDir);
+        try (ZipFile zipFile = new ZipFile(archive)) {
+            Files.createDirectories(targetPath);
+            Enumeration<? extends ZipEntry> entries = zipFile.entries();
+            while (entries.hasMoreElements()) {
+                ZipEntry entry = entries.nextElement();
+                String entryName = entry.getName();
+
+                // 防止 Zip Slip：确保解压路径仍在 targetDir 下
+                Path resolved = targetPath.resolve(entryName).toAbsolutePath().normalize();
+                if (!resolved.startsWith(targetPath.toAbsolutePath().normalize())) {
+                    throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, "Bad zip entry: " + entryName);
+                }
+
+                if (entry.isDirectory()) {
+                    Files.createDirectories(resolved);
+                } else {
+                    Files.createDirectories(resolved.getParent());
+                    try (InputStream is = zipFile.getInputStream(entry);
+                         OutputStream os = Files.newOutputStream(resolved)) {
+                        byte[] buffer = new byte[8192];
+                        int len;
+                        while ((len = is.read(buffer)) != -1) {
+                            os.write(buffer, 0, len);
+                        }
+                    }
+                }
+            }
+        } catch (IOException e) {
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR, e.getMessage());
+        }
+    }
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/CategoryRelationRepositoryImpl.java

@@ -1,5 +1,6 @@
 package com.datamate.operator.infrastructure.persistence.Impl;
 
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.repository.CrudRepository;
 import com.datamate.operator.domain.model.CategoryRelation;
 import com.datamate.operator.domain.repository.CategoryRelationRepository;
@@ -23,10 +24,17 @@ public class CategoryRelationRepositoryImpl extends CrudRepository<CategoryRelat
     }
 
     @Override
-    public void batchInsert(String operatorId, List<Integer> categories) {
+    public void batchInsert(String operatorId, List<String> categories) {
         List<CategoryRelation> categoryRelations = categories.stream()
             .map(category -> new CategoryRelation(category, operatorId))
             .toList();
         mapper.insert(categoryRelations);
     }
+
+    @Override
+    public void deleteByOperatorId(String operatorId) {
+        LambdaQueryWrapper<CategoryRelation> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(CategoryRelation::getOperatorId, operatorId);
+        mapper.delete(queryWrapper);
+    }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/OperatorRepositoryImpl.java

@@ -17,8 +17,8 @@ public class OperatorRepositoryImpl extends CrudRepository<OperatorMapper, Opera
     private final OperatorMapper mapper;
 
     @Override
-    public List<Operator> findAllOperators() {
+    public List<OperatorDto> findAllOperators() {
-        return mapper.selectList(null);
+        return OperatorConverter.INSTANCE.fromEntityToDto(mapper.selectList(null));
     }
 
     @Override
@@ -30,4 +30,9 @@ public class OperatorRepositoryImpl extends CrudRepository<OperatorMapper, Opera
     public void insertOperator(OperatorDto operator) {
         mapper.insert(OperatorConverter.INSTANCE.fromDtoToEntity(operator));
     }
+
+    @Override
+    public void deleteOperator(String id) {
+        mapper.deleteById(id);
+    }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/OperatorViewRepositoryImpl.java

@@ -22,7 +22,7 @@ public class OperatorViewRepositoryImpl extends CrudRepository<OperatorViewMappe
 
     @Override
     public List<OperatorView> findOperatorsByCriteria(Integer page, Integer size, String operatorName,
-                                                      List<Integer> categories, Boolean isStar) {
+                                                      List<String> categories, Boolean isStar) {
         QueryWrapper<OperatorView> queryWrapper = Wrappers.query();
         queryWrapper.in(CollectionUtils.isNotEmpty(categories), "category_id", categories)
             .like(StringUtils.isNotBlank(operatorName), "operator_name", operatorName)
@@ -37,7 +37,7 @@ public class OperatorViewRepositoryImpl extends CrudRepository<OperatorViewMappe
     }
 
     @Override
-    public Integer countOperatorsByCriteria(String operatorName, List<Integer> categories, Boolean isStar) {
+    public Integer countOperatorsByCriteria(String operatorName, List<String> categories, Boolean isStar) {
         QueryWrapper<OperatorView> queryWrapper = Wrappers.query();
         queryWrapper.in(CollectionUtils.isNotEmpty(categories),"category_id", categories)
             .like(StringUtils.isNotBlank(operatorName), "operator_name", operatorName)

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/CategoryDto.java

@@ -3,14 +3,22 @@ package com.datamate.operator.interfaces.dto;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.time.LocalDateTime;
+
 @Setter
 @Getter
 public class CategoryDto {
-    private Integer id;
+    private String id;
 
     private String name;
 
+    private String value;
+
+    private long count;
+
     private String type;
 
-    private Integer parentId;
+    private String parentId;
+
+    private LocalDateTime createdAt;
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/CategoryRelationDto.java

@@ -6,7 +6,7 @@ import lombok.Setter;
 @Setter
 @Getter
 public class CategoryRelationDto {
-    private Integer categoryId;
+    private String categoryId;
 
     private String operatorId;
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/CategoryTreeResponse.java

@@ -12,11 +12,11 @@ import java.util.List;
 @Setter
 @NoArgsConstructor
 public class CategoryTreeResponse {
-    private Integer id;
+    private String id;
 
     private String name;
 
     private Integer count;
 
-    private List<SubCategory> categories = new ArrayList<>();
+    private List<CategoryDto> categories = new ArrayList<>();
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/OperatorDto.java

@@ -14,32 +14,32 @@ import java.util.List;
 @Getter
 @Setter
 public class OperatorDto {
-  private String id;
+    private String id;
 
-  private String name;
+    private String name;
 
-  private String description;
+    private String description;
 
-  private String version;
+    private String version;
 
-  private String inputs;
+    private String inputs;
 
-  private String outputs;
+    private String outputs;
 
-  private List<Integer> categories;
+    private List<String> categories;
 
-  private String runtime;
+    private String runtime;
 
-  private String settings;
+    private String settings;
 
-  private String fileName;
+    private String fileName;
 
-  private Boolean isStar;
+    private Boolean isStar;
 
-  @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
+    @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-  private LocalDateTime createdAt;
+    private LocalDateTime createdAt;
 
-  @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
+    @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
-  private LocalDateTime updatedAt;
+    private LocalDateTime updatedAt;
 }
 

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/OperatorsListPostRequest.java

@@ -19,7 +19,7 @@ public class OperatorsListPostRequest {
 
   private Integer size;
 
-  private List<Integer> categories = new ArrayList<>();
+  private List<String> categories = new ArrayList<>();
 
   private String operatorName;
 

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/SubCategory.java

@@ -1,18 +0,0 @@
-package com.datamate.operator.interfaces.dto;
-
-import lombok.Getter;
-import lombok.Setter;
-
-@Getter
-@Setter
-public class SubCategory {
-    private long id;
-
-    private String name;
-
-    private long count;
-
-    private String type;
-
-    private long parentId;
-}

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/dto/UploadOperatorRequest.java

@@ -0,0 +1,34 @@
+package com.datamate.operator.interfaces.dto;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.web.multipart.MultipartFile;
+
+/**
+ * 上传文件请求
+ * 用于分块上传文件时的请求参数封装，支持大文件分片上传功能
+ */
+@Getter
+@Setter
+public class UploadOperatorRequest {
+    /** 预上传返回的id，用来确认同一个任务 */
+    private String reqId;
+
+    /** 文件编号，用于标识批量上传中的第几个文件 */
+    private int fileNo;
+
+    /** 文件名称 */
+    private String fileName;
+
+    /** 文件总分块数量 */
+    private int totalChunkNum;
+
+    /** 当前分块编号，从1开始 */
+    private int chunkNo;
+
+    /** 上传的文件分块内容 */
+    private MultipartFile file;
+
+    /** 文件分块的校验和（十六进制字符串），用于验证文件完整性 */
+    private String checkSumHex;
+}

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/rest/CategoryController.java

@@ -1,11 +1,9 @@
 package com.datamate.operator.interfaces.rest;
 
-import com.datamate.common.infrastructure.common.Response;
 import com.datamate.common.interfaces.PagedResponse;
 import com.datamate.operator.application.CategoryService;
 import com.datamate.operator.interfaces.dto.CategoryTreeResponse;
 import lombok.RequiredArgsConstructor;
-import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
@@ -20,8 +18,8 @@ public class CategoryController {
     private final CategoryService categoryService;
 
     @GetMapping("/tree")
-    public ResponseEntity<Response<PagedResponse<CategoryTreeResponse>>> categoryTreeGet() {
+    public PagedResponse<CategoryTreeResponse> categoryTreeGet() {
         List<CategoryTreeResponse> allCategories = categoryService.getAllCategories();
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(allCategories)));
+        return PagedResponse.of(allCategories);
     }
 }

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/rest/LabelController.java

@@ -1,40 +0,0 @@
-package com.datamate.operator.interfaces.rest;
-
-import com.datamate.common.infrastructure.common.Response;
-import com.datamate.common.interfaces.PagedResponse;
-import com.datamate.operator.application.LabelService;
-import com.datamate.operator.interfaces.dto.LabelDto;
-import lombok.RequiredArgsConstructor;
-import org.springframework.http.ResponseEntity;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
-
-@RestController
-@RequestMapping("/labels")
-@RequiredArgsConstructor
-public class LabelController {
-    private final LabelService labelService;
-
-    @GetMapping
-    public ResponseEntity<Response<PagedResponse<LabelDto>>> labelsGet(@RequestParam("page") Integer page,
-                                                                       @RequestParam("size") Integer size,
-                                                                       @RequestParam("keyword") String keyword) {
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(labelService.getLabels(page, size, keyword))));
-    }
-
-    @PutMapping("/{id}")
-    public ResponseEntity<Response<Object>> labelsIdPut(@PathVariable("id") String id,
-                                                        @RequestBody List<LabelDto> updateLabelDtoRequest) {
-        labelService.updateLabel(id, updateLabelDtoRequest);
-        return ResponseEntity.ok(Response.ok(null));
-    }
-
-    @PostMapping
-    public ResponseEntity<Response<Object>> labelsPost(@RequestBody LabelDto labelsPostRequest) {
-        labelService.createLabels(labelsPostRequest);
-        return ResponseEntity.ok(Response.ok(null));
-    }
-}
-
-

backend/services/operator-market-service/src/main/java/com/datamate/operator/interfaces/rest/OperatorController.java

@@ -1,14 +1,13 @@
 package com.datamate.operator.interfaces.rest;
 
-import com.datamate.common.infrastructure.common.Response;
 import com.datamate.common.interfaces.PagedResponse;
 import com.datamate.operator.application.OperatorService;
 import com.datamate.operator.interfaces.dto.OperatorDto;
 import com.datamate.operator.interfaces.dto.OperatorsListPostRequest;
+import com.datamate.operator.interfaces.dto.UploadOperatorRequest;
 import lombok.RequiredArgsConstructor;
-import org.springframework.http.ResponseEntity;
+import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
-import org.springframework.web.multipart.MultipartFile;
 
 import java.util.List;
 
@@ -19,34 +18,48 @@ public class OperatorController {
     private final OperatorService operatorService;
 
     @PostMapping("/list")
-    public ResponseEntity<Response<PagedResponse<OperatorDto>>> operatorsListPost(@RequestBody OperatorsListPostRequest request) {
+    public PagedResponse<OperatorDto> operatorsListPost(@RequestBody OperatorsListPostRequest request) {
         List<OperatorDto> responses = operatorService.getOperators(request.getPage(), request.getSize(),
                 request.getCategories(), request.getOperatorName(), request.getIsStar());
         int count = operatorService.getOperatorsCount(request.getCategories(), request.getOperatorName(),
                 request.getIsStar());
         int totalPages = (count + request.getSize() + 1) / request.getSize();
-        return ResponseEntity.ok(Response.ok(PagedResponse.of(responses, request.getPage(), count, totalPages)));
+        return PagedResponse.of(responses, request.getPage(), count, totalPages);
     }
 
     @GetMapping("/{id}")
-    public ResponseEntity<Response<OperatorDto>> operatorsIdGet(@PathVariable("id") String id) {
+    public OperatorDto operatorsIdGet(@PathVariable("id") String id) {
-        return ResponseEntity.ok(Response.ok(operatorService.getOperatorById(id)));
+        return operatorService.getOperatorById(id);
     }
 
     @PutMapping("/{id}")
-    public ResponseEntity<Response<OperatorDto>> operatorsIdPut(@PathVariable("id") String id,
+    public OperatorDto operatorsIdPut(@PathVariable("id") String id,
                                                                 @RequestBody OperatorDto updateOperatorRequest) {
-        return ResponseEntity.ok(Response.ok(operatorService.updateOperator(id, updateOperatorRequest)));
+        return operatorService.updateOperator(id, updateOperatorRequest);
     }
 
     @PostMapping("/create")
-    public ResponseEntity<Response<OperatorDto>> operatorsCreatePost(@RequestBody OperatorDto createOperatorRequest) {
+    public OperatorDto operatorsCreatePost(@RequestBody OperatorDto createOperatorRequest) {
-        return ResponseEntity.ok(Response.ok(operatorService.createOperator(createOperatorRequest)));
+        return operatorService.createOperator(createOperatorRequest);
     }
 
     @PostMapping("/upload")
-    public ResponseEntity<Response<OperatorDto>> operatorsUploadPost(@RequestPart(value = "file") MultipartFile file,
+    public OperatorDto operatorsUploadPost(@RequestBody UploadOperatorRequest request) {
-                                                                     @RequestParam(value = "description") String description) {
+        return operatorService.uploadOperator(request.getFileName());
-        return ResponseEntity.ok(Response.ok(operatorService.uploadOperator(file, description)));
+    }
+
+    @PostMapping(value = "/upload/pre-upload", produces = MediaType.APPLICATION_JSON_VALUE)
+    public String preUpload() {
+        return operatorService.preUpload();
+    }
+
+    @PostMapping("/upload/chunk")
+    public void chunkUpload(@ModelAttribute UploadOperatorRequest request) {
+        operatorService.chunkUpload(request);
+    }
+
+    @DeleteMapping("/{id}")
+    public void operatorDelete(@PathVariable("id") String id) {
+        operatorService.deleteOperator(id);
     }
 }

deployment/docker/datamate/docker-compose.yml

@@ -9,6 +9,8 @@ services:
       - dataset_volume:/dataset
       - flow_volume:/flow
       - log_volume:/var/log/datamate
+      - operator-upload-volume:/operators/upload
+      - operator-runtime-volume:/operators/extract
     networks: [ datamate ]
     depends_on:
       - datamate-database
@@ -71,6 +73,7 @@ services:
       - log_volume:/var/log/datamate
       - dataset_volume:/dataset
       - flow_volume:/flow
+      - operator-runtime-volume:/opt/runtime/datamate/ops/user
     networks: [ datamate ]
 
   # 4) mineru
@@ -109,6 +112,10 @@ volumes:
     name: datamate-frontend-log-volume
   database_log_volume:
     name: datamate-database-log-volume
+  operator-upload-volume:
+    name: datamate-operator-upload-volume
+  operator-runtime-volume:
+    name: datamate-operator-runtime-volume
 
 networks:
   datamate:

deployment/helm/datamate/charts/frontend/templates/configmap.yaml

@@ -0,0 +1,28 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: datamate-nginx-conf
+data:
+  backend.conf: |
+    server {
+        listen       80;
+        server_name  0.0.0.0;
+
+        access_log /var/log/datamate/frontend/access.log main;
+        error_log /var/log/datamate/frontend/error.log notice;
+
+        client_max_body_size 1024M;
+
+        location /api/ {
+            proxy_pass http://datamate-backend:8080/api/;
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        }
+
+        location / {
+            root   /opt/frontend;
+            try_files $uri $uri/ /index.html;
+        }
+    }
+

deployment/helm/datamate/charts/ray-cluster/values.yaml

@@ -113,63 +113,11 @@ head:
   securityContext: {}
   # Optional: The following volumes/volumeMounts configurations are optional but recommended because
   # Ray writes logs to /tmp/ray/session_latests/logs instead of stdout/stderr.
-  volumes:
+  volumes: []
-    - name: log-volume
+  volumeMounts: []
-      hostPath:
-        path: /opt/datamate/data/log
-        type: DirectoryOrCreate
-    - name: dataset-volume
-      hostPath:
-        path: /opt/datamate/data/dataset
-        type: DirectoryOrCreate
-    - name: flow-volume
-      hostPath:
-        path: /opt/datamate/data/flow
-        type: DirectoryOrCreate
-  volumeMounts:
-    - mountPath: /tmp/ray
-      name: log-volume
-      subPath: ray/head
-    - mountPath: /dataset
-      name: dataset-volume
-    - mountPath: /flow
-      name: flow-volume
   # sidecarContainers specifies additional containers to attach to the Ray pod.
   # Follows standard K8s container spec.
-  sidecarContainers:
+  sidecarContainers: []
-    - name: runtime
-      image: datamate-runtime
-      imagePullPolicy: IfNotPresent
-      command:
-        - python
-        - /opt/runtime/datamate/operator_runtime.py
-        - --port
-        - "8081"
-      env:
-        - name: MYSQL_HOST
-          value: "datamate-database"
-        - name: MYSQL_PORT
-          value: "3306"
-        - name: MYSQL_USER
-          value: "root"
-        - name: MYSQL_PASSWORD
-          value: "password"
-        - name: MYSQL_DATABASE
-          value: "datamate"
-        - name: PDF_FORMATTER_BASE_URL
-          value: "http://datamate-mineru:9001"
-      ports:
-        - containerPort: 8081
-      volumeMounts:
-        - mountPath: /tmp/ray
-          name: log-volume
-          subPath: ray/head
-        - mountPath: /var/log/datamate
-          name: log-volume
-        - mountPath: /dataset
-          name: dataset-volume
-        - mountPath: /flow
-          name: flow-volume
   # See docs/guidance/pod-command.md for more details about how to specify
   # container command for head Pod.
   command: []
@@ -260,27 +208,8 @@ worker:
   securityContext: {}
   # Optional: The following volumes/volumeMounts configurations are optional but recommended because
   # Ray writes logs to /tmp/ray/session_latests/logs instead of stdout/stderr.
-  volumes:
+  volumes: []
-    - name: log-volume
+  volumeMounts: []
-      hostPath:
-        path: /opt/datamate/data/log
-        type: DirectoryOrCreate
-    - name: dataset-volume
-      hostPath:
-        path: /opt/datamate/data/dataset
-        type: DirectoryOrCreate
-    - name: flow-volume
-      hostPath:
-        path: /opt/datamate/data/flow
-        type: DirectoryOrCreate
-  volumeMounts:
-    - mountPath: /tmp/ray
-      name: log-volume
-      subPath: ray/worker
-    - mountPath: /dataset
-      name: dataset-volume
-    - mountPath: /flow
-      name: flow-volume
   # sidecarContainers specifies additional containers to attach to the Ray pod.
   # Follows standard K8s container spec.
   sidecarContainers: []

deployment/helm/datamate/values.yaml

@@ -40,11 +40,17 @@ dataVolume: &dataVolume
     path: /opt/datamate/data/mysql
     type: DirectoryOrCreate
 
+operatorVolume: &operatorVolume
+  name: operator-volume
+  hostPath:
+    path: /opt/datamate/data/operator
+
 backend:
   volumes:
     - *datasetVolume
     - *flowVolume
     - *logVolume
+    - *operatorVolume
   volumeMounts:
     - name: dataset-volume
       mountPath: /dataset
@@ -52,14 +58,22 @@ backend:
       mountPath: /flow
     - name: log-volume
       mountPath: /var/log/datamate
+    - name: operator-volume
+      mountPath: /operators
 
 frontend:
   volumes:
     - *logVolume
+    - name: datamate-nginx-conf
+      configMap:
+        name: datamate-nginx-conf
   volumeMounts:
     - name: log-volume
       mountPath: /var/log/datamate/frontend
       subPath: frontend
+    - mountPath: /etc/nginx/conf.d/backend.conf
+      name: datamate-nginx-conf
+      subPath: backend.conf
 
 database:
   volumes:
@@ -81,3 +95,76 @@ database:
       mountPath: /docker-entrypoint-initdb.d
     - name: mysql-utf8-config
       mountPath: /etc/mysql/conf.d
+
+ray-cluster:
+  head:
+    volumes:
+      - *datasetVolume
+      - *flowVolume
+      - *logVolume
+      - *operatorVolume
+    volumeMounts:
+      - mountPath: /tmp/ray
+        name: log-volume
+        subPath: ray/head
+      - mountPath: /dataset
+        name: dataset-volume
+      - mountPath: /flow
+        name: flow-volume
+      - mountPath: /opt/runtime/datamate/ops/user
+        name: operator-volume
+        subPath: extract
+    sidecarContainers:
+      - name: runtime
+        image: datamate-runtime
+        imagePullPolicy: IfNotPresent
+        args:
+          - python
+          - /opt/runtime/datamate/operator_runtime.py
+          - --port
+          - "8081"
+        env:
+          - name: MYSQL_HOST
+            value: "datamate-database"
+          - name: MYSQL_PORT
+            value: "3306"
+          - name: MYSQL_USER
+            value: "root"
+          - name: MYSQL_PASSWORD
+            value: "password"
+          - name: MYSQL_DATABASE
+            value: "datamate"
+          - name: PDF_FORMATTER_BASE_URL
+            value: "http://datamate-mineru:9001"
+        ports:
+          - containerPort: 8081
+        volumeMounts:
+          - mountPath: /tmp/ray
+            name: log-volume
+            subPath: ray/head
+          - mountPath: /var/log/datamate
+            name: log-volume
+          - mountPath: /dataset
+            name: dataset-volume
+          - mountPath: /flow
+            name: flow-volume
+          - mountPath: /opt/runtime/datamate/ops/user
+            name: operator-volume
+            subPath: extract
+  worker:
+    volumes:
+      - *datasetVolume
+      - *flowVolume
+      - *logVolume
+      - *operatorVolume
+    volumeMounts:
+      - mountPath: /tmp/ray
+        name: log-volume
+        subPath: ray/worker
+      - mountPath: /dataset
+        name: dataset-volume
+      - mountPath: /flow
+        name: flow-volume
+      - mountPath: /opt/runtime/datamate/ops/user
+        name: operator-volume
+        subPath: extract

scripts/db/data-cleaning-init.sql

@@ -3,7 +3,7 @@ USE datamate;
 CREATE TABLE IF NOT EXISTS t_clean_template
 (
     id          varchar(64) primary key not null unique,
-    name        varchar(64),
+    name        varchar(64) unique,
     description varchar(256),
     created_at  timestamp default current_timestamp,
     updated_at  timestamp default current_timestamp,
@@ -13,7 +13,7 @@ CREATE TABLE IF NOT EXISTS t_clean_template
 CREATE TABLE IF NOT EXISTS t_clean_task
 (
     id                varchar(64) primary key,
-    name              varchar(64),
+    name              varchar(64) unique,
     description       varchar(256),
     status            varchar(256),
     src_dataset_id    varchar(64),

scripts/db/data-operator-init.sql

@@ -3,7 +3,7 @@ USE datamate;
 CREATE TABLE IF NOT EXISTS t_operator
 (
     id          varchar(64) primary key,
-    name        varchar(64),
+    name        varchar(64) unique,
     description varchar(256),
     version     varchar(256),
     inputs      varchar(256),
@@ -18,15 +18,17 @@ CREATE TABLE IF NOT EXISTS t_operator
 
 CREATE TABLE IF NOT EXISTS t_operator_category
 (
-    id        int primary key auto_increment,
+    id        varchar(64) primary key,
-    name      varchar(64),
+    name      varchar(64) unique ,
+    value     varchar(64) unique ,
     type      varchar(64),
-    parent_id int
+    parent_id varchar(64),
+    created_at timestamp default current_timestamp
 );
 
 CREATE TABLE IF NOT EXISTS t_operator_category_relation
 (
-    category_id int,
+    category_id varchar(64),
     operator_id varchar(64),
     primary key (category_id, operator_id)
 );
@@ -41,7 +43,7 @@ SELECT o.id     AS operator_id,
        runtime,
        settings,
        is_star,
-       created_at,
+       o.created_at AS created_at,
        updated_at,
        toc.id   AS category_id,
        toc.name AS category_name
@@ -49,21 +51,21 @@ FROM t_operator_category_relation tocr
          LEFT JOIN t_operator o ON tocr.operator_id = o.id
          LEFT JOIN t_operator_category toc ON tocr.category_id = toc.id;
 
-INSERT IGNORE INTO t_operator_category(id, name, type, parent_id)
+INSERT IGNORE INTO t_operator_category(id, name, value, type, parent_id)
-VALUES (1, '模态', 'predefined', 0),
+VALUES ('64465bec-b46b-11f0-8291-00155d0e4808', '模态', 'modal',  'predefined', '0'),
-       (2, '语言', 'predefined', 0),
+       ('873000a2-65b3-474b-8ccc-4813c08c76fb', '语言', 'language', 'predefined', '0'),
-       (3, '文本', 'predefined', 1),
+       ('d8a5df7a-52a9-42c2-83c4-01062e60f597', '文本', 'text', 'predefined', '64465bec-b46b-11f0-8291-00155d0e4808'),
-       (4, '图片', 'predefined', 1),
+       ('de36b61c-9e8a-4422-8c31-d30585c7100f', '图片', 'image', 'predefined', '64465bec-b46b-11f0-8291-00155d0e4808'),
-       (5, '音频', 'predefined', 1),
+       ('42dd9392-73e4-458c-81ff-41751ada47b5', '音频', 'audio', 'predefined', '64465bec-b46b-11f0-8291-00155d0e4808'),
-       (6, '视频', 'predefined', 1),
+       ('a233d584-73c8-4188-ad5d-8f7c8dda9c27', '视频', 'video', 'predefined', '64465bec-b46b-11f0-8291-00155d0e4808'),
-       (7, '多模态', 'predefined', 1),
+       ('4d7dbd77-0a92-44f3-9056-2cd62d4a71e4', '多模态', 'multimodal', 'predefined', '64465bec-b46b-11f0-8291-00155d0e4808'),
-       (8, 'Python', 'predefined', 2),
+       ('9eda9d5d-072b-499b-916c-797a0a8750e1', 'Python', 'python', 'predefined', '873000a2-65b3-474b-8ccc-4813c08c76fb'),
-       (9, 'Java', 'predefined', 2),
+       ('b5bfc548-8ef6-417c-b8a6-a4197c078249', 'Java', 'java', 'predefined', '873000a2-65b3-474b-8ccc-4813c08c76fb'),
-       (10, '来源', 'predefined', 0),
+       ('16e2d99e-eafb-44fc-acd0-f35a2bad28f8', '来源', 'origin', 'predefined', '0'),
-       (11, '系统预置', 'predefined', 10),
+       ('96a3b07a-3439-4557-a835-525faad60ca3', '系统预置', 'predefined', 'predefined', '16e2d99e-eafb-44fc-acd0-f35a2bad28f8'),
-       (12, '用户上传', 'predefined', 10),
+       ('ec2cdd17-8b93-4a81-88c4-ac9e98d10757', '用户上传', 'customized', 'predefined', '16e2d99e-eafb-44fc-acd0-f35a2bad28f8'),
-       (13, '收藏状态', 'predefined', 0),
+       ('d8482257-7ee6-41a0-a914-8363c7db1db0', '收藏状态', 'starStatus', 'predefined', '0'),
-       (14, '已收藏', 'predefined', 13);
+       ('79f2d35a-3b6c-4846-a892-2f2015f48f24', '已收藏', 'isStar', 'predefined', 'd8482257-7ee6-41a0-a914-8363c7db1db0');
 
 INSERT IGNORE INTO t_operator
 (id, name, description, version, inputs, outputs, runtime, settings, file_name, is_star)
@@ -116,7 +118,7 @@ INSERT IGNORE INTO t_operator_category_relation(category_id, operator_id)
 SELECT c.id, o.id
 FROM t_operator_category c
 CROSS JOIN t_operator o
-WHERE c.id IN (3, 8, 11)
+WHERE c.id IN ('d8a5df7a-52a9-42c2-83c4-01062e60f597', '9eda9d5d-072b-499b-916c-797a0a8750e1', '96a3b07a-3439-4557-a835-525faad60ca3')
 AND o.id IN ('TextFormatter', 'FileWithShortOrLongLengthFilter', 'FileWithHighRepeatPhraseRateFilter',
             'FileWithHighRepeatWordRateFilter', 'FileWithHighSpecialCharRateFilter', 'FileWithManySensitiveWordsFilter',
             'DuplicateFilesFilter', 'DuplicateSentencesFilter', 'AnonymizedCreditCardNumber', 'AnonymizedIdNumber',
@@ -129,7 +131,7 @@ INSERT IGNORE INTO t_operator_category_relation(category_id, operator_id)
 SELECT c.id, o.id
 FROM t_operator_category c
        CROSS JOIN t_operator o
-WHERE c.id IN (4, 8, 11)
+WHERE c.id IN ('de36b61c-9e8a-4422-8c31-d30585c7100f', '9eda9d5d-072b-499b-916c-797a0a8750e1', '96a3b07a-3439-4557-a835-525faad60ca3')
   AND o.id IN ('ImgFormatter', 'ImgBlurredImagesCleaner', 'ImgBrightness', 'ImgContrast', 'ImgDenoise',
                'ImgDuplicatedImagesCleaner', 'ImgPerspectiveTransformation', 'ImgResize', 'ImgSaturation',
                'ImgShadowRemove', 'ImgSharpness', 'ImgSimilarImagesCleaner', 'ImgTypeUnify');
@@ -138,5 +140,5 @@ INSERT IGNORE INTO t_operator_category_relation(category_id, operator_id)
 SELECT c.id, o.id
 FROM t_operator_category c
        CROSS JOIN t_operator o
-WHERE c.id IN (7, 8, 11)
+WHERE c.id IN ('4d7dbd77-0a92-44f3-9056-2cd62d4a71e4', '9eda9d5d-072b-499b-916c-797a0a8750e1', '96a3b07a-3439-4557-a835-525faad60ca3')
-  AND o.id IN ('FileExporter', 'UnstructuredFormatter', 'ExternalPDFFormatter');
+  AND o.id IN ('FileExporter', 'UnstructuredFormatter');

scripts/images/backend/Dockerfile

@@ -14,10 +14,9 @@ RUN cd DataX && \
 FROM maven:3-amazoncorretto-21-debian AS builder
 
 COPY backend/ /opt/backend
-COPY scripts/images/backend/settings.xml /opt/backend
 
 RUN cd /opt/backend && \
-    mvn -U clean package -s settings.xml -Dmaven.test.skip=true
+    mvn -U clean package -Dmaven.test.skip=true
 
 
 FROM openjdk:21-jdk-slim
@@ -25,7 +24,7 @@ FROM openjdk:21-jdk-slim
 RUN apt-get update && \
     apt-get install -y vim wget curl nfs-common rsync python3 python3-pip python-is-python3 dos2unix && \
     apt-get clean && \
-    rm -rf /var/lib/apy/lists/*
+    rm -rf /var/lib/apt/lists/*
 
 COPY --from=builder /opt/backend/services/main-application/target/data-mate.jar /opt/backend/data-mate.jar
 COPY --from=datax-builder /DataX/target/datax/datax /opt/datax

scripts/images/backend/settings.xml

@@ -1,68 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"
-          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-          xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 https://maven.apache.org/xsd/settings-1.0.0.xsd">
-    <!-- 本地仓库路径（可选，默认在 ~/.m2/repository） -->
-    <localRepository>${user.home}/.m2/repository</localRepository>
-
-    <!-- 阿里云镜像配置 -->
-    <mirrors>
-        <mirror>
-            <id>aliyun-maven</id>
-            <name>Aliyun Maven Repository</name>
-            <url>https://maven.aliyun.com/repository/public</url>
-            <mirrorOf>central,jcenter,google,spring,spring-plugin,gradle-plugin</mirrorOf>
-        </mirror>
-    </mirrors>
-
-    <!-- 使用 Java 21 编译配置（可选，但推荐） -->
-    <profiles>
-        <profile>
-            <id>java21</id>
-            <activation>
-                <activeByDefault>true</activeByDefault>
-                <jdk>21</jdk>
-            </activation>
-            <properties>
-                <maven.compiler.source>21</maven.compiler.source>
-                <maven.compiler.target>21</maven.compiler.target>
-                <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-            </properties>
-        </profile>
-        <!-- 激活阿里云仓库（可选，增强依赖解析） -->
-        <profile>
-            <id>aliyun-repos</id>
-            <repositories>
-                <repository>
-                    <id>aliyun-public</id>
-                    <name>Aliyun Public Repository</name>
-                    <url>https://maven.aliyun.com/repository/public</url>
-                    <releases>
-                        <enabled>true</enabled>
-                    </releases>
-                    <snapshots>
-                        <enabled>false</enabled> <!-- 默认关闭快照版本 -->
-                    </snapshots>
-                </repository>
-            </repositories>
-            <pluginRepositories>
-                <pluginRepository>
-                    <id>aliyun-plugin</id>
-                    <name>Aliyun Plugin Repository</name>
-                    <url>https://maven.aliyun.com/repository/public</url>
-                    <releases>
-                        <enabled>true</enabled>
-                    </releases>
-                    <snapshots>
-                        <enabled>false</enabled>
-                    </snapshots>
-                </pluginRepository>
-            </pluginRepositories>
-        </profile>
-    </profiles>
-
-    <activeProfiles>
-        <activeProfile>aliyun-repos</activeProfile> <!-- 激活阿里云仓库 -->
-        <activeProfile>java21</activeProfile>       <!-- 激活 Java 21 配置 -->
-    </activeProfiles>
-</settings>

scripts/images/runtime/Dockerfile

@@ -2,6 +2,8 @@ FROM python:3.11
 
 COPY runtime/python-executor /opt/runtime
 COPY runtime/ops /opt/runtime/datamate/ops
+COPY runtime/ops/user /opt/runtime/user
+COPY scripts/images/runtime/start.sh /opt/runtime/start.sh
 
 ENV PYTHONPATH=/opt/runtime/datamate/
 
@@ -12,12 +14,13 @@ RUN apt update \
 
 WORKDIR /opt/runtime
 
-ENV HF_HUB_DISABLE_XET=1
+RUN pip install -e . --trusted-host mirrors.huaweicloud.com -i https://mirrors.huaweicloud.com/repository/pypi/simple \
-
+    && pip install -r /opt/runtime/datamate/ops/requirements.txt --trusted-host mirrors.huaweicloud.com -i https://mirrors.huaweicloud.com/repository/pypi/simple \
-RUN pip install -e . \
-    && pip install -r /opt/runtime/datamate/ops/requirements.txt \
     && pip cache purge
 
-RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
+RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
+    && chmod +x /opt/runtime/start.sh
 
 EXPOSE 8081
+
+ENTRYPOINT ["/opt/runtime/start.sh"]

scripts/images/runtime/start.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+set -e
+
+cp -r /opt/runtime/user/* /opt/runtime/datamate/ops/user
+
+echo "Starting main application..."
+exec "$@"