监听文件事件

最后更新时间: 2018-02-21 | 作者: AberSheeran | 捐助

最近在完善Maltose的时候,遇到了一个坑,因为我一开始是每一分钟扫描一次源目录,所以更新很不及时。查了查资料重写,写下此篇以作记录。

Watchdog

由于没找到能直接用的标准库,所以选择用了这个简单易用的第三方库。

调用它很简单,官方给的例子如下

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
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()

忽略掉其他细枝末节,关键在observer.schedule(event_handler, path, recursive=True)这一行,它的含义是

在path这个路径下触发更改事件时,将事件传递给event_handler进行处理。

那么我们来自定义个event_handler。首先这个类需要从FileSystemEventHandler这个事件处理基类继承,并重写事件处理方法。

事件处理方法如下:

建立如下的子类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from watchdog.events import FileSystemEventHandler

class MonitorFileEventHandler(FileSystemEventHandler):

    def on_modified(self, event):
        file = os.path.basename(event.src_path)
        if file.split(".")[-1] == "html":
            logging.info("Update article {}".format(file.replace(".html", "")))
            update_blog(file.replace(".html", ""))

    def on_deleted(self, event):
        file = os.path.basename(event.src_path)
        if file.split(".")[-1] == "html":
            logging.info("Delete article {}".format(file.replace(".html", "")))
            delete_blog(file.replace(".html", ""))

这个类只对更改和删除做出反应。你当然可以自己复写如上方法来做自己想做的事情。

event有如下的属性:

你可以在复写的方法里使用它们来做到一些有趣的事情。

然后我又编写了一个类来调用这个事件类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from watchdog.observers import Observer

class MonitorFile:
    def __init__(self, path):
        self.observer=Observer()
        self.observer.schedule(MonitorFileEventHandler(), path, recursive=True)
        self.observer.start()

    def __del__(self):
        self.observer.stop()
        self.observer.join()

接下来只需要实例化一下这个类,然后保持自身不被kill就OK了。

类似于这样

1
2
3
4
5
6
main = MonitorFile("D:/")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    pass

也欢迎去我的轮子maltose里看Watchdog的使用