#!/usr/bin/python -tt## SILAN# Simple IRC Log Analyzer# http://slackforge.com/silan/## Version 0.0.1# An extremely simple IRC log analyzer --;# Only supports dircproxy logs, because that's what I use## Copyright (c) 2007 Shidayu <shidayu@gmail.com>## Permission is hereby granted, free of charge, to any person obtaining a copy# of this software and associated documentation files (the "Software"), to deal# in the Software without restriction, including without limitation the rights# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell# copies of the Software, and to permit persons to whom the Software is# furnished to do so, subject to the following conditions:## The above copyright notice and this permission notice shall be included in# all copies or substantial portions of the Software.## THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN# THE SOFTWARE.### MODULE IMPORTS#import osimport reimport statimport sysimport globimport datetimeimport stringimport time## CONSTANTS## Application information_APP_VERSION = '0.0.1'# Regular expressions_RE_TIMESTAMP = '\[(\d\d) ([A-Z][a-z][a-z]) (\d\d):(\d\d)\]'_RE_LINESPLIT = '^' + _RE_TIMESTAMP + ' ([^\n\r\f\v]*)$' # Group up log line_RE_SELFNICK = '--- You changed your nickname to (\S*)'_RE_NICKCHANGE = '--- (\S*) \([^\t\n\r\f\v)]*\) changed nickname to (\S*)'_RE_QUIT = '<-- (\S*) \([^\t\n\r\f\v)]*\) quit from IRC: ([^\n\r\f\v]*)$'_RE_JOIN = '--> (\S*) \([^\t\n\r\f\v)]*\) joined the channel$'_RE_PART = '<-- (\S*) \([^\t\n\r\f\v)]*\) left the channel$'_RE_MESSAGE = '<([^ \t\n\r\f\v!]*)![^\t\n\r\f\v>]*> ([^\n\r\f\v]*)$'_RE_DESCRIBE = '\* (\S*) ([^\n\r\f\v]*)'_RE_TOPIC = '--- (\S*) \([^\t\n\r\f\v)]*\) changed topic: ([^\t\n\r\f\v]*)$'_RE_KICK = '<-- (\S*) kicked off by (\S*) \([^\t\n\r\f\v)]*\): ([^\n\r\f\v]*)$'_RE_KICKSELF = '<-- Kicked off by (\S*) \([^\t\n\r\f\v)]*\): ([^\n\r\f\v]*)$' # When "I" get kicked_RE_NICEWORD = '[^a-zA-Z0-9]*'_RE_CHANNEL = '^(\S*).log$' # Parse name of channel from filename of log# File pattern matching (for glob)_GLOB_SERVERLOG = 'Server.log'_GLOB_CHANLOG = '#*.log'# Months to translate from dircproxy's format_MONTHS = { 'Jan': 1, 'Feb': 2, 'Mar': 3, 'Apr': 4, 'May': 5, 'Jun': 6, 'Jul': 7, 'Aug': 8, 'Sep': 9, 'Oct': 10, 'Nov': 11, 'Dec': 12 }## DISPLAY PROGRAM INFORMATION#print '\n\nSILAN', _APP_VERSION, '- Simple IRC Log Analyzer - http://slackforge.com/silan/'print 'Copyright (c) 2007 Shidayu <shidayu@gmail.com>'## CLASSES## UserChannelActivity: Encapsulate activity for one user in one channelclass UserChannelActivity: """Activity data for one user in one channel""" def __init__(self): """Default constructor""" self.channel = None self._universal_defaults() def __init__(self, channel): """Constructor with optional channel instance provided""" self.channel = channel self._universal_defaults() def _universal_defaults(self): self.messages = [] self.describes = [] self.topics = [] self.joincount = 0 self.partcount = 0 self.words = {} self.kicks_delivered = [] self.kicks_received = [] def increment_word(self, word): wordregex = re.compile(_RE_NICEWORD) for nicerword in wordregex.split(word): nicerword = string.lower(nicerword) if nicerword not in self.words: self.words[nicerword] = 1 else: self.words[nicerword] += 1# User: Track IRC activity for a particular userclass User: """IRC activity for a particular user""" def __init__(self): """Default constructor""" self.nick_primary = None self.nick_alternates = [] self.channels = {} self.quit_messages = [] def __init__(self, primary): """Constructor with optional primary nickname specified""" self.nick_primary = primary self.nick_alternates = [] self.channels = {} self.quit_messages = [] def is_known_as(self, alternate): """Return 1 if this IRCUser could be known as the provided argument""" if alternate == self.nick_primary or alternate in self.nick_alternates: return 1 else: return 0# Channel: Track IRC activity for a particular channelclass Channel: """IRC activity for a particular channel""" def __init__(self): """Default constructor""" self.name = None self.log = None self._universal_defaults() def __init__(self, args): """Constructor with optional arguments specified A Channel can be instantiated with a string argument representing the channel's name, or with a tuple argument in the format of (channelname, logfilename) """ if isinstance(args, tuple): self.name = args[0] self.log = Log(args[1]) else: self.name = args[0] self.log = None self._universal_defaults() def _universal_defaults(self): self.users = {} self.topics = [] self.activity_total = 0 self.activity_bymonth = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0} self.activity_byday = {0: 0, 1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0} self.activity_byhour = {0: 0, 1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0, 13: 0, 14: 0, 15: 0, 16: 0, 17: 0, 18: 0, 19: 0, 20: 0, 21: 0, 22: 0, 23: 0} def add_new_topic(self, dttm, user, topic): if len(self.topics) == 0 or self.topics[len(self.topics) - 1] != topic: self.topics.append(HistoricalStatement(dttm, user, topic)) def most_active_day_count(self): top = 0 for day in self.activity_byday: if self.activity_byday[day] > top: top = self.activity_byday[day] return top def most_active_hour_count(self): top = 0 for hour in self.activity_byhour: if self.activity_byhour[hour] > top: top = self.activity_byhour[hour] return top# StampedStatement: Topics, describes, and messages fall under this categoryclass StampedStatement: """A timestamped bit of information such as a topic, describe, and message""" def __init__(self): """Default constructor""" self.dttm = None self.text = None def __init__(self, dttm, text): """Constructor with optional date/time and text provided""" self.dttm = dttm self.text = text# HistoricalStatement: StampedStatment including a user instanceclass HistoricalStatement(StampedStatement): """Extended StampedStatement including user instance for topic history storage""" def __init__(self): """Default constructor, calls parent constructor""" StampedStatement.__init__(self) self.user = None def __init__(self, dttm, user, text): """Constructor with optional date/time, user instance, and text provided""" StampedStatement.__init__(self, dttm, text) self.user = user# Log: About an IRC logclass Log: """IRC log file information and contents""" def __init__(self, filename): """Constructor, filename of log must be passed as an argument The log file will immediately be read into self.contents """ self.filename = filename self.dttm_start = self._find_start() self.lines = self._prepare_log() def _read_log(self): """Simply reads in a text file as a list and returns it""" print "Reading from file", self.filename + "..." sys.stdout.flush() loghandler = open(self.filename, 'r') loglist = loghandler.readlines() loghandler.close() return loglist def _first_line(self): """Read in the first line of the log and return it""" loghandler = open(self.filename, 'r') firstline = loghandler.readline() # readline is SINGULAR! loghandler.close() return firstline def _find_start(self): """Determines the time the log started by combining the date/time the file was created and the date/time of the first line in the log """ logstat = os.stat(self.filename) logctime = logstat.st_ctime logcdttm = datetime.datetime.fromtimestamp(logctime) logyear = logcdttm.year firststamp = re.compile('^' + _RE_TIMESTAMP + ' .*$').match( self._first_line()) # See CONSTANTS above for what's in _RE_TIMESTAMP firststampday = int(firststamp.group(1)) firststampmonth = int(convert_month(firststamp.group(2))) # Turns 'Oct' into '10' firststamphour = int(firststamp.group(3)) firststampminute = int(firststamp.group(4)) return datetime.datetime(logyear, firststampmonth, firststampday, firststamphour, firststampminute) def _prepare_log(self): """Loops through all lines in the log, splits the timestamp out, and returns a new list of StampedStatement()s with the line's text and a 'nice' date/time timestamp """ original = self._read_log() stampedlist = [] dttm_last = self.dttm_start print "Processing timestamps from", self.filename + "..." sys.stdout.flush() splitregex = re.compile(_RE_LINESPLIT) # See CONSTANTS above for what's in _RE_LINESPLIT for line in original: splitmatch = splitregex.match(line) if not splitmatch: sys.exit('\nError: Could not find timestamp in log entry. Are timestamps properly enabled?\n') lineday = int(splitmatch.group(1)) linemonth = int(convert_month(splitmatch.group(2))) linehour = int(splitmatch.group(3)) lineminute = int(splitmatch.group(4)) linetext = splitmatch.group(5) # # Takes a fairly good guess as to whether a year has advanced in # the log. This may not always be accurate if, say, a year or more # passes between entries in the log. # if linemonth < dttm_last.month or (linemonth == dttm_last.month and lineday < dttm_last.day): # The preceding: Does it look like we've gone back in time? dttm_last = datetime.datetime(dttm_last.year + 1, linemonth, lineday, linehour, lineminute) else: dttm_last = datetime.datetime(dttm_last.year, linemonth, lineday, linehour, lineminute) stampedlist.append(StampedStatement(dttm_last, linetext)) return stampedlist## FUNCTIONS#def conjure_user(allusers, nick): """Return user instance a nickname belongs to. If none, one is created.""" existinguser = locate_nick(allusers, nick) if not existinguser: newuser = User(nick) return newuser else: return existinguserdef conjure_user_with_alt(allusers, nick1, nick2): """Return User() instance one or both of the nicks belong to. If none, one is created with nick1 being primary and nick2 being an alternate. If there exists one but either nick1 or nick2 is not listed as the primary or an alternate, it is added to the list of alternates. """ existinguser = locate_nick(allusers, nick1) if not existinguser: existinguser = locate_nick(allusers, nick2) if not existinguser: newuser = User(nick1) newuser.nick_alternates.append(nick2) return newuser else: if existinguser.nick_primary == nick1 and not existinguser.is_known_as(nick2): existinguser.nick_alternates.append(nick2) elif existinguser.nick_primary == nick2 and not existinguser.is_known_as(nick1): existinguser.nick_alternates.append(nick1) return existinguserdef locate_nick(allusers, nick): """Attempts to locate a nickname and return its User instance. If not there, None.""" if nick in allusers: return allusers[nick] else: for user in allusers: if allusers[user].is_known_as(nick): return allusers[user] return Nonedef convert_month(month): """Takes in three-letter month from dircproxy logs and returns two-digit number.""" return _MONTHS[month]def find_channels(): """Searches current working directory for all channel logs and returns a Dictionary of Channel()s based on this. """ chanlogs = glob.glob(_GLOB_CHANLOG) # See CONSTANTS above for what's in _GLOB_CHANLOG # Remove logs which are not regular files for filename in chanlogs: filestat = os.stat(filename) filemode = filestat.st_mode if not stat.S_ISREG(filemode): chanlogs.remove(filename) # Return None if there are... none if len(chanlogs) < 1: return None else: chandict = {} chanregex = re.compile(_RE_CHANNEL) for filename in chanlogs: channame = chanregex.match(filename).group(1) chandict[channame] = Channel((channame, filename)) return chandictdef conjure_chanact(user, channel): """Looks for a UserChannelActivity() belonging to channel in the an instance of User(). If one is not found, a new one is created. """ if channel in user.channels: return user.channels[channel] else: return UserChannelActivity(channel)## INITIALIZATION#print "Locating server log..."sys.stdout.flush()serverfile = glob.glob(_GLOB_SERVERLOG)[0]if not serverfile: sys.exit('\nError: Unable to locate server log file in the current directory.\n')server = Log(serverfile)print "Locating channel logs..."sys.stdout.flush()channels = find_channels()if not channels: sys.exit('\nError: Unable to locate any channel log files in the current directory.\n')users = {}selfnick = Nonenickchgregex = re.compile(_RE_NICKCHANGE)selfnickregex = re.compile(_RE_SELFNICK)quitregex = re.compile(_RE_QUIT)print "Parsing lines of server log..."sys.stdout.flush()for line in server.lines: if nickchgregex.match(line.text): ncrmatch = nickchgregex.match(line.text) oldnick = ncrmatch.group(1) newnick = ncrmatch.group(2) processed_user = conjure_user_with_alt(users, oldnick, newnick) users[processed_user.nick_primary] = processed_user elif selfnickregex.match(line.text): snrmatch = selfnickregex.match(line.text) newself = snrmatch.group(1) if not selfnick: selfnick = newself selfuser = conjure_user(users, selfnick) if selfuser.nick_primary != selfnick: if selfuser.nick_primary in users: users.remove(selfuser) if selfuser.is_known_as(selfnick): selfuser.nick_alternates.remove(selfnick) selfuser.nick_alternates.append(selfuser.nick_primary) selfuser.nick_primary = selfnick if selfnick not in users: users[selfnick] = selfuser else: processed_user = conjure_user_with_alt(users, selfnick, newself) users[selfnick] = processed_user elif quitregex.match(line.text): qrmatch = quitregex.match(line.text) quitnick = qrmatch.group(1) quitmsg = qrmatch.group(2) processed_user = conjure_user(users, quitnick) processed_user.quit_messages.append(StampedStatement(line.dttm, quitmsg))joinregex = re.compile(_RE_JOIN)partregex = re.compile(_RE_PART)msgregex = re.compile(_RE_MESSAGE)descregex = re.compile(_RE_DESCRIBE)topicregex = re.compile(_RE_TOPIC)kickregex = re.compile(_RE_KICK)kickselfregex = re.compile(_RE_KICKSELF)for channel in channels: print "Parsing lines of", channel, "channel log..." sys.stdout.flush() for line in channels[channel].log.lines: if msgregex.match(line.text): mrmatch = msgregex.match(line.text) nick = mrmatch.group(1) message = mrmatch.group(2) processed_user = conjure_user(users, nick) processed_chanact = conjure_chanact(processed_user, channel) processed_chanact.messages.append(StampedStatement(line.dttm, message)) for word in string.split(message): processed_chanact.increment_word(word) elif descregex.match(line.text): drmatch = descregex.match(line.text) nick = drmatch.group(1) describe = drmatch.group(2) processed_user = conjure_user(users, nick) processed_chanact = conjure_chanact(processed_user, channel) processed_chanact.describes.append(StampedStatement(line.dttm, describe)) for word in string.split(describe): processed_chanact.increment_word(word) elif kickregex.match(line.text): krmatch = kickregex.match(line.text) recipient = krmatch.group(1) deliverer = krmatch.group(2) reason = krmatch.group(3) processed_recipient = conjure_user(users, recipient) processed_deliverer = conjure_user(users, deliverer) recipient_chanact = conjure_chanact(processed_recipient, channel) deliverer_chanact = conjure_chanact(processed_deliverer, channel) recipient_chanact.kicks_received.append(HistoricalStatement(line.dttm, processed_deliverer, reason)) deliverer_chanact.kicks_delivered.append(HistoricalStatement(line.dttm, processed_recipient, reason)) elif kickselfregex.match(line.text): ksrmatch = kickselfregex.match(line.text) deliverer = ksrmatch.group(1) reason = ksrmatch.group(2) processed_recipient = conjure_user(users, selfnick) processed_deliverer = conjure_user(users, deliverer) recipient_chanact = conjure_chanact(processed_recipient, channel) deliverer_chanact = conjure_chanact(processed_deliverer, channel) recipient_chanact.kicks_received.append(HistoricalStatement(line.dttm, processed_deliverer, reason)) deliverer_chanact.kicks_delivered.append(HistoricalStatement(line.dttm, processed_recipient, reason)) elif topicregex.match(line.text): trmatch = topicregex.match(line.text) nick = trmatch.group(1) topic = trmatch.group(2) processed_user = conjure_user(users, nick) processed_chanact = conjure_chanact(processed_user, channel) processed_chanact.topics.append(StampedStatement(line.dttm, topic)) channels[channel].add_new_topic(line.dttm, processed_user, topic) elif joinregex.match(line.text): jrmatch = joinregex.match(line.text) nick = jrmatch.group(1) processed_user = conjure_user(users, nick) processed_chanact = conjure_chanact(processed_user, channel) processed_chanact.joincount += 1 elif partregex.match(line.text): prmatch = partregex.match(line.text) nick = prmatch.group(1) processed_user = conjure_user(users, nick) processed_chanact = conjure_chanact(processed_user, channel) processed_chanact.partcount += 1 if msgregex.match(line.text) or descregex.match(line.text) or topicregex.match(line.text) or joinregex.match(line.text) or partregex.match(line.text): processed_user.channels[channel] = processed_chanact users[processed_user.nick_primary] = processed_user channels[channel].users[processed_user.nick_primary] = processed_user if kickregex.match(line.text) or kickselfregex.match(line.text): processed_recipient.channels[channel] = recipient_chanact processed_deliverer.channels[channel] = deliverer_chanact users[processed_recipient.nick_primary] = processed_recipient users[processed_deliverer.nick_primary] = processed_deliverer channels[channel].users[processed_recipient.nick_primary] = processed_recipient channels[channel].users[processed_deliverer.nick_primary] = processed_deliverer if msgregex.match(line.text) or descregex.match(line.text) or topicregex.match(line.text) or joinregex.match(line.text) or partregex.match(line.text) or kickregex.match(line.text) or kickselfregex.match(line.text): channels[channel].activity_total += 1 channels[channel].activity_bymonth[line.dttm.month] += 1 channels[channel].activity_byhour[line.dttm.hour] += 1 channels[channel].activity_byday[line.dttm.weekday()] += 1for channel in channels: chanhtml = channel + '.html' print "Generating statistics page for", channel, " (" + chanhtml +")..." sys.stdout.flush() chanlogstart = channels[channel].log.dttm_start out_startdate = channels[channel].log.dttm_start.strftime('%A %B %d, %Y') out_enddate = channels[channel].log.lines[len(channels[channel].log.lines) - 1].dttm.strftime('%A %B %d, %Y') out_logdays = (channels[channel].log.lines[len(channels[channel].log.lines) - 1].dttm - channels[channel].log.dttm_start).days out_numusers = len(channels[channel].users) dotw_mostactivity = channels[channel].most_active_day_count() hour_mostactivity = channels[channel].most_active_hour_count() chanout = open(chanhtml, 'w') chanout.write("<!DOCTYPE html PUBLIC \"-//W3C//DTD XHTML 1.0 Strict//EN\" \"http://www.w3.org/TR/xhtml1/DTD/xhtml1.dtd\">\<html xmlns=\"http://www.w3.org/1999/xhtml\" xml:lang=\"en\">\<head>\ <meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\" />\ <link rel=\"stylesheet\" href=\"styles/silan.css\" type=\"text/css\" />\ <title>Statistics for " + channel + "</title>\</head>\<body>\ <div id=\"header\">\ <h1>Statistics for " + channel + "</h1>\ <p>Generated from logs recorded by " + selfnick + "</p>\ <p>Statistics cover a " + str(out_logdays) + " day period between " + out_startdate + " and " + out_enddate + "</p>\ <p>" + str(out_numusers) + " user(s) participated in " + channel + " throughout the period covered</p>\ </div>\ <div id=\"main\">\ <div class=\"section\">\ <h2>Activity by Day of the Week</h2>") if dotw_mostactivity != 0: chanout.write("\ <table class=\"activity_graph\">\ <tr>") for day in [6, 0, 1, 2, 3, 4, 5]: chanout.write("\ <td>\ <div class=\"column\">\ <div class=\"gap\" style=\"height: " + str(100 - (int(float(channels[channel].activity_byday[day]) / float(dotw_mostactivity) * 100))) + "px;\">\ </div>\ <div class=\"bar\" style=\"height: " + str(int(float(channels[channel].activity_byday[day]) / float(dotw_mostactivity) * 100)) + "px;\">\ " + str(channels[channel].activity_byday[day]) + "\ </div>\ <div class=\"info\">\ " + {6: 'Sun', 0: 'Mon', 1: 'Tue', 2: 'Wed', 3: 'Thu', 4: 'Fri', 5: 'Sat'}[day] + "\ </div>\ </div>\ </td>") chanout.write("\ </tr>\ </table>") else: chanout.write("\ <p>There was no activity in this channel!</p>") chanout.write("\ </div>\ <div class=\"section\">\ <h2>Activity by Hour (" + time.tzname[0] + ")</h2>") if hour_mostactivity != 0: chanout.write("\ <table class=\"activity_graph\">\ <tr>") for hour in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23]: chanout.write("\ <td>\ <div class=\"small_column\">\ <div class=\"gap\" style=\"height: " + str(100 - (int(float(channels[channel].activity_byhour[hour]) / float(hour_mostactivity) * 100))) + "px;\">\ </div>\ <div class=\"bar\" style=\"height: " + str(int(float(channels[channel].activity_byhour[hour]) / float(hour_mostactivity) * 100)) + "px;\">\ " + str(channels[channel].activity_byhour[hour]) + "\ </div>\ <div class=\"info\">\ " + str(hour) + "\ </div>\ </div>\ </td>") chanout.write("\ </tr>\ </table>") else: chanout.write("\ <p>There was no activity in this channel!</p>") chanout.write("\ </div>\ <div id=\"footer\">\ <p class=\"pageinfo\"></p>\ </div>\</body>\</html>") chanout.close()