如何使用watchdog将新创建的文件名存入变量并供其他函数调用？

我希望使用watchdog监控某一文件夹中新建的文件，并将获取到的文件名作为其他函数的输入参数。目前我使用的代码如下：

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format=' %(message)s')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

该代码会在控制台输出类似如下内容：

Created file: ./test_file.h5
Modified directory: .
Modified file: ./test_file.h5
Modified directory: .

我需要实现的是仅捕获文件创建事件，将新创建的文件名存入变量（而非输出到控制台），以便将其作为其他函数的输入。请问是否有可行的实现方法？

当然有可行的实现方法！你目前用的LoggingEventHandler是watchdog自带的通用日志处理器，会输出所有类型的文件系统事件，但我们可以自定义一个专属的事件处理器，精准捕获文件创建事件，还能轻松把文件名传递给其他函数。

核心思路 #

1. 继承watchdog的FileSystemEventHandler基础类（LoggingEventHandler本身也是继承它的）
2. 重写on_created方法——这个方法是watchdog专门用来处理「文件/文件夹创建」事件的钩子
3. 在方法里过滤掉文件夹创建事件，只处理文件
4. 获取新创建文件的路径，要么存入变量，要么直接传递给你的目标函数

完整代码示例 #

import sys
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 示例：假设这是你需要传入文件名的业务函数
def process_new_file(file_path):
    """处理新创建的文件，这里写你的业务逻辑"""
    print(f"开始处理文件: {file_path}")
    # 比如读取HDF5文件、数据解析等操作都可以放在这里
    # with h5py.File(file_path, 'r') as f:
    #     ...

class FileCreationHandler(FileSystemEventHandler):
    def __init__(self):
        # 如果需要存储所有创建的文件路径，可以初始化一个列表
        self.created_files = []

    def on_created(self, event):
        # 只处理文件创建事件，忽略文件夹
        if not event.is_directory:
            new_file_path = event.src_path
            print(f"捕获到新创建的文件: {new_file_path}")
            
            # 1. 将文件名存入类属性，方便后续其他地方调用
            self.created_files.append(new_file_path)
            
            # 2. 直接将文件名传入你的业务函数
            process_new_file(new_file_path)

if __name__ == "__main__":
    # 获取监控路径，默认监控当前目录
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    
    # 创建自定义事件处理器
    event_handler = FileCreationHandler()
    observer = Observer()
    # 递归监控子文件夹，如果不需要可以把recursive设为False
    observer.schedule(event_handler, path, recursive=True)
    
    observer.start()
    print(f"开始监控文件夹: {path}")
    
    try:
        # 保持程序运行，直到按下Ctrl+C
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("停止监控")
        observer.stop()
    observer.join()
    
    # 如果需要查看所有捕获到的创建文件，可以在这里打印
    print(f"监控期间创建的文件列表: {event_handler.created_files}")

关键细节说明 #

• 过滤文件夹事件：通过event.is_directory判断触发事件的是文件还是文件夹，避免误处理文件夹创建
• 获取文件路径：event.src_path会返回新创建文件的完整路径，直接即可使用
• 灵活传递参数：你可以选择把文件名存入类属性（比如self.created_files）供后续批量处理，也可以直接在on_created里调用业务函数，实现实时处理
• 过滤特定格式文件：如果只需要处理特定后缀的文件（比如.h5），可以在on_created里加个判断：

  if new_file_path.endswith('.h5'):
      process_new_file(new_file_path)
  

运行方式 #

和你原来的代码一样，直接运行脚本并指定监控路径即可：

python your_script.py /path/to/monitor

如果不指定路径，默认监控当前工作目录。

内容的提问来源于stack exchange，提问作者mikanim