如何修改Python Watchdog代码以显示修改文件的进程？

Watchdog本身只能捕获文件系统的创建、修改、删除事件，但无法直接获取触发事件的进程信息。要实现这个需求，需要结合psutil库来查询当前访问目标文件的进程。

实现步骤 #

1. 先安装依赖库：

pip install psutil watchdog

2. 自定义事件处理器，重写对应的事件方法，在事件触发时通过psutil查找关联进程：

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

logging.basicConfig(level=logging.INFO, format='%(message)s')
path = sys.argv[1] if len(sys.argv) > 1 else '.'

class ProcessAwareEventHandler(FileSystemEventHandler):
    def _get_process_info(self, file_path):
        """获取访问指定文件的进程信息"""
        process_info = []
        try:
            # 遍历所有进程，查找打开了目标文件的进程
            for proc in psutil.process_iter(['pid', 'name', 'exe']):
                try:
                    for item in proc.open_files():
                        if item.path == file_path:
                            process_info.append(f"PID: {proc.pid}, 进程名: {proc.name()}, 路径: {proc.exe()}")
                except (psutil.AccessDenied, psutil.NoSuchProcess):
                    # 跳过无权限访问或已结束的进程
                    continue
        except Exception as e:
            logging.error(f"获取进程信息失败: {str(e)}")
        return process_info

    def on_created(self, event):
        if not event.is_directory:
            proc_info = self._get_process_info(event.src_path)
            logging.info(f"文件创建: {event.src_path}")
            if proc_info:
                logging.info(f"关联进程: {', '.join(proc_info)}")

    def on_modified(self, event):
        if not event.is_directory:
            proc_info = self._get_process_info(event.src_path)
            logging.info(f"文件修改: {event.src_path}")
            if proc_info:
                logging.info(f"关联进程: {', '.join(proc_info)}")

    def on_deleted(self, event):
        if not event.is_directory:
            logging.info(f"文件删除: {event.src_path}")
            # 文件删除后可能无法通过open_files查到进程，可根据需求调整逻辑

if __name__ == "__main__":
    event_handler = ProcessAwareEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

注意事项 #

• 权限问题：部分系统或进程需要管理员/root权限才能访问其打开的文件列表，遇到AccessDenied错误时会自动跳过这些进程。
• 延迟问题：文件修改事件触发时，进程可能已经关闭了文件句柄，导致无法捕获到对应进程信息，可根据业务需求调整查询时机。
• 系统差异：psutil在Windows、Linux、macOS上都能工作，但部分进程信息的展示可能略有不同。

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