os.stat(absname) except EnvironmentError as err: if err.errno != errno.ENOENT: raise else: if not stat.S_ISREG(st.st_mode): continue fid = self.get_file_id(st) ls.append((fid, absname)) # check existent files for fid, file in list(self._files_map.items()): try: st = os.stat(file.name) except EnvironmentError as err: if err.errno == errno.ENOENT: self.unwatch(file, fid) else: raise else: if fid != self.get_file_id(st): # same name but different file (rotation); reload it. self.unwatch(file, fid) self.watch(file.name) # add new ones for fid, fname in ls: if fid not in self._files_map: self.watch(fname) def readlines(self, file): """Read file lines since last access until EOF is reached and invoke callback. """ while True: lines = file.readlines(self._sizehint) if not lines: break self._callback(file.name, lines) def watch(self, fname): try: file = self.open(fname) fid = self.get_file_id(os.stat(fname)) except EnvironmentError as err: if err.errno != errno.ENOENT: raise else: self.log("watching logfile %s" % fname) self._files_map[fid] = file def unwatch(self, file, fid): # File no longer exists. If it has been renamed try to read it # for the last time in case we're dealing with a rotating log # file. self.log("un-watching logfile %s" % file.name) del self._files_map[fid] with file: lines = self.readlines(file) if lines: self._callback(file.name, lines) @staticmethod def get_file_id(st): if os.name == 'posix': return "%xg%x" % (st.st_dev, st.st_ino) else: return "%f" % st.st_ctime def close(self): for id, file in self._files_map.items(): file.close() self._files_map.clear()# ===================================================================# --- tests# ===================================================================if __name__ == '__main__': import unittest import atexit TESTFN = '$testfile.log' TESTFN2 = '$testfile2.log' PY3 = sys.version_info[0] == 3 if PY3: def b(s): return s.encode("latin-1") else: def b(s): return s class TestLogWatcher(unittest.TestCase): def setUp(self): def callback(filename, lines): self.filename.append(filename) for line in lines: self.lines.append(line) self.filename = [] self.lines = [] self.file = open(TESTFN, 'w') self.watcher = LogWatcher(os.getcwd(), callback) def tearDown(self): self.watcher.close() self.remove_test_files() def write_file(self, data): self.file.write(data) self.file.flush() @staticmethod @atexit.register def remove_test_files(): for x in [TESTFN, TESTFN2]: try: os.remove(x) except EnvironmentError: pass def test_no_lines(self): self.watcher.loop(blocking=False) def test_one_line(self): self.write_file('foo') self.watcher.loop(blocking=False) self.assertEqual(self.lines, [b"foo"]) self.assertEqual(self.filename, [os.path.abspath(TESTFN)]) def test_two_lines(self): self.write_file('foo\n') self.write_file('bar\n') self.watcher.loop(blocking=False) self.assertEqual(self.lines, [b"foo\n", b"bar\n"]) self.assertEqual(self.filename, [os.path.abspath(TESTFN)]) def test_new_file(self): with open(TESTFN2, "w") as f: f.write("foo") self.watcher.loop(blocking=False) self.assertEqual(self.lines, [b"foo"]) self.assertEqual(self.filename, [os.path.abspath(TESTFN2)]) def test_file_removed(self): self.write_file("foo") try: os.remove(TESTFN) except EnvironmentError: # necessary on Windows pass self.watcher.loop(blocking=False) self.assertEqual(self.lines, [b"foo"]) def test_tail(self): MAX = 10000 content = '\n'.join([str(x) for x in range(0, MAX)]) self.write_file(content) # input < BUFSIZ (1 iteration) lines = self.watcher.tail(self.file.name, 100) self.assertEqual(len(lines), 100) self.assertEqual(lines, [b(str(x)) for x in range(MAX-100, MAX)]) # input > BUFSIZ (multiple iterations) lines = self.watcher.tail(self.file.name, 5000) self.assertEqual(len(lines), 5000) self.assertEqual(lines, [b(str(x)) for x in range(MAX-5000, MAX)]) # input > file's total lines lines = self.watcher.tail(self.file.name, MAX + 9999) self.assertEqual(len(lines), MAX) self.assertEqual(lines, [b(str(x)) for x in range(0, MAX)]) # self.assertRaises(ValueError, self.watcher.tail, self.file.name, 0) LogWatcher.tail(self.file.name, 10) def test_ctx_manager(self): with self.watcher: pass test_suite = unittest.TestSuite() test_suite.addTest(unittest.makeSuite(TestLogWatcher)) unittest.TextTestRunner(verbosity=2).run(test_suite)Sample: watchSampleLog.py#!/usr/bin/env pythondef callback(filename, lines): print(filename, lines)if __name__ == '__main__': from LogWatcher import LogWatcher lw = LogWatcher.LogWatcher("./logs/", callback) try: lw.loop() except KeyboardInterrupt: pass