Python 实现 linux 命令 tail

在某些场景需要实时读取文件的新增内容，使用linux的命令行tail不是很容易操作，所有有了python版本的tail，方便直接加入到脚本。

一. 源码内容

• tail.py 源码

import os
import sys
import time

class Tail:
    def __init__(self, tailed_file, interval_seconds=1.0, start_position="end"):
        self.tailed_file = os.path.abspath(tailed_file)
        self.check_file_validity()
        self.callback = sys.stdout.write
        if start_position not in ("start", "end"):
            raise Exception("start_position must be: start or end")
        self.start_position = start_position
        self.interval_seconds = interval_seconds

    def follow(self):
        deleted_once = False
        while True:
            if not self.check_file_validity(raise_error=False):
                deleted_once = True
                time.sleep(self.interval_seconds)
                continue
            with open(self.tailed_file, encoding="utf-8") as f:
                if deleted_once:
                    f.seek(0, os.SEEK_SET)
                elif self.start_position == "start":
                    f.seek(0, os.SEEK_SET)
                elif self.start_position == "end":
                    f.seek(0, os.SEEK_END)
                while True:
                    if not self.check_file_validity(raise_error=False):
                        deleted_once = True
                        break
                    line = f.readline()
                    if not line:
                        time.sleep(self.interval_seconds)
                    else:
                        self.callback(line)

    def register_callback(self, func):
        self.callback = func

    def check_file_validity(self, raise_error=True):
        if not os.access(self.tailed_file, os.F_OK):
            err = "File '%s' does not exist" % self.tailed_file
            if raise_error:
                raise Exception(err)
            else:
                print(err)
                return False
        if not os.access(self.tailed_file, os.R_OK):
            err = "File '%s' not readable" % self.tailed_file
            if raise_error:
                raise Exception(err)
            else:
                print(err)
                return False
        if os.path.isdir(self.tailed_file):
            err = "File '%s' is a directory" % self.tailed_file
            if raise_error:
                raise Exception(err)
            else:
                print(err)
                return False
        return True

二. 使用样例

打印到标准输出

• example.py

import tail

t = tail.Tail("/tmp/able")
t.follow()

自定义处理函数

同时还可以指定从文件开头，还是文件末尾读取，以及读取的间隔时间。

• example.py

import tail

def process_line(line):
    line = line.strip("\n")
    print("processing...", line)

t = tail.Tail("/tmp/able", interval_seconds=2.0, start_position="start")
t.register_callback(process_line)
t.follow()

• python example.py 输出结果示例(Ctrl+C停止)

example line
example line
File '/tmp/able' does not exist
File '/tmp/able' does not exist
File '/tmp/able' does not exist
File '/tmp/able' does not exist
example line
example line
^CTraceback (most recent call last):
  File "programs/py38/example.py", line 5, in <module>
    t.follow()
  File "/Users/yucs/programs/py38/tail.py", line 36, in follow
    time.sleep(self.interval_seconds)
KeyboardInterrupt