怎么用Python完成tail命令

本篇內(nèi)容主要講解“怎么用Python完成tail命令”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“怎么用Python完成tail命令”吧!

1.第一版--從文件尾部讀取實(shí)時(shí)數(shù)據(jù)

主要思路是: 打開(kāi)文件, 把指針移動(dòng)到文件最后, 然后有數(shù)據(jù)則輸出數(shù)據(jù), 無(wú)數(shù)據(jù)則休眠一段時(shí)間.

import time
import sys

from typing import Callable, NoReturn


class Tail(object):
    def __init__(
            self,
            file_name: str,
            output: Callable[[str], NoReturn] = sys.stdout.write,
            interval: int = 1
    ):
        self.file_name: str = file_name
        self.output: Callable[[str], NoReturn] = output
        self.interval: int = interval

    def __call__(self):
        with open(self.file_name) as f:
            f.seek(0, 2)  # 從文件結(jié)尾處開(kāi)始seek
            while True:
                line: str = f.readline()
                if line:
                    self.output(line)  # 使用print都會(huì)每次都打印新的一行
                else:
                    time.sleep(self.interval)


if __name__ == '__main__':
    filename: str = sys.argv[0]
    Tail(filename)()

之后只要做如下調(diào)用即可:

python xxx.py filename

2.第二版--實(shí)現(xiàn)tail -f

tail -f默認(rèn)先讀取最后10行數(shù)據(jù),再?gòu)奈募膊孔x取實(shí)時(shí)數(shù)據(jù).如果對(duì)于小文件,可以先讀取所有文件內(nèi)容,并輸出最后10行, 但是讀取全文再獲取最后10行的性能不高, 而從后滾10行的邊界條件也很復(fù)雜, 先看先讀取全文再獲取最后10行的實(shí)現(xiàn):

import time
import sys

from typing import Callable, NoReturn


class Tail(object):
    def __init__(
            self,
            file_name: str,
            output: Callable[[str], NoReturn] = sys.stdout.write,
            interval: int = 1
    ):
        self.file_name: str = file_name
        self.output: Callable[[str], NoReturn] = output
        self.interval: int = interval

    def __call__(self):
        with open(self.file_name) as f:
            self.read_last_line(f)
            while True:
                line: str = f.readline()
                if line:
                    self.output(line)  # 使用print都會(huì)每次都打印新的一行
                else:
                    time.sleep(self.interval)

    def read_last_line(self, f):
        last_lines = f.readlines()[-10:]
        for line in last_lines:
            self.output(line)

if __name__ == '__main__':
    filename: str = sys.argv[0]
    Tail(filename)()

可以看到實(shí)現(xiàn)很簡(jiǎn)單, 相比第一版只多了個(gè)read_last_line的函數(shù), 接下來(lái)就要解決性能的問(wèn)題了, 當(dāng)文件很大的時(shí)候, 這個(gè)邏輯是不行的, 特別是有些日志文件經(jīng)常有幾個(gè)G大, 如果全讀出來(lái)內(nèi)存就爆了. 而在Linux系統(tǒng)中, 沒(méi)有一個(gè)接口可以指定指針跳到倒數(shù)10行, 只能使用如下方法來(lái)模擬輸出倒數(shù)10行:

• 首先游標(biāo)跳轉(zhuǎn)到最新的字符, 保存當(dāng)前游標(biāo), 然后預(yù)估一行數(shù)據(jù)的字符長(zhǎng)度, 最好偏多, 這里我按1024字符長(zhǎng)度為一行來(lái)處理

• 然后利用seek的方法,跳轉(zhuǎn)到seek(-1024 * 10, 2)的字符, 這就是我們預(yù)估的倒數(shù)10行內(nèi)的內(nèi)容

• 接著對(duì)內(nèi)容進(jìn)行判斷, 如果跳轉(zhuǎn)的字符長(zhǎng)度小于 10 * 1024, 則證明整個(gè)文件沒(méi)有10行, 則采用原來(lái)的read_last_line方法.

• 如果跳轉(zhuǎn)到字符長(zhǎng)度等于1024 * 10, 則利用換行符計(jì)算已取字符長(zhǎng)度共有多少行,如果行數(shù)大于10,那只輸出最后10行,如果只讀了4行,則繼續(xù)讀6*1024,直到讀滿10行為止

通過(guò)以上步奏, 就把倒數(shù)10行的數(shù)據(jù)計(jì)算好了可以打印出來(lái), 可以進(jìn)入追加數(shù)據(jù)了, 但是這時(shí)候文件內(nèi)容可能發(fā)生改變了, 我們的游標(biāo)也發(fā)生改變了, 這時(shí)候要把游標(biāo)跳回到剛才保存的游標(biāo),防止漏打或者重復(fù)打印數(shù)據(jù).

