Python实现linux的tail功能

基于python3，简单实现tail -n、tail -f功能

原理

1. 把文件大小分页进行读取，这样读取大日志就无需全部加载到内存中
2. 文件倒读输出

代码实现

#!/usr/bin/env python
# coding: utf-8
"""
原理：
    https://yq.aliyun.com/articles/60684
功能：
    实现tail -n
    实现tail -f
其它实现：
    http://www.cnblogs.com/bufferfly/p/4878688.html
    https://github.com/shengxinjing/my_blog/issues/11
BUG：
    重定向相同数据到日志文件里，使用>而不是>>的话，输入无法打印出来
"""

import os
import sys
import time

PAGE = 4096

class Tail:
    def __init__(self, filename, callback=sys.stdout.write):
        self.filename = filename
        self.callback = callback

    def reverse(self, n=10):
        """
        实现 tail -n
        """
        with open(self.filename, 'rb') as f:
            f_len = f.seek(0, 2)
            rem = f_len % PAGE
            page_n = f_len // PAGE
            r_len = rem if rem else PAGE
            while True:
                # 如果读取的页大小>=文件大小，直接读取数据输出
                if r_len >= f_len:
                    f.seek(0)
                    lines = f.readlines()[::-1]
                    break

                f.seek(-r_len, 2)
                # print('f_len: {}, rem: {}, page_n: {}, r_len: {}'.format(f_len, rem, page_n, r_len))
                lines = f.readlines()[::-1]
                count = len(lines) -1   # 末行可能不完整，减一行，加大读取量

                if count >= n:  # 如果读取到的行数>=指定行数，则退出循环读取数据
                    break
                else:   # 如果读取行数不够，载入更多的页大小读取数据
                    r_len += PAGE
                    page_n -= 1

        for line in lines[:n]:
            self.callback(line.decode('utf-8'))

    def follow(self):
        """
        实现 tail －f
        """
        with open(self.filename, 'rb') as fd:
            pos = fd.seek(0, 2)  # 打开文件时大小
            try:
                while True:
                    curr_pos = fd.seek(0,2)
                    # print('pos: {}, curr_pos: {}'.format(pos, curr_pos))
                    if pos > curr_pos:  # 表示文件数据减少或清空
                        pos = fd.seek(0, 2)
                        # time.sleep(0.3)
                        continue

                    line = fd.readline()
                    if line:
                        self.callback(line.decode('utf-8'))
                    # time.sleep(0.1)
            except KeyboardInterrupt as e:
                pass

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print('Usage: {} [ -f | -# ] file'.format(sys.argv[0]), file=sys.stderr)
        raise SystemExit(1)

    if not os.path.isfile(sys.argv[2]):
        print('File does not exist.')
        raise SystemExit(1)
    else:
        tail = Tail(sys.argv[2])

    if '-f' == sys.argv[1]:
        tail.reverse()
        tail.follow()
    elif '-' in sys.argv[1]:
        try:
            n = int(sys.argv[1].strip('-'))
            tail.reverse(n)
        except ValueError:
            print('Unknown command {!r}'.format(sys.argv[1]), file=sys.stderr)
            raise SystemExit(1)
    else:
        print('Unknown command {!r}'.format(sys.argv[1]), file=sys.stderr)
        raise SystemExit(1)

测试

1. 生成测试数据

with open('test.log', 'w+') as f:
    for l in range(1, 21):
        print('This is line {}'.format(l), file=f )

2. 读取测试

[tail] python test_data.py                                            15:49:04
[tail] python tail.py -5 test.log                                     15:50:09
This is line 20
This is line 19
This is line 18
This is line 17
This is line 16