Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Quick-and-dirty IRCCloud log export tool
import requests
import json
import sys
from collections import deque
from concurrent.futures import ThreadPoolExecutor as Pool
from datetime import datetime
from os import mkdir
from os.path import isdir
def debug(*args, **kwargs):
print(*args, file=sys.stderr, **kwargs)
relevant_events = {'joined_channel', 'you_joined_channel', 'parted_channel',
'you_parted_channel', 'kicked_channel', 'you_kicked_channel',
'quit', 'quit_server', 'nickchange', 'you_nickchange',
'user_channel_mode', 'channel_mode', 'notice', 'buffer_msg',
'buffer_me_msg', 'channel_topic'}
program, email, password, dest = sys.argv
debug('authenticating as %s with password %s' % (email, password))
auth_response =
'email': email,
'password': password
cookies = {'session': auth_response.json()['session']}
debug('authenticated: %r' % cookies)
initial_buffer_events = []
debug('getting stream' % cookies)
stream = requests.get('', cookies=cookies, stream=True)
for line in stream.iter_lines():
message = json.loads(line.decode('utf-8'))
debug('message: %s' % message['type'])
if message['type'] == 'oob_include':
debug('requesting buffers from %s' % message['url'])
buffers = requests.get('' + message['url'], cookies=cookies)
initial_buffer_events = buffers.json()
class Connection:
def __init__(self, name, id): = name = id
self.channels = []
self.queries = []
self.console = None
def add_buffer(self, buffer):
if buffer.type == 'console':
self.console = buffer
elif buffer.type == 'channel':
elif buffer.type == 'conversation':
class Buffer:
def __init__(self, name, id, type, start_of_history=0): = name = id
self.type = type
self.start_of_history = start_of_history = deque()
def add_event(self, event):
if isinstance(initial_buffer_events, dict):
connections = {}
buffers = {}
for event in initial_buffer_events:
if event['type'] == 'makeserver':
connections[event['cid']] = Connection(event['name'], event['cid'])
elif event['type'] == 'makebuffer':
buffer = Buffer(event['name'], event['bid'], event['buffer_type'], start_of_history=event['min_eid'])
buffers[event['bid']] = buffer
elif event['type'] in relevant_events:
def fill_buffer(cid, buffer):
buffer_events = True
if len( > 0:
beforeid =[0]['eid']
beforeid = None
while buffer_events:
debug('requesting events before %r on %r (%r) ...' % (beforeid,,
params = {
'cid': cid,
'num': 1000,
if beforeid: params['beforeid'] = beforeid
buffer_events = requests.get(
buffer_events = buffer_events.json()
for event in reversed(buffer_events):
if event['type'] in relevant_events:
if buffer_events:
beforeid = buffer_events[0]['eid']
beforeid = 0
if buffer.start_of_history ==[0]['eid']: buffer_events = []
for cid, connection in connections.items():
pool = Pool(16)
for x in buffer: fill_buffer(cid, buffer), connection.channels + connection.queries + [connection.console]):
for connection in connections.values():
debug('writing logs for %r (cid %r)' % (,
cbase = '%s/%s' % (dest,
for buffer_type in ('channels', 'queries'):
if not isdir(cbase): mkdir(cbase)
if not isdir('%s/%s' % (cbase, buffer_type)): mkdir('%s/%s' % (cbase, buffer_type))
for buffer in getattr(connection, buffer_type):
debug('writing logs for %r (bid %r)' % (,
bbase = '%s/%s/%s' % (cbase, buffer_type,
if not isdir(bbase): mkdir(bbase)
for event in
time = datetime.fromtimestamp(event['eid'] / 10 ** 6)
with open('%s/%s.txt' % (bbase, time.strftime('%Y-%m-%d')), 'a', encoding='utf-8') as f:
def log(line):
print('%s %s' % (time.strftime('%H:%M:%S'), line), file=f)
if event['type'] in {'buffer_msg', 'notice'}:
log('<%s> %s' % (event['from'] if 'from' in event else event['server'], event['msg']))
elif event['type'] == 'buffer_me_msg':
log('* %s %s' % (event['from'] if 'from' in event else event['server'], event['msg']))
elif event['type'] in {'joined_channel', 'you_joined_channel'}:
log('*** %s (%s) has joined %s' % (event['nick'], event['hostmask'], event['chan']))
elif event['type'] in {'parted_channel', 'you_parted_channel'}:
log('*** %s (%s) has parted %s (%s)' % (event['nick'], event['hostmask'], event['chan'], event['msg']))
elif event['type'] in {'kicked_channel', 'you_kicked_channel'}:
log('*** %s kicked %s from %s (%s)' % (event['kicker'], event['nick'], event['chan'], event['msg']))
elif event['type'] in {'quit', 'quit_server'}:
log('*** %s has quit (%s)' % (event['nick'], event['msg']))
elif event['type'] in {'nickchange', 'you_nickchange'}:
log('*** %s is now known as %s' % (event['oldnick'], event['newnick']))
elif event['type'] == 'channel_mode':
log('*** %s sets mode %s' % (event['from'] if 'from' in event else event['server'], event['diff']))
elif event['type'] == 'user_channel_mode':
log('*** %s sets mode %s %s' % (event['from'] if 'from' in event else event['server'], event['diff'], event['nick']))
elif event['type'] == 'channel_topic':
log('*** %s changed the topic to: %s' % (event['author'], event['topic']))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.