面试宝典_Python.运维开发.0004.用Python实现tail实时输出新增日志?

Posted 2020-08-30

面试题目:

1. 用PYTHON实现tail -f功能,默认显示最后15行,实时输出新增行?

解题思路:

1. 此需求在很多场景中都有遇到,而且在各大群中也被讨论过,虽然有现成的模版如pyinotify等模块实现,但面试更想通过你的解题思路来判断这场面试,具体到tail需要实现2个功能,一个实时输出新增内容,一个默认输出前15行,前者直接循环打开文件读取,全局变量中记录上次读取的位置,下一次循环seek到上次的位置读取即可,而对于默认显示前15行的做法是假设一行1000个字节,循环读取,当文件总长度小于1000时则从开头开始读取分割行取出后10行即可,即使不够10行也没关系,当文件总长度大于等于1000时,如果分割后大于10行,则读取分割行取出后10行即可,如果小于10行则继续向前1000个读取,以此类推~

具体实现:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# 51CTOBG: http://xmdevops.blog.51cto.com/
# Purpose:
#
"""
from __future__ import absolute_import
# 说明: 导入公共模块
import os 
import sys
import time
import chardet
# 说明: 导入其它模块
# 


if __name__ == ‘__main__‘:
    seekps = 0;
    target = ‘data.txt‘;

    try:
        # 指针跳到最后作为起读位置
        with open(target, ‘r+b‘) as f:
            f.seek(0, os.SEEK_END)
            seekps = f.tell()
            print ‘notice: file length is‘, seekps
            rest_lines = []
            # 默认读取15行
            line_reads = 15
            # 假设默认每行1000字节
            line_bytes = 1000
            ‘‘‘
            1. 当f_length < line_bytes,就seek到0开始读取,读完break
            2. 当f_length > line_bytes,换行符数n表示n+1行
               => 当n+1 >= line_reads,取出其中前line_reads个元素
               => 当n+1 <  line_reads,继续向前seek到count*line_reads位置读取
            ‘‘‘
            count = 1
            while True:
                if seekps <= line_bytes*count:
                    f.seek(0)
                    rest_lines = f.read().split(os.linesep)[-line_reads:]
                    break
                f.seek(-1*line_bytes*count, 2)
                rest_lines = f.read().split(os.linesep)
                if len(rest_lines)>=line_reads:
                    rest_lines = rest_lines[-line_reads:]
                    break
                else:
                    count += 1

            for line in rest_lines:
                code = chardet.detect(line).get(‘encoding‘)
                line = line.decode(code).encode(sys.stdout.encoding)
                sys.stdout.write(‘‘.join([line.strip(), os.linesep]))
                sys.stdout.flush()

    except Exception, e:
        print ‘notice: open file with error({0})‘.format(e)
        exit()

    print ‘notice: start position is‘, seekps
    while True:
        try:
            with open(target, ‘r+b‘) as f:
                f.seek(0, os.SEEK_END)
                # 防止数据被意外截断
                if f.tell()<seekps:
                    f.seek(f.tell())
                else:
                    f.seek(seekps)

                while True:
                    line = f.readline()
                    if not line.strip():
                        break
                    code = chardet.detect(line).get(‘encoding‘)
                    line = line.decode(code).encode(sys.stdout.encoding)
                    sys.stdout.write(‘‘.join([line.strip(), os.linesep]))
                    sys.stdout.flush()
                seekps = f.tell()
        except Exception, e:
            print ‘notice: open file with error({0})‘.format(e)
            break
        time.sleep(0.1)

有图有像:

本文出自 “满满李 - 运维开发之路” 博客，请务必保留此出处http://xmdevops.blog.51cto.com/11144840/1895119