为什么Python Watchdog在文件变更时会触发多个看似错误的事件？

大部分文本编辑器的保存逻辑采用「临时文件替换策略」避免保存过程异常损坏原文件，单次保存的实际磁盘操作顺序为：

1. 写入修改后的内容到一个新的临时文件（触发FileCreatedEvent，同时文件所属目录变更触发DirModifiedEvent）
2. 删除原有源文件，将临时文件重命名为原文件名称（触发原文件的FileDeletedEvent，同时文件所属目录再次变更触发DirModifiedEvent）
   这就是你观察到4次事件、且文件事件为创建+删除而非修改的核心原因。你的代码中虽然过滤了目录事件，但两次非目录的文件事件仍然会各自触发一次停止-构建-启动流程，加上如果构建过程本身有对源目录的写入操作、或者进程终止的延迟导致事件堆积，就会出现重复执行4次的现象。

你需要做三处核心调整：

1. 增加防抖逻辑：设置最短触发间隔，同一次保存产生的多次事件仅触发一次处理流程
2. 简化事件处理逻辑：复用已经封装好的start_server方法，避免重复执行停止、构建操作
3. 可选：增加文件后缀过滤，忽略不需要监听的临时文件、隐藏文件变更

修改后的参考代码如下：

import os
import time
from http.server import HTTPServer, SimpleHTTPRequestHandler
from multiprocessing import Process
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# Dict containing config
config = {
    [...]
    "source_dir": "/path/to/source",
    "build_dir": "/path/to/build",
    # 新增：需要监听的源文件后缀，可根据你的项目实际情况调整
    "allowed_suffixes": (".md", ".html", ".css", ".js"),
    # 新增：防抖间隔（单位：秒，可根据你的构建速度调整）
    "debounce_interval": 1
}


# Run a local dev server
def create_server():
    os.chdir(config["build_dir"])
    httpd = HTTPServer(("", 8000), SimpleHTTPRequestHandler)
    httpd.serve_forever()


def build():
    # 你的原有构建逻辑
    [...]


def dev():
    class DevServer(FileSystemEventHandler):
        p = None
        # 新增：记录上次触发处理的时间
        last_trigger = 0

        def start_server(self):
            if self.p:
                self.stop_server()
            build()
            print("Starting server...")
            self.p = Process(target=create_server)
            self.p.start()

        def stop_server(self):
            if self.p:
                print("Stopping server...")
                self.p.terminate()
                # 等待进程完全退出，避免端口占用
                self.p.join()

        def on_any_event(self, event):
            # 过滤目录事件
            if event.is_directory:
                return
            # 过滤不符合后缀的文件
            if not event.src_path.endswith(config["allowed_suffixes"]):
                return
            # 防抖判断：小于间隔时间的事件直接忽略
            now = time.time()
            if now - self.last_trigger < config["debounce_interval"]:
                return
            self.last_trigger = now
            # 直接调用start_server即可，内部已经实现了停止-构建-启动的完整流程
            self.start_server()

    devserver = DevServer()
    devserver.start_server()

    observer = Observer()
    observer.schedule(devserver, config["source_dir"], recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        # 退出时主动停止服务器
        devserver.stop_server()
    observer.join()

其他可以优化的点：

• 可以在停止服务器后判断端口是否真的释放，再启动新服务，避免偶发的端口占用报错
• 如果构建过程较长，可以加锁避免上一次构建还没完成就触发下一次构建

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