How to use mattermostdriver - 9 common examples
To help you get started, we’ve selected a few mattermostdriver examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
anl-cyberscience / LQMToolset / lqmt / tools / to_mattermost / tool.py View on Github 
def __init__(self, config):
"""
ToMattermost tool.
:param config: configuration file
"""
super().__init__(config, [AlertAction.get('OtherAction')])
self._logger = logging.getLogger("LQMT.ToMattermost.{0}".format(self.getName()))
self.client = mattermostdriver.Driver({
'scheme': self._config.scheme,
'url': self._config.url,
'port': self._config.port,
'login_id': self._config.login,
'password': self._config.password
})
self.tool_user_id = 'Unknown'def __init__(self, **kwargs):
"""Initialize Mattermost class which extends Chat ABC
Args:
**kwargs: args to pass to Chat ABC
team_channel : (String) required for posting to channel
channel_name : (String) required for posting to channel
"""
super().__init__(**kwargs)
self.inst = Driver({
'url': self.url,
'verify': False,
'token': self.api_key,
'username': self.username,
'port': kwargs.get('port', 8065)
})
self.team_name = kwargs.get("team_name", None)
self.channel_name = kwargs.get("channel_name", None)
self.filepath = kwargs.get("filepath", None)def main():
print("Creating Mattermost Driver...")
driver_options = {
'url': config.URL,
'login_id': config.USERNAME,
'password': config.PASSWORD,
'port': config.PORT
}
driver = Driver(driver_options)
print("Authenticating...")
driver.login()
driver.users.get_user('me')
print("Successfully authenticated.")
print("Retrieving Coffee Buddies participants...")
team_name = config.TEAM_NAME
channel_name = config.CHANNEL_NAME
members = utils.get_channel_members(driver, team_name, channel_name)
print("Successfully retrieved Coffee Buddies participants.")
print("Preparing participants database...")
utils.create_users(members)
utils.create_pairs(members)
print("Succesfully prepared participants database.")super().__init__(config, opsdroid=opsdroid)
_LOGGER.debug(_("Starting Mattermost connector"))
self.name = "mattermost"
self.token = config["token"]
self.url = config["url"]
self.team_name = config["team-name"]
self.scheme = config.get("scheme", "https")
self.port = config.get("port", 8065)
self.verify = config.get("ssl-verify", True)
self.timeout = config.get("connect-timeout", 30)
self.request_timeout = None
self.mfa_token = None
self.debug = False
self.listening = True
self.mm_driver = Driver(
{
"url": self.url,
"token": self.token,
"scheme": self.scheme,
"port": self.port,
"verify": self.verify,
"timeout": self.timeout,
"request_timeout": self.request_timeout,
"mfa_token": self.mfa_token,
"debug": self.debug,
}"""File upload
Args:
**kwargs:
filepath: (String) optional filepath to check for files to upload
Returns:
"""
print("[*] doing post message")
try:
self.inst.login()
team_id = self.inst.teams.get_team_by_name(self.team_name)['id']
channel_id = self.inst.channels.get_channel_by_name(
channel_name=self.channel_name, team_id=team_id)['id']
except exceptions.NoAccessTokenProvided as er:
print(f"[!] NoAccessTokenProvided {er}")
logger.exception()
except exceptions.InvalidOrMissingParameters as er:
print(f"[!] InvalidOrMissingParameters {er}")
logger.exception()
try:
if isfile(self.filepath):
file_ids = [
self.inst.files.upload_file(
channel_id=channel_id, files={
'files': (
basename(
self.filepath), open(
join_abs(
self.filepath), 'rb'))})['file_infos'][0]['id']]self.filepath), open(
join_abs(
self.filepath), 'rb'))})['file_infos'][0]['id']]
self.inst.posts.create_post(options={
'channel_id': channel_id,
'message': f"Recon Data {datetime.datetime.now()}",
'file_ids': file_ids
})
elif isdir(self.filepath):
file_location = abspath(self.filepath)
self._upload_files(file_location, channel_id)
except exceptions.ContentTooLarge as er:
print(f"[!] ContentTooLarge {er}")
logger.exception()
except exceptions.ResourceNotFound as er:
print(f"[!] ResourceNotFound {er}")
logger.exception()
except OSError as er:
print(f"[!] File not found {er}")
logger.exception()async def connect(self):
"""Connect to the chat service."""
_LOGGER.info(_("Connecting to Mattermost"))
login_response = self.mm_driver.login()
_LOGGER.info(login_response)
if "id" in login_response:
self.bot_id = login_response["id"]
if "username" in login_response:
self.bot_name = login_response["username"]
_LOGGER.info(_("Connected as %s"), self.bot_name)
self.mm_driver.websocket = Websocket(
self.mm_driver.options, self.mm_driver.client.token
)
_LOGGER.info(_("Connected successfully"))**kwargs:
filepath: (String) optional filepath to check for files to upload
Returns:
"""
print("[*] doing post message")
try:
self.inst.login()
team_id = self.inst.teams.get_team_by_name(self.team_name)['id']
channel_id = self.inst.channels.get_channel_by_name(
channel_name=self.channel_name, team_id=team_id)['id']
except exceptions.NoAccessTokenProvided as er:
print(f"[!] NoAccessTokenProvided {er}")
logger.exception()
except exceptions.InvalidOrMissingParameters as er:
print(f"[!] InvalidOrMissingParameters {er}")
logger.exception()
try:
if isfile(self.filepath):
file_ids = [
self.inst.files.upload_file(
channel_id=channel_id, files={
'files': (
basename(
self.filepath), open(
join_abs(
self.filepath), 'rb'))})['file_infos'][0]['id']]
self.inst.posts.create_post(options={
'channel_id': channel_id,self.inst.posts.create_post(options={
'channel_id': channel_id,
'message': f"Recon Data {datetime.datetime.now()}",
'file_ids': file_ids
})
elif isdir(self.filepath):
file_location = abspath(self.filepath)
self._upload_files(file_location, channel_id)
except exceptions.ContentTooLarge as er:
print(f"[!] ContentTooLarge {er}")
logger.exception()
except exceptions.ResourceNotFound as er:
print(f"[!] ResourceNotFound {er}")
logger.exception()
except OSError as er:
print(f"[!] File not found {er}")
logger.exception()
