增加静默模式，适用于无网络或者不想要接收消息场景

Polaris-F · Polaris-F · commit 0c0601e586a9 · 2025-05-13T21:37:52.000+08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,4 +4,10 @@
 
 
 ### 修复
-- MSG_PUSH_TOKEN环境变量读取问题 
+- MSG_PUSH_TOKEN环境变量读取问题 
+
+## 2025年5月12日 TODO
+
+1. CLI 命令增加一个 监听单个任务的功能：
+    输入 taskflow monitor 'XXX', 其中'XXX'表示需要监听的任务的进程名称，针对已经有一个且只有一个任务已经在跑的场景，通过 monitor 在任务结束发送消息。
+<!-- 2. 发送消息改为  -->
diff --git a/README.md b/README.md
@@ -107,14 +107,17 @@ python -m build
 - name: "任务1-数据准备"
   command: "python scripts/prepare_data.py --input data/raw --output data/processed"
   status: "pending"
+  # silent: false  # 默认会发送消息通知
 
 - name: "任务2-模型训练"
   command: "python scripts/train_model.py --data data/processed --epochs 10"
   status: "pending"
+  silent: true  # 静默模式，不发送消息通知
 
 - name: "任务3-结果评估"
   command: "python scripts/evaluate.py --model-path models/latest.pt"
   status: "pending"
+  # silent: false  # 默认会发送消息通知
 ```
 
 ### 2. （方法一）使用Python API (推荐使用方法二、三)
@@ -133,7 +136,8 @@ task_manager.run()
 # 您也可以动态添加任务
 task_manager.add_task_by_config(
     name="额外任务", 
-    command="echo '这是一个动态添加的任务'"
+    command="echo '这是一个动态添加的任务'",
+    silent=True  # 设置为静默模式，不发送消息通知
 )
 ```
 
@@ -214,14 +218,43 @@ taskflow examples/tasks.yaml
 - name: "示例任务"
   command: "python script.py"
   status: "pending"  # pending, running, completed, failed
