🐛 could cause task updates to fail

Author: MoeexT

runtime/datamate-python/app/core/security.py

@@ -69,14 +69,18 @@ def preserve_sensitive_values(
             original_nested = original_config.get(key, {})
             preserved_config[key] = preserve_sensitive_values(new_value, original_nested)
         elif isinstance(new_value, list):
-            # Process list items (preserve original if matching structure)
             original_list = original_config.get(key, [])
-            preserved_config[key] = [
-                preserve_sensitive_values(new_item, orig_item) 
-                if isinstance(new_item, dict) and isinstance(orig_item, dict)
-                else new_item
-                for new_item, orig_item in zip(new_value, original_list)
-            ] if len(new_value) == len(original_list) else new_value
+            preserved_list = []
+            for i, new_item in enumerate(new_value):
+                if i < len(original_list):
+                    orig_item = original_list[i]
+                    if isinstance(new_item, dict) and isinstance(orig_item, dict):
+                        preserved_list.append(preserve_sensitive_values(new_item, orig_item))
+                    else:
+                        preserved_list.append(new_item)
+                else:
+                    preserved_list.append(new_item)
+            preserved_config[key] = preserved_list
         else:
             # Keep non-sensitive/non-masked values
             preserved_config[key] = new_value
@@ -139,18 +143,17 @@ def mask_sensitive_info(text: str) -> str:
 
     for field in SENSITIVE_FIELDS:
         patterns = [
-            rf'"{field}"\s*:\s*"[^"]*"',
             rf'"{field}"\s*:\s*"[^"]*"',
             rf'{field}\s*=\s*[^\s,\]]+',
             rf"'{field}'\s*:\s*'[^']*'",
         ]
 
         for pattern in patterns:
             if '"' in pattern:
-                masked_text = re.sub(pattern, f'"{field}": "{MASK_PATTERN}"', masked_text)
+                masked_text = re.sub(pattern, f'"{field}": "{MASK_PATTERN}"', masked_text, flags=re.IGNORECASE)
             elif "'" in pattern:
-                masked_text = re.sub(pattern, f"'{field}': '{MASK_PATTERN}'", masked_text)
+                masked_text = re.sub(pattern, f"'{field}': '{MASK_PATTERN}'", masked_text, flags=re.IGNORECASE)
             else:
-                masked_text = re.sub(pattern, f'{field}={MASK_PATTERN}', masked_text)
+                masked_text = re.sub(pattern, f'{field}={MASK_PATTERN}', masked_text, flags=re.IGNORECASE)
 
     return masked_text

runtime/datamate-python/app/module/collection/client/datax_client.py

@@ -119,25 +119,19 @@ def run_datax_job(self):
         Returns:
             执行结果字典
         """
-        # 创建配置文件
-        self.create__config_file()
+        if not self.execution.started_at:
+            self.execution.started_at = datetime.now()
+        
         try:
-            # 构建命令
+            self.create__config_file()
             cmd = [self.python_path, str(self.datax_main), str(self.config_file_path)]
             cmd_str = ' '.join(cmd)
             logger.info(f"执行命令: {cmd_str}")
-            if not self.execution.started_at:
-                self.execution.started_at = datetime.now()
-            # 执行命令并写入日志
             with open(self.execution.log_path, 'w', encoding='utf-8') as log_f:
-                # 写入头信息
                 self.write_header_log(cmd_str, log_f)
-                # 启动datax进程
                 exit_code = self._run_process(cmd, log_f)
-                # 记录结束时间
                 self.execution.completed_at = datetime.now()
                 self.execution.duration_seconds = (self.execution.completed_at - self.execution.started_at).total_seconds()
-                # 写入结束信息
                 self.write_tail_log(exit_code, log_f)
             if exit_code == 0:
                 logger.info(f"DataX 任务执行成功: {self.execution.id}")
@@ -150,17 +144,17 @@ def run_datax_job(self):
                 logger.error(self.execution.error_message)
         except Exception as e:
             self.execution.completed_at = datetime.now()
-            self.execution.duration_seconds = (self.execution.completed_at - self.execution.started_at).total_seconds()
+            if self.execution.started_at:
+                self.execution.duration_seconds = (self.execution.completed_at - self.execution.started_at).total_seconds()
             self.execution.error_message = f"执行异常: {e}"
             self.execution.status = TaskStatus.FAILED.name
             logger.error(f"执行异常: {e}", exc_info=True)
+            with open(self.execution.log_path, 'w', encoding='utf-8') as log_f:
+                log_f.write(f"任务执行失败: {e}\n")
 
-        # 根据同步模式更新任务状态
         if self.task.sync_mode == SyncMode.ONCE:
-            # 一次性任务：使用执行结果作为最终状态
             self.task.status = self.execution.status
         else:
-            # 定时任务：恢复为 PENDING 状态，等待下次执行
             self.task.status = TaskStatus.PENDING.name
 
     def rename_collection_result(self):

runtime/datamate-python/app/module/collection/interface/collection.py

@@ -315,13 +315,14 @@ async def update_task(
         template = template.scalar_one_or_none()
         if template:
             # Use preserved config to generate DataX config
+            config_obj = CollectionConfig(**preserved_config)
             DataxClient.generate_datx_config(
-                CollectionConfig(**preserved_config), 
+                config_obj, 
                 template, 
                 task.target_path
             )
-            # Save preserved config to database
-            task.config = json.dumps(preserved_config)
+            # Save the modified config (with regenerated job) to database
+            task.config = json.dumps(config_obj.model_dump())
 
     # 如果任务处于 FAILED 状态，修改后重置为 PENDING，允许重新执行
     if task.status == TaskStatus.FAILED.name: