feat/clean

Makefile

@@ -231,8 +231,8 @@ else
     	echo -n "Enter choice (default: 2): "; \
     	read DELETE_VOLUMES_CHOICE; \
     	export DELETE_VOLUMES_CHOICE; \
-	fi
-	@$(MAKE) label-studio-$(INSTALLER)-uninstall DELETE_VOLUMES_CHOICE=$$DELETE_VOLUMES_CHOICE; \
+	fi; \
+	$(MAKE) label-studio-$(INSTALLER)-uninstall DELETE_VOLUMES_CHOICE=$$DELETE_VOLUMES_CHOICE; \
 	$(MAKE) milvus-$(INSTALLER)-uninstall DELETE_VOLUMES_CHOICE=$$DELETE_VOLUMES_CHOICE; \
 	$(MAKE) deer-flow-$(INSTALLER)-uninstall DELETE_VOLUMES_CHOICE=$$DELETE_VOLUMES_CHOICE; \
 	$(MAKE) datamate-$(INSTALLER)-uninstall DELETE_VOLUMES_CHOICE=$$DELETE_VOLUMES_CHOICE

backend/api-gateway/src/main/java/com/datamate/gateway/ApiGatewayApplication.java

@@ -45,6 +45,13 @@ public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
             .route("python-service", r -> r.path("/api/rag/**", "api/models/**")
                  .uri("http://datamate-backend-python:18000"))
 
+            // 数据评估服务路由
+            .route("data-operator", r -> r.path("/api/operators/**", "api/categories/**")
+                .uri("http://datamate-backend-python:18000"))
+
+            .route("data-cleaning", r -> r.path("/api/cleaning/**")
+                .uri("http://datamate-backend-python:18000"))
+
             .route("deer-flow-frontend", r -> r.path("/chat/**")
                 .uri("http://deer-flow-frontend:3000"))
 

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