分析完畢后, 就可以開(kāi)始重構(gòu)read_last_line函數(shù)了.

import time
import sys

from typing import Callable, List, NoReturn


class Tail(object):
    def __init__(
            self,
            file_name: str,
            output: Callable[[str], NoReturn] = sys.stdout.write,
            interval: int = 1,
            len_line: int = 1024
    ):
        self.file_name: str = file_name
        self.output: Callable[[str], NoReturn] = output
        self.interval: int = interval
        self.len_line: int = len_line

    def __call__(self, n: int = 10):
        with open(self.file_name) as f:
            self.read_last_line(f, n)
            while True:
                line: str = f.readline()
                if line:
                    self.output(line)  # 使用print都會(huì)每次都打印新的一行
                else:
                    time.sleep(self.interval)

    def read_last_line(self, file, n):
        read_len: int = self.len_line * n

        # 跳轉(zhuǎn)游標(biāo)到最后
        file.seek(0, 2)
        # 獲取當(dāng)前結(jié)尾的游標(biāo)位置
        now_tell: int = file.tell()
        while True:
            if read_len > file.tell():
                # 如果跳轉(zhuǎn)的字符長(zhǎng)度大于原來(lái)文件長(zhǎng)度,那就把所有文件內(nèi)容打印出來(lái)
                file.seek(0) # 由于read方法是按照游標(biāo)進(jìn)行打印, 所以要重置游標(biāo)
                last_line_list: List[str] = file.read().split('\n')[-n:]
                # 重新獲取游標(biāo)位置
                now_tell: int = file.tell()
                break
            # 跳轉(zhuǎn)到我們預(yù)估的字符位置
            file.seek(-read_len, 2)
            read_str: str = file.read(read_len)
            cnt: int = read_str.count('\n')
            if cnt >= n:
                # 如果獲取的行數(shù)大于要求的行數(shù),則獲取前n行的行數(shù)
                last_line_list: List[str] = read_str.split('\n')[-n:]
                break
            else:
                # 如果獲取的行數(shù)小于要求的行數(shù),則預(yù)估需要獲取的行數(shù),繼續(xù)獲取
                if cnt == 0:
                    line_per: int = read_len
                else:
                    line_per: int = int(read_len / cnt)
                read_len = line_per * n

        for line in last_line_list:
            self.output(line + '\n')
        # 重置游標(biāo),確保接下來(lái)打印的數(shù)據(jù)不重復(fù)
        file.seek(now_tell)


if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("-f", "--filename")
    parser.add_argument("-n", "--num", default=10)
    args, unknown = parser.parse_known_args()
    if not args.filename:
        raise RuntimeError('filename args error')
    Tail(args.filename)(int(args.num))

3.第三版--優(yōu)雅的讀取輸出日志文件

可以發(fā)現(xiàn)實(shí)時(shí)讀取那塊的邏輯性能還是很差, 如果每秒讀一次文件,實(shí)時(shí)性就太慢了,把間隔改小了,則處理器占用太多. 性能最好的情況是如果能得知文件更新再進(jìn)行打印文件, 那性能就能得到保障了.慶幸的是,在Linux中inotify提供了這樣的功能. 此外,日志文件有一個(gè)特點(diǎn)就是會(huì)進(jìn)行l(wèi)ogrotate,如果日志被logrotate了,那我們就需要重新打開(kāi)文件,并進(jìn)一步讀取數(shù)據(jù), 這種情況也可以利用到inotify, 當(dāng)inotify獲取到文件重新打開(kāi)的事件時(shí),我們就重新打開(kāi)文件,再讀取.

import os
import sys

from typing import Callable, List, NoReturn

import pyinotify

multi_event = pyinotify.IN_MODIFY | pyinotify.IN_MOVE_SELF  # 監(jiān)控多個(gè)事件


