我有一个日志文件正在写的另一个进程，我想观察变化。每次发生更改时，我都希望将新数据读入并对其进行一些处理。
最好的方法是什么?我希望在PyWin32库中有某种钩子。我找到了win32文件。函数FindNextChangeNotification，但不知道如何要求它监视特定的文件。
如果有人做过类似的事情，我真的很感激能听到…
[编辑]我应该提到我追求的是一种不需要轮询的解决方案。
[编辑]诅咒!这似乎不能在映射的网络驱动器上工作。我猜windows不会像在本地磁盘上那样“听到”任何对文件的更新。
当前回答
正如您在Tim Golden的文章(由Horst Gutmann指出)中看到的，WIN32相对复杂，它监视目录，而不是单个文件。
我建议你研究一下IronPython，它是一个。net python实现。 在IronPython中，你可以使用所有的。net功能——包括
System.IO.FileSystemWatcher
它用一个简单的Event接口处理单个文件。
其他回答
对我来说，最简单的解决方法就是使用看门狗的工具watchmedo
从https://pypi.python.org/pypi/watchdog我现在有一个进程，在一个目录中查找sql文件，并在必要时执行它们。
watchmedo shell-command \
--patterns="*.sql" \
--recursive \
--command='~/Desktop/load_files_into_mysql_database.sh' \
.
如果您想要一个多平台解决方案，那么请检查QFileSystemWatcher。 下面是一个示例代码(未消毒):
from PyQt4 import QtCore
@QtCore.pyqtSlot(str)
def directory_changed(path):
print('Directory Changed!!!')
@QtCore.pyqtSlot(str)
def file_changed(path):
print('File Changed!!!')
fs_watcher = QtCore.QFileSystemWatcher(['/path/to/files_1', '/path/to/files_2', '/path/to/files_3'])
fs_watcher.connect(fs_watcher, QtCore.SIGNAL('directoryChanged(QString)'), directory_changed)
fs_watcher.connect(fs_watcher, QtCore.SIGNAL('fileChanged(QString)'), file_changed)
在对蒂姆·戈尔登的剧本进行了一些修改后，我有了以下似乎工作得相当不错的内容:
import os
import win32file
import win32con
path_to_watch = "." # look at the current directory
file_to_watch = "test.txt" # look for changes to a file called test.txt
def ProcessNewData( newData ):
print "Text added: %s"%newData
# Set up the bits we'll need for output
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed from something",
5 : "Renamed to something"
}
FILE_LIST_DIRECTORY = 0x0001
hDir = win32file.CreateFile (
path_to_watch,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
# Open the file we're interested in
a = open(file_to_watch, "r")
# Throw away any exising log data
a.read()
# Wait for new data and call ProcessNewData for each new chunk that's written
while 1:
# Wait for a change to occur
results = win32file.ReadDirectoryChangesW (
hDir,
1024,
False,
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE,
None,
None
)
# For each change, check to see if it's updating the file we're interested in
for action, file in results:
full_filename = os.path.join (path_to_watch, file)
#print file, ACTIONS.get (action, "Unknown")
if file == file_to_watch:
newText = a.read()
if newText != "":
ProcessNewData( newText )
它可能需要进行更多的错误检查，但对于简单地观察日志文件并在将其输出到屏幕之前对其进行一些处理，这样做效果很好。
谢谢大家的投入-很棒的东西!
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)上获得二进制文件。
如果轮询对您来说足够好，我只观察“修改的时间”文件统计是否发生变化。阅读方法:
os.stat(filename).st_mtime
(还要注意，Windows本机更改事件解决方案并不在所有情况下都有效，例如在网络驱动器上。)
import os
class Monkey(object):
def __init__(self):
self._cached_stamp = 0
self.filename = '/path/to/file'
def ook(self):
stamp = os.stat(self.filename).st_mtime
if stamp != self._cached_stamp:
self._cached_stamp = stamp
# File has changed, so do something...
