3

ファイルに追加された新しい行を表示しようとしています。だから私が持っていると想像してくださいsample_file.txt：

 1. line 1
 2. line 2
 3. line 3

このファイルに新しい行があるかどうかを確認し、その行を表示したい (すべてのファイルを再度印刷せずに)

#!/usr/bin/python

import os
import pyinotify
from datetime import datetime
import time

PATH = os.path.join(os.path.expanduser('~/'), 'sample_file.txt')

m = pyinotify.WatchManager()
m.add_watch(PATH, pyinotify.ALL_EVENTS, rec=True)
notifier = pyinotify.Notifier(m, pyinotify.ProcessEvent(), 0,  0, 100)

f = open(PATH)
for line in f.readlines():
    print line

while True:
    time.sleep(5)
    try:
        if notifier.check_events():
            # RIGHT HERE:
            # HOW TO DO SOMETHING LIKE
            # f.last() ???
            print f.next()
        else:
            print 'waiting for a line....'
    except KeyboardInterrupt:
        notifier.stop()
        break

私が考えていたのは、whileループの前のどこかですべての行を読み取り、次の行を出力することですが、コードに何か問題がありf.next()、ループに入った直後にチェックします。

4

1 に答える 1

4

次の 2 つの問題に対処します。

  • tailファイルに実装する方法、
  • pyinotifyそしてモジュールの使い方。

ファイルのテール

コードでは、次のことを行う必要があります。

  • readまたはreadlinesを使用して、できるだけ多くの行全体を読んでみてください。
  • を使用して印刷できるようになるまで、ファイルを最後の不完全な行の先頭まで巻き戻しますseek。

これは、たとえば次のように変換されます。

f = open(PATH)
for line in f.readlines():
    print line[:-1]

while True:
    time.sleep(5)
    try:
        line_start = f.tell()
        new_lines = f.read()
        last_n = new_lines.rfind('\n')
        if last_n >= 0:
            # We got at least one full line
            line_start += last_n + 1
            print new_lines[:last_n]
        else:
            # No complete line yet
            print 'no line'
        f.seek(line_start)
    except KeyboardInterrupt:
        notifier.stop()
        break

ここでさらに例を見つけることができますが、改行で終わらないファイルへの追加に対応していないものもあります。

そして、ここにいくつかの選択肢があります Python でログファイルを追跡するにはどうすればよいですか?

Pyinotify ループ

pyinotifyまた、ドキュメントで説明されているように、 のイベント ハンドラ内にコードを移動する必要があります。

check_events処理するイベントがある場合は戻りますTrueが、実際にはイベントを処理しないため、イベントTrueが処理されるまで常に戻ります。

また、while/sleepループを避けるようにしてください。Inotify は、リソースを犠牲にすることなく、イベントを受信するとすぐに処理する機能を追加します。/ループは反応性が低くなりますwhile。sleep

以下は、 のショート チュートリアルの最初の 2 つの方法pyinotifyです。

1. 終わりのない監視

他のイベント ループがない場合は、これが最も反応性が高いため、この方法をお勧めします。

PATH = os.path.join(os.path.expanduser('~/'), 'experiments', 'testfile')

class EventHandler(pyinotify.ProcessEvent):
    def __init__(self, *args, **kwargs):
        super(EventHandler, self).__init__(*args, **kwargs)
        self.file = open(PATH)
        self.position = 0
        self.print_lines()

    def process_IN_MODIFY(self, event):
        print 'event received'
        self.print_lines()

    def print_lines(self):
        new_lines = self.file.read()
        last_n = new_lines.rfind('\n')
        if last_n >= 0:
            self.position += last_n + 1
            print new_lines[:last_n]
        else:
            print 'no line'
        self.file.seek(self.position)

wm = pyinotify.WatchManager()
handler = EventHandler()
notifier = pyinotify.Notifier(wm, handler)
wm.add_watch(PATH, pyinotify.IN_MODIFY, rec=True)
notifier.loop()

2. 定期的なモニタリング

すでに処理ループがある場合は、process_events定期的に呼び出すだけです。EventHandlerクラスは方法 1 と同じですが、呼び出しの代わりに通知notifier.loop()機能に小さなタイムアウトを追加し、独自のイベント ループを実装します。

...

wm = pyinotify.WatchManager()
handler = EventHandler()
notifier = pyinotify.Notifier(wm, handler, timeout=10)
wm.add_watch(PATH, pyinotify.IN_MODIFY, rec=True)

while True:
    # Do something unrelated to pyinotify
    time.sleep(5)

    notifier.process_events()
    #loop in case more events appear while we are processing
    while notifier.check_events():
        notifier.read_events()
        notifier.process_events()
于 2013-04-23T21:22:02.543 に答える