IRC Bot
UnalloBot is currently under complete re-write using the Eggdrop IRC bot engine. All functions will be re-implemented as eggdrop modules in C or TCL/TK. These modules may call external scripts or programs in other languages as needed.
UnalloBot is an IRC bot that facilitates a number of functions for Unallocated Space. It started as a way to check to see if the space was open and grew from there.
Related: IRC & Minecraft
- Version 2.5 Major addition to 2.5 is the ability to add new commands and edit old commands on the fly by editing botfunc.py then echoing "update" into the irc named pipe.
- Version 2.7 Saw the inclusion of command approximation and the removal of several command aliases which can still be used via the aforementioned approximation (Levenshtein)
bot.py - this file no longer exists???
#!/usr/bin/env python
import socket,urllib,sys,threading,time,serial,botfunc,os
class pipein(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
threading.Thread.daemon = True
def redcheck(self,data):
if data not in cflist:
cflist.append(data)
if len(cflist)>=3: cflist.pop(0) #why is this one less than it looks like?
return True
else:
return False
def run (self):
global irc
try:
while True:
tmp=sys.stdin.readline().strip()
if tmp != "":
if tmp == "update":
reload(botfunc)
elif pipein().redcheck(tmp):
irc.send('PRIVMSG #unallocatedspace :\001ACTION '+tmp+'\001\r\n')
print tmp.strip()
#time.sleep(1)
except:
print "Something fucked up! Threaded edtion."
print sys.exc_info()[1]
cflist=[]
while True:
try:
# network = 'irc.choopa.net'
# network = 'efnet.port80.se'
# network = 'chat.efnet.org'
network = 'irc.shoutcast.com'
port=6667
irc=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
irc.connect((network,port))
# irc = socket.create_connection(("irc.paraphysics.net","6667"))
irc.send('NICK UnalloBot\r\n')
irc.send('USER UnalloBot UnalloBot UnalloBot :Unallocated Bot\r\n')
irc.send('VERSION 2\r\n')
time.sleep(15)
irc.send('JOIN #unallocatedspace\r\n')
# print irc.recv(4096)
# print irc.recv(4096)
# print irc.recv(4096)
# print irc.recv(4096)
pipein().start()
while True:
data=irc.recv(4096)
print(data)
if data.find('PING')!=-1:
irc.send ('PONG '+data.split()[1]+'\r\n')
# time.sleep(2)
elif data.find('PRIVMSG #unallocatedspace :!')!=-1:
data=data[data.find(' :!')+3:].strip().replace("'",'')
data=data.replace('`','')#move to one liner eventually
command,u,data=data.partition(" ")
command=command.lower()
if command in botfunc.commands:
botfunc.send(irc,botfunc.commands[command](data)) #command exists, run it
else:
botfunc.send(irc,botfunc.commands[botfunc.closest(botfunc.commands,command)](data)) #command does not exist, approximte if possible
elif data.find('PRIVMSG #unallocatedspace :')!=-1:
os.system("echo '.' > traffic_monitor"); #place a dot in the traffic file to let stoplight monitor know we had activity.
except:
print "Something fucked up!"
print sys.exc_info()[1]
time.sleep(30)
botfunc.py
####################################################################
# botfunc.py Unallocated Space Last edit: 7/18/2013
# Created: Whodafuqkno
# TOFIX: Document functions better.
# Recent Changes: Added function descriptions and basic handling
# documentation
# Fixed tweet function with new API
# When editing this file, please put your handle and date below
# for rudimentary change tracking purposes.
# 7/18/2013 - Hunterkll
####################################################################
####################################################################
# Imports: Socket, urllib, time, serial, os, string, sys, pywapi
# pywapi - http://code.google.com/p/python-weather-api/ - weather
# tweepy - http://tweepy.github.io/ - twitter API access
# All others - basic communication needs
####################################################################
import socket,urllib,time,serial,os,string,sys,pywapi,tweepy
from xml.dom import minidom
from random import choice
####################################################################
# Help Function - returns available commands to IRC channel upon
# invocation of !help. !help <parameter> returns the help text
# of that specific command
####################################################################
def help(data):
data=data.replace('!','')
helps={
'status':'!status returns the current status of the space. (open/closed)',
'sign':'!sign returns or updates the text displayed on the prolite LED sign.',
'tweet':'!tweet returns the latest tweet on the @Unallocated twitter account.',
'site':'!site returns the latest blog post on http://unallocatedspace.org/',
'rollcall':'!rollcall lists Unallocated Space members that have checked into the space with their UAS member smart cards during the current session (opening to closing)',
'phone':'!phone prints the phone number of the space. 512-943-2827. ooh how meta.',
'weather':'!weather returns the weather conditions outside the space.',
'address':address(None),
'request':'!request is used to make improvment or feature requests for UnalloBot.',
'wiki':'!wiki when used with text returns a link most closely related to the search term on the Unallocated Wiki.',
'link':'!link returns various links to Unallocated Space related web pages. List can be altered here: http://www.unallocatedspace.org/wiki/Links',
'video':'!video queries the Unallocated Space youtube account and returns the top result.',
'nowplaying':'!nowplaying returns the current music track information playing on the sound system',
'alert':'!alert plays a short series of flashes on the stop sign'
}
if data in helps:
return helps[data]
elif data in help_alias and help_alias[data] in helps:
ret="!%s is an alias. "%data
return ret+helps[help_alias[data]]
else:
return 'Available commands are: !'+' !'.join(helps.keys())
####################################################################
# Returns the open/close status of the space upon invocation of
# !status in the IRC channel. Parameter is ignored.
# reads from /tmp for now, need to get this read from a tcp socket
####################################################################
def status(data):
return open('/tmp/status').read()[1:]
####################################################################
# Return the latest tweet from the UAS twitter account upon
# invocatino of !tweet in the IRC channel. Parameter is ignored.
# Pulls from twitter API as per twitter API docs
####################################################################
def tweet(data):
# xmldoc=url_get('https://api.twitter.com/1/statuses/user_timeline.rss?screen_name=unallocated','dom')
# return "Last Tweet: "+xmldoc.getElementsByTagName("item")[0].getElementsByTagName("title")[0].toxml()[20:-8]
My_consumer_key='-redacted-'
My_consumer_secret='-redacted-'
My_access_token_key='-redacted-'
My_access_token_secret='-redacted-'
Tauth = tweepy.OAuthHandler(My_consumer_key, My_consumer_secret)
Tauth.set_access_token(My_access_token_key, My_access_token_secret)
api = tweepy.API(Tauth)
mentions = api.mentions_timeline(count=1)
#print tweet.text
#print "Tweet URL: https://twitter.com/%s/status/%s" % (tweet.author.screen_name, tweet.id)
for mention in mentions:
return "Last Tweet: "+mention.author.screen_name+" said: "+mention.text
return "Twitter access is not currently working"
####################################################################
# Pulls latest post from the UAS irc site upon invocation of !site
# in the IRC channel. Parameter is ignored. Pulls in via RSS feed
# returning only the first item.
####################################################################
def site(data):
xmldoc=url_get('http://www.unallocatedspace.org/uas/feed/rss/','dom')
title=xmldoc.getElementsByTagName('item')[0].getElementsByTagName('title')[0].toxml()[7:-8]
link= xmldoc.getElementsByTagName('item')[0].getElementsByTagName('link')[0].toxml()[6:-7]
return 'Last Post: %s - %s' % (title, link)
####################################################################
# Returns currently playing song from the music system. upon
# invocation of !nowplaying in IRC. Broken - no remote machine to
# pull data from. Unknown remote machine code/implementation.
####################################################################
def nowplaying(data):
try:
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('-redacted-',8000))
message=s.recv(1024)
s.close()
except socket.error:
message="The music machine is not on, or is not responding."
return ''.join([x for x in message if ord(x) < 128]).strip()
####################################################################
# Returns current sign data in IRC upon invocatino of !sign.
# Parameter is sent over tcp socket to listening daemon to write to
# sign. Need to review remote code.
####################################################################
def sign(data):
try:
if data=="":
message='The last sign update read as: '+open('/tmp/sign','r').read()
else:
if '<FO>' in data:
message="<FO> is not allowed"
else:
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('-redacted-',9001))
s.sendall(data)
message=s.recv(1024)
s.close()
if sys.argv[0] == "/uas/responder/imap_responder.py" and "Updating sign to " in message:
os.system("echo 'Mobile update to sign: "+message+"' > /uas/irc/irc ")
except socket.error:
message="Failed to update sign"
return message
####################################################################
# Returns current weather data for the space's location to irc
# upon invocation of !weather. Parameter is ignored. Uses python
# weather api referenced in header to return 2 sources of data.
####################################################################
def weather(data):
noaa_result = pywapi.get_weather_from_noaa('KBWI')
yahoo_result = pywapi.get_weather_from_yahoo('21144')
data+= "Yahoo says: " + string.lower(yahoo_result['condition']['text']) + " and " + yahoo_result['condition']['temp'] + "C now at the space." + " And our friend NOAA reports that: " + string.lower(noaa_result['weather']) + " and " + noaa_result['temp_f'] + "F now at the space"
return data
####################################################################
# Returns list of UAS related links to irc upon invocation of
# !link. Pulls from a wiki page and returns the data to irc.
# Need to test and verify this code to see functionality & document
####################################################################
def links(data):
data=data.lower()
xmldoc = url_get('http://www.unallocatedspace.org/wiki/Links','dom')
results=xmldoc.getElementsByTagName('p')[1].getElementsByTagName('a')
for dat in results:
if data in str(dat.toxml()).lower().split('{0}')[1:-1]:
return 'Link for '+data+': '+str(dat.getAttribute('href'))
return 'Nothing could be found.'
####################################################################
# Returns a single search result for the passed parameter for
# videos in the unallocated space youtube account upon invocation
# of !youtube <parameter>. Need to verify youtube API for correct
# way to handle this type of search. Is there a youtube API?
# TOFIX: check to see if we're even doing this properly?
####################################################################
def youtube(data):
if data != '':
xmldoc = url_get('https://gdata.youtube.com/feeds/api/videos?author=TheUnallocated&max-results=1&q=%s' % data, 'dom')
if '>0<' not in xmldoc.getElementsByTagName('openSearch:totalResults')[0].toxml():
title=xmldoc.getElementsByTagName('entry')[0].getElementsByTagName('title')[0].toxml()[19:-8]
link =xmldoc.getElementsByTagName('entry')[0].getElementsByTagName('link')[0].getAttribute('href')[0:-22]
return title+': '+link
return 'Nothing has been found'
else:
return 'Use this command with text to query the Unallocated Space youtube account for video.'
####################################################################
# Returns the currently checked in members of the space to irc
# upon invocation of !rollcall. Currently reads from a flatfile
# database at a fixed path. Checks to see if space is open first
#
# TOFIX: Read from a socket instead. Completely seperate checkin
# from bot and just pull data over TCP instead. Reduce reliance on
# Filesystem paths/locations.
####################################################################
def rollcall(data):
status=open("/tmp/status").read()
if status[0:1]=="+":
peps=open("/uas/checkin/ciusers")
peeeps=[]
for line in peps:
peeeps.append(line.split(':')[0])
peeps=', '.join(peeeps)
if peeps.strip() != "":
per=(1.0*sum(1 for line in open('/uas/checkin/ciusers'))/sum(1 for line in open('/uas/checkin/users'))*100)
per=int(per) if data[0:2] != '-v' else per
return "The following people have checked into the space this session. "+peeps[0:]+" - "+str(per)+"% of carded members."
else:
return "No one has checked into the space this session."
else:
return "The space is closed, Rollcall is not allowed"
###################################################################
# Returns a summary of current events, conditions, and info about
# the space to irc upon invocation of !goingson. Currently
# calls other functions of the bot to return current data.
#
# TOFIX: passing null to weather causes a crash. currently
# returning an empty string instead.
# TOFIX: reduce sleep times between function calls to lowest
# sane level without causing bot disconnection or crashes, yet
# leave enough safety margin to combat said effects. Increased
# to 0.5 from 0.1 during debugging.
###################################################################
def goingson(data):
os.system("echo '"+status("")+"' > /uas/irc/irc ")
time.sleep(.5)
os.system("echo '"+rollcall("")+"' > /uas/irc/irc ")
time.sleep(.5)
os.system("echo '"+sign("")+"' > /uas/irc/irc ")
time.sleep(.5)
os.system("echo '"+site(None)+"' > /uas/irc/irc ")
time.sleep(.5)
os.system("echo '"+tweet("")+"' > /uas/irc/irc ")
time.sleep(.5)
os.system("echo '"+weather("")+"' > /uas/irc/irc ")
return ""
###################################################################
# Returns the first result of a search of our wiki, the search
# query being the text following the invocation of !wiki in the irc
# channel. The invocation here simply passes the query string into
# a pre-formed search URL. Returned result is scraped via a search
# of a specific HTML tag and the following link.
#
# Currently functional with no known bugs.
###################################################################
def wiki(data):
if data:
data=url_get('http://www.unallocatedspace.org/wik/index.php?title=Special:Search&search=%s&&fulltext=Search' % data)
if "There were no results matching the query." in data:
return "Nothing could be found."
elif "<ul class='mw-search-results'>\n<li><a href=" in data:
data=data[data.find("<ul class='mw-search-results'>\n<li><a href=")+44:]
data=data[0:data.find('" title')]
return 'Closest Match: http://www.unallocatedspace.org'+data
return 'Something strange occured.'
else:
return 'Use this command with text to search the wiki.'
###################################################################
# Send in a feature request. Merely takes the passed data if it's
# not a null string and appends it to the requests.txt file in the
# bot's current exeuction directory.
#
# TOFIX: Should make the bot path aware and have subdirectories
# or specific fixed paths for things like requests file, data
# files read in, or named pipes.
###################################################################
def request(data):
if data != "":
f=open('requests.txt','a')
f.write(data+"\n\n")
f.close()
return "Thank you for the feature request."
else:
return "Use this command with text to request new features."
################################################################
# Sends a flashing light pattern alert on the traffic light
# hanging inside the space.
# Calls an external python script via a system call to handle
# this.
################################################################
def alert(data):
os.system("python /uas/ippower/alert.py &")
return "Alert Sent."
################################################################
# Extra definition for roll call. Just calls the rollcall
# Function.
################################################################
def trollcall(data):
trollface=rollcall(data)
return trollface.replace('the space this session. ','your mom. ')
################################################################
# Return the phone number to the space
################################################################
def phone(data):
return "Call us at 512-943-2827!"
################################################################
# Return the address to the space
################################################################
def address(data):
return "512 Shaw CT Suite 105, Severn MD 21144"
################################################################
# Magic 8ball! Upon invocation of !8ball - if the parameter in
# question has a ? mark in it and is not empty, return
# a random result from the choices defined below.
################################################################
def eightball(data):
if data!='' and '?' in data:
return choice(['It is certain.','It is decidedly so.','Without a doubt.','Yes. definitely.','You may rely on it.','As I see it, yes.','Most likely.','Outlook good.','Signs point to yes.','Yes.','Reply hazy, try again.','Ask again later.','Better not tell you now.','Cannot predict now.','Concentrate and ask again.','Don\'t count on it.','My reply is no.','My sources say no.','Outlook not so good.','Very doubtful.'])
else:
return 'I can do nothing unless you ask me a question....'
#######################################################
# Aliases, give multiple names to single commands
#######################################################
def command_aliaser(commandz):
helpals={}
commands={}
for k in commandz:
for c in k.split():
commands[c]=commandz[k]
helpals[c]=getattr(commandz[k],'func_name')
return [helpals,commands]
help_alias,commands = command_aliaser({
'status space':status,
'rollcall':rollcall,
'tweet':tweet,
'site blog':site,
'sign':sign,
'phone':phone,
'address':address,
'weather temp':weather,
'goingson allthethings':goingson,
'request':request,
'alert':alert,
'help commands':help,
'wiki':wiki,
'links':links,
'video youtube':youtube,
'nowplaying music song':nowplaying,
'trollcall':trollcall,
'8ball eightball magic8ball magiceightball':eightball
})
responder_commands={'status':status,'rollcall':rollcall,'tweet':tweet,'site':site,'sign':sign,'weather':weather,'address':address}
#########################################################
# Send text to IRC.
#########################################################
def send(irc, text):
if text.strip() != "":
irc.send('PRIVMSG #unallocatedspace :\001ACTION '+str(text).strip()+'\001\r\n')
#irc.send('PRIVMSG #unallocatedspace :\001ACTION '+tmp+'\001\r\n')
#########################################################
# Retrieve a URL and return the data from the page
#########################################################
def url_get(data,type="raw"):
dat = urllib.urlopen(data)
if type == "dom":
data=minidom.parse(dat)
else:
data=dat.read()
dat.close()
return data
#########################################################
# Reload the bot code for any updates - can be done while
# the bot is live.
#########################################################
def update(data):
reload(botfunc)
#########################################################
# Match A to B length precision.
#########################################################
def levenshtein(a,b):
n, m = len(a), len(b)
if n > m:
a,b = b,a
n,m = m,n
current = range(n+1)
for i in range(1,m+1):
previous, current = current, [i]+[0]*n
for j in range(1,n+1):
add, delete = previous[j]+1, current[j-1]+1
change = previous[j-1]
if a[j-1] != b[i-1]:
change = change + 1
current[j] = min(add, delete, change)
return current[n]
##########################################################
# Find the closest match to 50% of the command and execute
# if such a match exists. !porncall matches 50% of
# !rollcall so rollcall will be executed
##########################################################
def closest(coms,u):
match = min(coms,key=lambda v:len(set(u)^set(v)))
return match if levenshtein(u, match) < len(match)/2 else "help"