使用Watchdog监听文件夹文件修改时自定义函数无法触发的问题

我用Watchdog库监听指定文件夹的文件修改事件，期望文件改动时自动运行Python脚本。已创建项目文件夹、激活venv，脚本也放置在目标文件夹内。

示例代码如下：

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import pandas as pd

def on_modified(event):
    print("modified")
    test()

def test():
    print("test")

if __name__ == "__main__":
    path = 'C:/Users/MyUser/Desktop/Watchdog_test'
    event_handler = FileSystemEventHandler()

    event_handler.on_modified = on_modified

    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    try:
        print("monitoring")
        while True:
            time.sleep(3)
    except KeyboardInterrupt:
        observer.stop()
        print("Done")
    observer.join()

当前代码能正常启动并打印monitoring，修改文件夹内文件时终端也会输出modified，但test()函数完全没被触发。实际场景中，我在on_modified里调用trigger_all_functions()执行多个业务函数，同样遇到调用失效的问题：

def on_modified(event):
    print("modified")
    trigger_all_functions()

def trigger_all_functions():
    function1(...)
    ... = function2(...)

问题原因及修复方案 #

1. 重复触发的事件干扰

Windows系统下，修改文件时Watchdog会多次触发on_modified事件（比如文件保存时会先创建临时文件再替换，触发多次修改事件），后续事件可能打断前一次的函数执行，导致看起来函数没运行。

修复方式：给on_modified添加防抖逻辑，忽略短时间内的重复事件：

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import pandas as pd

last_trigger_time = 0
DEBOUNCE_DELAY = 1  # 防抖间隔，单位秒

def on_modified(event):
    global last_trigger_time
    current_time = time.time()
    # 跳过重复触发的事件
    if current_time - last_trigger_time < DEBOUNCE_DELAY:
        return
    last_trigger_time = current_time
    
    print("modified")
    test()

def test():
    print("test")

if __name__ == "__main__":
    path = 'C:/Users/MyUser/Desktop/Watchdog_test'
    event_handler = FileSystemEventHandler()
    event_handler.on_modified = on_modified

    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    try:
        print("monitoring")
        while True:
            time.sleep(3)
    except KeyboardInterrupt:
        observer.stop()
        print("Done")
    observer.join()

2. 函数执行异常被静默

如果test()或trigger_all_functions()内部存在报错，Watchdog默认不会打印异常信息，会让你误以为函数没被调用。

修复方式：给函数调用添加try-except捕获异常并打印：

def on_modified(event):
    print("modified")
    try:
        test()
    except Exception as e:
        print(f"调用test()出错: {e}")

def test():
    print("test")
    # 若有业务逻辑，这里的报错会被捕获并输出

3. 采用继承FileSystemEventHandler的规范写法

直接给event_handler.on_modified赋值的方式虽可行，但更规范的做法是继承FileSystemEventHandler并重写方法，避免潜在的事件冲突：

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import pandas as pd

class MyHandler(FileSystemEventHandler):
    last_trigger_time = 0
    DEBOUNCE_DELAY = 1

    def on_modified(self, event):
        current_time = time.time()
        if current_time - self.last_trigger_time < self.DEBOUNCE_DELAY:
            return
        self.last_trigger_time = current_time
        
        print("modified")
        try:
            self.test()
        except Exception as e:
            print(f"函数执行出错: {e}")

    def test(self):
        print("test")

if __name__ == "__main__":
    path = 'C:/Users/MyUser/Desktop/Watchdog_test'
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    try:
        print("monitoring")
        while True:
            time.sleep(3)
    except KeyboardInterrupt:
        observer.stop()
        print("Done")
    observer.join()

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