如何复用Watchdog的FolderWatcher类，避免重复创建重写on_created的类？

我正在使用Watchdog库监控不同文件夹，两个文件夹有不同的行为：

• 在alpha文件夹中，当有新文件创建时，将其移动到destination_alpha。
• 在beta文件夹中，当有新文件创建时，将其传入一个方法。

目前实现第一种行为的代码片段：

import shutil
import time
from watchdog.events import FileSystemHandler
from watchdog.observers import Observer

class FolderWatcher(FileSystemHandler):
    '''Overrides the on_created method to take action when a file is created.'''
    def on_created(self, event):
        shutil.move(event.src_path, '/destination_alpha')

event_handler = FolderWatcher()
folder_alpha_observer = Observer()
folder_alpha_observer.schedule(event_handler, '/folder_alpha')

try:
    while True:
        time.sleep(1)
finally: 
    folder_alpha_observer.stop()
    folder_alpha_observer.join()

第二种行为的代码片段：

import time
from watchdog.events import FileSystemHandler
from watchdog.observers import Observer

class SecondFolderWatcher(FileSystemHandler):
    '''Overrides the on_created method to take action when a file is created.'''
    def on_created(self, event):
        imported_method(event.src_path)

second_folder_watcher = SecondFolderWatcher()
folder_beta_observer = Observer()
folder_beta_observer.schedule(second_folder_watcher, '/folder_beta')

try:
    while True:
        time.sleep(1)
finally: 
    folder_beta_observer.stop()
    folder_beta_observer.join()

为每个on_created操作创建全新的类不够优雅，请问有什么优化思路？

完全可以复用同一个类，通过注入回调函数的方式让不同实例拥有不同的行为，无需重复创建类。

1. 通用事件处理类实现 #

定义一个接受回调函数的通用DynamicFolderWatcher类，在on_created方法中调用传入的回调（可额外添加非文件事件过滤，避免处理文件夹创建）：

import shutil
import time
from watchdog.events import FileSystemHandler
from watchdog.observers import Observer

class DynamicFolderWatcher(FileSystemHandler):
    def __init__(self, on_created_callback):
        self.on_created_callback = on_created_callback

    def on_created(self, event):
        # 过滤掉文件夹创建事件，只处理文件
        if not event.is_directory:
            self.on_created_callback(event.src_path)

2. 为不同文件夹创建实例 #

• 针对alpha文件夹，传入移动文件的逻辑作为回调：

def move_to_alpha(src_path):
    shutil.move(src_path, '/destination_alpha')

alpha_handler = DynamicFolderWatcher(move_to_alpha)
alpha_observer = Observer()
alpha_observer.schedule(alpha_handler, '/folder_alpha')

• 针对beta文件夹，传入目标方法作为回调：

# 假设imported_method是已导入的目标方法
def call_imported_method(src_path):
    imported_method(src_path)

beta_handler = DynamicFolderWatcher(call_imported_method)
beta_observer = Observer()
beta_observer.schedule(beta_handler, '/folder_beta')

3. 同时启动多个观察者 #

要同时监控两个文件夹，需要将每个观察者放在独立线程中运行：

import threading

def run_observer(observer):
    observer.start()
    try:
        while True:
            time.sleep(1)
    finally:
        observer.stop()
        observer.join()

# 启动两个线程分别运行观察者
threading.Thread(target=run_observer, args=(alpha_observer,)).start()
threading.Thread(target=run_observer, args=(beta_observer,)).start()

额外简化：用Lambda表达式缩短代码 #

对于逻辑简单的回调，可直接用lambda表达式替代单独定义函数，比如alpha文件夹的回调可以简化为：

alpha_handler = DynamicFolderWatcher(lambda path: shutil.move(path, '/destination_alpha'))

这种方式既避免了重复创建类，又能灵活定义每个文件夹的行为，代码更简洁易维护。

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