Python: SlackにインポートするためのCSVファイルを、Slackのエクスポートしたデータ(ZIPファイル)から作成する

旧ワークスペースの一部のチャンネルのみ、新ワークスペース(新WS)にCSVファイルで移行するためのPythonスクリプトを書いたので紹介する。

準備することは、

旧ワークスペースからエクスポートしたデータ (アーカイブしたZIPファイル)
ワークスペースのチャンネル名と新WSのチャンネル名 (新規作成でも可)
旧ワークスペースのユーザーID、新WSのユーザー名とユーザーID (新WSに移行したい全ての人が参加している前提)
各オプションのオンオフ

である。コードは以下の通り

# ZIPファイルへのPATH
filepath = r'Sample Slack export Jan 01 2020 - Jul 20 2022.zip'

# 出力するCSVファイル名
csvpath = 'ForImport.csv'

#旧ワークスペースのチャンネル名 → 新ワークスペースのチャンネル名
channel_mapping = {'生物実験':'new-生物実験',
                   '物理実験':'new-物理実験',
                   '装置':'new-装置'}

# 旧ワークスペースのユーザーID → [新ワークスペースのユーザー名・ユーザーID]
name_mapping = {'UWGKT5M5I': ['Hakase Shinonome','U01CKGMWGBM'], 
                'U01AC7YM2QW': ['Nichijo Isezaki','U01WG2KJB6D']}

show_file_link = True
show_thread_id = True
show_quote_post = True
ignore_not_user = False

変換コード

import os
import json
import zipfile

def get_text(l, name_mapping):
    text = l['text'].replace(r'"',r'\"') # デリミタに " を使うのでエスケープする
        
    # ファイルが添付された投稿の場合、旧WSのファイルへのリンクをFile: と書く
    if show_file_link and 'files' in l: 
        for file in l['files']:
            if 'permalink' not in file: 
                continue
            # 添付ファイルに permalink がある場合
            text+='\nFile: {}'.format(file['permalink'].replace('\\',''))
    
    # 投稿の引用の場合、To: 以下に投稿を引用形式で書く
    if show_thread_id and 'attachments' in l: 
        for attachment in l['attachments']:
            if 'fallback' not in attachment or 'ts' not in attachment: 
                continue
            # fallbackがある場合
            text+='\nTo:\n{}'.format('> '+attachment['fallback'].replace('\n','\n> ')).replace(r'"',r'\"')

    # スレッドの場合、スレッドの最初の投稿には Thread top: と、 スレッド内の投稿には In thread: と書く
    # 番号はタイムスタンプの下4桁を使う
    if show_quote_post and 'thread_ts' in l:
        if 'reply_count' in l:
            text+='\nThread top: {}'.format(l['thread_ts'].split(".")[0][-4:])
        else:
            text+='\nIn thread: {}'.format(l['thread_ts'].split(".")[0][-4:])
        
    # @ユーザーIDを置き換える
    for key,name in name_mapping.items():
        text = text.replace(f'<@{key}>',f'<@{name[1]}>') 
    return text

texts = {}

with zipfile.ZipFile(filepath, 'r') as myzip:
    infolist = myzip.infolist()
    
    # ユーザーリストのチェック
    users = json.load(myzip.open('users.json'))
    for userid in name_mapping.keys():
        assert(userid in [user['id'] for user in users])# 旧WSにユーザーが存在しない
    print("★旧WSでの名前 -> 新WSでの名前")
    for user in users:
        if user['id'] in name_mapping:
            if 'real_name' in user:
                print(user['real_name'],'->',name_mapping[user['id']][0],'OK?')
            else:
                print(user['profile']['real_name'],'->',name_mapping[user['id']][0],'OK?')
        
    # チャンネルリストのチェック
    channels = json.load(myzip.open('channels.json'))
    print("★旧WSでのチャンネル名 -> 新WSでのチャンネル名")
    for channelname in channel_mapping.keys():
        assert(channelname in [channel['name'] for channel in channels])# 旧WSにチャンネルが存在しない
        print(channelname,"->",channel_mapping[channelname])
    
    # 投稿を取得
    for info in infolist:
        filename_utf_8 = info.filename.encode('cp437').decode('utf-8')
        channel_utf_8 = filename_utf_8.split('/')[0]
        if '/' in filename_utf_8 and '.json' in filename_utf_8 and channel_utf_8 in list(channel_mapping.keys()):
            obj = json.load(myzip.open(info.filename))
            for l in obj:
                if 'text' not in l:continue # テキストがなければ無視
                if ignore_not_user == True:
                    assert(l['user'] in name_mapping) # ユーザーIDリストに存在しない投稿者がいる
                line_csv = '"{}","{}","{}","{}"\n'.format(l['ts'],channel_mapping[channel_utf_8],name_mapping[l['user']][0],get_text(l,name_mapping))
                texts[l['ts']] = line_csv

# 仕様上タイムスタンプでソート
texts_sorted = dict(sorted(texts.items(), key=lambda x:float(x[0])))

# CSVファイル出力
with open(csvpath,'w',encoding='utf_8') as f:
    for text in texts_sorted.values():
        f.write(text)

# ログ出力
print("★ログ:",len(texts_sorted),"件の投稿を出力しました")