我有一个日志文件正在写的另一个进程,我想观察变化。每次发生更改时,我都希望将新数据读入并对其进行一些处理。

最好的方法是什么?我希望在PyWin32库中有某种钩子。我找到了win32文件。函数FindNextChangeNotification,但不知道如何要求它监视特定的文件。

如果有人做过类似的事情,我真的很感激能听到…

[编辑]我应该提到我追求的是一种不需要轮询的解决方案。

[编辑]诅咒!这似乎不能在映射的网络驱动器上工作。我猜windows不会像在本地磁盘上那样“听到”任何对文件的更新。


当前回答

正如您在Tim Golden的文章(由Horst Gutmann指出)中看到的,WIN32相对复杂,它监视目录,而不是单个文件。

我建议你研究一下IronPython,它是一个。net python实现。 在IronPython中,你可以使用所有的。net功能——包括

System.IO.FileSystemWatcher

它用一个简单的Event接口处理单个文件。

其他回答

watchfiles (https://github.com/samuelcolvin/watchfiles)是一个Python API和CLI,它使用了用Rust编写的Notify (https://github.com/notify-rs/notify)库。

rust实现目前(2022-10-09)支持:

Linux / Android: inotify macOS: FSEvents或kqueue,参见特性 Windows: ReadDirectoryChangesW FreeBSD / NetBSD / OpenBSD / DragonflyBSD: kqueue 所有平台:投票

可以在PyPI (https://pypi.org/project/watchfiles/)和conda-forge (https://github.com/conda-forge/watchfiles-feedstock)上获得二进制文件。

似乎没有人张贴fswatch。它是一个跨平台的文件系统监视器。只要安装它,运行它,并按照提示。

我在python和golang程序中使用过它,它只是工作。

import inotify.adapters
from datetime import datetime


LOG_FILE='/var/log/mysql/server_audit.log'


def main():
    start_time = datetime.now()
    while True:
        i = inotify.adapters.Inotify()
        i.add_watch(LOG_FILE)
        for event in i.event_gen(yield_nones=False):
            break
        del i

        with open(LOG_FILE, 'r') as f:
            for line in f:
                entry = line.split(',')
                entry_time = datetime.strptime(entry[0],
                                               '%Y%m%d %H:%M:%S')
                if entry_time > start_time:
                    start_time = entry_time
                    print(entry)


if __name__ == '__main__':
    main()

下面是Kender代码的简化版本,它似乎做了同样的技巧,并且没有导入整个文件:

# Check file for new data.

import time

f = open(r'c:\temp\test.txt', 'r')

while True:

    line = f.readline()
    if not line:
        time.sleep(1)
        print 'Nothing New'
    else:
        print 'Call Function: ', line

这是一个检查文件更改的示例。这可能不是最好的方法,但肯定是一条捷径。

方便的工具,重新启动应用程序时,已作出更改的源。我在玩pygame的时候做了这个,这样我就可以看到文件保存后立即发生的效果。

当在pygame中使用时,确保'while'循环中的东西被放置在你的游戏循环中,也就是更新或其他什么。否则你的应用将陷入无限循环,你将看不到游戏的更新。

file_size_stored = os.stat('neuron.py').st_size

  while True:
    try:
      file_size_current = os.stat('neuron.py').st_size
      if file_size_stored != file_size_current:
        restart_program()
    except: 
      pass

如果你想要重启代码我在网上找到的。在这儿。(与问题无关,但可能会派上用场)

def restart_program(): #restart application
    python = sys.executable
    os.execl(python, python, * sys.argv)

让电子做你想让它们做的事情。