如何实现与 tail -F 等效的 Python 代码？
- 2025-03-12 08:54:00
- admin 原创
- 32
问题描述：
观察一个不断增长的文件尾部是否存在某些关键字的 Python 方式是什么？
在 shell 中我可能会说：
tail -f "$file" | grep "$string" | while read hit; do
#stuff
done
解决方案 1：
嗯，最简单的方法就是不断地从文件中读取内容，检查有什么新内容，并测试是否命中。
import time
def watch(fn, words):
fp = open(fn, 'r')
while True:
new = fp.readline()
# Once all lines are read this just returns ''
# until the file changes and a new line appears
if new:
for word in words:
if word in new:
yield (word, new)
else:
time.sleep(0.5)
fn = 'test.py'
words = ['word']
for hit_word, hit_sentence in watch(fn, words):
print "Found %r in line: %r" % (hit_word, hit_sentence)
readline如果您知道数据将以行形式出现，那么此解决方案有效。
如果数据是某种流，则需要一个缓冲区，该缓冲区大于word您要查找的最大值，然后先填充它。这样会变得有点复杂……
解决方案 2：
def tail(f):
f.seek(0, 2)
while True:
line = f.readline()
if not line:
time.sleep(0.1)
continue
yield line
def process_matches(matchtext):
while True:
line = (yield)
if matchtext in line:
do_something_useful() # email alert, etc.
list_of_matches = ['ERROR', 'CRITICAL']
matches = [process_matches(string_match) for string_match in list_of_matches]
for m in matches: # prime matches
m.next()
while True:
auditlog = tail( open(log_file_to_monitor) )
for line in auditlog:
for m in matches:
m.send(line)
我用它来监控日志文件。在完整实现中，我将 list_of_matches 保存在配置文件中，以便可以将其用于多种用途。我的增强功能列表包括对正则表达式的支持，而不是简单的“in”匹配。
解决方案 3：
编辑：正如下面的评论所指出的，O_NONBLOCK不适用于磁盘上的文件。如果其他人来寻找来自套接字或命名管道或其他进程的尾部数据，这仍然会有所帮助，但它并没有回答实际提出的问题。原始答案仍保留在下面以供后人参考。（调用 tail 和 grep 会起作用，但无论如何都不是答案。）
要么打开文件O_NONBLOCK，要么使用它select来轮询读取可用性，然后read读取新数据和字符串方法来过滤文件末尾的行...或者只是使用subprocess模块并让tail并grep为您完成工作，就像在 shell 中一样。
解决方案 4：
您可以使用 select 来轮询文件中的新内容。
def tail(filename, bufsize = 1024):
fds = [ os.open(filename, os.O_RDONLY) ]
while True:
reads, _, _ = select.select(fds, [], [])
if 0 < len(reads):
yield os.read(reads[0], bufsize)
解决方案 5：
看起来有一个这样的包：https://github.com/kasun/python-tail
解决方案 6：
如果您无法将问题限制为基于行的读取，则需要借助于块。
这应该有效：
import sys
needle = "needle"
blocks = []
inf = sys.stdin
if len(sys.argv) == 2:
inf = open(sys.argv[1])
while True:
block = inf.read()
blocks.append(block)
if len(blocks) >= 2:
data = "".join((blocks[-2], blocks[-1]))
else:
data = blocks[-1]
# attention, this needs to be changed if you are interested
# in *all* matches separately, not if there was any match ata all
if needle in data:
print "found"
blocks = []
blocks[:-2] = []
if block == "":
break
挑战在于确保即使针被两个块边界分隔开，也能匹配针。
解决方案 7：
以下代码：
适用于Linux
用途
inotify不需要第三方库
使用
ctypes库
import os
import ctypes
import ctypes.util
# Constants for inotify
IN_MODIFY = 0x00000002
# Load libc
libc = ctypes.CDLL(ctypes.util.find_library('c'))
# Define inotify functions from libc
inotify_init = libc.inotify_init
inotify_add_watch = libc.inotify_add_watch
inotify_read = libc.read
# Define the tail_file function
def tail_file(filepath):
# Initialize inotify
inotify_fd = inotify_init()
# Add a watch for file modification events
wd = inotify_add_watch(inotify_fd, filepath.encode(), IN_MODIFY)
# Open the file and move to the end
with open(filepath, 'r') as file:
file.seek(0, os.SEEK_END)
while True:
# Create a buffer to store inotify events
buffer = os.read(inotify_fd, 1024) # This will block until the file is modified
# Read and print any new lines from the file
line = file.readline()
if line:
print(line, end='')
tail_file('your_file.txt')
解决方案 8：
据我所知，Python 函数列表中没有与“tail”等效的函数。解决方案是使用 tell()（获取文件大小）和 read() 来计算结束行。
这篇博客文章（不是我写的）写出了这个功能，我觉得很合适！
http://www.manugarg.com/2007/04/real-tailing-in-python.html
解决方案 9：
您可以使用pytailf：简单的 python tail -f 包装器
from tailf import tailf
for line in tailf("myfile.log"):
print line
解决方案 10：
如果您只需要一个简单的 Python 3 解决方案来处理文本文件的行，并且不需要 Windows 支持，那么这对我来说很有效：
import subprocess
def tailf(filename):
#returns lines from a file, starting from the beginning
p = subprocess.Popen(['tail', '-n', '+1', '-F', filename], stdout=subprocess.PIPE, universal_newlines=True)
for line in p.stdout:
yield line
for line in tailf("logfile"):
#do stuff
它会阻止等待新行被写入，因此如果不进行一些修改，它不适合异步使用。
解决方案 11：
您可以使用collections.deque来实现尾部。
来自http://docs.python.org/library/collections.html#deque-recipes ...
def tail(filename, n=10):
'Return the last n lines of a file'
return deque(open(filename), n)
当然，这会读取整个文件的内容，但这是实现 tail 的一种简洁的方法。