@@ -48,7 +48,7 @@ public int countOperatorByStar(boolean isStar) {
 
     @Override
     public boolean operatorInTemplateOrRunning(String operatorId) {
-        return mapper.operatorInTemplate(operatorId) > 0 && mapper.operatorInUnstopTask(operatorId) > 0;
+        return mapper.operatorInTemplate(operatorId) > 0 || mapper.operatorInUnstopTask(operatorId) > 0;
     }
 
     @Override

deployment/docker/datamate/docker-compose.yml

@@ -35,6 +35,8 @@ services:
       - flow_volume:/flow
       - log_volume:/var/log/datamate
       - graph_data_volume:/data/rag_storage
+      - operator-upload-volume:/operators/upload
+      - operator-runtime-volume:/operators/extract
     networks: [ datamate ]
     depends_on:
       - datamate-database

deployment/helm/datamate/values.yaml

@@ -143,13 +143,16 @@ backend-python:
     - *datasetVolume
     - *flowVolume
     - *logVolume
+    - *operatorVolume
   volumeMounts:
     - name: dataset-volume
       mountPath: /dataset
     - name: flow-volume
       mountPath: /flow
     - name: log-volume
       mountPath: /var/log/datamate
+    - name: operator-volume
+      mountPath: /operators
 
 gateway:
   env:

frontend/public/config/error-code.json

@@ -1,20 +1,25 @@
 {
+  "0": "成功",
+  "cleaning.0001": "清洗任务不存在",
+  "cleaning.0002": "清洗任务名称重复",
+  "cleaning.0003": "清洗模板不存在",
+  "cleaning.0004": "清洗模板名称重复",
+  "cleaning.0005": "算子输入输出类型不匹配",
+  "cleaning.0006": "执行器类型无效",
+  "cleaning.0007": "数据集不存在",
+  "cleaning.0008": "文件系统错误",
+  "cleaning.0009": "设置解析错误",
+  "cleaning.0010": "任务ID不能为空",
+  "operator.0001": "算子不存在",
+  "operator.0002": "算子正在使用中",
+  "operator.0003": "无法删除预置算子",
+  "operator.0004": "不支持的文件类型",
+  "operator.0005": "解析算子包失败",
+  "operator.0006": "缺少必要的字段",
   "400": "请求参数错误",
   "401": "登录已过期，请重新登录",
   "403": "没有权限访问该资源",
   "404": "请求的资源不存在",
   "500": "服务器内部错误，请稍后重试",
-  "502": "网关错误",
-  "op.0001": "不支持的文件类型",
-  "op.0002": "算子中缺少元数据文件",
-  "op.0003": "缺少必要的字段",
-  "op.0004": "settings字段解析失败",
-  "op.0005": "算子ID已存在",
-  "op.0006": "算子名称已存在",
-  "op.0007": "算子已被编排在模板或未完成的任务中",
-  "op.0008": "预置算子无法删除",
-  "clean.0001": "清洗任务名称重复",
-  "clean.0002": "任务列表为空",
-  "clean.0003": "算子输入输出不匹配",
-  "clean.0004": "算子执行器不匹配"
+  "502": "网关错误"
 }

frontend/src/pages/DataCleansing/Create/components/ParamConfig.tsx

@@ -26,11 +26,11 @@ const ParamConfig: React.FC<ParamConfigProps> = ({
   onParamChange,
 }) => {
   if (!param) return null;
-  let defaultVal: any = param.defaultVal;
+  let defaultVal: any = operator.overrides?.[paramKey] ?? param.defaultVal;
   if (param.type === "range") {
 
-    defaultVal = Array.isArray(param.defaultVal)
-      ? param.defaultVal
+    defaultVal = Array.isArray(defaultVal)
+      ? defaultVal
       : [
           param?.properties?.[0]?.defaultVal,
           param?.properties?.[1]?.defaultVal,
@@ -217,6 +217,7 @@ const ParamConfig: React.FC<ParamConfigProps> = ({
           key={paramKey}
         >
           <Switch
+            value={value}
             checkedChildren={param.checkedLabel}
             unCheckedChildren={param.unCheckedLabel}
             defaultChecked={String(param.defaultVal).toLowerCase() === 'true'}

frontend/src/pages/OperatorMarket/Detail/components/ChangeLog.tsx

@@ -22,7 +22,7 @@ export default function ChangeLog({ operator }) {
             )}
           </div>
           <ul className="space-y-2">
-            {release.changelog.map((change, changeIndex) => (
+            {release.changelog?.map((change, changeIndex) => (
               <li key={changeIndex} className="flex items-start gap-2">
                 <ChevronRight className="w-4 h-4 text-gray-400 mt-0.5 flex-shrink-0" />
                 <span className="text-gray-700">{change}</span>

frontend/src/pages/OperatorMarket/Home/OperatorMarket.tsx

@@ -76,14 +76,10 @@ export default function OperatorMarketPage() {
   };
 
   const handleDeleteOperator = async (operator: OperatorI) => {
-    try {
-      await deleteOperatorByIdUsingDelete(operator.id);
-      message.success(t("operatorMarket.home.operations.messages.deleteSuccess"));
-      fetchData();
-      await initCategoriesTree();
-    } catch (error) {
-      message.error(t("operatorMarket.home.operations.messages.deleteFailed"));
-    }
+    await deleteOperatorByIdUsingDelete(operator.id);
+    message.success(t("operatorMarket.home.operations.messages.deleteSuccess"));
+    fetchData();
+    await initCategoriesTree();
   };
 
   const handleStar = async (operator: OperatorI) => {

frontend/src/pages/OperatorMarket/operator.const.tsx

@@ -148,10 +148,10 @@ export const mapOperator = (op: OperatorI, t: (key: string) => string) => {
         label: t("operatorMarket.const.language"),
         value: "Python",
       },
-      {
-        label: t("operatorMarket.const.function"),
-        value: functionLabel,
-      },
+      // {
+      //   label: t("operatorMarket.const.function"),
+      //   value: functionLabel,
+      // },
     ],
   };
 };
@@ -198,4 +198,4 @@ export const formatBytes = (bytes: number | null | undefined, decimals: number =
   // 4. 格式化数值并拼接单位
   // parseFloat 用于去掉末尾多余的 0 (例如 "1.20 MB" -> "1.2 MB")
   return `${parseFloat((bytes / Math.pow(k, i)).toFixed(dm))} ${sizes[i]}`;
-};
+};

frontend/vite.config.ts

@@ -13,54 +13,25 @@
   },
   server: {
     host: "0.0.0.0",
-    proxy: (() => {
-      const pythonProxyConfig = {
-        target: "http://localhost:18000",
+    proxy: {
+      "^/api": {
+        target: "http://localhost:8080", // 本地后端服务地址
         changeOrigin: true,
         secure: false,
-        configure: (proxy: { on: (event: string, handler: (arg: unknown) => void) => void }) => {
-          proxy.on("proxyReq", (proxyReq: unknown) => {
-            (proxyReq as { removeHeader: (name: string) => void }).removeHeader("referer");
-            (proxyReq as { removeHeader: (name: string) => void }).removeHeader("origin");
+        rewrite: (path) => path.replace(/^\/api/, "/api"),
+        configure: (proxy, options) => {
+          // proxy 是 'http-proxy' 的实例
+          proxy.on("proxyReq", (proxyReq, req, res) => {
+            // 可以在这里修改请求头
+            proxyReq.removeHeader("referer");
+            proxyReq.removeHeader("origin");
           });
-          proxy.on("proxyRes", (proxyRes: unknown) => {
-            const res = proxyRes as { headers: Record<string, unknown> };
-            delete res.headers["set-cookie"];
-            res.headers["cookies"] = "";
+          proxy.on("proxyRes", (proxyRes, req, res) => {
+            delete proxyRes.headers["set-cookie"];
+            proxyRes.headers["cookies"] = ""; // 清除 cookies 头
           });
         },
-      };
-
-      const javaProxyConfig = {
-        target: "http://localhost:8080",
-        changeOrigin: true,
-        secure: false,
-        configure: (proxy: { on: (event: string, handler: (arg: unknown) => void) => void }) => {
-          proxy.on("proxyReq", (proxyReq: unknown) => {
-            (proxyReq as { removeHeader: (name: string) => void }).removeHeader("referer");
-            (proxyReq as { removeHeader: (name: string) => void }).removeHeader("origin");
-          });
-          proxy.on("proxyRes", (proxyRes: unknown) => {
-            const res = proxyRes as { headers: Record<string, unknown> };
-            delete res.headers["set-cookie"];
-            res.headers["cookies"] = "";
-          });
-        },
-      };
-
-      // Python 服务: rag, synthesis, annotation, evaluation, models
-      const pythonPaths = ["rag", "synthesis", "annotation", "data-collection", "evaluation", "models"];
-      // Java 服务: data-management, knowledge-base
-      const javaPaths = ["data-management", "knowledge-base", "operators"];
-
-      const proxy: Record<string, object> = {};
-      for (const p of pythonPaths) {
-        proxy[`/api/${p}`] = pythonProxyConfig;
-      }
-      for (const p of javaPaths) {
-        proxy[`/api/${p}`] = javaProxyConfig;
-      }
-      return proxy;
-    })(),
+      },
+    },
   },
 });

runtime/datamate-python/app/core/exception/codes.py

@@ -86,6 +86,26 @@ def __init__(self):
     RATIO_ALREADY_EXISTS: Final = ErrorCode("ratio.0003", "Task already exists", 400)
     RATIO_DELETE_FAILED: Final = ErrorCode("ratio.0004", "Failed to delete task", 500)
 
+    # ========== 清洗模块 ==========
+    CLEANING_TASK_NOT_FOUND: Final = ErrorCode("cleaning.0001", "Cleaning task not found", 404)
+    CLEANING_NAME_DUPLICATED: Final = ErrorCode("cleaning.0002", "Cleaning task name is duplicated", 400)
+    CLEANING_TEMPLATE_NOT_FOUND: Final = ErrorCode("cleaning.0003", "Cleaning template not found", 404)
+    CLEANING_TEMPLATE_NAME_DUPLICATED: Final = ErrorCode("cleaning.0004", "Cleaning template name is duplicated", 400)
+    CLEANING_INVALID_OPERATOR_INPUT: Final = ErrorCode("cleaning.0005", "Invalid operator input/output types", 400)
+    CLEANING_INVALID_EXECUTOR_TYPE: Final = ErrorCode("cleaning.0006", "Invalid executor type", 400)
+    CLEANING_DATASET_NOT_FOUND: Final = ErrorCode("cleaning.0007", "Dataset not found", 404)
+    CLEANING_FILE_SYSTEM_ERROR: Final = ErrorCode("cleaning.0008", "File system error", 500)
+    CLEANING_SETTINGS_PARSE_ERROR: Final = ErrorCode("cleaning.0009", "Settings parse error", 400)
+    CLEANING_TASK_ID_REQUIRED: Final = ErrorCode("cleaning.0010", "Task ID is required", 400)
+
+    # ========== 算子市场模块 ==========
+    OPERATOR_NOT_FOUND: Final = ErrorCode("operator.0001", "Operator not found", 404)
+    OPERATOR_IN_INSTANCE: Final = ErrorCode("operator.0002", "Operator is in use", 400)
+    OPERATOR_CANNOT_DELETE_PREDEFINED: Final = ErrorCode("operator.0003", "Cannot delete predefined operator", 400)
+    OPERATOR_UNSUPPORTED_FILE_TYPE: Final = ErrorCode("operator.0004", "Unsupported file type", 400)
+    OPERATOR_PARSE_FAILED: Final = ErrorCode("operator.0005", "Failed to parse operator package", 400)
+    OPERATOR_FIELD_NOT_FOUND: Final = ErrorCode("operator.0006", "Required field is missing", 400)
+
     # ========== 系统模块 ==========
     SYSTEM_MODEL_NOT_FOUND: Final = ErrorCode("system.0006", "Model configuration not found", 404)
     SYSTEM_MODEL_HEALTH_CHECK_FAILED: Final = ErrorCode("system.0007", "Model health check failed", 500)

runtime/datamate-python/app/core/exception/middleware.py

@@ -69,7 +69,7 @@ async def dispatch(self, request: Request, call_next):
         except Exception as exc:
             # 捕获所有未处理的异常
             logger.error(
-                f"Unhandled exception occurred at {request.method} {request.url.path}",
+                f"Unhandled exception occurred at {request.method} {request.url.path}", exc,
                 exc_info=True
             )
             return self._error_response(

runtime/datamate-python/app/db/models/__init__.py

@@ -21,6 +21,17 @@
     EvaluationItem
 )
 
+from .operator import (
+    Operator,
+    Category,
+    CategoryRelation,
+    OperatorRelease
+)
+
+from .chunk_upload import (
+    ChunkUploadPreRequest
+)
+
 __all__ = [
     "Dataset",
     "DatasetTag",
@@ -32,4 +43,9 @@
     "LabelingProject",
     "EvaluationTask",
     "EvaluationItem",
+    "Operator",
+    "Category",
+    "CategoryRelation",
+    "OperatorRelease",
+    "ChunkUploadPreRequest",
 ]

runtime/datamate-python/app/db/models/chunk_upload.py

@@ -0,0 +1,38 @@
+"""
+Chunk Upload Database Model
+分片上传数据库模型
+"""
+from sqlalchemy import Column, String, Integer, DateTime
+from sqlalchemy.sql import func
+
+from app.db.models.base_entity import Base
+
+
+class ChunkUploadPreRequest(Base):
+    """分片上传预请求"""
+    __tablename__ = "t_chunk_upload_request"
+
+    id = Column(String(36), primary_key=True, comment="请求ID")
+    total_file_num = Column(Integer, nullable=False, comment="总文件数")
+    uploaded_file_num = Column(Integer, nullable=True, comment="已上传文件数")
+    upload_path = Column(String(512), nullable=False, comment="文件路径")
+    timeout = Column(DateTime, nullable=False, comment="上传请求超时时间")
+    service_id = Column(String(64), nullable=True, comment="上传请求所属服务ID")
+    check_info = Column(String(512), nullable=True, comment="业务信息")
+
+    def increment_uploaded_file_num(self):
+        """增加已上传文件数"""
+        if self.uploaded_file_num is None:
+            self.uploaded_file_num = 1
+        else:
+            self.uploaded_file_num += 1
+
+    def is_upload_complete(self) -> bool:
+        """检查是否已完成上传"""
+        return (self.uploaded_file_num is not None and
+                self.uploaded_file_num == self.total_file_num)
+
+    def is_request_timeout(self) -> bool:
+        """检查是否已超时"""
+        from datetime import datetime
+        return self.timeout is not None and datetime.utcnow() > self.timeout