算子将抽取与落盘固定到流程中 (#134)

* feature: 将抽取动作移到每一个算子中

* feature: 落盘算子改为默认执行

* feature: 优化前端展示

* feature: 使用pyproject管理依赖

Author: hhhhsc701

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

@@ -150,6 +150,7 @@ const OperatorFlow: React.FC<OperatorFlowProps> = ({
                   max={selectedOperators.length}
                   defaultValue={index + 1}
                   className="w-10 h-6 text-xs text-center"
+                  style={{ width: 60 }}
                   autoFocus
                   onBlur={(e) => handleIndexChange(operator.id, e.target.value)}
                   onKeyDown={(e) => {

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

@@ -227,19 +227,16 @@ export default function FileTable({result, fetchTaskResult}) {
       dataIndex: "status",
       key: "status",
       filters: [
-        { text: "已完成", value: "已完成" },
-        { text: "失败", value: "失败" },
-        { text: "处理中", value: "处理中" },
+        { text: "已完成", value: "COMPLETED" },
+        { text: "失败", value: "FAILED" },
       ],
       onFilter: (value: string, record: any) => record.status === value,
       render: (status: string) => (
         <Badge
           status={
             status === "COMPLETED"
               ? "success"
-              : status === "FAILED"
-              ? "error"
-              : "processing"
+              : "error"
           }
           text={TaskStatusMap[status as TaskStatus].label}
         />
@@ -248,6 +245,7 @@ export default function FileTable({result, fetchTaskResult}) {
     {
       title: "操作",
       key: "action",
+      width: 200,
       render: (_text: string, record: any) => (
         <div className="flex">
           {record.status === "COMPLETED" ? (

runtime/ops/filter/file_with_high_repeat_phrase_rate_filter/process.py

@@ -33,6 +33,7 @@ def __init__(self, *args, **kwargs):
 
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         start = time.time()
+        self.read_file_first(sample)
         sample[self.text_key] = self._file_with_high_repeat_phrase_rate_filter(sample[self.text_key],
                                                                                sample[self.filename_key])
         logger.info(f"fileName: {sample[self.filename_key]}, "

runtime/ops/filter/file_with_high_repeat_word_rate_filter/process.py

@@ -30,6 +30,7 @@ def _extract_word(input_data):
 
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         start = time.time()
+        self.read_file_first(sample)
         sample[self.text_key] = self._file_with_high_repeat_word_rate_filter(sample[self.text_key],
                                                                              sample[self.filename_key])
         logger.info(f"fileName: {sample[self.filename_key]}, "

runtime/ops/filter/file_with_high_special_char_rate_filter/process.py

@@ -26,6 +26,7 @@ def __init__(self, *args, **kwargs):
 
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         start = time.time()
+        self.read_file_first(sample)
         sample[self.text_key] = self._file_with_high_special_char_rate_filter(sample[self.text_key],
                                                                               sample[self.filename_key])
         logger.info(f"fileName: {sample[self.filename_key]}, "

runtime/ops/filter/img_advertisement_images_cleaner/process.py

@@ -105,6 +105,7 @@ def resize_img(self, image):
 
     def execute(self, sample: Dict[str, Any]):
         start = time.time()
+        self.read_file_first(sample)
         file_name = sample[self.filename_key]
         file_type = "." + sample[self.filetype_key]
         img_bytes = sample[self.data_key]

runtime/ops/filter/img_blurred_images_cleaner/process.py

@@ -27,6 +27,7 @@ def __init__(self, *args, **kwargs):
 
     def execute(self, sample: Dict[str, Any]):
         start = time.time()
+        self.read_file_first(sample)
         img_bytes = sample[self.data_key]
         file_name = sample[self.filename_key]
         file_type = "." + sample[self.filetype_key]

runtime/ops/filter/img_duplicated_images_cleaner/process.py

@@ -61,6 +61,7 @@ def compute_md5(self, img_bytes: bytes) -> str:
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         """重复图片去重算子执行入口"""
         start = time.time()
+        self.read_file_first(sample)
         file_name = sample[self.filename_key]
         self.task_uuid = sample.get("instance_id") if not self.task_uuid else self.task_uuid
         img_data = self._duplicate_images_filter(file_name, sample[self.data_key])

runtime/ops/filter/img_similar_images_cleaner/process.py

@@ -227,6 +227,7 @@ def determine_similar_images(self, file_features: List, p_hash: str, des_matrix:
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         """去除相似图片算子执行入口"""
         start = time.time()
+        self.read_file_first(sample)
         file_name = sample[self.filename_key]
         img_bytes = sample[self.data_key]
         data = bytes_to_numpy(img_bytes) if img_bytes else np.array([])

runtime/ops/filter/remove_duplicate_file/process.py

@@ -150,6 +150,7 @@ def determine_similar_text(self, file_features: List, text_minhash: MinHash, fil
 
     def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
         start = time.time()
+        self.read_file_first(sample)
         file_name = sample[self.filename_key]
         self.task_uuid = sample.get("instance_id") if not self.task_uuid else self.task_uuid
         sample[self.text_key] = self.deduplicate_files(sample, file_name)