使用Watchdog监控文件夹时重复输出日志的问题排查

问题原因 #

出现重复日志的核心原因并非Watchdog本身的bug，而是由以下两点导致：

1. 文件系统底层行为：Windows/macOS/Linux的文件系统在处理文件写入、保存时，会拆分多个操作步骤（比如编辑器保存时会先创建临时文件，写入内容后替换原文件，最后删除临时文件），每个步骤都会触发独立的文件事件。
2. 编辑器自动保存机制：多数编辑器会在输入过程中自动触发多次保存，或保存时执行多轮写入操作，进而导致多个修改事件被捕获。

解决方案 #

1. 自定义防抖事件处理器 #

通过防抖（Debounce）机制，在指定时间窗口内对同一个文件的重复修改事件只处理一次，以下是修改后的代码：

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

class DebouncedEventHandler(FileSystemEventHandler):
    def __init__(self, debounce_time=0.5):
        self.debounce_time = debounce_time
        self.last_event_time = {}

    def on_modified(self, event):
        if event.is_directory:
            return
        current_time = time.time()
        file_path = event.src_path
        # 检查是否在防抖时间窗口内，避免重复触发
        if file_path in self.last_event_time:
            elapsed = current_time - self.last_event_time[file_path]
            if elapsed < self.debounce_time:
                return
        self.last_event_time[file_path] = current_time
        logging.info(f'Modified file: {file_path}')

    def on_created(self, event):
        if not event.is_directory:
            logging.info(f'Created file: {event.src_path}')

    def on_deleted(self, event):
        if not event.is_directory:
            logging.info(f'Deleted file: {event.src_path}')

    def on_moved(self, event):
        if not event.is_directory:
            logging.info(f'Moved file: from {event.src_path} to {event.dest_path}')

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')

    path = sys.argv[1] if len(sys.argv) > 1 else '.'

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

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

2. 过滤临时文件 #

很多编辑器保存时会生成临时文件（比如VS Code的.tmp文件、Office的~$开头文件），可以在事件处理器中直接忽略这些文件：

def on_modified(self, event):
    if event.is_directory:
        return
    file_path = event.src_path
    # 过滤常见临时文件格式
    if file_path.endswith('.tmp') or file_path.startswith('~$'):
        return
    # 后续防抖逻辑...

效果说明 #

使用防抖处理器后，同一个文件在0.5秒内的多次修改事件会被合并成一条日志，可直接解决单次操作触发多条重复日志的问题。

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