我有一个日志文件正在写的另一个进程，我想观察变化。每次发生更改时，我都希望将新数据读入并对其进行一些处理。
最好的方法是什么?我希望在PyWin32库中有某种钩子。我找到了win32文件。函数FindNextChangeNotification，但不知道如何要求它监视特定的文件。
如果有人做过类似的事情，我真的很感激能听到…
[编辑]我应该提到我追求的是一种不需要轮询的解决方案。
[编辑]诅咒!这似乎不能在映射的网络驱动器上工作。我猜windows不会像在本地磁盘上那样“听到”任何对文件的更新。
当前回答
最好和最简单的解决方案是使用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)
其他回答
正如您在Tim Golden的文章(由Horst Gutmann指出)中看到的，WIN32相对复杂，它监视目录，而不是单个文件。
我建议你研究一下IronPython，它是一个。net python实现。 在IronPython中，你可以使用所有的。net功能——包括
System.IO.FileSystemWatcher
它用一个简单的Event接口处理单个文件。
因为我已经全局安装了它，所以我最喜欢的方法是使用nodemon。如果你的源代码在src中，你的入口点是src/app.py，那么就像这样简单:
nodemon -w 'src/**' -e py,html --exec python src/app.py
．.． 其中-e py,html允许您控制要监视更改的文件类型。
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
...
最好和最简单的解决方案是使用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)
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()
