91

ブロックやロックを行わずに、tail -F などの出力を Python で利用できるようにしたいと考えています。ここでそれを行うための非常に古いコードをいくつか見つけましたが、今では同じことを行うためのより良い方法またはライブラリが必要だと考えています。誰か知っていますか？

理想的には、tail.getNewData()より多くのデータが必要になるたびに呼び出すことができるようなものがあるとよいでしょう。

4

13 に答える 13

78

ノンブロッキング

Linuxを使用している場合（Windowsはファイルでのselectの呼び出しをサポートしていないため）、selectモジュールと一緒にサブプロセスモジュールを使用できます。

import time
import subprocess
import select

f = subprocess.Popen(['tail','-F',filename],\
        stdout=subprocess.PIPE,stderr=subprocess.PIPE)
p = select.poll()
p.register(f.stdout)

while True:
    if p.poll(1):
        print f.stdout.readline()
    time.sleep(1)

これにより、出力パイプで新しいデータがポーリングされ、使用可能になったときに出力されます。通常、time.sleep(1)とprint f.stdout.readline()は有用なコードに置き換えられます。

ブロッキング

追加のselectモジュール呼び出しなしでサブプロセスモジュールを使用できます。

import subprocess
f = subprocess.Popen(['tail','-F',filename],\
        stdout=subprocess.PIPE,stderr=subprocess.PIPE)
while True:
    line = f.stdout.readline()
    print line

これにより、追加された新しい行も出力されますが、テールプログラムが閉じられるまで、おそらく。でブロックされf.kill()ます。

于 2012-09-21T02:09:27.790 に答える
51

sh モジュールの使用(pip install sh):

from sh import tail
# runs forever
for line in tail("-f", "/var/log/some_log_file.log", _iter=True):
    print(line)

[アップデート]

=True を指定した sh.tail_iterはジェネレーターであるため、次のことができます。

import sh
tail = sh.tail("-f", "/var/log/some_log_file.log", _iter=True)

次に、次のように「getNewData」を実行できます。

new_data = tail.next()

テールバッファが空の場合、データが増えるまでブロックされることに注意してください（あなたの質問から、この場合に何をしたいのか明確ではありません）。

[アップデート]

これは -f を -F に置き換えれば機能しますが、Python ではロックされます。可能であれば、必要なときに新しいデータを取得するために呼び出すことができる関数を持つことにもっと興味があります。– エリ

while True ループ内に末尾呼び出しを配置し​​、最終的な I/O 例外をキャッチするコンテナー ジェネレーターは、-F とほぼ同じ効果があります。

def tail_F(some_file):
    while True:
        try:
            for line in sh.tail("-f", some_file, _iter=True):
                yield line
        except sh.ErrorReturnCode_1:
            yield None

ファイルにアクセスできなくなった場合、ジェネレーターは None を返します。ただし、ファイルにアクセスできる場合は、新しいデータが作成されるまでブロックされます。この場合、あなたが何をしたいのか、私にはまだ不明です。

Raymond Hettinger のアプローチはかなり良いようです。

def tail_F(some_file):
    first_call = True
    while True:
        try:
            with open(some_file) as input:
                if first_call:
                    input.seek(0, 2)
                    first_call = False
                latest_data = input.read()
                while True:
                    if '\n' not in latest_data:
                        latest_data += input.read()
                        if '\n' not in latest_data:
                            yield ''
                            if not os.path.isfile(some_file):
                                break
                            continue
                    latest_lines = latest_data.split('\n')
                    if latest_data[-1] != '\n':
                        latest_data = latest_lines[-1]
                    else:
                        latest_data = input.read()
                    for line in latest_lines[:-1]:
                        yield line + '\n'
        except IOError:
            yield ''

ファイルにアクセスできなくなった場合、または新しいデータがない場合、このジェネレーターは '' を返します。

[アップデート]

最後から 2 番目の回答は、データがなくなるたびに見えるファイルの先頭に回り込みます。– エリ

2 番目は、テール プロセスが終了するたびに、最後の 10 行を出力すると思い-fます。これは、I/O エラーが発生したときです。Unixのtail --follow --retryような環境で考えられるほとんどの場合、動作はこれとはかけ離れたものではありません。

おそらく、質問を更新して、本当の目標 (テールを模倣したい理由 --retry) を説明すると、より良い答えが得られるでしょう。

最後の回答は、実際には末尾に続くのではなく、実行時に利用可能なものを読み取るだけです。– エリ

もちろん、tail はデフォルトで最後の 10 行を表示します... file.seek を使用してファイル ポインタをファイルの末尾に配置できます。適切な実装は読者の演習として残します。

私見 file.read() アプローチは、サブプロセスベースのソリューションよりもはるかにエレガントです。

于 2012-09-21T01:25:39.760 に答える
27

実際、ファイルへの唯一の移植可能な方法は、ファイルから読み取り、 0が返された場合にtail -f（の後に）再試行することです。さまざまなプラットフォームのユーティリティは、プラットフォーム固有のトリック（BSDなど）を使用して、ファイルを永久に効率的にテールします。を必要とせずに。sleepreadtailkqueuesleep

