Pythonでファイルの変更を監視する(OS非依存・PyGObject使用)

ライブラリGLibのメインループには定期的に関数を呼び出す仕組みがあり、「PyGTK上のタイマー処理について」で扱っている。これを用いてファイルのタイムスタンプを定期的に取得し、前回の結果と比較するようにすることで、ファイルが変更されるのを監視し、更新されたときに処理を実行するといったことも行える。この方法はOSに依存しない(GLibに依存するのみ)。
なお、Linuxカーネルにはファイル/ディレクトリの様々な変更を監視するinotifyという機能がある*1が、これはLinux固有の機能で、他の(広い意味での)UNIX系OSでは使えない。
(2010/4/12)「GIOライブラリのファイル変更監視機能を用いる」では、PyGObjectのGIOライブラリの言語バインディングを用いてより高度な変更監視を行う方法についてを扱っている。

最終更新日時の取得に関するメモ

Pythonで最終更新日時を取得するには

os.stat([ファイルのパス]).st_mtime

で取り出す。
(2010/2/27)os.path.getmtime()のほうが楽に取得できる。

例

テキスト入力欄にファイルの絶対パスを入力して「Watch」ボタン(トグルボタン)を押すと監視が開始され、1秒に1回*2最終更新日時をチェックし、ファイルが上書きされるかtouchコマンドによりタイムスタンプが変更されるかしたときにテキストビュー内に通知メッセージを表示する。「Watch」ボタンをもう一度押して元に戻すと監視も解除する。
下のプログラムはPyGObjectの他PyGTKも用いている(どちらかのパッケージがなければインストールしておく必要がある)。
[任意]ファイル名: filewatchtest.py

#! /usr/bin/python
# -*- encoding: utf-8 -*-

import time
import sys
import os
try:
  import pygtk
  pygtk.require('2.0')
except:
  pass
try:
  import gtk
except:
  print >> sys.stderr, 'Error: PyGTK is not installed'
  sys.exit(1)
try:
  from glib import timeout_add_seconds as glib_timeout_add_seconds
  from glib import source_remove as glib_source_remove
except:
  try:
    from gobject import timeout_add_seconds as glib_timeout_add_seconds
    from gobject import source_remove as glib_source_remove
  except:
    print >> sys.stderr, 'Error: cannot import GLib functions'
    sys.exit(1)


