Some very basic code to search for a content in Quassel logs

quassel_logs.py

# -*- coding: utf-8 -*-
"""
Very basic code to search within logs from Quassel IRC client

Tested on Quassel v0.10.0 (dist-575f27e)
"""

import sqlite3
import os
import json
import time

LOGS_DBFILE = os.path.expanduser("~/.config/quassel-irc.org/quassel-storage.sqlite")
SERVER_HOSTNAME = 'irc.freenode.net'
CHANNEL = "#ubuntu" # Can be a username for its private messages
CONTENT_TO_SEARCH = "test"  # Used in WHERE with LIKE %content%

def get_user_name(userid, usernames):
    """Return display name from given userid"""
    return usernames.get(userid, 'unknown')

def timestamp_to_time(timestamp):
    return time.strftime("[%H:%M.%S]", time.localtime(timestamp))

def timestamp_to_datetime(timestamp):
    return time.strftime("[%a %b %d %H:%M:%S %Y]", time.localtime(timestamp))


def get_server_id(conn, hostname):
    """Return finded id for given hostname"""
    object_id = None
    for serverid,adminid,networkid,hostname in conn.execute("SELECT serverid,userid,networkid,hostname FROM ircserver"):
        if hostname == SERVER_HOSTNAME:
            object_id = serverid

    assert (object_id is None) == False

    return object_id


def get_usernames(conn):
    """Return a Dict of users id:name"""
    usernames = {}
    for senderid, sendername in conn.execute("SELECT * FROM sender"):
        name = sendername.split('!~')[0]
        usernames[senderid] = name

    return usernames


def get_buffers(conn):
    """Return a Dict of buffers id:name"""
    buffers = {}
    for bufferid,userid,networkid,buffername,buffercname in conn.execute("SELECT bufferid,userid,networkid,buffername,buffercname FROM buffer"):
        buffers[bufferid] = buffername

    return buffers


def get_buffers_for_channel(conn, buffers, channel):
    channel_buffers = []
    for k,v in buffers.items():
        if v.strip() == channel.strip():
            channel_buffers.append(k)
    return channel_buffers

def get_messages_for_buffer(conn, usernames, bufferids, content=None, limit=(0,300)):
    """Return a list of message dicts for given buffer list ids"""
    messages = []
    queryset = "SELECT messageid,time,bufferid,type,flags,senderid,message FROM backlog"
    queryset += ' WHERE type=1'
    if content is not None:
        queryset += " AND message LIKE '%{}%'".format(content)
    if limit:
        queryset += ' LIMIT {},{}'.format(limit[0], limit[1])

    for messageid,msgtime,bufferid,msgtype,flags,senderid,message in conn.execute(queryset):
        messages.append({
            'messageid': messageid,
            'datetime': timestamp_to_datetime(msgtime),
            'bufferid': bufferid,
            'msgtype': msgtype,
            'flags': flags,
            'senderid': get_user_name(senderid, usernames),
            'message': message
        })

    return messages


conn = sqlite3.connect(LOGS_DBFILE)
c = conn.cursor()


# Serverid to search for
SERVERID = get_server_id(conn, SERVER_HOSTNAME)

# Relations needed to display usernames
_USERNAMES = get_usernames(conn)


# Buffer are equivalent to the irc client windows, so a buffer can be a window
# on channel #lambda, or message from hostname or a window for an user private
# messages. Since message are parented with buffer, they are needed to crawl.
_BUFFER = get_buffers(conn)

channel_buffers = get_buffers_for_channel(conn, _BUFFER, CHANNEL)

messages = get_messages_for_buffer(conn, _USERNAMES, channel_buffers, content=CONTENT_TO_SEARCH, limit=None)
#print json.dumps(messages, indent=4)
for item in messages:
    print item['datetime'], item['senderid'], ':', item['message']
print
print "Count:", len(messages)


c.close()