-
Notifications
You must be signed in to change notification settings - Fork 8
/
slack_auto_export.py
executable file
·111 lines (91 loc) · 3.35 KB
/
slack_auto_export.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
#!/usr/bin/env python
import time
import json
import os
import slacker
import click
class SlackAutoExport(object):
def __init__(self, token, verbose=False):
self.slack = slacker.Slacker(token)
self.verbose = verbose
def _get_channels_list(self):
return {c["name"]: c for c in
self.slack.channels.list().body["channels"]}
def _get_channel_history(self, channel_id, request_pause_period=0.5):
latest = None
has_more = True
messages = []
while has_more:
m = self.slack.channels.history(
channel_id,
count=1000,
latest=latest
)
messages.extend(m.body['messages'])
if self.verbose:
print("{}: Retrieved {} messages from channel {}".format(
self.__class__.__name__,
len(messages), [c_name for c_name, c in self.channels.items() if c['id'] == channel_id],
))
if m.body['messages']:
latest = m.body['messages'][-1]['ts']
has_more = m.body["has_more"]
time.sleep(request_pause_period)
return messages
def _get_history(self):
channels = self._get_channels_list()
history = {}
for channel_name in channels:
channel_id = channels[channel_name]["id"]
history[channel_name] = self._get_channel_history(channel_id)
return history
def _get_users(self):
return {c["name"]: c for c in
self.slack.users.list().body["members"]}
@property
def history(self):
if not hasattr(self, "_history"):
self._history = self._get_history()
return self._history
@property
def channels(self):
if not hasattr(self, "_channels"):
self._channels = self._get_channels_list()
return self._channels
@property
def users(self):
if not hasattr(self, "_users"):
self._users = self._get_users()
return self._users
def write_history(self, output_dir):
os.makedirs(os.path.join(output_dir, "channels"), exist_ok=True)
for channel_name, data in self.history.items():
self._write_json_file(
data,
(
output_dir,
"channels",
"{}.json".format(channel_name)
)
)
self._write_json_file(self.channels, (output_dir, "channels.json"))
self._write_json_file(self.users, (output_dir, "users.json"))
def _write_json_file(self, obj, paths):
filepath = os.path.join(*paths)
with open(filepath, "w+") as f:
json.dump(obj, f, indent=4)
if self.verbose:
print("Wrote {}".format(filepath))
@click.command()
@click.option('-t', "--token", type=click.STRING, required=True,
help="Slack API token")
@click.option("-o", "--output-dir", type=click.Path(), required=True,
help="Output directory for JSON files")
@click.option('-q', '--quiet', is_flag=True,
help="Set this if you don't want progress printed"
" to your terminal.")
def main(token, output_dir, quiet):
s = SlackAutoExport(token=token, verbose=not quiet)
s.write_history(output_dir)
if __name__ == '__main__':
main()