面試寶典_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)

有圖有像: