1
0
Fork 0

Compare commits

...

5 commits
v0.1 ... master

Author SHA1 Message Date
fanir fc1a9a04d5 Added helper function msgsplit
Added command fefe
Added MapQuest and Fefe to .help
Minor fix
2014-03-21 09:57:47 +01:00
fanir c079093af9 Added command MapQuest
Some minor fixes
2014-03-19 10:20:23 +01:00
fanir f7289a1ac4 added commands help and forecast
implemented usage of bot-nick as cmdprefix
changed min time for waiting between channel privmsgs to 0.1sec
changed the way pircbot.query() chooses between NOTICE or PRIVMSG
some minor fixes and commenting
2014-03-07 15:21:38 +01:00
fanir 421c206719 redone logging of exceptions
added more try ... except
added choose command
done some commenting
2014-03-06 16:44:14 +01:00
fanir e1b0e83e98 redone configuration: is now in an external file, parsed with configobj
redone logging, now using module logging of the standard library
added argument "checkconf" for viewing the configuration
added configuration options for external apis
added option to wait between channel privmsgs
added DuckDuckGo command
prepared Forecast command
2014-03-05 15:44:04 +01:00
3 changed files with 461 additions and 171 deletions

5
.gitignore vendored Normal file
View file

@ -0,0 +1,5 @@
*.conf
__pycache__
*.pyc
*.pyo
*.nogit

95
bot.conf.example Normal file
View file

@ -0,0 +1,95 @@
###########
### BOT ###
###########
[Network]
# The Server name to connect to. Duh!
Server = fanir.de
# "Default" is 6667. An often used port for SSL would be 6697, if SSL would be
# supported. Maybe in a future, far far away...
Port = 6667
# Serverpassword. Empty in most cases.
ServerPasswd =
# A comma-seperated list of channels to join.
# The channels must be enclosed with quotation marks ("")!
Channels = "#bots", "#your_channel"
# The encodings to try when getting messages from IRC. Will be tried in the given order.
# The first one is used to encode data when sending stuff.
# The list given shoud do just fine in most networks, I assume.
# Also comma seperated.
Encodings = utf-8, latin-1, iso-8859-1, cp1252
[Bot]
# The list of nicknames to try. If the first one is not aviable, it will
# try the second, and so on...
# You should specfy at least two nicks.
Nicknames = chalkbot, chalkbot_, chalkbot__
# Also known as username. Some IRC-internal.
# By default, the nickname will be used as ident.
Ident = chalkbot
Realname = A ChalkBot Instance
# Command for registering with nickserv, without leading slash.
NickservCommand =
Modes = +iB
[Permissions]
# List of users (hostmasks as regex) and the commands they are allowed to execute.
# * can be used instead of commands to allow every command. You should append a trailing comma.
# All command names should be lowercase. Add "help" as a command if you want the user
# to see all aviable commands in the help message (is automatically included with "*").
yournickname\!yourusername@.* = *,
\!@some other user = join, part, invite
.*\!murderer@(localhost|127(\.0){2}\.1) = die,
[Behavior]
# The prefix for commands for the bot.
CommandPrefix = !
# Which way should be used to speak to users?
# "" (Nothing) means the type of the incoming message should be used.
# One of: NOTICE, PRIVMSG, "" (Nothing)
QueryType =
# With how much information do you want to be annoyed?
# DEBUG spams most, FATAL least. WARNING should be a good tradeoff.
# One of: DEBUG, INFO, WARNING, ERROR, CRITICAL
Loglevel = WARNING
# Time to wait between two PRIVMSGS in seconds.
# Can prevent the bot from running into flood limits.
MsgWaitTime = 0.1
# The time the parser for incoming data should wait between each attempt to read new data in seconds.
# High values will certainly make the bot reply slowly while very low values increads cpu load and therefore will perform badly on slow machines.
# You should keep it between 1 and 0.001 seconds.
# For gods sake, don't ever set it to 0!
ParserWaitTime = 0.05
#########################
### EXTERNAL SERVICES ###
#########################
[DuckDuckGo]
Active = 1
[Forecast.io]
Active = 0
ApiKey = your_api_key
[MapQuest]
Active = 1

532
main.py
View file

@ -26,55 +26,68 @@
# SETTINGS CAN BE FOUND AT THE BOTTOM OF THE FILE
import sys, string, socket, re, signal
import sys, string, socket, re, signal, json, logging
from random import choice
from time import sleep
from time import sleep, time, strftime, localtime
from select import poll, POLLIN, POLLPRI, POLLOUT,\
POLLERR, POLLHUP, POLLNVAL
from threading import Thread, Event
from urllib.request import urlopen
from urllib.parse import quote_plus
from configobj import ConfigObj
bot = None
class log():
DEBUG, INFO, WARNING, ERROR, FATAL, SILENT =\
0, 1, 2, 3, 4, 5
levels = {
0: "[DEBUG] ", 1: "[INFO] ", 2: "[WARNING] ",
3: "[ERROR] ", 4: "[FATAL] ", 5: "[SILENT] "
}
def show(level, msg):
if level in range(log.DEBUG, log.SILENT):
if LOGLEVEL <= level: print("".join((log.levels[level], msg)))
else:
raise ValueError("That's not a loglevel!")
logging.basicConfig(format="[%(asctime)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
class pircbot():
def __init__(self, nicknames, server, port=6667,
ident="pircbot", realname="pircbot", serverpasswd="",
encodings=("utf-8", "latin-1"), users="",
query_type="", command_prefix=".",
parser_wait_time=0.1):
self.server = server
self.port = port
self.serverpasswd = serverpasswd
self.encodings = encodings
def __init__( self,
server,
nicknames,
ident = "pircbot",
realname = "pircbot",
port = 6667,
serverpasswd = "",
encodings = ("utf-8", "latin-1"),
query_type = "",
command_prefix = "!",
msg_wait_time = 0,
parser_wait_time = 0.1,
users = "",
duckduckgo_cfg = {"Active": "0"},
forecast_cfg = {"Active": "0"},
mapquest_cfg = {"Active": "0"},
logger = logging.getLogger(logging.basicConfig())
):
self.log = logger
self.nicknames = nicknames
self.ident = ident
self.realname = realname
self.users = users
self.cmdprefix = command_prefix
self.server = server
self.port = int(port)
self.serverpasswd = serverpasswd
self.encodings = encodings
self.query_type = query_type.upper()
self.cmdprefix = command_prefix
self.msg_wait_time = float(msg_wait_time)
if self.msg_wait_time < 0.1:
self.log.info("msg_wait_time ist zu klein, nutze 0.1 Sekunden")
self.msg_wait_time = 0.1
self.parser_wait_time = float(parser_wait_time)
self.parser_wait_time=parser_wait_time
self.users = users
self.user = {}
self.duckduckgo_cfg = duckduckgo_cfg
self.forecast_cfg = forecast_cfg
self.mapquest_cfg = mapquest_cfg
self.user = {"mask":"", "nick":"", "ident":"", "host":""}
self.socket = None
self.recvbuffer = bytearray(1024)
@ -86,45 +99,54 @@ class pircbot():
self.ready = False
self.mode_reply = {}
self.last_msg_ts = time()
def connect(self):
# connect
self.socket = socket.socket()
log.show(log.DEBUG, "--- SOCKET OPENING ---")
self.log.debug("--- SOCKET OPENING ---")
try:
self.socket.connect((self.server, self.port))
except socket.error as e:
log.show(log.FATAL, "Fehler: %s" % e)
return
except Exception as e:
self.log.critical("Fehler beim Verbinden: %s", e)
self.disconnect(send_quit=False)
# start getting data
self.recvloop = Thread(target=self.recv, name="recvloop")
self.recvloop.start()
try:
self.recvloop = Thread(target=self.recv, name="recvloop")
self.recvloop.start()
except Exception as e:
self.log.critical("Fehler beim Starten des Empfangs von Daten vom IRC: %s", e)
self.disconnect(send_quit=False)
# optionally send a server password
if self.serverpasswd != "": self.send("PASS %s" % self.serverpasswd)
# get a nick
self.send("NICK %s" % self.nicknames.pop(0))
# set user data
self.send("USER %s 0 * :%s" % (self.ident, self.realname))
try:
# optionally send a server password
if self.serverpasswd != "": self.send("PASS %s" % self.serverpasswd)
# get a nick
self.send("NICK %s" % self.nicknames.pop(0))
# set user data
self.send("USER %s 0 * :%s" % (self.ident, self.realname))
except Exception as e:
self.log.critical("Fehler beim IRC-Handshake: %s", e)
self.disconnect(send_quit=False)
# implements irc command QUIT and (more or less) clean exiting
def disconnect(self, reason="", send_quit=True):
if send_quit:
try:
self.send("QUIT :%s" % reason)
sleep(1)
except: pass
try: self.send("QUIT :%s" % reason)
except Exception as e: self.log.critical("Fehler beim Senden der QUIT-Nachricht: %s", e)
sleep(1)
self.die_event.set()
ctr = 0
while self.recvloop.is_alive() and self.parseloop.is_alive() and ctr < 15:
ctr += 1
sleep(1)
log.show(log.DEBUG, "--- SOCKET CLOSING ---")
self.log.debug("--- SOCKET CLOSING ---")
try:
self.socket.shutdown(socket.SHUT_RDWR)
self.socket.close()
except: pass
except Exception as e: self.log.warning("Fehler beim Schließen des Sockets: %s", e)
### threaded functions ###
@ -134,25 +156,34 @@ class pircbot():
"""
Loop for reciving data
"""
log.show(log.DEBUG, "--- RECVLOOP STARTING ---")
self.parseloop = Thread(target=self.parser, name="parser")
self.log.debug("--- RECVLOOP STARTING ---")
try: self.parseloop = Thread(target=self.parser, name="parser")
except Exception as e:
self.log.error("Fehler beim Vorbereiten des Threads zum Parsen von Nachrichten von IRC: %s", e)
p = poll()
p.register(self.socket.fileno(), POLLIN)
try: p.register(self.socket.fileno(), POLLIN)
except Exception as e:
self.log.critical("Fehler, der eigentlich nie auftreten sollte: %s", e)
while not self.die_event.is_set():
ap = p.poll(1000)
if (self.socket.fileno(), POLLIN) in ap:
self.recvbuffer.extend(self.socket.recv(1024))
try: self.recvbuffer.extend(self.socket.recv(1024))
except Exception as e:
self.log.critical("Konnte keine Daten vom IRC lesen. Verbindung tot? Fehler: %s", e)
if not self.parseloop.is_alive():
self.parseloop = Thread(target=self.parser, name="parser")
self.parseloop.start()
log.show(log.DEBUG, "--- RECVLOOP EXITING ---")
try:
self.parseloop = Thread(target=self.parser, name="parser")
self.parseloop.start()
except Exception as e:
self.log.critical("Fehler beim Starten des Threads zum Parsen von Nachrichten vom IRC: %s", e)
self.log.debug("--- RECVLOOP EXITING ---")
# loop for parsing incoming data
def parser(self):
"""
Loop for parsing incoming data
"""
log.show(log.DEBUG, "--- PARSELOOP STARTING ---")
self.log.debug("--- PARSELOOP STARTING ---")
while not self.die_event.is_set():# and self.recvbuffer.endswith(b"\r\n"):# != b"":
if self.recvbuffer.endswith(b"\r\n"):
# get and decode line from buffer
@ -179,17 +210,21 @@ class pircbot():
params = head
params.append(larg.strip())
log.show(log.DEBUG, " > %s" % rawline)
self.log.debug(" > %s" % rawline)
# PING
if command == "PING":
self.send("PONG %s" % params[0])
# PRIVMSG and NOTICE
elif command == "PRIVMSG" or command == "NOTICE":
if params[1][0] == self.cmdprefix:
args = []
for v in params[1][1:].split(" "):
if v!="": args.append(v)
elif command == "PRIVMSG" or command == "NOTICE" and self.ready:
args = []
for v in params[1].split(" "):
if v!="": args.append(v)
if len(args)>1 and args[0][0:len(self.user["nick"])] == self.user["nick"]:
args.pop(0)
args[0] = "".join((self.cmdprefix, args[0]))
if args[0][0] == self.cmdprefix:
args[0] = args[0][1:]
rp = self.on_command(command, prefix, origin, params[0], args[0].lower(), args[1:])
if rp not in (None, ""):
self.reply(origin, params[0], rp, command)
@ -213,11 +248,11 @@ class pircbot():
self.send("NICK %s" % self.nicknames.pop(0))
# KILL
elif command == "KILL":
log.show(log.WARNING, "Got killed by %s: %s" % (params[0], params[1:]))
self.log.warning("Got killed by %s: %s" % (params[0], params[1:]))
self.disconnect(send_quit=False)
else:
sleep(self.parser_wait_time)
log.show(log.DEBUG, "--- PARSELOOP EXITING ---")
self.log.debug("--- PARSELOOP EXITING ---")
### helper functions ###
@ -237,12 +272,18 @@ class pircbot():
return textstring.decode(self.encodings[0], 'ignore')
# splits messages into parts with a specific maximal length
def msgsplit(self, msg, length=400):
return [msg[i:i+length] for i in range(0, len(msg), length)]
# checks if a given user may execute a given command
def check_privileges(self, usermask, command):
def check_privileges(self, usermask, command, log=True):
for user, privs in self.users.items():
if re.search(user, usermask, re.IGNORECASE) != None:
if command.lower() in privs or "*" in privs:
return True
if log: self.log.info("Unauthorized call of %s from user %s", command, usermask)
return False
@ -259,12 +300,12 @@ class pircbot():
cnt = 0
while not self.ready and cnt<retry_times: sleep(interval)
if not self.ready:
log.show(log.WARNING, "Connection did not get ready in time (%sx %s seconds), \"%s\" not executed" % (retry_times, interval, command))
self.log.warning("Connection did not get ready in time (%sx %s seconds), \"%s\" not executed" % (retry_times, interval, command))
return True
exec("ret = %s" % command)
return False
else:
log.show(log.FATAL, "exec_on_ready() called with an invalid command: %s" % command)
self.log.critical("exec_on_ready() called with an invalid command: %s" % command)
return True
def exec_retry(self, command, times=5, interval=1, wait_for_ready=True):
@ -273,14 +314,14 @@ class pircbot():
while not self.ready: sleep(interval)
cnt = 0
exec("while %s and cnt<times: sleep(interval)" % command)
else: log.show(log.FATAL, "exec_retry() called with an invalid command.")
else: self.log.critical("exec_retry() called with an invalid command.")
def send(self, data):
log.show(log.DEBUG, "< %s" % data)
self.log.debug("< %s" % data)
try: self.socket.send(self.encode("".join((data, "\r\n"))))
except BrokenPipeError as e:
log.show(log.FATAL, e)
except Exception as e:
self.log.critical("Fehler beim Senden von Daten: %s", e)
self.disconnect(send_quit=False)
# decides whether to reply to user or to channel
@ -295,11 +336,17 @@ class pircbot():
# replies to channel by PRIVMSG
def chanmsg(self, channel, msg):
while self.last_msg_ts + self.msg_wait_time > time():
sleep(0.1)
self.last_msg_ts = time()
self.send("".join(("PRIVMSG ", channel, " :", msg)))
# replies to user by NOTICE or PRIVMSG
def query(self, nick, msg, in_query_type):
self.send("".join((self.query_type if self.query_type!="" else in_query_type, " ", nick, " :", msg)))
def query(self, nick, msg, in_query_type="", force_query_type=""):
self.send("".join((force_query_type if force_query_type!="" else
(self.query_type if self.query_type!="" else
(in_query_type if in_query_type!="" else
"PRIVMSG")), " ", nick, " :", msg)))
def nick(self, nick):
self.exec_on_ready("".join(('self.send("JOIN %s" % "', channel, '")')))
@ -330,37 +377,214 @@ class pircbot():
Command is the command for the bot.
Params contains a list of originally space separated parameters.
"""
# hello
if command == "hello":
self.log.info("Command from %s: %s", origin["mask"], command)
numparams = len(params)
### INTERNAL ###
# help [me]
if command == "help":
if numparams == 1 and params[0].startswith("me"):
rply = "\
Nope, you're on your own."
else:
rply = "\
Command Prefix is \"%s\" or \"%s\"\n\
\n\
Commands:\n\
hello [<...>]\n\
say <text>\n\
choose <choice1>, <choice2>[, <choice3>[, ...]] -- Let the bot decide!\n\
DuckDuckGo, ddg <query> -- Ask the DuckDuckGo Instant Answer API\n\
Forecast, fc <query> -- Query Forecast.io\n\
MapQuest, mq, OpenStreetMap, osm <query> -- Get lat and lon for a place\n\
Fefe [<id>] -- Show a given or the lastest post in Fefes Blog\n\
" % (
self.cmdprefix,
self.user["nick"]
)
if self.check_privileges(origin["mask"], command, log=False):
rply += " \n\
~ For the aristocrats ~\n\
join <channel>\n\
part <channel>\n\
mode ±<modes>\n\
die [<quitmsg>]"
for line in rply.splitlines(): self.query(origin["nick"], line[20:], force_query_type="PRIVMSG")
# hello [<...>]
elif command == "hello":
greeting = "".join(("Hi " + origin["nick"] +"!"))
return greeting
# say
if command == "say":
# say <text>
elif command == "say":
return " ".join(params)
# choose <choice1>, <choice2>[, <choice3>[, ...]]
elif command == "choose":
choices = " ".join(params).split(", ")# if numparams>1 else params.split(", ")
if choices[0] == "":
return "Whaddayawant? (Seperated by \", \")"
elif len(choices) == 1:
return "Such a difficult question... I don't know..."
else:
return choice(choices)
### EXTERNAL ###
# DuckDuckGo, ddg <query>
elif command in ("duckduckgo", "ddg") and self.duckduckgo_cfg["Active"] == "1":
if numparams==0:
return "You didn't ask anything..."
try: rp = urlopen("https://api.duckduckgo.com/?q=%s&format=json&no_html=1&no_redirect=1&t=pircbot:chalkbot"
% quote_plus(" ".join(params)))
except Exception as e:
self.log.error("Error while querying DuckDuckGo: %s" % e)
return "Error while querying DuckDuckGo: %s" % e
if rp.getcode() == 200:
used_fields = (
"Heading", "AbstractText", "AbstractSource", "AbstractURL",
"AnswerType", "Answer",
"Definition", "DefinitionSource", "DefinitionURL",
)
rj = json.loads(str(rp.readall(), "utf-8"))
empty_field_counter = 0
for elem in [v for v in used_fields if v in rj]:
if rj[elem] not in ("", []):
self.reply(origin, source, "%s: %s" % (elem, rj[elem]), in_query_type)
else:
empty_field_counter+=1
if empty_field_counter == len(used_fields):
return "No suitable reply from DuckDuckGo for query %s" % " ".join(params)
else:
return "(Results from DuckDuckGo <https://duckduckgo.com>)"
else:
return "Error while querying DuckDuckGo, got HTTP-Status %i" % rp.getcode()
# Forecast, fc <query>
elif command in ("forecast", "fc") and self.forecast_cfg["Active"] == "1":
if numparams==2:
try: rp = urlopen("https://api.forecast.io/forecast/%s/%s?units=si&exclude=minutely,hourly,daily"
% (self.forecast_cfg["ApiKey"], quote_plus(",".join(params))))
except Exception as e:
self.log.error("Error while querying Forecast.io: %s" % e)
return "Error while querying Forecast.io: %s" % e
if rp.getcode() == 200:
rj = json.loads(str(rp.readall(), "utf-8"))
self.reply(origin, source,
strftime("CURRENTLY (%d.%m.%Y %H:%M:%S)", localtime(rj["currently"]["time"])), in_query_type)
self.reply(origin, source,
"Summary: %s" % rj["currently"]["summary"], in_query_type)
self.reply(origin, source,
"Temperature: %s °C" % rj["currently"]["temperature"], in_query_type)
self.reply(origin, source,
"Apparent Temperature: %s °C" % rj["currently"]["apparentTemperature"], in_query_type)
self.reply(origin, source,
"Dew Point: %s °C" % rj["currently"]["dewPoint"], in_query_type)
self.reply(origin, source,
"Wind Speed: %s m/s" % rj["currently"]["windSpeed"], in_query_type)
self.reply(origin, source,
"Cloud Cover: %s %%" % (rj["currently"]["cloudCover"]*100), in_query_type)
self.reply(origin, source,
"Precipitation Probability: %s %%" % (rj["currently"]["precipProbability"]*100), in_query_type)
if "precipIntensity" in rj["currently"]: self.reply(origin, source,
"Precipitation Intensity: %s mm/h" % rj["currently"]["precipIntensity"], in_query_type)
if "precipType" in rj["currently"]: self.reply(origin, source,
"Precipitation Type: %s" % rj["currently"]["precipType"], in_query_type)
if "visibility" in rj["currently"]: self.reply(origin, source,
"Visibility: %s km" % rj["currently"]["visibility"], in_query_type)
self.reply(origin, source,
"Humidity: %s %%" % (rj["currently"]["humidity"]*100), in_query_type)
self.reply(origin, source,
"Pressure: %s hPa" % rj["currently"]["pressure"], in_query_type)
self.reply(origin, source,
"Ozone: %s DU" % rj["currently"]["ozone"], in_query_type)
if "nearestStormDistance" in rj["currently"]: self.reply(origin, source,
"Nearest Storm Distance: %s km" % rj["currently"]["nearestStormDistance"], in_query_type)
self.reply(origin, source,
"Sources: %s" % ", ".join(rj["flags"]["sources"]), in_query_type)
return "(Powered by Forecast <http://forecast.io/>)"
else:
return "Error while querying Forecast.io, got HTTP-Status %i" % rp.getcode()
else:
return "Usage: %s <lat> <lon>" % command
# MapQuest, mq, OpenStreetMap, osm <query>
elif command in ("mapquest", "mq", "openstreetmap", "osm") and self.mapquest_cfg["Active"] == "1":
if numparams==0:
return "You didn't ask anything..."
try: rp = urlopen("http://open.mapquestapi.com/nominatim/v1/search?q=%s&format=json"
% quote_plus(" ".join(params)))
except Exception as e:
self.log.error("Error while querying MapQuest: %s" % e)
return "Error while querying MapQuest: %s" % e
if rp.getcode() == 200:
used_fields = (
"display_name",
"lat", "lon"
)
rj = json.loads(str(rp.readall(), "utf-8"))
if len(rj) == 0:
return "No suitable reply from MapQuest for query %s" % " ".join(params)
rj = rj[0]
self.reply(origin, source, "Display Name: %s" % rj["display_name"], in_query_type)
self.reply(origin, source, "Lat: %s" % rj["lat"], in_query_type)
self.reply(origin, source, "Lon: %s" % rj["lon"], in_query_type)
return "(Nominatim Search Courtesy of MapQuest <http://www.mapquest.com/>)"
else:
return "Error while querying MapQuest, got HTTP-Status %i" % rp.getcode()
# Fefe <id>
elif command == "fefe":
if numparams>1:
return "Waddayawannasee? (The one and only optional argument needed is the ID of the blog post. If it is omitted, the last post will be shown.)"
else:
try:
if numparams==1:
rp = urlopen("http://blog.fefe.de/?ts=%s" % quote_plus(params[0]))
else:
rp = urlopen("http://blog.fefe.de/")
except Exception as e:
self.log.error("Error while querying Fefe: %s" % e)
return "Error while querying Fefe: %s" % e
if rp.getcode() == 200:
try:
rpd = str(rp.readall(), "utf-8")
m = re.search(r"<li>(.*?)<\/(?:li|ul)>", "".join(rpd.splitlines())).groups()[0]
print(m+"\n\n")
m = " ".join((re.split(r"<.+?>", m)))[6:]
if len(m)>400:
mmlen = 400
for l in self.msgsplit(m):
print(l)
self.reply(origin, source, l, in_query_type)
except Exception as e:
self.log.warning("Suspectious things happend while handling a response from fefes blog, but it's probably just an incorrect blogpost-id: %s" % e)
return "Nothing matched... (If you're sure your id was correct, see error log)"
### IRC ###
# join <channel>
elif command == "join":
if len(params)>0:
if numparams>0:
if self.check_privileges(origin["mask"], command): self.join(params[0])
else: self.query(origin["nick"], "You cannot do that!", in_query_type)
# part <channel>
elif command == "part":
if len(params)>0:
if numparams>0:
if self.check_privileges(origin["mask"], command): self.part(params[0])
else: self.query(origin["nick"], "You cannot do that!", in_query_type)
# mode ±<modes>
elif command == "mode":
if self.check_privileges(origin["mask"], command):
if len(params)==0:
if numparams==0:
self.mode_reply["to"] = origin["nick"]
self.mode_reply["type"] = in_query_type
self.get_modes()
else:
self.set_mode(" ".join(params) if len(params)>0 else params)
self.set_mode(" ".join(params) if numparams>0 else params)
else: self.query(origin["nick"], "You cannot do that!", in_query_type)
# die [<quitmsg>]
elif command == "die":
if self.check_privileges(origin["mask"], command):
self.disconnect("".join(params) if len(params)>0 else "".join((origin["nick"], " shot me, dying now... Bye...")))
self.disconnect("".join(params) if numparams>0 else "".join((origin["nick"], " shot me, dying now... Bye...")))
else: self.query(origin["nick"], "Go die yourself!", in_query_type)
else:
replies = [
@ -375,21 +599,56 @@ class pircbot():
def parseargs():
import argparse
p = argparse.ArgumentParser(
description = "i think my desc is missing")
p.add_argument("action", default="help", choices = ["start", "stop"], help="What to do?")
description = "guess what? i think my desc is still missing!")
p.add_argument("action",
default = "help",
choices = ["start", "stop", "checkconf"],
help = "What to do?")
p.add_argument("--loglevel", "-l",
choices = ["critical", "error", "warning", "info", "debug"],
help = "Verbosity of logging")
#p.add_argument("--daemon", "-d", type = bool, choices = [1, 0], default=1, help="Daemonize, Default: 1")
return p.parse_args()
def main():
# get the logging-module logger
log = logging.getLogger(__name__)
global bot
# parse command line and config
args = parseargs()
cfg = ConfigObj("bot.conf")
# set the loglevel
nll = getattr(logging, cfg["Behavior"]["Loglevel"].upper(), None)
if not isinstance(nll, int): raise ValueError('Invalid log level: %s' % cfg["Behavior"]["Loglevel"])
if args.loglevel != None: nll = getattr(logging, args.loglevel.upper(), None)
if not isinstance(nll, int):
raise ValueError('Invalid log level: %s' % args.loglevel)
log.setLevel(nll)
# do awesome stuff!
if args.action == "start":
try:
bot = pircbot(nicknames=NICKNAMES, ident=IDENT, realname=REALNAME,
server=SERVER, port=PORT, serverpasswd=SERVERPASSWD,
encodings=ENCODINGS, users=USERS,
query_type=QUERY_TYPE, command_prefix=COMMAND_PREFIX,
parser_wait_time=PARSER_WAIT_TIME)
bot = pircbot(
nicknames = cfg["Bot"]["Nicknames"],
ident = cfg["Bot"]["Ident"],
realname = cfg["Bot"]["Realname"],
server = cfg["Network"]["Server"],
port = cfg["Network"]["Port"],
serverpasswd = cfg["Network"]["ServerPasswd"],
encodings = cfg["Network"]["Encodings"],
query_type = cfg["Behavior"]["QueryType"],
command_prefix = cfg["Behavior"]["CommandPrefix"],
msg_wait_time = cfg["Behavior"]["MsgWaitTime"],
parser_wait_time = cfg["Behavior"]["ParserWaitTime"],
users = cfg["Permissions"],
duckduckgo_cfg = cfg["DuckDuckGo"],
forecast_cfg = cfg["Forecast.io"],
mapquest_cfg = cfg["MapQuest"],
logger = log,
)
bot.connect()
# wait for the bot to become ready
while bot.ready and not bot.die_event.is_set() == False:
@ -397,97 +656,28 @@ def main():
if not bot.die_event.is_set():
# set modes and join channels
bot.set_mode(MODES)
for channel in CHANNELS:
bot.set_mode(cfg["Bot"]["Modes"])
for channel in cfg["Network"]["Channels"]:
bot.join(channel)
# while bot is active, do nothing
while not bot.die_event.is_set():
sleep(1)
except KeyboardInterrupt:
log.show(log.INFO, "Got Ctrl-C, dying now...")
log.info("Got Ctrl-C, dying now...")
bot.disconnect("Ouch! Got shot by Ctrl-C, dying now... See you!")
except Exception as e:
log.exception("Fehler: %s", e)
bot.disconnect("Fehler: %s" % e)
log.debug("--- MAIN EXITING ---")
elif args.action == "stop": print("nope!")
log.show(log.DEBUG, "--- MAIN EXITING ---")
elif args.action == "checkconf":
for section, settings in cfg.items():
print("".join(("[", section, "]")))
for e in settings:
print("%20s : %s" % (e, settings[e]))
return 0
if __name__ == '__main__':
################
### SETTINGS ###
################
# Also known as username. Some IRC-internal.
# By default, the nickname will be used as ident.
IDENT = "chalkbot"
# The list of nicknames to try. If the first one is not aviable, it will
# try the second, and so on...
# You should specfy at least two nicks.
NICKNAMES = ["chalkbot", "chalkbot_", "chalkbot__"]
REALNAME = "A ChalkBot Instance"
# Command for registering with nickserv, without leading slash.
NICKSERVCMD = ""
MODES = "+iB"
# The Server name to connect to. Duh!
SERVER = "fanir.de"
#SERVER = "localhost"
# "Default" is 6667. An often used port for SSL would be 6697, if SSL would be
# supported. Maybe in a future, far far away...
PORT = 6667
# Serverpassword. Empty in most cases.
SERVERPASSWD = ""
# A comma-seperated list of channels to join, enclosed by braces.
CHANNELS = ["#bots"]
# The encodings to try when getting messages from IRC. Will be tried in the given order.
# The first one is used to encode data when sending stuff.
# The list given shoud do just fine in most networks, I assume.
# Also comma seperated and enclosed by braces.
ENCODINGS = ['utf-8', 'latin-1', 'iso-8859-1', 'cp1252']
# List of users (hostmasks as regex) and the commands they are allowed to execute.
# "*" Can be used instead of commands to allow every command.
# All command names should be lowercase.
USERS = {
"Fanir\!fanir@.*": ["*"],
"\!@some weird user that cannot exist": ["join", "part", "invite"],
}
# The prefix for commands for the bot.
COMMAND_PREFIX = "."
# Which way should be used to speak to users?
# "" means the type of the incoming message should be used.
# One of: "NOTICE", "PRIVMSG", ""
QUERY_TYPE = ""
# With how much information do you want to be annoyed?
# DEBUG spams most, FATAL least. WARNING should be a good tradeoff.
# One of: log.DEBUG, log.INFO, log.WARNING, log.ERROR, log.FATAL
LOGLEVEL = log.DEBUG
# The time the parser for incoming data should wait between each attempt to read new data in seconds.
# High values will certainly make the bot reply slowly while very low values increads cpu load and therefore will perform badly on slow machines.
# You should keep it between 1 and 0.001 seconds.
# For gods sake, don't set it to 0!
PARSER_WAIT_TIME = 0.05
#######################
### END OF SETTINGS ###
#######################
main()