-
Notifications
You must be signed in to change notification settings - Fork 2
/
slackbot.py
115 lines (94 loc) · 3.92 KB
/
slackbot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
# -*- coding: utf-8 -*-
from __future__ import division
from __future__ import unicode_literals
from collections import namedtuple
import json
import logging
import time
from collections import namedtuple
from slackclient import SlackClient
class SlackBot(object):
def __init__(self, settingsFilePath='settings.json'):
self.s = {}
self.sclient = {}
### Not necessary, moved to settings?
# self.botcheck = '' #<@' + s['bot']['id'] + '>: '
with file(settingsFilePath, 'r') as settingsfile:
self.s = json.load(settingsfile)
logging.debug(self.s)
self.lastping = int(time.time())
self.sclient = SlackClient(self.s["token"])
self.CommandBody = namedtuple('CommandBody', 'action help')
self.botcheck = '' # <@' + s['bot']['id'] + '>: '
self.commands = {}
#"""{ u'channel': u'G1FS1CJ84',
# u'team': u'T05311JTT',
# u'text': u'<@U1FRJ3WMU>: lol',
# u'ts': u'1465583194.000034',
# u'type': u'message',
# u'user': u'U0LJ6Q4S0'}""" ### Typical structure of a command packet
def help(self, msg):
output = self.sclient.api_call('chat.postMessage',
as_user='true',
channel=msg['channel'],
text=self.commands["help"].help)
logging.debug(output)
def generateHelp(self):
helptext = 'Commands are:\n'
for c in self.commands:
helptext += "\t" + self.botcheck + self.commands[c].help + "\n"
helptext += "\t" + self.botcheck + "help [this help text]\n"
self.commands['help'] = self.CommandBody(action=self.help, help=helptext)
def get_bot_id(self):
api_call = self.sclient.api_call("users.list")
if api_call.get('ok'):
# retrieve all users so we can find our bot
users = api_call.get('members')
for user in users:
if 'name' in user and user.get('name') == self.s["bot"]["name"]:
self.s["bot"]["id"] = user.get('id')
self.botcheck = '<@' + self.s['bot']['id'] + '>: '
return ({user['name']: user.get('id')})
else:
return "could not find bot user with the name " + s["bot"]["name"]
def autoping(self,last):
### hardcode the interval to 3 seconds
now = int(time.time())
if last + 3 < now:
self.sclient.server.ping()
return now
def addCommand(self, command, action, help):
self.commands[command] = self.CommandBody(action=action, help=help)
def sendReply(self, msg):
text = msg['text'][len(self.botcheck):]
pos = len(text)
try:
pos = text.index(' ')
except ValueError:
pass
cmd = text[:pos] # grab command string up to first space
logging.info('cmd ="' + cmd + '"')
if cmd in self.commands:
self.commands[cmd].action(msg)
def monitor(self):
#self.sclient = SlackClient(s["token"])
logging.info("Connecting as " + self.s["bot"]["name"])
if self.sclient.rtm_connect():
logging.info("...Connected!")
logging.debug(self.get_bot_id())
self.generateHelp()
last_ping = int(time.time())
while True:
messages = self.sclient.rtm_read()
# logging.debug(messages)
last_ping = self.autoping(last_ping)
for message in messages:
if all(k in message for k in ('type', 'text')) \
and message['type'] == 'message' \
and 'bot_id' not in message \
and self.botcheck in message['text']:
logging.debug(message)
self.sendReply(message)
time.sleep(1)
else:
logging.info("Connection Failed, invalid token?")