import logging
import os
from ssl import SSLContext
from typing import Dict
from typing import List
from typing import Optional
from typing import Union
from slack_sdk import WebClient
from slack_sdk.http_retry import RetryHandler
from .functions.common import get_channels_list
from .functions.common import get_file_volumes
from .functions.common import get_replies
from .functions.common import test_connection
from .functions.common import test_downloader
from .functions.common import test_uploader
class CommonDryRunClient(WebClient):
def get_channels_list(self) -> List[Dict]:
return get_channels_list(client=self)
def get_replies(self, channel_id: str, ts: str) -> List[Dict]:
return get_replies(client=self, channel_id=channel_id, ts=ts)
def get_file_volumes(
self, channel_ids: Optional[Union[str, List[str]]] = None
) -> float:
return get_file_volumes(client=self, channel_ids=channel_ids, auto_join=False)
class CommonNoLocalVolumeClient(CommonDryRunClient):
def get_file_volumes(
self,
channel_ids: Optional[Union[str, List[str]]] = None,
auto_join: bool = False,
) -> float:
return get_file_volumes(
client=self, channel_ids=channel_ids, auto_join=auto_join
)
def test_connection(self) -> None:
return test_connection(client=self)
def test_downloader(self) -> None:
return test_downloader(client=self)
def test_uploader(self) -> None:
return test_uploader(client=self)
[docs]class CommonClient(CommonNoLocalVolumeClient):
BASE_URL = "https://www.slack.com/api/"
TYPE = "bot"
[docs] def __init__(
self,
local_data_dir: str,
token: Optional[str] = None,
base_url: str = BASE_URL,
timeout: int = 30,
ssl: Optional[SSLContext] = None,
proxy: Optional[str] = None,
headers: Optional[dict] = None,
user_agent_prefix: Optional[str] = None,
user_agent_suffix: Optional[str] = None,
team_id: Optional[str] = None,
logger: Optional[logging.Logger] = None,
retry_handlers: Optional[List[RetryHandler]] = None,
):
if token is not None and self.TYPE == "bot" and not token.startswith("xoxb-"):
raise ValueError("token have to be start with xoxb-")
if token is not None and self.TYPE == "user" and not token.startswith("xoxp-"):
raise ValueError("token have to be start with xoxp-")
super().__init__(
token=token,
base_url=base_url,
timeout=timeout,
ssl=ssl,
proxy=proxy,
headers=headers,
user_agent_prefix=user_agent_prefix,
user_agent_suffix=user_agent_suffix,
team_id=team_id,
logger=logger,
retry_handlers=retry_handlers,
)
self.local_data_dir: str = local_data_dir
os.makedirs(os.path.join(self.local_data_dir, "files"), exist_ok=True)
os.makedirs(os.path.join(self.local_data_dir, "channels"), exist_ok=True)
os.makedirs(os.path.join(self.local_data_dir, "bookmarks"), exist_ok=True)
os.makedirs(os.path.join(self.local_data_dir, "emojis"), exist_ok=True)
[docs]class UploaderClientABC(CommonClient):
pass
[docs]class DownloaderClientABC(CommonClient):
pass