RokcetChatのREST APIをPythonでなしかいじったものです。
Public,Privateで使い分けが必要になります（いまいっぽ。。。）。なので利用者にはその使い分けを意識せずに情報をとれるようにしてみました。
原理としては
responseの存在により
- publicチャンネル→private用API
- privateチャンネル→public用API
の組み合わせになった際にはresponseの有無で
格納処理をスルーするようにすることで
なんとかしている仕組みにしてみました。
#!/opt/anaconda3/bin/python3
# -*- coding: utf-8 -*-
'''RocketChat Channelメンテナンス
RocketChatのチャンネル管理を行う
Todo:
* まだRedmineとRocketChatのみ。他のOSSに対しても同様に作る
def __init__(self, HEADERS, URL):
def _getChannelPublicMap(self):
def _getChannelPrivateMap(self):
def exchangeMapkeyToList(self, map):
def getChannelMap(self):
def getChannelUserMap(self, list_channelname):
def getDifftimeLastUpdateSec(self, _targetTime):
def _getChannel_id(self, channelname):
def sendMessageToRocketChat(self, channel, msg):
def closeTargetChannel(self, roomname):
def _ISOtimeToDatetime(self, target):
def _CreateMapFromChannelIDtoChannelname(self):
def _CreateMapFromChannelnameToChannelID(self, self._CreateMapFromChannelIDtoChannelname()):
def _judgeRocketChatMessage(self, target_date, limit):
def _JudgeDeleteChannelMessages(self, roomname, LIMIT):
def JudgeDeleteChannelMessages(self, LIMIT):
'''
################################################
# library
################################################
import dateutil
import json
import pandas as pd
import requests
import sys
from datetime import date
from datetime import datetime
from datetime import timedelta
from dateutil import parser
from pprint import pprint
from pytz import timezone
################################################
# 環境変数取得
################################################
################################################
# RocketChatChannelManager
################################################
class RocketChatChannelManager(object):
def __init__(self, HEADERS, URL):
'''RESTを呼ぶ形式
classの __init__処理
REST APIでCallするために HEADERSとURLを共有する。
RedmineXXXXXManager classとはことなりインスタンスは
生成しない。
'''
# 引数チェック 型
if not isinstance(HEADERS, dict):
print(f'引数：HEADERSの型が正しくありません dict <-> {type(HEADERS)}')
raise TypeError
# 引数チェック 型
if not isinstance(URL, str):
print(f'引数：URLの型が正しくありません str <-> {type(URL)}')
raise TypeError
# パラメータ共有
self.HEADERS = HEADERS
self.URL = URL
def _getChannelPublicMap(self):
'''パブリックチャネルのリストと最終更新時間のマップ
パブリックチャンネル名とチャンネル最終更新時間のマップを作成する。
Args:
Returns:
map: パブリックチャンネル名と最終更新時間のマップ
Raises:
API実行時のエラー
Examples:
>>> map = self._getChannelPublicMap()
Note:
publicとprivateで取得関数が異なるという。。。
'''
# 結果格納
_map = {}
# API定義
API = f'{self.URL}/api/v1/channels.list'
# 取得処理
response = None
try:
response = requests.get(
API,
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
return False
else:
for l in response.json()['channels']:
_map[l['name']] = l['_updatedAt']
# mapを返す
return _map
def _getChannelPrivateMap(self):
'''プライベートチャネルのリストと最終更新時間のマップ
プライベート名とチャンネル最終更新時間のマップを作成する。
Args:
Returns:
map: プライベートチャンネル 名と最終更新時間のマップ
Raises:
API実行時のエラー
Examples:
>>> map = self._getChannelPrivateMap()
Note:
publicとprivateで取得関数が異なるという。。。
'''
# 結果格納
_map = {}
# API定義
API = f'{self.URL}/api/v1/groups.listAll'
# 取得処理
try:
response = requests.get(
API,
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
return False
finally:
for l in response.json()['groups']:
_map[l['name']] = l['_updatedAt']
# mapを返す
return _map
def exchangeMapkeyToList(self, map):
'''mapのkeyを要素とするlistを生成する
ちょっとめんどい変換なのでヘルパー関数として作成したもの
'''
# 引数チェック 型
if not isinstance(map, dict):
print(f'引数：mapの型が正しくありません dict <-> {type(map)}')
raise TypeError
# 入れ物
_list = []
# mapループ
for key in map.keys():
_list.append(key)
return _list
def getChannelMap(self):
'''チャンネル一覧およびチャンネルの最終更新時間を取得する
パブリック、プライベート両方のチャンネルをまとめて処理する
Args:
Returns:
map: チャンネル名と所属ユーザリストのマップ
Raises:
API実行時のエラー
Examples:
>>> map_ = R.getChannelMap()
Note:
self._getChannelPubliclist()
self._getChannelPrivatelist()
パブリック、プライベートまとめて取得
'''
# public,privateそれぞれ取得
_map_public = self._getChannelPublicMap()
_map_private = self._getChannelPrivateMap()
# mapを結合して返す
if ((_map_public) and (_map_private)):
_map_public.update(_map_private)
return _map_public
# public Channelのみの場合
elif _map_public :
return _map_public
# private Channelのみの場合
elif _map_private :
return _map_private
else:
return {}
def getChannelUserMap(self, list_channelname):
'''指定チャンネルの登録ID一覧
listに格納したチャンネルに所属するユーザ一覧を
チャンネル名と参加しているユーザリストのマップを返す
パブリック、プライベートをまとめて実施
Args:
list_channelname(list): 探索対象のチャンネル名リスト
Returns:
map: チャンネル名をKeyとする所属ユーザリストのマップ
Raises:
API実行時のエラー
Examples:
>>> map = getChannelUserMap(['aaaa','bbbb'])
Note:
'''
# 引数チェック 型
if not isinstance(list_channelname , list):
print(f'引数：list_channelnameの型が正しくありません list <-> {type(list_channelname)}')
raise TypeError
# 結果全体格納するMap
_map = {}
# MSG送信API定義
# パブリックもプライベートもまとめて実施
APIS = [f'{self.URL}/api/v1/channels.members',
f'{self.URL}/api/v1/groups.members']
# 1000人は超えないだろう。。。から
COUNT = '1000'
# 対象チャンネル名リストでループ
for channel in list_channelname:
# MSG組み立て
msg = (('roomName', channel),('count',COUNT),)
# API発行
for api in APIS:
try:
response = requests.get(
api,
params=msg,
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
return False
else:
# ユーザたちを格納するList
_list = []
# 結果を得られた場合のみ格納
if response:
# 所属するユーザlistを生成
for l in response.json()['members']:
_list.append(f'{l["username"]}')
# mapにchannel名をKeyにしてユーザリストを格納
_map[channel] = _list
# mapを返す
return _map
def getDifftimeLastUpdateSec(self, _targetTime):
'''最終更新時間からの経過秒を返す
Public,Privateそれぞれ指定が可能
Args:
_targetTime(str): 比較したい時間 ISO時間フォーマット
Returns:
list: ユーザ一覧を格納したlist
Raises:
API実行時のエラー
Examples:
>>> list_AllUser = R.getAllUserList()
Note:
'''
# 引数チェック 型
if not isinstance(_targetTime, str):
print(f'引数：_targetTimeの型が正しくありません str <-> {type(_targetTime)}')
raise TypeError
# 今時間生成
jst_now = datetime.now(timezone('Asia/Tokyo'))
target = parser.parse(_targetTime).astimezone(timezone('Asia/Tokyo'))
# いま時間とターゲット時間の差分を秒で返す
return (jst_now - target).total_seconds()
def _getChannel_id(self, channelname):
'''Channel名の _id情報を取得する
チャンネル名からチャンネルIDを取得する
RocketChatAPIではチャンネル名ではなくチャンネルIDを
要求するケースが多数ある。
Args:
channelname: チャンネル名
Returns:
str: チャンネル名に対するチャンネルID
Raises:
API実行時のエラー
Examples:
>>> R._getChannel_id('general')
Note:
'''
# 引数チェック 型
if not isinstance(channelname, str):
print(f'引数：channelの型が正しくありません str <-> {type(channelname)}')
raise TypeError
# ユーザ情報取得API定義
API = f'{self.URL}/api/v1/rooms.info'
# MSG組み立て
msg = {'roomName': channelname,}
# MSG送信
try:
response = requests.get(
API,
params=msg,
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
return False
else:
if response.json()['success']:
return response.json()['room']['_id']
else:
return False
def sendMessageToRocketChat(self, channel, msg):
'''指定チャネルにメッセージを送る
指定チャンネルにメッセージを送信する
Args:
channel: チャンネル名
msg: 送信メッセージ
Returns:
処理結果, HTTP ステータスコード
Raises:
API実行時のエラー
Examples:
'>>> R.getUser_id('geneal', 'こんにちわ')
Note:
'''
# 引数チェック 型
if not isinstance(channel, str):
print(f'引数：channelの型が正しくありません str <-> {type(channel)}')
raise TypeError
if not isinstance(msg, str):
print(f'引数：msgの型が正しくありません str <-> {type(msg)}')
raise TypeError
# MSG送信API定義
API = f'{self.URL}/api/v1/chat.postMessage'
# MSG組み立て
msg = {'channel': channel,
'text' : msg,}
# 指定チャンネルが存在する場合のみ実行
if self._getChannel_id(channel):
# MSG送信
try:
response = requests.post(
API,
data=json.dumps(msg),
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
return False
else:
pprint(f'Status code: {response.status_code}')
return True
else:
print(f'指定したチャンネルが存在しません: {channel}')
return False
def closeTargetChannel(self, roomname):
'''パブリック、プライベート区別なくチャンネルを削除する
指定したチャンネル名を削除する
Args:
roomname(str): 削除するチャンネル名
Returns:
Raises:
API実行時のエラー
Examples:
>>> R.closeTargetChannel('テストチャンネル')
Note:
まとめて消す仕様ではない、1チャンネルづつターゲットで
'''
# 引数チェック 型
if not isinstance(roomname, str):
print(f'引数：roomnameの型が正しくありません str <-> {type(roomname)}')
raise TypeError
# 削除API定義
# パブリックもプライベートも区別なくまとめて実施
APIS = [f'{self.URL}/api/v1/channels.delete',
f'{self.URL}/api/v1/groups.delete']
# MSG組み立て
msg = {'roomId': self._getChannel_id(roomname)}
# まとめてチャンネル削除を遂行
for API in APIS:
try:
response = requests.post(API,
data=json.dumps(msg),
headers=self.HEADERS,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
else:
# 結果を得られた場合のみ処理コードを返す
if response:
return response.json()['success']
def _ISOtimeToDatetime(self, target):
'''ISOフォーマット時刻文字列をJST変換してdatetime型で返す
Args:
target: str ISO形式のUTC時刻文字列
Returns:
datetime: JST変換後
Raises:
API実行時のエラー
Examples:
>>> self._ISOtimeToDatetime('2021-01-20T00:23:10.256Z')
Note:
'''
return parser.parse(target).astimezone(timezone('Asia/Tokyo'))
def _CreateMapFromChannelIDtoChannelname(self):
'''チャンネルIDに対するチャンネル名をもつmapを生成する。
チャンネルIDをkeyにしてチャンネル名をValueに持つmapを生成する。
RocketChatから還元される情報が何かとチャンネルIDで返してくるのだが
還元する立場だとチャンネルIDだとわかりにくい問題がある。
チャンネル名に置き換えることでデータ利便性を上げる。
-> cf. _getChannel_id(self, channelname): チャンネル名からチャンネルIDを取得
Args:
Returns:
map: Key:チャンネルID、Value: チャンネル名
Raises:
API実行時のエラー
Examples:
>>> _MapChannelIDtoChannelName = self._CreateMapFromChannelIDtoChannelname()
Note:
つどつどAPIを叩いて情報収集する仕掛けだとレスポンス懸念あり。
mapを予め作成し変換パフォーマンスを向上させる。
TODO: チャンネル名 -> チャンネルIDのmapも作っておくべきかもしれない。
'''
# パブリック、プライベート合算でチャンネル名を取得する
## Class内メソッドを使ってチャンネル名をまとめて取得
_map = self.getChannelMap()
# 蓄積するDataFrame生成
channelMap = {}
# 処理ループ
for key in _map.keys():
_key = self._getChannel_id(key)
channelMap[_key] = key
# 蓄積結果を返す
return channelMap
def _CreateMapFromChannelnameToChannelID(self, self._CreateMapFromChannelIDtoChannelname()):
'''ChannelID->ChannelNameのmapを利用してChannelName->ChannelID mapを生成する
内包を使用してkey/valueを反転させる
Args:
map: map ChannelID->ChannelKeyマップ
Returns:
map: map key/valueを反転させたmap
Raises:
API実行時のエラー
Examples:
>>> _map = self._CreateMapFromChannelnameToChannelID(self._CreateMapFromChannelIDtoChannelname())
Note:
'''
# ChannelID -> Channel NameMap
_map = self._CreateMapFromChannelIDtoChannelname
# ChannelID -> Channel NameMapのKey/Valueを反転させる
swap_map = {v: k for k, v in _map.items()}
return swap_map
def _judgeRocketChatMessage(self, target_date, limit):
'''メッセージ作成日付から保管する、しない判定を行う
limitで指定した期間のMSGを保管する、しない判定を行いTrue/Falseで返す。
日付差分計算はdatetime型のサポートにより行う。timedeltaオブジェクトを使用し
差分日付けに対する判定処理を行う。
Args:
target_date: datetime 判定対象の時間データ
limit : int RocketChatメッセージ保存期間
Returns:
True/False: Boolean True 保存、False 削除対象
Raises:
API実行時のエラー
Examples:
>>> self._judgeRocketChatMessage(target_datetime, 10)
Note:
'''
today = date.today()
diff_date = timedelta(limit)
return (today - target_date.date() > diff_date)
def _JudgeDeleteChannelMessages(self, roomname, LIMIT):
'''roomnameに対しLIMIT超過日数を超えたメッセージに削除判別フラグを設定したデータを生成する。
指定したroomnameに対し、メッセージ作成日からの日数が
LIMITを超過している場合に削除判定フラグをつけて
DataFrameを生成する。
Args:
roomname: str 探索対象のチャンネル名
LIMIT: int 保存期間（日数）
Returns:
df: DataFrame: ['チャンネル','MSG_ID','更新時間','削除対象','MSG']
Raises:
API実行時のエラー
Examples:
>>> _df = self._JudgeDeleteChannelMessages(key, LIMIT)
Note:
_CreateMapFromChannelIDtoChannelname
_MapChannelIDtoChannelName
_ISOtimeToDatetime
_judgeRocketChatMessage
'''
# 引数チェック 型
if not isinstance(roomname, str):
print(f'引数：roomnameの型が正しくありません str <-> {type(roomname)}')
raise TypeError
if not isinstance(LIMIT, int):
print(f'引数：LIMITの型が正しくありません int <-> {type(LIMIT)}')
raise TypeError
# MSG抽出API定義
# パブリックもプライベートも区別なくまとめて実施
APIS = [f'{self.URL}/api/v1/channels.messages',
f'{self.URL}/api/v1/groups.messages']
# MSG組み立て
channel_id = self._getChannel_id(roomname)
params = (
('roomId', channel_id),
)
## この書き方だと失敗する
#params = (
# ('roomId', channel_id)
#)
# 変換Map作成
_MapChannelIDtoChannelName = self._CreateMapFromChannelIDtoChannelname()
# 両パターンに対応する入れ物を用意
_list = []
for API in APIS:
pprint(f'API={API}')
try:
response = requests.get(API,
headers=self.HEADERS,
params=params,)
except Exception as e:
print(f'API実行エラー: {API}')
print(f'Error: {e}')
else:
# 結果をログっぽく返す
# 結果を得られた場合のみログを返す
pprint(f'response={response}')
if response:
# 結果チェック
pprint(response)
pprint(len(response.json()['messages']))
# 削除対象判定結果をDataFrameに組み込んでで返す
for _ in response.json()['messages']:
_list.append([_MapChannelIDtoChannelName[_['rid']],
_['_id'],
self._ISOtimeToDatetime(_['_updatedAt']),
self._judgeRocketChatMessage(self._ISOtimeToDatetime(_['_updatedAt']), LIMIT),
_['msg']])
# DataFrameにして結果を返す
df= pd.DataFrame(_list)
df.columns = ['チャンネル','MSG_ID','更新時間','削除対象','MSG']
return df
def JudgeDeleteChannelMessages(self, LIMIT):
'''削除対象フラグを持ったDataFrameをまとめて1つのDataFrameにする。
サブメソッド _JudgeDeleteChannelMessagesから得られる
パブリック、プライベート両方から得られたDataFrameを蓄積する。
Args:
LIMIT: int 保存期間（日数）
Returns:
df: DataFrame 蓄積したDataFrame
Raises:
API実行時のエラー
Examples:
Note:
self._JudgeDeleteChannelMessages(key, LIMIT): 削除対象フラグをつけたDataFrameを作成
'''
if not isinstance(LIMIT, int):
print(f'引数：LIMITの型が正しくありません int <-> {type(LIMIT)}')
raise TypeError
# パブリック、プライベート合算でチャンネル名を取得する
## Class内メソッドを使ってチャンネル名をまとめて取得
_map = self.getChannelMap()
# 蓄積するDataFrame生成
df = pd.DataFrame(index=[])
# 処理ループ
for key in _map.keys():
_df = self._JudgeDeleteChannelMessages(key, LIMIT)
df = pd.concat([df, _df], axis=0)
# 蓄積結果を返す
return df.reset_index(drop=True)