pytail.py

#!/usr/bin/env python
# Pytail
# ------
# Dumb python program that lets you tail a file while interactively filtering it
import os
import re
import select
import subprocess
import sys
import time

from blessed import Terminal

term = Terminal()

# Clear the screen
print(f"{term.home}{term.clear}")

poll = select.epoll()
p = subprocess.Popen(['/usr/bin/tail', '-f', sys.argv[1]], stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE)

# Register stdout of tail process with epoll
poll.register(p.stdout.fileno())

val = ''
has_filter = False
filter_val = []

# <https://blessed.readthedocs.io/en/stable/keyboard.html#inkey>
with term.cbreak():
    while val.lower() != 'q':

        # Don't hang on wait, just check if process is closed
        pid, _ = os.waitpid(p.pid, os.WNOHANG)

        # Very short timeout on polling to see if there's data on tail's stdout
        for fd, _ in poll.poll(0.01):

            # 32768 is 2^15 -- some arbitary power of 2
            line = os.read(fd, 32768).decode('utf-8').strip()

            # If the tail line matches the filter, output the filter
            if re.search(''.join(filter_val), line):
                print(line)

        # This *shouldn't* happen: the tail process is closed...
        if pid:
            raise RuntimeError('"tail" process (%d) closed', pid)

        # Print instructions at the top of the string + currently active filter
        with term.location(0, 0):
            msg = 'Press "q" to quit, '
            if not has_filter:
                msg += '"f" to filter'
            else:
                msg += f"filtering using: '{''.join(filter_val)}'"
            print(term.on_green(msg + term.clear_eol))

        # Short timeout getting key input
        val = term.inkey(timeout=0.01)

        if has_filter:
            if val.is_sequence:
                # Turn off filter when user hits "backspace"
                if val.name == 'KEY_DELETE':
                    filter_val = []
                    has_filter = False
            else:
                filter_val.append(val)

        # Turn on the filter when user hits "f"
        if not has_filter and val.lower() == 'f':
            has_filter = True

        time.sleep(0.1)

A quick example of filtering apache logs to only show "GET" lines -- by typing "f" followed by "GET" while filtering: