Slackからエクスポートしたjsonファイルを特定のユーザーだけフィルタしてCSVに変換した
はじめに
Slackにはデータエクスポート出来る機能があり、"もしもの時"はエクスポート機能に頼ることになると思います。
今回、弊社にも"もしもの時"が訪れたのでSlackのデータをエクスポートしましたが、全てjsonファイルで情シス以外のメンバーが閲覧するのは難しそうだったので、jsonから変換する方法を探っていました。
注意点
Slakの全ての会話(DM、プライベートチャンネル含む)をエクスポートするには事前に承認が必要となります。
"もしもの時"が来た場合にはSlackへリクエストし、必要に応じて法律が許す範囲内で利用しましょう。
参考情報
json→CSV変換の方法を探る過程で、↑に出会いました。
ただ、上記のconvert.py
でjson→CSVに変換すると以下動作となります。
・特定ユーザーだけ絞って出力できない
・投稿時間が無い
・出来上がるCSVはチャンネル単位
特定のユーザーで絞りたいし投稿時間は欲しいしチャンネルが分かれると面倒、ということで少しコードを追記修正しました。
(以降クソコード失礼)
追記修正した部分
特定のユーザーのデータだけCSVに出力したい
definesでSlackのユーザーIDを指定。
# defines =====================
#抽出したいSlackユーザーのユーザーIDを指定して下さい
SLACK_USER_ID = "hogehoge"
メッセージを処理する過程で、jsonに含まれるSlackのユーザーIDが↑で指定したIDじゃなければスキップ。
if item[USER_KEY] == SLACK_USER_ID:
lines += get_line_text(users, item,channel)
投稿時間を追加し日付と共にJSTへ変換
SlackからエクスポートしたjsonファイルにはtsというUnixTimeのデータが含まれています。
ただUnixTimeじゃ人間は分からんので変換します。
(クソコード失礼定期)
# jsonからtsを取得→文字列→hh:mm:ssまで変換
JST = datetime.timezone(+datetime.timedelta(hours=9))
ts = f'{item[TS_KEY]}'.replace('"', '\"')
ts1 = float(ts)
ts2 = int(ts1)
ts3 =datetime.datetime.fromtimestamp(ts2,tz=JST)
time = ts3.time().strftime('%X')
date = ts3.date().strftime('%Y/%m/%d')
出力されるCSVを1つにまとめる
↓をfor文の外に出しておしまい。
# 変換した情報をCSVへ書き込み
out_file_path = f"{output_dir}/{SLACK_USER_ID}_postlist.csv"
f = open(out_file_path, 'w')
f.write(lines)
f.close()
出力するCSVにチャンネル名を追加
関数にチャンネル名渡してそのまま返してもらっています。
def get_line_text(users, item, channel):
return f'{date},{ts4},{name},{channel}"\n'
それを出力するlinesに追加しているだけ。
lines += get_line_text(users, item,channel)
DMとグループDMをはずした
times
は外したいとか逆にhoge_
だけ集計したい等にも対応出来ます。
#DMは除外
if channel.startswith('D0'):
continue
#GroupDMも除外
if channel.startswith('mpdm'):
continue
実際のコード
from cgitb import text
import sys
import os
import glob
import json
import datetime
# defines =====================
ID_KEY = 'id'
TEXT_KEY = 'text'
TS_KEY = 'ts'
USER_KEY = 'user'
PROFILE_KEY = 'profile'
REAL_NAME_KEY = 'real_name'
DISPLAY_NAME_KEY = 'display_name'
FILES_KEY = 'files'
URL_KEY = 'url_private'
OUT_PUT_DIR_NAME = 'slack_csv_output'
USER_FILE_NAME = 'users.json'
#抽出したいSlackユーザーのユーザーIDを指定して下さい
SLACK_USER_ID = "hogehoge"
# functions =====================
# jsonファイルをjson辞書に変換
def json_file_to_data(full_path):
f = open(full_path, 'r')
converted = json.load(f)
f.close()
return converted
# ユーザー情報を取得
def get_users(source_dir):
users_json = json_file_to_data(source_dir)
users = {}
for user in users_json:
name = user[PROFILE_KEY][DISPLAY_NAME_KEY]
if not name:
name = user[PROFILE_KEY][REAL_NAME_KEY]
id = user[ID_KEY]
users[id] = name
return users
# 1メッセージのjson辞書データをカンマ区切りの1行データに変換
def get_line_text(users, item, channel):
text = f'{item[TEXT_KEY]}'.replace('"', '\"')
# jsonからtsを取得→文字列→hh:mm:ssまで変換
JST = datetime.timezone(+datetime.timedelta(hours=9))
ts = f'{item[TS_KEY]}'.replace('"', '\"')
ts1 = float(ts)
ts2 = int(ts1)
ts3 =datetime.datetime.fromtimestamp(ts2,tz=JST)
time = ts3.time().strftime('%X')
date = ts3.date().strftime('%Y/%m/%d')
name = ''
if USER_KEY in item.keys():
user_id = item[USER_KEY]
if user_id in users.keys():
name = users[user_id]
urls = ''
if FILES_KEY in item.keys():
for attachmentFile in item[FILES_KEY]:
if not URL_KEY in attachmentFile.keys():
continue
url = f"{attachmentFile[URL_KEY]}".replace('"', '\"')
urls += f'{url}\n'
return f'{date},{time},{name},{text3},{channel}"\n'
# 失敗手続き
def failed(text):
print(f'[error] {text}')
print('failed...')
exit()
# core logics =====================
# 引数からソースフォルダ情報取得
argv = sys.argv
if len(argv) < 2:
failed('Please add argument of work directory')
source_dir = argv[1]
if not os.path.exists(source_dir):
failed(f'not exists directory: {source_dir}')
print(f'Source directory > {source_dir}')
# 出力フォルダの作成
output_dir = f'{source_dir}/../{OUT_PUT_DIR_NAME}'
if os.path.exists(output_dir):
failed(f'already exists output directory: {output_dir}')
print(f'Create output dir > {output_dir}/')
os.makedirs(output_dir)
# jsonファイルを省いたチャンネル名のフォルダ一覧とユーザー一覧の取得
channels = sorted(os.listdir(path=source_dir))
channels = [x for x in channels if not x.endswith('.json')]
users = get_users(f'{source_dir}/{USER_FILE_NAME}')
#ヘッダー書き込み
lines = "date,time,name,channel\n"
# channelフォルダ単位でループ
for channel in channels:
#DMは除外
if channel.startswith('D0'):
continue
#GroupDMも除外
if channel.startswith('mpdm'):
continue
json_files = sorted(glob.glob(f"{source_dir}/{channel}/*.json"))
# 日付名のjsonファイル単位でループ
for file_full_path in json_files:
file_name = os.path.split(file_full_path)[1]
date = file_name.replace('.json', '')
json_dic = json_file_to_data(file_full_path)
# メッセージ単位ループ
for item in json_dic:
if not TEXT_KEY in item.keys():
continue
if not USER_KEY in item.keys():
continue
if item[USER_KEY] == SLACK_USER_ID:
lines += get_line_text(users, item,channel)
print("Done : " + channel)
# 変換した情報をCSVへ書き込み
out_file_path = f"{output_dir}/{SLACK_USER_ID}_postlist.csv"
f = open(out_file_path, 'w')
f.write(lines)
f.close()
print(f'{len(channels)} channels converted.')
実行!
実行するとCSVファイルがSlackユーザーID_postlist.csv
という名前で出力されます🙌
お し ま い
Discussion