class InotifyEventHandler(pyinotify.ProcessEvent):  # 定制化事件處理類，注意繼承
    """
    執(zhí)行inotify event的封裝
    """
    f: 'open()'
    filename: str
    path: str
    wm: 'pyinotify.WatchManager'
    output: Callable

    def my_init(self, **kargs):
        """pyinotify.ProcessEvent要求不能直接繼承__init__, 而是要重寫(xiě)my_init, 我們重寫(xiě)這一段并進(jìn)行初始化"""

        # 獲取文件
        filename: str = kargs.pop('filename')
        if not os.path.exists(filename):
            raise RuntimeError('Not Found filename')
        if '/' not in filename:
            filename = os.getcwd() + '/' + filename
        index = filename.rfind('/')
        if index == len(filename) - 1 or index == -1:
            raise RuntimeError('Not a legal path')

        self.f = None
        self.filename = filename
        self.output: Callable = kargs.pop('output')
        self.wm = kargs.pop('wm')
        # 只監(jiān)控路徑,這樣就能知道文件是否移動(dòng)
        self.path = filename[:index]
        self.wm.add_watch(self.path, multi_event)

    def read_line(self):
        """統(tǒng)一的輸出方法"""
        for line in self.f.readlines():
            self.output(line)

    def process_IN_MODIFY(self, event):
        """必須為process_事件名稱，event表示事件對(duì)象, 這里表示監(jiān)控到文件發(fā)生變化, 進(jìn)行文件讀取"""
        if event.pathname == self.filename:
            self.read_line()

    def process_IN_MOVE_SELF(self, event):
        """必須為process_事件名稱，event表示事件對(duì)象, 這里表示監(jiān)控到文件發(fā)生重新打開(kāi), 進(jìn)行文件讀取"""
        if event.pathname == self.filename:
            # 檢測(cè)到文件被移動(dòng)重新打開(kāi)文件
            self.f.close()
            self.f = open(self.filename)
            self.read_line()

    def __enter__(self) -> 'InotifyEventHandler':
        self.f = open(self.filename)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.f.close()


class Tail(object):
    def __init__(
            self,
            file_name: str,
            output: Callable[[str], NoReturn] = sys.stdout.write,
            interval: int = 1,
            len_line: int = 1024
    ):
        self.file_name: str = file_name
        self.output: Callable[[str], NoReturn] = output
        self.interval: int = interval
        self.len_line: int = len_line

        wm = pyinotify.WatchManager()  # 創(chuàng)建WatchManager對(duì)象
        inotify_event_handler = InotifyEventHandler(
            **dict(filename=file_name, wm=wm, output=output)
        )  # 實(shí)例化我們定制化后的事件處理類, 采用**dict傳參數(shù)
        wm.add_watch('/tmp', multi_event)  # 添加監(jiān)控的目錄，及事件
        self.notifier = pyinotify.Notifier(wm, inotify_event_handler)  # 在notifier實(shí)例化時(shí)傳入,notifier會(huì)自動(dòng)執(zhí)行
        self.inotify_event_handle: 'InotifyEventHandler' = inotify_event_handler

    def __call__(self, n: int = 10):
        """通過(guò)inotify的with管理打開(kāi)文件"""
        with self.inotify_event_handle as i:
            # 先讀取指定的行數(shù)
            self.read_last_line(i.f, n)
            # 啟用inotify的監(jiān)聽(tīng)
            self.notifier.loop()

    def read_last_line(self, file, n):
        read_len: int = self.len_line * n

        # 獲取當(dāng)前結(jié)尾的游標(biāo)位置
        file.seek(0, 2)
        now_tell: int = file.tell()
        while True:
            if read_len > file.tell():
                # 如果跳轉(zhuǎn)的字符長(zhǎng)度大于原來(lái)文件長(zhǎng)度,那就把所有文件內(nèi)容打印出來(lái)
                file.seek(0)
                last_line_list: List[str] = file.read().split('\n')[-n:]
                # 重新獲取游標(biāo)位置
                now_tell: int = file.tell()
                break
            file.seek(-read_len, 2)
            read_str: str = file.read(read_len)
            cnt: int = read_str.count('\n')
            if cnt >= n:
                # 如果獲取的行數(shù)大于要求的行數(shù),則獲取前n行的行數(shù)
                last_line_list: List[str] = read_str.split('\n')[-n:]
                break
            else:
                # 如果獲取的行數(shù)小于要求的行數(shù),則預(yù)估需要獲取的行數(shù),繼續(xù)獲取
                if cnt == 0:
                    line_per: int = read_len
                else:
                    line_per: int = int(read_len / cnt)
                read_len = line_per * n

        for line in last_line_list:
            self.output(line + '\n')
        # 重置游標(biāo),確保接下來(lái)打印的數(shù)據(jù)不重復(fù)
        file.seek(now_tell)


if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("-f", "--filename")
    parser.add_argument("-n", "--num", default=10)
    args, unknown = parser.parse_known_args()
    if not args.filename:
        raise RuntimeError('filename args error')
    Tail(args.filename)(int(args.num))

可以看到, 從原本的open打開(kāi)文件改為用inotify打開(kāi)文件(這時(shí)候會(huì)調(diào)用my_init方法進(jìn)行初始化), 打開(kāi)后還是運(yùn)行我們打開(kāi)原來(lái)n行的代碼, 然后就交給inotify運(yùn)行. 在inotify運(yùn)行之前, 我們把重新打開(kāi)文件方法和打印文件方法都掛載在inotifiy對(duì)應(yīng)的事件里, 之后inotify運(yùn)行時(shí), 會(huì)根據(jù)對(duì)應(yīng)的事件執(zhí)行對(duì)應(yīng)的方法。

到此，相信大家對(duì)“怎么用Python完成tail命令”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！