class MainWindow(gtk.Window):
  """
  メインウィンドウ
  """
  def __init__(self, *args, **kwargs):
    gtk.Window.__init__(self, *args, **kwargs)
    # ショートカットキー(アクセラレータ)
    self.__accelgroup = gtk.AccelGroup()
    self.add_accel_group(self.__accelgroup)
    # メニュー項目
    self.__item_quit = gtk.ImageMenuItem(gtk.STOCK_QUIT, self.__accelgroup)
    self.__menu_file = gtk.Menu()
    self.__menu_file.add(self.__item_quit)
    self.__item_file = gtk.MenuItem('_File')
    self.__item_file.props.submenu = self.__menu_file
    self.__menubar = gtk.MenuBar()
    self.__menubar.append(self.__item_file)
    # ファイル名のテキスト入力欄
    self.__entry = gtk.Entry()
    # 参照ボタン
    self.__btn_browse = gtk.Button('...')
    # 監視開始/停止ボタン
    self.__btn_watch = gtk.ToggleButton('Watch')
    self.__btn_watch.props.sensitive = False
    # メッセージ出力用テキストビュー
    self.__textbuf = gtk.TextBuffer()
    self.__textview = gtk.TextView(self.__textbuf)
    self.__textview.props.editable = False
    self.__sw = gtk.ScrolledWindow()
    self.__sw.add(self.__textview)
    self.__sw.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_AUTOMATIC)
    # レイアウト用コンテナ
    self.__hbox = gtk.HBox()
    self.__hbox.pack_start(self.__entry)
    self.__hbox.pack_start(self.__btn_browse, expand=False, fill=False)
    self.__hbox.pack_start(self.__btn_watch, expand=False, fill=False)
    self.__vbox = gtk.VBox()
    self.__vbox.pack_start(self.__menubar, expand=False, fill=False)
    self.__vbox.pack_start(self.__hbox, expand=False, fill=False)
    self.__vbox.pack_start(self.__sw)
    # シグナル
    self.connect('delete_event', gtk.main_quit)
    self.__item_quit.connect('activate', gtk.main_quit)
    self.__entry.connect('changed', self.__on_entry_changed)
    self.__btn_browse.connect('clicked', self.__on_button_browse_clicked)
    self.__id_btnsig = self.__btn_watch.connect('toggled', self.__on_button_watch_toggled)
    # ウィンドウ
    self.add(self.__vbox)
    self.set_size_request(400, 300)
  def __on_entry_changed(self, widget):
    """
    テキスト入力欄の中身が変更されたときの処理
    """
    self.__btn_watch.props.sensitive = os.access(widget.props.text, os.R_OK)
  def __on_button_browse_clicked(self, widget):
    """
    ボタンがクリックされたときの処理
    """
    opendlg = gtk.FileChooserDialog(title='Select file', parent=self, buttons=(gtk.STOCK_CANCEL, gtk.RESPONSE_CANCEL, gtk.STOCK_OPEN, gtk.RESPONSE_ACCEPT))
    opendlg.set_local_only(True)
    if opendlg.run() == gtk.RESPONSE_ACCEPT:
      self.__entry.set_text(opendlg.get_filename())
    opendlg.destroy()
  def __on_button_watch_toggled(self, widget):
    """
    監視ボタンの状態が切り替えられたときの処理
    """
    if widget.props.active:
      # 監視開始
      try:
        self.__mtime = os.path.getmtime(self.__entry.props.text)
      except OSError, (errno, msg):
        self.__append_log_and_scroll('cannot stat "%s"\n errno: %d\n message: %s\n' % (self.__entry.props.text, errno, msg))
        # トグルボタンを元に戻すがその間ハンドラは呼ばれないようにする
        # connect()の戻り値をhandler_block()/handler_unblock()に指定
        self.__btn_watch.handler_block(self.__id_btnsig)
        self.__btn_watch.props.active = False
        self.__btn_watch.handler_unblock(self.__id_btnsig)
        return
      self.__entry.props.sensitive = False
      self.__btn_browse.props.sensitive = False
      self.__id_watch = glib_timeout_add_seconds(1, self.__to_watch_file)
      self.__append_log_and_scroll('start watching "%s" (%s)\n' % (self.__entry.props.text, time.ctime(None)))
    else:
      # 監視停止
      glib_source_remove(self.__id_watch)
      self.__append_log_and_scroll('stop watching "%s"\n' % self.__entry.props.text)
      self.__entry.props.sensitive = True
      self.__btn_browse.props.sensitive = True
  def __append_log_and_scroll(self, text):
    """
    テキストビューに文字列を追加して最後にスクロール
    """
    self.__textbuf.place_cursor(self.__textbuf.get_end_iter())
    self.__textbuf.insert_at_cursor(text)
    self.__textview.scroll_to_mark(self.__textbuf.get_insert(), 0)
  def __to_watch_file(self):
    """
    ファイルの最終更新日時をチェック
    """
    try:
      new_mtime = os.path.getmtime(self.__entry.props.text)
    except OSError, (errno, msg):
      self.__append_log_and_scroll('cannot stat "%s"\n errno: %d\n message: %s\n' % (self.__entry.props.text, errno, msg))
      self.__btn_watch.props.active = False  # ここでsource_remove()されるので
      return False                           # 戻り値はFalseでなくてもOK
    # 新しく取得したタイムスタンプが以前記憶したものと異なれば更新されている
    if self.__mtime != new_mtime:
      self.__mtime = new_mtime  # 新しいものを記憶
      # 更新されたときに行う処理をここに記述
      self.__append_log_and_scroll('file "%s" has been updated (%s)\n' % (self.__entry.props.text, time.ctime(self.__mtime)))
    # source_remove()されるまで無限に繰り返す
    return True;

class PyGTKFileWatchTest:
  """
  ファイル変更監視のテスト
  """
  def main(self):
    """
    アプリケーションのメイン処理
    """
    win = MainWindow()
    win.show_all()
    gtk.main()


if __name__ == '__main__':
  app = PyGTKFileWatchTest()
  app.main()

(2010/2/27)PyGObjectのプロパティを用いるように修正した他、os.path.getmtime()の使用など、一部を微調整(動作は以前と同一)

関連記事:

• PyGTK上のタイマー処理について
• PythonでPyGObjectを用いずにファイルの変更を監視する
• GIOライブラリのファイル変更監視機能を用いる

参考URL:

• Python ライブラリリファレンス: 14.1.4 ファイルとディレクトリ
• PyGObject リファレンス: glib Functions

*1:Pythonから利用するのであればpyinotifyというパッケージから利用できるようだ

*2:glib.timeout_add_seconds()やglib.timeout_add()の引数でタイムアウト関数の呼び出し間隔が調整でき、監視の細かさ/変更時の反応の良さに関係してくる