python监听并获取文件内容

Watchdog介绍

Watchdog是一个Python API和shell工具，用于监控文件系统事件。它能够监控文件和目录的创建、修改、移动和删除等事件，并触发相应的回调函数。这对于需要实时响应文件系统变化的应用非常有用，比如自动编译系统、文件同步服务、日志监控等。

应用场景

• 日志监控：自动处理或分析新生成的日志条目，如错误日志监控和报警。
• 自动测试：在开发环境中，自动运行测试套件，当源代码文件发生改变时。
• 文件同步：在多个目录或设备间自动同步文件更改。
• 备份系统：监控特定文件或目录，一旦有变动立即进行备份。
• Web开发：自动重启服务器或重新加载配置，当项目文件发生变化时。

弊端

• 资源消耗：长时间运行的监视进程可能会占用一定的系统资源，特别是当监控大量文件或目录时。
• 事件风暴：在短时间内产生大量文件系统事件（如大量日志写入）时，可能会引发大量的回调执行，影响性能甚至导致程序响应迟缓。
• 平台兼容性：虽然Watchdog努力提供跨平台支持，但在不同操作系统上的表现可能有差异，特别是在一些低级文件系统特性上。
• 复杂的文件操作可能被遗漏：快速连续的文件操作（如快速创建和删除文件）可能不会被准确捕获，因为操作系统或文件系统的事件通知机制可能有延迟。
• 依赖轮询：在某些平台上，Watchdog可能依赖于轮询机制来检测文件系统变化，这会增加CPU使用率并可能导致事件检测的延迟。

使用Watchdog

安装：

pip install watchdog

使用watchdog监控日志文件变化，包括创建及内容监控，当文件发生变化时触发on_modified。会根据文件的上次指针位置读取内容并输出。

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


class LogMonitor(FileSystemEventHandler):
    def __init__(self, log_directory):
        super().__init__()
        self.log_directory = log_directory
        self.last_position = {}
        self.initialize_file_positions()

    def initialize_file_positions(self):
        """初始化时，将所有.log文件的末尾指针位置存入字典"""
        for filename in os.listdir(self.log_directory):
            if filename.endswith('.log'):
                filepath = os.path.join(self.log_directory, filename)
                if os.path.isfile(filepath):
                    self.last_position[filepath] = os.path.getsize(filepath)

    def on_modified(self, event):
        """当文件被修改时触发"""
        # print("文件被修改：", event.src_path)
        if not event.is_directory and event.src_path.endswith('.log'):
            with open(event.src_path, 'rb') as file:
                # 如果是第一次读取该文件，或者文件被截断（大小变小），重置读取位置
                if event.src_path not in self.last_position or os.path.getsize(event.src_path) < self.last_position[
                    event.src_path]:
                    file.seek(0, 2)
                    self.last_position[event.src_path] = file.tell()
                    print(file.tell())
                else:
                    file.seek(self.last_position[event.src_path], 0)

                while True:
                    print("文件被修改：", )
                    new_content = file.readline()
                    if not new_content:  # 如果没有更多内容可读，则跳出循环
                        break

                    print(new_content.decode('utf-8'), end="")
                self.last_position[event.src_path] = file.tell()


def monitor_log_directory(path='.'):
    """监控指定目录下的.log文件"""
    event_handler = LogMonitor(path)
    observer = Observer()
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(0.5)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == "__main__":
    log_directory = os.path.dirname(__file__)  # 替换目录
    monitor_log_directory(log_directory)

监听目录中的文件：

在文件中输入内容并保存：

输出：

a
b
c
e