したがって、tail -f（プラットフォーム固有のハックに頼ることなく）最小公分母の実装を使用する必要があるため、Pythonで純粋に適切なものを実装することはおそらく良い考えではありません。簡単subprocessに開いて別のスレッドの行を反復処理することで、Pythontail -fでノンブロッキング操作を簡単に実装できます。tail

実装例：

import threading, Queue, subprocess
tailq = Queue.Queue(maxsize=10) # buffer at most 100 lines

def tail_forever(fn):
    p = subprocess.Popen(["tail", "-f", fn], stdout=subprocess.PIPE)
    while 1:
        line = p.stdout.readline()
        tailq.put(line)
        if not line:
            break

threading.Thread(target=tail_forever, args=(fn,)).start()

print tailq.get() # blocks
print tailq.get_nowait() # throws Queue.Empty if there are no lines to read
于 2012-09-21T01:59:54.063 に答える
15

だから、これはかなり遅れていますが、私は再び同じ問題に遭遇しました.今はもっと良い解決策があります. pygtailを使用するだけです:

Pygtail は、読み取られていないログ ファイルの行を読み取ります。ローテーションされたログ ファイルも処理します。logcheck の logtail2 ( http://logcheck.org )に基づく

于 2015-08-04T01:56:33.317 に答える
11

理想的には、より多くのデータが必要になるたびに呼び出すことができる tail.getNewData() のようなものが必要です

私たちはすでにそれを持っていて、とても素敵です。さらにデータが必要なときはいつでもf.read()を 呼び出すだけです。前回の読み取りが中断したところから読み取りを開始し、データ ストリームの最後まで読み取ります。

f = open('somefile.log')
p = 0
while True:
    f.seek(p)
    latest_data = f.read()
    p = f.tell()
    if latest_data:
        print latest_data
        print str(p).center(10).center(80, '=')

行ごとに読み取るには、 f.readline()を使用します。読み取り中のファイルが、部分的に読み取られた行で終了する場合があります。f.tell()で現在のファイル位置を見つけ、f.seek()を使用してファイル ポインタを不完全な行の先頭に戻すことで、そのケースを処理します。作業コードについては、この ActiveState レシピを参照してください。

于 2012-09-21T02:16:53.017 に答える
5

もう 1 つのオプションは、独自のモジュールで使用できる とtailheadの両方の Python バージョンtailとユーティリティと API を提供するライブラリです。head

元々はモジュールに基づいていましたがtailer、その主な利点は、ファイルをパスで追跡できることです。つまり、ファイルが再作成されたときに状況を処理できます。さらに、さまざまなエッジケースのバグ修正がいくつかあります。

于 2016-02-23T06:59:53.397 に答える
0

「AWK」コマンドを使用することもできます。
詳細については、http: //www.unix.com/shell-programming-scripting/41734-how-print-specific-lines-awk.html
を参照してください。 awk を使用して、最後の行、最後の数行、またはファイル。
これは python から呼び出すことができます。

于 2014-02-08T17:59:05.983 に答える
0 
# -*- coding:utf-8 -*-
import sys
import time


class Tail():
    def __init__(self, file_name, callback=sys.stdout.write):
        self.file_name = file_name
        self.callback = callback

    def follow(self, n=10):
        try:
            # 打开文件
            with open(self.file_name, 'r', encoding='UTF-8') as f:
            # with open(self.file_name,'rb') as f:
                self._file = f
                self._file.seek(0, 2)
                # 存储文件的字符长度
                self.file_length = self._file.tell()
                # 打印最后10行
                self.showLastLine(n)
                # 持续读文件 打印增量
                while True:
                    line = self._file.readline()
                    if line:
                        self.callback(line)
                    time.sleep(1)
        except Exception as e:
            print('打开文件失败，囧，看看文件是不是不存在，或者权限有问题')
            print(e)

    def showLastLine(self, n):
        # 一行大概100个吧 这个数改成1或者1000都行
        len_line = 100
        # n默认是10，也可以follow的参数传进来
        read_len = len_line * n
        # 用last_lines存储最后要处理的内容
        while True:
            # 如果要读取的1000个字符，大于之前存储的文件长度
            # 读完文件，直接break
            if read_len > self.file_length:
                self._file.seek(0)
                last_lines = self._file.read().split('\n')[-n:]
                break
            # 先读1000个 然后判断1000个字符里换行符的数量
            self._file.seek(-read_len, 2)
            last_words = self._file.read(read_len)
            # count是换行符的数量
            count = last_words.count('\n')

            if count >= n:
                # 换行符数量大于10 很好处理，直接读取
                last_lines = last_words.split('\n')[-n:]
                break
            # 换行符不够10个
            else:
                # break
                # 不够十行
                # 如果一个换行符也没有，那么我们就认为一行大概是100个
                if count == 0:

                    len_perline = read_len
                # 如果有4个换行符，我们认为每行大概有250个字符
                else:
                    len_perline = read_len / count
                # 要读取的长度变为2500，继续重新判断
                read_len = len_perline * n
        for line in last_lines:
            self.callback(line + '\n')


if __name__ == '__main__':
    py_tail = Tail('test.txt')
    py_tail.follow(1)
于 2021-09-17T07:26:20.130 に答える