Дампит юзеров и чаты (10к сообщений из каждого).
py main.py xoxb-KEY-KEY-KEY
Python:
import asyncio
from os import mkdir, listdir
from os.path import exists
from sys import argv
from time import ctime, sleep
from aiofiles import open as aopen
from aiohttp import ClientSession
class SlackMessages:
def __init__(self, key: str) -> None:
assert key.startswith('xoxb-') or key.startswith('xoxp-'), 'Key starts wrong'
if key.startswith('xoxb-'):
assert len(key) > 52 and len(key) < 59, 'Wrong key len'
assert len(key.split('-')) == 4, 'Wrong struct? Must be: xoxb-D-ID-N'
elif key.startswith('xoxp-'):
assert len(key) > 44 and len(key) < 90, 'Wrong key len'
assert len(key.split('-')) == 5, 'Wrong struct? Must be: xoxb-D-ID-N-N'
print('[+] Token passed tests')
self.key = key
self.b_url = 'https://slack.com'
self.users_db = {}
self.already_downloaded = listdir(key)
def is_critical(self, resp: dict) -> bool:
if resp['ok']:
return False
critical_errors = {
'access_denied': 'Access Dined',
'account_inactive': 'Inactive Account',
'deprecated_endpoint': 'Need update code',
'method_deprecated': 'Need update request method',
'not_authed': 'No token? Need code review',
'token_expired': 'Token is experied... Look for new one',
'token_revoked': 'Our token is fired...',
'two_factor_setup_required': 'For first need setup 2FA',
'internal_error': 'Not our bad. Internal error...',
'service_unavailable': 'Slack unavilable',
'invalid_post_type': 'Invalid POST type',
'invalid_form_data': 'Invalid form data',
'invalid_arguments': 'Invalid Arguments'
}
'''medium_errors = {
'no_permission': 'User cant view this resource',
'request_timeout': 'Request timeout'
}'''
if resp['error'] in critical_errors.keys():
print('[!!!]', critical_errors[resp['error']])
exit(0)
if resp['error'] == 'ratelimited':
sleep(60)
return False
else:
print(resp['error'])
return True
return False
async def get_members(self) -> bool:
passed = 0
params = {
'token': self.key,
'limit': 100,
'cursor': ''
}
async with aopen(f'{self.key}/SLACK_MEMBERS_LIST.csv', 'w') as mem_list:
await mem_list.write('User ID,Real Name,Job,Email,Phone,Skype,Bot,Deleted,App\n') # noqa
while True:
print(' '*50, end='\r')
print(f'[...] Parsed: {passed}', end='\r')
resp = {}
async with ClientSession(base_url=self.b_url) as ses:
async with ses.get('/api/users.list', params=params) as r:
resp: dict = await r.json()
self.is_critical(resp)
params['cursor'] = resp.get(
'response_metadata', {}
).get('next_cursor', False)
passed += len(resp.get('members', []))
for member in resp.get('members', []):
member_d = []
prof = member['profile']
member_d.append(member['id'])
member_d.append(prof.get('real_name', ''))
member_d.append(prof.get('title', ''))
member_d.append(prof.get('email', ''))
member_d.append(prof.get('phone', ''))
member_d.append(prof.get('skype', ''))
member_d.append(str(member['is_bot']))
member_d.append(str(member["deleted"]))
member_d.append(str(member["is_app_user"]))
await mem_list.write(','.join(member_d)+'\n')
self.users_db[member['id']] = prof["real_name"]
if not params['cursor']:
break
sleep(3)
return True
async def get_messages(self, uid: str, name: str) -> bool:
print(f'[...] Fetching {name}', end='\r')
params = {
'limit': 500,
'channel': uid,
'token': self.key,
'cursor': ''
}
for i in range(5):
async with ClientSession(base_url=self.b_url) as ses:
async with ses.get(
'/api/conversations.history', params=params
) as r:
msgs = await r.json()
self.is_critical(msgs)
params['cursor'] = msgs.get(
'response_metadata', {}
).get('next_cursor', False)
msgs = msgs['messages']
if not msgs:
return False
async with aopen(f'{self.key}/{name}.txt', 'a') as msgs_s:
for msg in msgs:
try:
user = self.users_db[msg['user']] if self.users_db.get(msg['user']) else msg['user'] # noqa
except Exception:
user = 'UNKNOWN'
try:
await msgs_s.write(f'[{ctime(int(float(msg["ts"])))}] {user}: {msg["text"]}\n') # noqa
await msgs_s.write('\n======MSG DETELMINER======\n')
except Exception as e:
print(e)
pass
if not params['cursor']:
break
sleep(3)
return True
async def get_chats(self):
chats = []
params = {
'limit': 100,
'token': self.key,
'cursor': ''
}
while True:
async with ClientSession(base_url=self.b_url) as ses:
async with ses.get(
'/api/conversations.list', params=params
) as r:
resp: dict = await r.json()
params['cursor'] = resp.get(
'response_metadata', {}
).get('next_cursor', False)
self.is_critical(resp)
for chat in resp['channels']:
if (chat['id'], chat['name']) not in chats:
if chat['name']+'.txt' in self.already_downloaded:
continue
yield (chat['id'], chat['name'])
if not params['cursor']:
break
sleep(3)
async def main() -> None:
client = SlackMessages(argv[1])
print('[...] Checking working directory')
if not exists(argv[1]):
mkdir(argv[1])
print('[...] Parsing users')
await client.get_members()
print('[+] Parsed')
async for chatid, chatname in client.get_chats():
print('\033cDumping', chatname)
await client.get_messages(chatid, chatname)
if __name__ == "__main__":
assert len(argv) == 2, 'Forgot token.\nmain.py TOKEN'
asyncio.run(main())