⌨️

Slackからエクスポートしたjsonファイルを特定のユーザーだけフィルタしてCSVに変換した

2022/12/15に公開

Python

Slack

tech

はじめに

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

実際のコード

convert.py　

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という名前で出力されます🙌

お　し　ま　い