我有一个日志文件正在写的另一个进程，我想观察变化。每次发生更改时，我都希望将新数据读入并对其进行一些处理。
最好的方法是什么?我希望在PyWin32库中有某种钩子。我找到了win32文件。函数FindNextChangeNotification，但不知道如何要求它监视特定的文件。
如果有人做过类似的事情，我真的很感激能听到…
[编辑]我应该提到我追求的是一种不需要轮询的解决方案。
[编辑]诅咒!这似乎不能在映射的网络驱动器上工作。我猜windows不会像在本地磁盘上那样“听到”任何对文件的更新。
当前回答
我不知道任何Windows特有的函数。您可以尝试每秒钟/分钟/小时获取文件的MD5哈希值(取决于您需要它的速度)，并将其与最后的哈希值进行比较。当它不同时，您知道文件已被更改，并读取最新的行。
其他回答
它不应该在windows上工作(也许与cygwin ?)，但对于unix用户，您应该使用“fcntl”系统调用。下面是Python中的一个例子。如果你需要用C语言写的话，基本上是相同的代码(相同的函数名)
import time
import fcntl
import os
import signal
FNAME = "/HOME/TOTO/FILETOWATCH"
def handler(signum, frame):
print "File %s modified" % (FNAME,)
signal.signal(signal.SIGIO, handler)
fd = os.open(FNAME, os.O_RDONLY)
fcntl.fcntl(fd, fcntl.F_SETSIG, 0)
fcntl.fcntl(fd, fcntl.F_NOTIFY,
fcntl.DN_MODIFY | fcntl.DN_CREATE | fcntl.DN_MULTISHOT)
while True:
time.sleep(10000)
这是一个检查文件更改的示例。这可能不是最好的方法，但肯定是一条捷径。
方便的工具，重新启动应用程序时，已作出更改的源。我在玩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)
让电子做你想让它们做的事情。
相关@4Oh4解决方案一个流畅的更改文件列表观看;
import os
import sys
import time
class Watcher(object):
running = True
refresh_delay_secs = 1
# Constructor
def __init__(self, watch_files, call_func_on_change=None, *args, **kwargs):
self._cached_stamp = 0
self._cached_stamp_files = {}
self.filenames = watch_files
self.call_func_on_change = call_func_on_change
self.args = args
self.kwargs = kwargs
# Look for changes
def look(self):
for file in self.filenames:
stamp = os.stat(file).st_mtime
if not file in self._cached_stamp_files:
self._cached_stamp_files[file] = 0
if stamp != self._cached_stamp_files[file]:
self._cached_stamp_files[file] = stamp
# File has changed, so do something...
file_to_read = open(file, 'r')
value = file_to_read.read()
print("value from file", value)
file_to_read.seek(0)
if self.call_func_on_change is not None:
self.call_func_on_change(*self.args, **self.kwargs)
# Keep watching in a loop
def watch(self):
while self.running:
try:
# Look for changes
time.sleep(self.refresh_delay_secs)
self.look()
except KeyboardInterrupt:
print('\nDone')
break
except FileNotFoundError:
# Action on file not found
pass
except Exception as e:
print(e)
print('Unhandled error: %s' % sys.exc_info()[0])
# Call this function each time a change happens
def custom_action(text):
print(text)
# pass
watch_files = ['/Users/mexekanez/my_file.txt', '/Users/mexekanez/my_file1.txt']
# watcher = Watcher(watch_file) # simple
if __name__ == "__main__":
watcher = Watcher(watch_files, custom_action, text='yes, changed') # also call custom action function
watcher.watch() # start the watch going
你也可以使用一个简单的库，叫做repyt，下面是一个例子:
repyt ./app.py
最好和最简单的解决方案是使用pygtail: https://pypi.python.org/pypi/pygtail
from pygtail import Pygtail
import sys
while True:
for line in Pygtail("some.log"):
sys.stdout.write(line)
