|
import json |
|
import os |
|
import requests |
|
|
|
ID_MAPPING_JSON = 'channel_id_mapping.json' |
|
|
|
|
|
class ChannelIdMapper: |
|
def __init__(self, token): |
|
self.token = token |
|
self._load_channel_id_mapping() |
|
|
|
def create_channel_dict(self): |
|
base_url = 'https://slack.com/api/conversations.list' |
|
headers = {'Authorization': f'Bearer {self.token}'} |
|
params = { |
|
'types': 'public_channel,private_channel', |
|
'limit': 1000 |
|
} |
|
response = requests.get(base_url, headers=headers, params=params) |
|
data = response.json() |
|
|
|
channel_dict = {} |
|
if data['ok']: |
|
channels = data['channels'] |
|
for channel in channels: |
|
channel_name = channel['name'] |
|
channel_id = channel['id'] |
|
channel_dict[channel_name] = channel_id |
|
else: |
|
print(f'Error: {data}') |
|
print(channel_dict) |
|
return channel_dict |
|
|
|
def _load_channel_id_mapping(self): |
|
if os.path.exists(ID_MAPPING_JSON): |
|
with open(ID_MAPPING_JSON, 'r') as f: |
|
channel_dict = json.load(f) |
|
else: |
|
channel_dict = self.create_channel_dict() |
|
with open(ID_MAPPING_JSON, 'w') as f: |
|
json.dump(channel_dict, f, indent=4) |
|
return channel_dict |
|
|
|
def get_channel_id(self, channel_name): |
|
channel_dict = self._load_channel_id_mapping() |
|
return channel_dict[channel_name] |