python watchdog与增量文件监听

Watch Dog

在Python中，我们可以使用watchdog来实现文件监控。下面将介绍watch dog的使用。首先可以使用pip进行安装。

pip install watchdog

watch dog使用观察者模型，主要参与角色有observer，event_handler以及被监听的路径。当被监听的路径出现相关变化之后，就会触发event_handler中定义的方法。示例代码如下：

import datetime
import time

from watchdog.observers import Observer
from watchdog.events import *


class FileEventHandler(FileSystemEventHandler):
    def __init__(self):
        FileSystemEventHandler.__init__(self)

    def on_moved(self, event):
        if event.is_directory:
            print("directory moved from {0} to {1}".format(event.src_path, event.dest_path))
        else:
            print("file moved from {0} to {1}".format(event.src_path, event.dest_path))

    def on_created(self, event):
        if event.is_directory:
            print("directory created:{0}".format(event.src_path))
        else:
            print("file created:{0}".format(event.src_path))

    def on_deleted(self, event):
        if event.is_directory:
            print("directory deleted:{0}".format(event.src_path))
        else:
            print("file deleted:{0}".format(event.src_path))

    def on_modified(self, event):
        if event.is_directory:
            print("directory modified:{0}".format(event.src_path))
        else:
            print("file modified:{0}".format(event.src_path))


if __name__ == '__main__':
    observer = Observer()
    event_handler = FileEventHandler()
    observer.schedule(event_handler, "./", False)
    observer.start()

    try:
        while True:
            print(datetime.datetime.now())
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

在watchdog中，observer实际上是一个守护线程，它并不会阻塞主线程的运行，所以我们这里在主线程中使用循环等待来查看相关效果。使用watchdog，我们主要完成的是继承FileSystemEventHandler类，然后覆盖相关方法，之后利用observer提供的schedule方法进行关联。需要注意的是，这里schedule有三个参数。第一个参数表示对应的事件Handler类；第二个参数表示需要监听的目录（而不是文件）；第三个参数表示是否递归监听。如果仅需要监听某一个文件，则需要在相关触发方法中进行判断。

增量文件监听

利用watchdog的文件监听功能，我们可以实现一个增量文件的监听功能。在业务场景中，日志文件是非常常见的。日志文件实际上就是一个增量文件，我们可以监听日志文件内容的增加，然后获取到增加的内容，下面我们就实现这个功能。

首先直接展示完整代码如下：

import time
import os
import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class FileUtil:
    @staticmethod
    def increment_read(begin_index: int, file: str):
        fd = open(file, 'r', encoding='utf-8')
        fd.seek(begin_index, 0)
        lines = fd.readlines()
        now_index = fd.tell()
        fd.close()
        return now_index, lines

    @staticmethod
    def raw_read(file: str):
        fd = open(file, 'r', encoding='utf-8')
        lines = fd.readlines()
        fd.close()
        return lines


class FileEventHandler(FileSystemEventHandler):
    def __init__(self, file: str):
        self.file = file
        self.now_index = os.path.getsize(file)
        FileSystemEventHandler.__init__(self)

    def on_modified(self, event):
        if not event.is_directory and os.path.basename(self.file) == os.path.basename(event.src_path):
            new_index, new_lines = FileUtil.increment_read(self.now_index, self.file)
            self.now_index = new_index
            print("[{0}] new lines: {1}".format(datetime.datetime.now(), new_lines))


class IncrementSpy:
    def __init__(self, directory: str, filename: str):
        self.directory = directory
        self.filename = filename
        self.file = os.path.join(directory, filename)
        self.observer = None
        print("[{0}] origin file content: {1}".format(datetime.datetime.now(), FileUtil.raw_read(self.file)))

    def execute(self):
        self.observer = Observer()
        event_handler = FileEventHandler(self.file)
        self.observer.schedule(event_handler, self.directory, False)
        self.observer.start()

    def close(self):
        if self.observer is not None:
            self.observer.stop()


if __name__ == '__main__':
    spy = IncrementSpy('./', 'test.txt')
    spy.execute()

    try:
        while True:
            print(datetime.datetime.now())
            time.sleep(1)
    except KeyboardInterrupt:
        spy.close()

我们可以准备一个test.txt文件，其中可以填一些初始内容。之后启动程序，同时使用编辑器对test.txt文件进行编辑，当我们完成编辑保存之后，在运行界面中就会出现相关的提示。同时我们在主线程中循环打印时间，目的是为了表现监听线程并不会阻塞主线程的执行。运行效果如下：

注意只能监听增量文件，并不能监听文件内容的修改（指delete/update）

接下来就开始介绍相关功能的实现。

首先我们实现了一个工具类FileUtil，其中提供两个静态方法，分别是用于读取文件的全部内容raw_read以及从指定index开始读取文件内容的increment_read，第二个增量读取的方法在读取内容的同时还会返回当前文件的index，方便下一次的增量读取。

之后我们实现watchdog中相关的处理类，即FileEventHandler。它继承FileSystemEventHandler，但是我们只实现了其中的on_modified方法，用于监听文件修改事件。由于我们这里只需要监听对应的文件，所以我们需要在这个方法中进行判断，只有event对应的src_path符合我们的要求时才进行动作。触发的动作就是进行增量读取，同时保存当前的index。

最后，我们将整个监听操作进行包装，实现IncrementSpy类。在构造方法中，它会首先读取文件的初始内容并进行输出，之后执行execute方法进行监听。在execute方法中，我们完成的主要是Observer对象的创建以及schedule的指定。对于使用者来说，只需要在初始化IncrementSpy类之后调用execute方法即可，就如同在main中的使用方法一样。

参考文章

1. python watchdog 详细讲解-云社区-华为云 (huaweicloud.com)
2. python中文件变化监控-watchd - 腾讯云开发者社区-腾讯云 (tencent.com)
3. python 实现增量的读取文件 - 简书 (jianshu.com)