问题：如何查看文件中的更改？
我有一个日志文件正在由另一个进程写入，我想监视它的更改。每次发生更改时，我都希望读入新数据以对其进行一些处理。
最好的方法是什么？我希望从PyWin32库中获得某种吸引。我找到了该win32file.FindNextChangeNotification功能，但不知道如何要求它观看特定文件。
如果有人做了这样的事情，我将不胜感激。
[编辑]我应该提到我在寻求不需要轮询的解决方案。
[编辑]诅咒！看来这在映射的网络驱动器上不起作用。我猜想Windows不会像在本地磁盘上那样“听到”文件的任何更新。
回答 0
您是否已经看过http://timgolden.me.uk/python/win32_how_do_i/watch_directory_for_changes.html上的可用文档？如果只需要它在Windows下运行，则第二个示例似乎正是您想要的（如果您将目录的路径与要观看的文件之一交换了）。
否则，轮询将可能是唯一真正与平台无关的选项。
注意：我还没有尝试过任何这些解决方案。
回答 1
您尝试使用看门狗了吗？
Python API库和Shell实用程序可监视文件系统事件。
目录监视变得容易
- 跨平台API。
- 一种外壳程序工具，用于响应目录更改而运行命令。
回答 2
如果轮询对您来说足够好，那么我只看“修改时间”文件的统计信息是否发生了变化。阅读：
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...回答 3
如果您需要多平台解决方案，请检查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)回答 4
它不能在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)回答 5
查看pyinotify。
inotify在较新的linux中替换了dnotify（来自较早的答案），并允许文件级而不是目录级的监视。
回答 6
在对Tim Golden的脚本进行了一些修改之后，我发现以下内容似乎运行良好：
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 )它可能可以处理更多的错误检查，但是对于简单地查看日志文件并在将其吐到屏幕上之前对其进行一些处理，这很好。
感谢大家的投入-很棒的东西！
回答 7
为了观看具有轮询和最小依赖性的单个文件，下面是一个基于Deestan（上图）的答案的充实示例：
import os
import sys
import time
class Watcher(object):
running = True
refresh_delay_secs = 1
# Constructor
def __init__(self, watch_file, call_func_on_change=None, *args, **kwargs):
self._cached_stamp = 0
self.filename = watch_file
self.call_func_on_change = call_func_on_change
self.args = args
self.kwargs = kwargs
# Look for changes
def look(self):
stamp = os.stat(self.filename).st_mtime
if stamp != self._cached_stamp:
self._cached_stamp = stamp
# File has changed, so do something...
print('File changed')
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:
print('Unhandled error: %s' % sys.exc_info()[0])
# Call this function each time a change happens
def custom_action(text):
print(text)
watch_file = 'my_file.txt'
# watcher = Watcher(watch_file) # simple
watcher = Watcher(watch_file, custom_action, text='yes, changed') # also call custom action function
watcher.watch() # start the watch going回答 8
检查我对类似问题的回答。您可以在Python中尝试相同的循环。该页面建议：
import time
while 1:
where = file.tell()
line = file.readline()
if not line:
time.sleep(1)
file.seek(where)
else:
print line, # already has newline另请参见问题tail（）使用Python的文件。
回答 9
对我来说，最简单的解决方案是使用看门狗工具watchmedo
现在从https://pypi.python.org/pypi/watchdog获得一个进程，该进程可以查找目录中的sql文件，并在必要时执行它们。
watchmedo shell-command \
--patterns="*.sql" \
--recursive \
--command='~/Desktop/load_files_into_mysql_database.sh' \
.回答 10
好吧，由于您使用的是Python，因此您可以打开一个文件并继续读取其中的行。
f = open('file.log')如果读取的行不为空，则对其进行处理。
line = f.readline()
if line:
// Do what you want with the line您可能会错过继续拨打readlineEOF的权限。在这种情况下，它将仅返回一个空字符串。并且，将某些内容添加到日志文件后，将根据需要从停止的位置继续读取。
如果您正在寻找使用事件或特定库的解决方案，请在问题中进行指定。否则，我认为这种解决方案就可以了。
回答 11
这是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回答 12
这是Tim Goldan脚本的另一种修改，该脚本可在unix类型上运行，并通过使用dict（file => time）添加了一个简单的文件修改监视程序。
用法：whateverName.py path_to_dir_to_watch
#!/usr/bin/env python
import os, sys, time
def files_to_timestamp(path):
files = [os.path.join(path, f) for f in os.listdir(path)]
return dict ([(f, os.path.getmtime(f)) for f in files])
if __name__ == "__main__":
path_to_watch = sys.argv[1]
print('Watching {}..'.format(path_to_watch))
before = files_to_timestamp(path_to_watch)
while 1:
time.sleep (2)
after = files_to_timestamp(path_to_watch)
added = [f for f in after.keys() if not f in before.keys()]
removed = [f for f in before.keys() if not f in after.keys()]
modified = []
for f in before.keys():
if not f in removed:
if os.path.getmtime(f) != before.get(f):
modified.append(f)
if added: print('Added: {}'.format(', '.join(added)))
if removed: print('Removed: {}'.format(', '.join(removed)))
if modified: print('Modified: {}'.format(', '.join(modified)))
before = after回答 13
正如您在由Horst Gutmann指出的Tim Golden的文章中所看到的，WIN32比较复杂，并且监视目录，而不是单个文件。
我想建议您研究IronPython，它是一个.NET python实现。借助IronPython，您可以使用所有.NET功能-包括
System.IO.FileSystemWatcher使用简单的事件接口处理单个文件。
回答 14
这是检查文件更改的示例。这样做可能不是最好的方法，但是肯定是很短的方法。
对源进行更改后重新启动应用程序的便捷工具。我在使用pygame玩游戏时做到了这一点，因此我可以看到文件保存后立即发生了效果。
在pygame中使用时，请确保“ while”循环中的内容已放置在您的游戏循环中，也称为update或其他内容。否则，您的应用程序将陷入无限循环，并且您将看不到游戏更新。
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)让电子做您想做的事情变得有趣。
回答 15
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed from something",
5 : "Renamed to something"
}
FILE_LIST_DIRECTORY = 0x0001
class myThread (threading.Thread):
def __init__(self, threadID, fileName, directory, origin):
threading.Thread.__init__(self)
self.threadID = threadID
self.fileName = fileName
self.daemon = True
self.dir = directory
self.originalFile = origin
def run(self):
startMonitor(self.fileName, self.dir, self.originalFile)
def startMonitor(fileMonitoring,dirPath,originalFile):
hDir = win32file.CreateFile (
dirPath,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
# 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_M in results:
full_filename = os.path.join (dirPath, file_M)
#print file, ACTIONS.get (action, "Unknown")
if len(full_filename) == len(fileMonitoring) and action == 3:
#copy to main file
...回答 16
这是一个示例，用于观察每秒写入不超过一行但通常少得多的输入文件。目标是将最后一行（最新写入）追加到指定的输出文件。我已从我的一个项目中复制了此内容，并删除了所有不相关的行。您必须填写或更改缺少的符号。
from PyQt5.QtCore import QFileSystemWatcher, QSettings, QThread
from ui_main_window import Ui_MainWindow # Qt Creator gen'd
class MainWindow(QMainWindow, Ui_MainWindow):
def __init__(self, parent=None):
QMainWindow.__init__(self, parent)
Ui_MainWindow.__init__(self)
self._fileWatcher = QFileSystemWatcher()
self._fileWatcher.fileChanged.connect(self.fileChanged)
def fileChanged(self, filepath):
QThread.msleep(300) # Reqd on some machines, give chance for write to complete
# ^^ About to test this, may need more sophisticated solution
with open(filepath) as file:
lastLine = list(file)[-1]
destPath = self._filemap[filepath]['dest file']
with open(destPath, 'a') as out_file: # a= append
out_file.writelines([lastLine])当然，并不是严格要求包含QMainWindow类。您可以单独使用QFileSystemWatcher。
回答 17
您还可以使用一个名为repyt的简单库，这是一个示例：
repyt ./app.py回答 18
似乎没有人发布fswatch。它是一个跨平台的文件系统监视程序。只需安装，运行并按照提示进行操作即可。
我已经将它与python和golang程序一起使用，并且可以正常工作。
回答 19
相关的@ 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回答 20
最好和最简单的解决方案是使用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)回答 21
我不知道Windows的任何特定功能。您可以尝试每秒钟/分钟/小时获取文件的MD5哈希值（取决于您需要的速度），并将其与最后一个哈希值进行比较。如果不同，您就知道文件已更改，并读出了最新的行。
回答 22
我会尝试这样的事情。
try:
f = open(filePath)
except IOError:
print "No such file: %s" % filePath
raw_input("Press Enter to close window")
try:
lines = f.readlines()
while True:
line = f.readline()
try:
if not line:
time.sleep(1)
else:
functionThatAnalisesTheLine(line)
except Exception, e:
# handle the exception somehow (for example, log the trace) and raise the same exception again
raw_input("Press Enter to close window")
raise e
finally:
f.close()循环检查自上次读取文件以来是否存在新行-如果存在，则将其读取并传递给functionThatAnalisesTheLine函数。如果不是，脚本将等待1秒钟，然后重试该过程。
