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

面试题目:

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)

有图有像:

wKioL1iXNFngMPcWAABFkWpclw8083.png

©著作权归作者所有：来自51CTO博客作者运维开发的原创作品，如需转载，请注明出处，否则将追究法律责任