面试宝典_Python.运维开发.000

• 2020 年 1 月 7 日
• 筆記

面试题目:

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)

有图有像: