1. 程式人生 > >python watchdog:監控檔案系統事件的Python庫

python watchdog:監控檔案系統事件的Python庫

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,也就是處理DirCreatedEventFileCreatedEvent事件,子類需重寫該方法

self.on_deleted(event)
Called when a file or directory is deleted,也就是處理DirDeletedEventFileDeletedEvent事件,子類需重寫該方法

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_pathdest_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()