目的
slackを使ってて標準で備えてない機能をBOTに実現させて便利につかいたい
できるようになったことは目次参照
作ったBOT
今ある機能
常駐
- slack上の投稿を監視してキーワードに反応し様々なアクションを返すようにした
https://github.com/hadacchi/slackagent/blob/master/agent.py#L270-L304- BOTのように常時接続させてメッセージを監視する場合はReal Time Messaging API | Slackを使う
- rtm_connectは自動接続機能をつけると謎の断に悩まされずに済んでオススメ
- sleep(1)しないとアクセス過多でslackから弾かれるよ
def startListen(self):
'connect to slack and start to listening'
realname, uname, uid = self._auth_test()
imid = self._get_im()
self._sc = slackclient.SlackClient(privatedata.bot_user_oauth_access_token)
if self._sc.rtm_connect(with_team_state=False, auto_reconnect=True):
while True:
try:
buf = self._sc.rtm_read()
except Exception as e:
self._logger.exception(e)
raise e
if len(buf) == 0:
time.sleep(1)
continue
for res in buf:
if 'type' in res:
if res['type'] == 'user_typing':
if random.uniform(0,1) > 0.8:
self._random_message(res['channel'], configuration.WAITING_MSGS)
continue
if res['type'] == 'message': # 発言形式を取るメッセージは全部これ
if res.get('channel') == imid[uname]:
# BOTへのprivate ch
if res.get('subtype') == 'bot_message':
continue
self._search_for_pic(res)
if 'subtype' not in res and 'attachments' not in res:
self._chooseAction(res['channel'], res)
else:
self._logger.debug(str(res))
time.sleep(1)
self._logger.info('now, disconeccted')
メッセージの一括削除
-
指定したchのメッセージを，最新10件を残して全て削除する
- 削除はchat.delete
-
BOTから見えるIMも削除対象として選択できる
- channels.listだけでなく，im.listも引いてくるのがポイント
https://github.com/hadacchi/slackagent/blob/master/agent.py#L165-L201
- channels.listだけでなく，im.listも引いてくるのがポイント
def _clear_log(self, fch, chname):
'clear log of the indicated ch'
chid = self._get_ch()
imid = self._get_im()
realname, uname, uid = self._auth_test()
if chname not in chid:
if chname not in imid:
if chname != realname or uname not in imid:
self._random_message(fch, configuration.NOCH_MSGS)
return
else:
targetid = imid[uname]
else:
targetid = imid[chname]
else:
targetid = chid[chname]
self._random_message(fch, configuration.CONFIRM_MSGS)
history, more = self._get_history(fch, targetid)
while True:
for msg in history[10:]:
r = requests.post( slackparams.del_url
, params = { 'token': privatedata.oauth_access_token
, 'channel': targetid
, 'ts': msg['ts']
, 'as_user': True
}
)
time.sleep(1)
if more:
history, more = self._get_history(fch, targetid)
else:
break
self._random_message(fch, configuration.FINISH_MSGS)
- im.listではユーザIDとIMのID（多分，2者通話なのでユーザIDペア毎に生成される）しか得られない
- users.listでユーザ名とユーザIDのペアを取得
- BOT側を固定すればユーザ毎にIMは一意に特定されるので，ユーザ名⇒IM IDの辞書を使うことでユーザ名で指定可能とする
https://github.com/hadacchi/slackagent/blob/master/agent.py#L98-L129
def _get_im(self):
'return a dict from uname to imid'
try:
with open(configuration.impkl, 'rb') as f:
ims = pickle.load(f)
except FileNotFoundError:
r = requests.get( slackparams.imlist_url
, params = {'token': privatedata.oauth_access_token}
)
response = json.loads(r.text)
if response['ok']:
ims = {res['user']:res['id'] for res in response['ims']}
with open(configuration.impkl,'wb') as f:
pickle.dump(ims,f)
return ims
else:
raise Exception(r.text)
try:
with open(configuration.usrpkl, 'rb') as f:
usrs = pickle.load(f)
except FileNotFoundError:
r = requests.get( slackparams.usrlist_url
, params = {'token': privatedata.oauth_access_token}
)
response = json.loads(r.text)
if response['ok']:
usrs = {res['name']:res['id'] for res in response['members']}
with open(configuration.usrpkl,'wb') as f:
pickle.dump(usrs,f)
else:
raise Exception(r.text)
- 画像付メッセージの画像を保存
- メッセージごと保存してもいいんだけど，多分読み返すことはないので画像だけ保存する機能にしてみた
https://github.com/hadacchi/slackagent/blob/master/agent.py#L246-L256
- メッセージごと保存してもいいんだけど，多分読み返すことはないので画像だけ保存する機能にしてみた
def _store_pic(self, uname, obj):
'画像添付を保存する'
r = requests.get(obj.get('image_url'))
if r.status_code == 200:
dname = configuration.IMGBASE + uname + '/'
try:
os.mkdir(dname)
except FileExistsError:
pass
with open(dname + obj.get('image_url').split('/')[-1], 'wb') as f:
f.write(r.content)