-  retry: 3  # 失败后重试次数
-  timeout: 3600  # 任务超时时间（秒）
-  depends_on: ["前置任务名称"]  # 依赖的任务
-  env:  # 环境变量设置
-    KEY1: "VALUE1"
-    KEY2: "VALUE2"
+  retry: 3  # 失败后重试次数 (TODO)
+  timeout: 3600  # 任务超时时间（秒）(TODO)
+  depends_on: ["前置任务名称"]  # 依赖的任务 (TODO)
+  silent: false  # 是否静默执行（不发送消息通知）
 ```
 
+### 静默模式
+
+MultiTaskFlow 支持静默模式，可以通过配置让某些任务不发送消息通知。这对于以下场景非常有用：
+
+- **中间过程任务**：对于工作流中的中间步骤，可能不需要收到每个步骤的通知
+- **调试阶段任务**：在开发和调试阶段，可以关闭消息通知以避免干扰
+- **高频执行任务**：对于频繁执行的任务，可以只关注最终结果而不是每次执行
+
+#### 配置静默模式：
+
+1. **在YAML配置文件中**：
+   ```yaml
+   - name: "静默任务"
+     command: "python script.py"
+     silent: true  # 设置为静默模式
+   ```
+
+2. **通过API动态添加**：
+   ```python
+   task_manager.add_task_by_config(
+       name="静默任务", 
+       command="echo '这是静默任务'",
+       silent=True
+   )
+   ```
+
+当任务设置为静默模式时：
+- 任务执行时不会发送消息通知
+- 任务信息仍会记录在日志文件中
+- 如果所有任务都是静默模式，任务流管理器结束时也不会发送总结报告
+
 ### 自定义通知
 
 您可以配置系统在任务状态变更时发送通知：
diff --git a/examples/tasks.yaml b/examples/tasks.yaml
@@ -6,24 +6,30 @@
 - name: "数据预处理"
   command: "python examples/scripts/data_preprocess.py --dataset coco --output-dir data/processed"
   status: "pending"
+  # silent: false  # 默认会发送消息通知
 
 - name: "模型训练-阶段1"
   command: "python examples/scripts/train_model.py --model yolov8 --epochs 5 --batch-size 32 --dataset data/processed"
   status: "pending"
+  silent: true  # 静默模式，不发送消息通知
 
 - name: "模型评估-阶段1"
   command: "python examples/scripts/evaluate_model.py --model-path checkpoints/yolov8_epoch_5.pt --dataset data/val"
   status: "pending"
+  # silent: false  # 默认会发送消息通知
 
 - name: "模型训练-阶段2"
   command: "python examples/scripts/train_model.py --model yolov8 --epochs 5 --batch-size 16 --resume checkpoints/yolov8_epoch_5.pt"
   status: "pending"
+  silent: true  # 静默模式，不发送消息通知
 
 - name: "模型评估-阶段2"
   command: "python examples/scripts/evaluate_model.py --model-path checkpoints/yolov8_epoch_5.pt --dataset data/val"
   status: "pending"
+  # silent: false  # 默认会发送消息通知
 
 # 非Python任务示例
 - name: "数据归档"
   command: "bash examples/scripts/archive_data.sh checkpoints/yolov8_epoch_5.pt results/"
-  status: "pending" 
+  status: "pending" 
+  # silent: false  # 默认会发送消息通知 
diff --git a/multitaskflow/process_monitor.py b/multitaskflow/process_monitor.py
@@ -161,6 +161,7 @@ class ProcessMonitor(Thread):
         process_cmd: 进程命令，用于查找进程
         pid: 进程ID
         start_time: 进程开始时间
+        silent: 是否静默执行（不发送消息通知）
         
     Methods:
         set_result: 设置任务执行结果和错误信息
@@ -182,7 +183,7 @@ class ProcessMonitor(Thread):
         """
     }
 
-    def __init__(self, process_name: str, process_cmd: str, logger: logging.Logger, start_time: datetime = None):
+    def __init__(self, process_name: str, process_cmd: str, logger: logging.Logger, start_time: datetime = None, silent: bool = False):
         """
         初始化进程监控器
         
@@ -191,6 +192,7 @@ def __init__(self, process_name: str, process_cmd: str, logger: logging.Logger,
             process_cmd: 进程命令（用于查找进程）
             logger: 日志记录器
             start_time: 开始时间，默认为当前时间
+            silent: 是否静默执行（不发送消息通知），默认为False
         """
         super().__init__()
         self.process_name = process_name
@@ -200,6 +202,7 @@ def __init__(self, process_name: str, process_cmd: str, logger: logging.Logger,
         self.daemon = True  # 设置为守护线程，随主线程退出
         self.return_code = None
         self.error_message = None
+        self.silent = silent
         load_dotenv()
         self.pid = self._find_process_pid()
 
@@ -302,6 +305,8 @@ def send_notification(self):
         """
         发送进程结束通知
         
+        如果设置了silent=True，将跳过消息发送，只记录日志
+        
         Returns:
             bool: 通知是否发送成功
         """
@@ -317,6 +322,14 @@ def send_notification(self):
             error_msg=error_msg
         )
         
+        # 记录任务完成日志
+        self.logger.info(f"任务 {self.process_name} 已{status}，运行时长: {self.get_duration()}")
+        
+        # 如果设置了静默模式，跳过消息发送
+        if self.silent:
+            self.logger.info("任务设置为静默模式，跳过消息发送")
+            return True
+            
         return Msg_push(self.MESSAGE_TEMPLATE["title"], content, self.logger)
 
 def setup_logger(name: str, log_dir: str = "logs") -> logging.Logger:
diff --git a/multitaskflow/task_flow.py b/multitaskflow/task_flow.py
@@ -63,21 +63,23 @@ class Task:
         end_time: 结束时间
         return_code: 命令返回值
         duration: 执行时长
+        silent: 是否静默执行（不发送消息通知）
     """
     
     STATUS_PENDING = "pending"
     STATUS_RUNNING = "running"
     STATUS_COMPLETED = "completed"
     STATUS_FAILED = "failed"
 
-    def __init__(self, name: str, command: str, status: str = STATUS_PENDING):
+    def __init__(self, name: str, command: str, status: str = STATUS_PENDING, silent: bool = False):
         """
         初始化任务实例
         
         Args:
             name: 任务名称
             command: 要执行的命令行字符串
             status: 初始状态，默认为"pending"
+            silent: 是否静默执行（不发送消息通知），默认为False
         """
         self.name = name
         self.command = command
@@ -89,6 +91,7 @@ def __init__(self, name: str, command: str, status: str = STATUS_PENDING):
         self.error_message = None
         self.duration = None
         self.monitor = None
+        self.silent = silent
 
     def to_dict(self) -> Dict[str, Any]:
         """
@@ -105,7 +108,8 @@ def to_dict(self) -> Dict[str, Any]:
             "end_time": self.end_time.strftime('%Y-%m-%d %H:%M:%S') if self.end_time else None,
             "duration": str(self.duration) if self.duration else "未完成",
             "return_code": self.return_code,
-            "error_message": self.error_message
+            "error_message": self.error_message,
+            "silent": self.silent
         }
 
     def update_duration(self):
@@ -241,6 +245,11 @@ def load_tasks(self):
         从配置文件加载初始任务
         
         配置文件应为YAML格式，包含任务列表，每个任务需指定名称和命令
+        任务可以包含以下参数：
+        - name: 任务名称（必需）
+        - command: 要执行的命令（必需）
+        - status: 任务状态（可选，默认为"pending"）
+        - silent: 是否静默执行（可选，默认为False，设为True时不发送消息通知）
         """
         try:
             with open(self.config_path, 'r', encoding='utf-8') as f:
@@ -253,7 +262,8 @@ def load_tasks(self):
                 task = Task(
                     name=task_config['name'],
                     command=task_config['command'],
-                    status=task_config.get('status', 'pending')
+                    status=task_config.get('status', 'pending'),
+                    silent=task_config.get('silent', False)
                 )
                 self.add_task(task)
                 
@@ -275,18 +285,19 @@ def add_task(self, task: Task):
             self.total_tasks += 1
         self.logger.info(f"新任务已添加: {task.name}")
 
-    def add_task_by_config(self, name: str, command: str):
+    def add_task_by_config(self, name: str, command: str, silent: bool = False):
         """
         通过参数添加新任务
         
         Args:
             name: 任务名称
             command: 要执行的命令
+            silent: 是否静默执行（不发送消息通知），默认为False
             
         Returns:
             Task: 新添加的任务实例
         """
-        task = Task(name=name, command=command)
+        task = Task(name=name, command=command, silent=silent)
         self.add_task(task)
         return task
 
@@ -381,6 +392,9 @@ def execute_task(self, task: Task) -> bool:
         self.logger.info(f"开始执行任务: {task.name}")
         self.logger.info(f"执行命令: {task.command}")
         
+        if task.silent:
+            self.logger.info("任务设置为静默模式，不会发送消息通知")
+            
         task.start()
         
         try:
@@ -399,7 +413,8 @@ def execute_task(self, task: Task) -> bool:
                 process_name=task.name,
                 process_cmd=task.command,
                 logger=self.logger,
-                start_time=task.start_time
+                start_time=task.start_time,
+                silent=task.silent  # 传递静默模式设置
             )
             task.monitor.start()
             
@@ -459,7 +474,8 @@ def check_new_tasks(self):
                     task = Task(
                         name=task_config['name'],
                         command=task_config['command'],
-                        status=task_config.get('status', 'pending')
+                        status=task_config.get('status', 'pending'),
+                        silent=task_config.get('silent', False)
                     )
                     self.add_task(task)
                     
@@ -518,6 +534,8 @@ def run(self):
     def stop(self):
         """
         停止任务流管理器并发送总结报告
+        
+        如果所有任务都是静默模式，则不发送总结报告
         """
         self.stop_event.set()
         self.logger.info("正在停止任务流管理器...")
@@ -526,15 +544,23 @@ def stop(self):
         if self.running:
             self.end_time = datetime.now()
             summary = self.generate_summary()
-            self.logger.info("发送任务总结报告...")
-            self.logger.info(summary)  # 在日志中也记录摘要
+            self.logger.info("任务总结报告:")
+            self.logger.info(summary)  # 在日志中记录摘要
             
-            # 发送总结报告
-            Msg_push(
-                title="任务流管理器执行报告",
-                content=summary,
-                logger=self.logger
-            )
+            # 检查是否所有任务都是静默模式
+            all_silent = all(task.silent for task in self.tasks if len(self.tasks) > 0)
+            
+            # 只有在非静默模式下才发送总结报告
+            if not all_silent:
+                # 发送总结报告
+                self.logger.info("发送任务总结报告...")
+                Msg_push(
+                    title="任务流管理器执行报告",
+                    content=summary,
+                    logger=self.logger
+                )
+            else:
+                self.logger.info("所有任务都设置为静默模式，跳过发送总结报告")
 
     def is_running(self) -> bool:
         """
