From 8844d22da0d69f9d799fcd581d7cd5d9da4f8383 Mon Sep 17 00:00:00 2001 From: hhhhsc <1710496817@qq.com> Date: Mon, 22 Dec 2025 19:18:54 +0800 Subject: [PATCH 1/2] fix: prevent deletion of predefined operators and improve error handling --- Makefile | 2 +- .../operator/application/OperatorService.java | 3 ++ .../CategoryRelationRepository.java | 2 + .../exception/OperatorErrorCode.java | 4 +- .../Impl/CategoryRelationRepositoryImpl.java | 9 ++++ backend/services/pom.xml | 6 +++ frontend/src/components/ActionDropdown.tsx | 2 + .../Detail/components/FileTable.tsx | 38 ++++++++++++++++ .../OperatorMarket/Home/components/List.tsx | 45 +------------------ runtime/datamate-python/app/core/config.py | 2 +- .../filter/remove_duplicate_file/process.py | 2 +- .../remove_duplicate_file/sql/sql_config.json | 2 +- .../python-executor/datamate/core/base_op.py | 31 ++++++++++--- .../python-executor/datamate/core/dataset.py | 5 ++- 14 files changed, 97 insertions(+), 56 deletions(-) diff --git a/Makefile b/Makefile index 183759314..9e34676da 100644 --- a/Makefile +++ b/Makefile @@ -155,7 +155,7 @@ endef # ========== Build Targets ========== # Valid build targets -VALID_BUILD_TARGETS := backend database frontend runtime backend-python deer-flow mineru mineru-npu +VALID_BUILD_TARGETS := backend database frontend runtime backend-python deer-flow mineru mineru-npu gateway # Generic docker build target with service name as parameter # Automatically prefixes image names with "datamate-" unless it's deer-flow diff --git a/backend/services/operator-market-service/src/main/java/com/datamate/operator/application/OperatorService.java b/backend/services/operator-market-service/src/main/java/com/datamate/operator/application/OperatorService.java index 87ed4334c..1542ea229 100644 --- a/backend/services/operator-market-service/src/main/java/com/datamate/operator/application/OperatorService.java +++ b/backend/services/operator-market-service/src/main/java/com/datamate/operator/application/OperatorService.java @@ -95,6 +95,9 @@ public void deleteOperator(String id) { if (operatorRepo.operatorInTemplateOrRunning(id)) { throw BusinessException.of(OperatorErrorCode.OPERATOR_IN_INSTANCE); } + if (relationRepo.operatorIsPredefined(id)) { + throw BusinessException.of(OperatorErrorCode.CANT_DELETE_PREDEFINED_OPERATOR); + } operatorRepo.deleteOperator(id); relationRepo.deleteByOperatorId(id); } diff --git a/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/CategoryRelationRepository.java b/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/CategoryRelationRepository.java index bd719a83a..146363b5f 100644 --- a/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/CategoryRelationRepository.java +++ b/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/repository/CategoryRelationRepository.java @@ -15,4 +15,6 @@ public interface CategoryRelationRepository extends IRepository categories); void deleteByOperatorId(String operatorId); + + boolean operatorIsPredefined(String operatorId); } diff --git a/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/exception/OperatorErrorCode.java b/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/exception/OperatorErrorCode.java index f94511b31..4a7676802 100644 --- a/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/exception/OperatorErrorCode.java +++ b/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/exception/OperatorErrorCode.java @@ -18,7 +18,9 @@ public enum OperatorErrorCode implements ErrorCode { SETTINGS_PARSE_FAILED("op.0004", "settings字段解析失败"), - OPERATOR_IN_INSTANCE("op.0005", "算子已被编排在模板或未完成的任务中"); + OPERATOR_IN_INSTANCE("op.0005", "算子已被编排在模板或未完成的任务中"), + + CANT_DELETE_PREDEFINED_OPERATOR("op.0006", "预置算子无法删除"); private final String code; private final String message; diff --git a/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/CategoryRelationRepositoryImpl.java b/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/CategoryRelationRepositoryImpl.java index 47fdc149f..9701e208a 100644 --- a/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/CategoryRelationRepositoryImpl.java +++ b/backend/services/operator-market-service/src/main/java/com/datamate/operator/infrastructure/persistence/Impl/CategoryRelationRepositoryImpl.java @@ -2,6 +2,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.extension.repository.CrudRepository; +import com.datamate.operator.domain.contants.OperatorConstant; import com.datamate.operator.domain.model.CategoryRelation; import com.datamate.operator.domain.repository.CategoryRelationRepository; import com.datamate.operator.infrastructure.converter.CategoryRelationConverter; @@ -48,4 +49,12 @@ public void deleteByOperatorId(String operatorId) { queryWrapper.eq(CategoryRelation::getOperatorId, operatorId); mapper.delete(queryWrapper); } + + @Override + public boolean operatorIsPredefined(String operatorId) { + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(CategoryRelation::getOperatorId, operatorId) + .eq(CategoryRelation::getCategoryId, OperatorConstant.CATEGORY_PREDEFINED_ID); + return this.exists(queryWrapper); + } } diff --git a/backend/services/pom.xml b/backend/services/pom.xml index fd4014609..1e599f2cf 100644 --- a/backend/services/pom.xml +++ b/backend/services/pom.xml @@ -64,6 +64,12 @@ + + org.springframework.boot + spring-boot-starter-log4j2 + ${spring-boot.version} + + com.baomidou mybatis-plus-spring-boot3-starter diff --git a/frontend/src/components/ActionDropdown.tsx b/frontend/src/components/ActionDropdown.tsx index ae4e0396e..9fe696bf4 100644 --- a/frontend/src/components/ActionDropdown.tsx +++ b/frontend/src/components/ActionDropdown.tsx @@ -68,6 +68,7 @@ const ActionDropdown = ({ + + + + ), + onFilter: (value: string, record: any) => + record.destName.toLowerCase().includes(value.toLowerCase()), + render: (text: string) => ( + {text?.replace(/\.[^/.]+$/, "")} + ), + }, { title: "文件类型", dataIndex: "srcType", diff --git a/frontend/src/pages/OperatorMarket/Home/components/List.tsx b/frontend/src/pages/OperatorMarket/Home/components/List.tsx index ea811483d..e0d4f3d94 100644 --- a/frontend/src/pages/OperatorMarket/Home/components/List.tsx +++ b/frontend/src/pages/OperatorMarket/Home/components/List.tsx @@ -1,54 +1,13 @@ -import { Button, List, Tag, Badge } from "antd"; -import { StarFilled } from "@ant-design/icons"; -import { Zap, Settings, X } from "lucide-react"; -import { useState } from "react"; +import { Button, List, Tag } from "antd"; import { useNavigate } from "react-router"; import { Operator } from "../../operator.model"; export function ListView({ operators = [], pagination, operations }) { const navigate = useNavigate(); - const [favoriteOperators, setFavoriteOperators] = useState>( - new Set([1, 3, 6]) - ); - const handleUpdateOperator = (operator: Operator) => { - navigate(`/data/operator-market/create/${operator.id}`); - }; + const handleViewOperator = (operator: Operator) => { navigate(`/data/operator-market/plugin-detail/${operator.id}`); }; - const handleToggleFavorite = (operatorId: number) => { - setFavoriteOperators((prev) => { - const newFavorites = new Set(prev); - if (newFavorites.has(operatorId)) { - newFavorites.delete(operatorId); - } else { - newFavorites.add(operatorId); - } - return newFavorites; - }); - }; - const getStatusBadge = (status: string) => { - const statusConfig = { - active: { - label: "活跃", - color: "green", - icon: , - }, - beta: { - label: "测试版", - color: "blue", - icon: , - }, - deprecated: { - label: "已弃用", - color: "gray", - icon: , - }, - }; - return ( - statusConfig[status as keyof typeof statusConfig] || statusConfig.active - ); - }; return ( query_sql = self.sql_dict.get("query_sql") for i in range(0, total_count, self.page_size): rows = connection.execute( - text(query_sql), {"task_uuid": self.task_uuid, "ge": self.page_size, "le": i}).fetchall() + text(query_sql), {"task_uuid": self.task_uuid, "file_name": file_name, "ge": self.page_size, "le": i}).fetchall() # 对应任务uuid，最后一页没有数据，跳出循环 if not rows: break diff --git a/runtime/ops/filter/remove_duplicate_file/sql/sql_config.json b/runtime/ops/filter/remove_duplicate_file/sql/sql_config.json index acef0ccc5..9eec96727 100644 --- a/runtime/ops/filter/remove_duplicate_file/sql/sql_config.json +++ b/runtime/ops/filter/remove_duplicate_file/sql/sql_config.json @@ -1,5 +1,5 @@ { - "query_sql": "SELECT * FROM operators_similar_text_features WHERE task_uuid = :task_uuid ORDER BY timestamp LIMIT :ge OFFSET :le", + "query_sql": "SELECT * FROM operators_similar_text_features WHERE task_uuid = :task_uuid AND file_name != :file_name ORDER BY timestamp LIMIT :ge OFFSET :le", "create_tables_sql": "CREATE TABLE IF NOT EXISTS operators_similar_text_features (id INT AUTO_INCREMENT PRIMARY KEY, task_uuid VARCHAR(255),file_feature TEXT,file_name TEXT,timestamp DATETIME);", "insert_sql": "INSERT INTO operators_similar_text_features (task_uuid, file_feature, file_name, timestamp) VALUES (:task_uuid, :file_feature, :file_name, :timestamp)", "query_task_uuid_sql": "SELECT * FROM operators_similar_text_features WHERE task_uuid = :task_uuid" diff --git a/runtime/python-executor/datamate/core/base_op.py b/runtime/python-executor/datamate/core/base_op.py index bef03ef6a..fd5f661df 100644 --- a/runtime/python-executor/datamate/core/base_op.py +++ b/runtime/python-executor/datamate/core/base_op.py @@ -5,6 +5,7 @@ import time import traceback import uuid +from pathlib import Path from typing import List, Dict, Any, Tuple import cv2 @@ -445,7 +446,7 @@ def execute(self, sample: Dict[str, Any]): return False if save_path: - self.save_file(sample, save_path) + save_path = self.save_file(sample, save_path) sample[self.text_key] = '' sample[self.data_key] = b'' sample[Fields.result] = True @@ -453,6 +454,7 @@ def execute(self, sample: Dict[str, Any]): file_type = save_path.split('.')[-1] sample[self.filetype_key] = file_type + file_name = os.path.basename(save_path) base_name, _ = os.path.splitext(file_name) new_file_name = base_name + '.' + file_type sample[self.filename_key] = new_file_name @@ -516,13 +518,28 @@ def get_medicalfile_handler(self, sample: Dict[str, Any]): def save_file(self, sample, save_path): # 以二进制格式保存文件 file_sample = sample[self.text_key].encode('utf-8') if sample[self.text_key] else sample[self.data_key] - with open(save_path, 'wb') as f: - f.write(file_sample) - # 获取父目录路径 - - parent_dir = os.path.dirname(save_path) + path_obj = Path(save_path).resolve() + parent_dir = path_obj.parent + stem = path_obj.stem # 文件名不含后缀 + suffix = path_obj.suffix # 后缀 (.txt) + + counter = 0 + current_path = path_obj + while True: + try: + # x 模式保证：如果文件存在则报错，如果不存在则创建。 + # 这个检查+创建的过程是操作系统级的原子操作，没有竞态条件。 + with open(current_path, 'xb') as f: + f.write(file_sample) + break + except FileExistsError: + # 文件已存在（被其他线程/进程抢占），更新文件名重试 + counter += 1 + new_filename = f"{stem}_{counter}{suffix}" + current_path = parent_dir / new_filename os.chmod(parent_dir, 0o770) - os.chmod(save_path, 0o640) + os.chmod(current_path, 0o640) + return str(current_path) def _get_from_data(self, sample: Dict[str, Any]) -> Dict[str, Any]: sample[self.data_key] = bytes(sample[self.data_key]) diff --git a/runtime/python-executor/datamate/core/dataset.py b/runtime/python-executor/datamate/core/dataset.py index 185752441..1058ed2c7 100644 --- a/runtime/python-executor/datamate/core/dataset.py +++ b/runtime/python-executor/datamate/core/dataset.py @@ -5,6 +5,7 @@ import os import importlib import sys +import uuid from abc import ABC, abstractmethod import pyarrow as pa @@ -119,12 +120,14 @@ def process(self, # 加载Ops module temp_ops = self.load_ops_module(op_name) + operators_cls_list.append(temp_ops) + if index == 0: init_kwargs["is_first_op"] = True if index == len(cfg_process) - 1: init_kwargs["is_last_op"] = True - operators_cls_list.append(temp_ops) + init_kwargs["instance_id"] = kwargs.get("instance_id", str(uuid.uuid4())) init_kwargs_list.append(init_kwargs) for cls_id, operators_cls in enumerate(operators_cls_list): From f8de9090c0a3b2000b9a71f8ae9142048390faea Mon Sep 17 00:00:00 2001 From: hhhhsc <1710496817@qq.com> Date: Mon, 22 Dec 2025 19:22:48 +0800 Subject: [PATCH 2/2] fix: prevent deletion of predefined operators and improve error handling --- .../com/datamate/operator/domain/contants/OperatorConstant.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/contants/OperatorConstant.java b/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/contants/OperatorConstant.java index 144389e3e..7c88041cd 100644 --- a/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/contants/OperatorConstant.java +++ b/backend/services/operator-market-service/src/main/java/com/datamate/operator/domain/contants/OperatorConstant.java @@ -30,6 +30,8 @@ public class OperatorConstant { public static String CATEGORY_STAR_ID = "51847c24-bba9-11f0-888b-5b143cb738aa"; + public static String CATEGORY_PREDEFINED_ID = "96a3b07a-3439-4557-a835-525faad60ca3"; + public static Map CATEGORY_MAP = new HashMap<>(); static {