python watchdog：监控文件系统事件的Python库

约定不等于承诺〃 2022-08-19 12:20 191阅读 0赞

**python watchdog：监控文件系统事件的Python库和shell工具**

watchdog用来监控指定目录/文件的变化，如添加删除文件或目录、修改文件内容、重命名文件或目录等，每种变化都会产生一个事件，且有一个特定的事件类与之对应，然后再通过事件处理类来处理对应的事件，怎么样处理事件完全可以自定义，只需继承事件处理类的基类并重写对应实例方法。

先给一个官网的例子：

import sys
    import time
    import logging
    from watchdog.observers import Observer
    from watchdog.events import LoggingEventHandler
    
    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 = LoggingEventHandler()
        observer = Observer()
        observer.schedule(event_handler, path, recursive=True)
        observer.start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

使用 `LoggingEventHandler()` 事件处理器，当指定目录发生任何变化都会打印消息到终端。

### 事件类(event)： ###

watchdog.events.FileSystemEvent(event_type, 
                                    src_path, 
                                    is_directory=False)

事件类基类，所有具体事件类的父类。当一个目录或文件变化时，就会产生一个特定事件，也就是该类的子类。

属性和方法：  
event.is\_directory  
该事件是否由一个目录触发

event.src\_path  
触发该事件的文件或目录路径

event.event\_type  
事件类型，为`moved`、`deleted`、`created`或`modified`的其中之一

event.key  
返回元组`(event_type, src_path, is_directory)`

#### watchdog.events.FileSystemEvent类的子类： ####

watchdog.events.FileDeletedEvent()   
文件被删除时触发该事件

watchdog.events.DirDeletedEvent()   
目录被删除时触发该事件

watchdog.events.FileCreatedEvent()   
文件被创建时触发该事件

watchdog.events.DirCreatedEvent()   
目录被创建时触发该事件

watchdog.events.FileModifiedEvent()   
文件被修改时触发该事件(修改文件内容、修改文件inode信息如权限和访问时间，都会触发该事件)

watchdog.events.DirModifiedEvent()   
目录被修改时触发该事件

watchdog.events.FileMovedEvent()   
文件被移动或重命名时触发该事件，因为涉及文件移动，所以除了`event.src_path`表示原路径，还有`event.dest_path`表示目的路径

watchdog.events.DirMovedEvent()   
目录被移动或重命名时触发该事件，因为涉及文件移动，所以除了`event.src_path`表示原路径，还有`event.dest_path`表示目的路径

### 事件处理类(event handler)： ###

watchdog.events.FileSystemEventHandler()

事件处理器的基类，用于处理事件，用户需继承该类，并在子类中重写对应方法。

类实例方法如下：  
self.dispatch(event)  
接收到一个事件后，通过该方法来决定该event由下面哪个方法处理

self.on\_any\_event(event)  
任何事件发生都会首先执行该方法，该方法默认为空，dispatch()方法会先执行该方法，然后再把event分派给其他方法处理

self.on\_moved(event)  
Called when a file or a directory is moved or renamed，也就是处理`DirMovedEvent`和`FileMovedEvent`事件，子类需重写该方法

self.on\_created(event)  
Called when a file or directory is created，也就是处理`DirCreatedEvent`和`FileCreatedEvent`事件，子类需重写该方法

self.on\_deleted(event)  
Called when a file or directory is deleted，也就是处理`DirDeletedEvent`和`FileDeletedEvent`事件，子类需重写该方法

self.on\_modified(event)  
Called when a file or directory is modified，也就是处理`DirModifiedEvent`和`FileModifiedEvent`事件，子类需重写该方法

#### watchdog默认提供的一些事件处理类 ####

watchdog.events.PatternMatchingEventHandler(patterns=None, 
                                                ignore_patterns=None, 
                                                ignore_directories=False, 
                                                case_sensitive=False)

该类会检查触发事件的`src_path`和`dest_path`（如果有的话），是否与patterns指定的模式匹配；ignore\_patterns是需要排除不处理的模式，如果路径匹配该模式则不处理；还有ignore\_directories为True则表示不处理由目录引起的事件；case\_sensitive为True则表示路径不区分大小写。如果需要按模式匹配处理事件，则可以继承该类，不过需要自己实现`on_moved()`、`on_created()`、`on_deleted()`、`on_modified()`这四个方法。

watchdog.events.RegexMatchingEventHandler(regexes=[r".*"], 
                                              ignore_regexes=[], 
                                              ignore_directories=False, 
                                              case_sensitive=False)

基本等同于PatternMatchingEventHandler()类，除了是使用正则，而不是模式匹配。

watchdog.events.LoggingEventHandler()

使用logging模块记录所有事件信息，见文章开头的列举的官网例子。

例子：自定义事件处理类

import time
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler
    
    class MyHandler(FileSystemEventHandler):
        def on_modified(self, event):
            if event.src_path == "/home/sapser/scripts/test.log":      #监控指定文件内容、权限等变化
                print "log file %s changed!" % event.src_path
    
    if __name__ == "__main__":
        event_handler = MyHandler()
        observer = Observer()
        observer.schedule(event_handler, path='.', recursive=False)
        observer.start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()<span style="font-weight: 700;">
    </span>

#### observer: ####

watchdog.observers.Observer(timeout=1)

该类实现了监控文件变化，触发对应的事件类，然后调用关联的事件处理类来处理事件。该类其实是`threading.Thread`的子类，通过`observer.start()`使之运行在一个线程中，不会阻塞主进程运行，然后可以调用`observer.stop()`来停止该线程

实例属性及方法：  
observer.schedule(event\_handler, path, recursive=False)  
监控指定路径`path`，该路径触发任何事件都会调用`event_handler`来处理，如果`path`是目录，则`recursive=True`则会递归监控该目录的所有变化。每一次调用schedule()对一个路径进行监控处理就叫做一个watch，schedule()方法会返回这个watch，接着可以对这个watch做其他操作，如为该watch增加多个event处理器等  
注：内部由一个字典handlers来保存所有watch，watch的值是一个集合，包含对应此watch的所有event handler：

handlers = { 
        watch1: set(event_handler1, event_handler2),
        watch2: set(event_handler),
    }

observer.add\_handler\_for\_watch(event\_handler, watch)  
添加一个新的事件处理器到watch中，watch是`ObservedWatch()`类或其子类的实例

observer.remove\_handler\_for\_watch(event\_handler, watch)  
从watch中移除一个事件处理器

observer.unschedule(watch)  
移除一个watch及这个watch上的所有事件处理器

observer.unschedule\_all()  
移除所有watch及关联的事件处理器

observer.on\_thread\_stop()  
等同于`observer.unschedule_all()`

observer.stop()  
调用该方法来停止observer线程

例子：为一个路径添加多个事件处理器

import time
    import logging
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler, LoggingEventHandler
    from watchdog.observers.api import ObservedWatch
    
    class MyHandler(FileSystemEventHandler):
        def on_modified(self, event):
            if event.src_path == "/home/sapser/scripts/test.log":
                print "log file %s changed!" % event.src_path
    
    if __name__ == "__main__":
        event_handler1 = MyHandler()
        observer = Observer()
        watch = observer.schedule(event_handler1, path='.', recursive=True)
    
        logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s - %(message)s',
                            datefmt='%Y-%m-%d %H:%M:%S')
        event_handler2 = LoggingEventHandler()  
        observer.add_handler_for_watch(event_handler2, watch)      #为watch新添加一个event handler
        observer.start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

详见: https://github.com/gorakhargosh/watchdog

**     **