Last active
December 23, 2015 22:00
-
-
Save otger/6700575 to your computer and use it in GitHub Desktop.
'tail -f' alternative which colors any given tag/word: usage: ctail path/to/file regex_to_color_1 regex_to_color_2 ... regex_to_color_N
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
# encoding: utf-8 | |
''' | |
ctail -- colored tail | |
ctail is a simple 'tail -f' alternative which colors given regular expressions matches | |
@author: Otger Ballester | |
@license: Apache License 2.0\nhttp://www.apache.org/licenses/LICENSE-2.0 | |
@contact: [email protected] | |
@deffield updated: Updated | |
''' | |
import sys | |
import os | |
import time | |
import re | |
from optparse import OptionParser | |
__all__ = [] | |
__version__ = 0.1 | |
__date__ = '2013-09-25' | |
__updated__ = '2013-10-03' | |
DEBUG = 0 | |
TESTRUN = 0 | |
PROFILE = 0 | |
class CLIError(Exception): | |
'''Generic exception to raise and log different fatal errors.''' | |
# def __init__(self, msg): | |
# super(CLIError).__init__(type(self)) | |
# self.msg = "E: %s" % msg | |
def __str__(self): | |
return self.message | |
def __unicode__(self): | |
return self.message | |
def __repr__(self, *args, **kwargs): | |
return self.message | |
colours = {} | |
# Regular | |
colours["Black"]="\033[0;30m" | |
colours["Red"]="\033[0;31m" | |
colours["Green"]="\033[0;32m" | |
colours["Yellow"]="\033[0;33m" | |
colours["Blue"]="\033[0;34m" | |
colours["Purple"]="\033[0;35m" | |
colours["Cyan"]="\033[0;36m" | |
colours["White"]="\033[0;37m" | |
#Bold | |
colours["BBlack"]="\033[1;30m" | |
colours["BRed"]="\033[1;31m" | |
colours["BGreen"]="\033[1;32m" | |
colours["BYellow"]="\033[1;33m" | |
colours["BBlue"]="\033[1;34m" | |
colours["BPurple"]="\033[1;35m" | |
colours["BCyan"]="\033[1;36m" | |
colours["BWhite"]="\033[1;37m" | |
# High Intensity | |
colours["IBlack"]="\033[0;90m" | |
colours["IRed"]="\033[0;91m" | |
colours["IGreen"]="\033[0;92m" | |
colours["IYellow"]="\033[0;93m" | |
colours["IBlue"]="\033[0;94m" | |
colours["IPurple"]="\033[0;95m" | |
colours["ICyan"]="\033[0;96m" | |
colours["IWhite"]="\033[0;97m" | |
# Bold High Intensity | |
colours["BIBlack"]="\033[1;90m" | |
colours["BIRed"]="\033[1;91m" | |
colours["BIGreen"]="\033[1;92m" | |
colours["BIYellow"]="\033[1;93m" | |
colours["BIBlue"]="\033[1;94m" | |
colours["BIPurple"]="\033[1;95m" | |
colours["BICyan"]="\033[1;96m" | |
colours["BIWhite"]="\033[1;97m" | |
colour_close = "\033[0m" | |
colour_list = ['BRed', 'BGreen', 'BYellow', 'BBlue', 'BPurple', 'BCyan', 'IRed', 'IGreen', 'IYellow', 'IBlue', 'IPurple', 'ICyan'] | |
def regexes_coloring(what, regexes, regexes_colors): | |
for regex in regexes: | |
what = re.sub(regex, '{c}\g<0>{cc}'.format(c=regexes_colors[regex], cc=colour_close), what) | |
return what | |
def seek_last_n_lines_position(fp, nlines=10): | |
''' | |
Return fp positioned at start of last nlines | |
Inspired in code found at: http://stackoverflow.com/questions/136168/get-last-n-lines-of-a-file-with-python-similar-to-tail | |
''' | |
BLOCKSIZE = 1023 | |
fp.seek(0, 2) | |
pending_lines = nlines | |
fp_pos = fp.tell() | |
block = -1 | |
while 1: | |
if fp_pos > BLOCKSIZE: | |
fp.seek(block*BLOCKSIZE, 2) | |
data = fp.read(BLOCKSIZE) | |
else: | |
## file too small to contain nlines lines, return positioned at start | |
fp.seek(0,0) | |
data = fp.read(fp_pos) | |
block=0 | |
lines_in_block = data.count('\n') | |
if pending_lines <= lines_in_block: | |
end = -1 | |
for _ in range(pending_lines): | |
end = data.rfind('\n', 0, end) | |
if block==0: | |
fp.seek(end +len('\n'),0) | |
else: | |
fp.seek(block*BLOCKSIZE + end +len('\n'),2) | |
return fp | |
elif block == 0: | |
fp.seek(0,0) | |
return fp | |
pending_lines -= lines_in_block | |
fp_pos -= BLOCKSIZE | |
block -= 1 | |
def main(argv=None): | |
'''Command line options.''' | |
datetime_regex = r'\d{4}[-.]?\d{2}[-.]?\d{2} \d{2}:\d{2}:\d{2}(?:,\d{3})?' | |
program_name = os.path.basename(sys.argv[0]) | |
program_version = "v0.1" | |
program_build_date = "%s" % __updated__ | |
program_version_string = '%%prog %s (%s)' % (program_version, program_build_date) | |
#program_usage = '''usage: spam two eggs''' # optional - will be autogenerated by optparse | |
program_longdesc = '''''' # optional - give further explanation about what the program does | |
program_license = "Copyright 2013 Otger Ballester (otger) \ | |
Licensed under the Apache License 2.0\nhttp://www.apache.org/licenses/LICENSE-2.0" | |
if argv is None: | |
argv = sys.argv[1:] | |
try: | |
# setup option parser | |
parser = OptionParser(version=program_version_string, epilog=program_longdesc, description=program_license) | |
parser.add_option("-d", action="store_true", dest="color_date", help="color date times in format: yyyy-mm-dd HH:MM:SS") | |
parser.add_option("-l", default=10, dest="nlines", type="int", help="show last lines from file [default: %default]") | |
# process options | |
(opts, args) = parser.parse_args(argv) | |
if len(args) < 1: | |
raise CLIError('Path to file must be provided') | |
file_path = args[0] | |
if not os.path.exists(file_path): | |
raise CLIError('File ({0}) does not exist'.format(file_path)) | |
if len(args) > 1: | |
regexes = args[1:] | |
else: regexes = [] | |
if opts.color_date: regexes.insert(0, datetime_regex) | |
color_mod = len(colour_list) | |
## requires python 2.7+ | |
## color_regexes = {regex: colours[colour_list[ix%color_mod]] for ix, regex in enumerate(regexes)} | |
color_regexes = dict((regex, colours[colour_list[ix%color_mod]]) for ix, regex in enumerate(regexes)) | |
# MAIN BODY # | |
fp = open(file_path, 'r') | |
#fp.seek(0, 2) | |
fp = seek_last_n_lines_position(fp, opts.nlines) | |
if regexes: | |
while True: | |
new = fp.readline() | |
# Once all lines are read this just returns '' | |
# until the file changes and a new line appears | |
if new: | |
sys.stdout.write(regexes_coloring(new, regexes, color_regexes)) | |
else: | |
time.sleep(0.05) | |
else: | |
while True: | |
new = fp.readline() | |
# Once all lines are read this just returns '' | |
# until the file changes and a new line appears | |
if new: | |
sys.stdout.write(new) | |
else: | |
time.sleep(0.05) | |
except KeyboardInterrupt: | |
##Clean exit when pressing Ctrl + c | |
sys.stdout.write('\n') | |
return 0 | |
except Exception, e: | |
import traceback | |
indent = len(program_name) * " " | |
sys.stderr.write(program_name + ": " + repr(e) + "\n") | |
sys.stderr.write(indent + " for help use --help\n") | |
if DEBUG: | |
print "\nTraceback:" | |
traceback.print_exc() | |
return 2 | |
if __name__ == "__main__": | |
sys.exit(main()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment