import logging import json import requests import msal class MSGraphApi(object): """Docstring for MSGraphApi. """ def __init__(self, config=None, flow="app"): # TODO: make these private vars self.config = {} self.app = None self.access_token = None if config is not None: self.config = config else: return self.flow = flow if self.flow == "obo": app = msal.PublicClientApplication( self.config["client_id"], authority=self.config["authority"] ) else: # no flow was given, assume app flow app = msal.ConfidentialClientApplication( self.config["client_id"], authority=self.config["authority"], client_credential=self.config["secret"], ) self.app = app def getAuth(self): """Docstring for getAuth. Get the MS Authorization to use the Graph API :returns: on success, the instance of this object - self on failure, None """ # The pattern to acquire a token looks like this. result = None # Firstly, looks up a token from cache if self.flow == "obo": result = self.app.acquire_token_silent( self.config["scope"], account=self.app.get_accounts(self.config["username"]), ) else: # Since we are looking for token for the current app, NOT for an end user, # notice we give account parameter as None. result = self.app.acquire_token_silent(self.config["scope"], account=None) if not result: logging.info( "No suitable token exists in cache. Let's get a new one from AAD." ) if self.flow == "obo": result = self.app.acquire_token_by_username_password( self.config["username"], self.config["password"], scopes=self.config["scope"], ) else: result = self.app.acquire_token_for_client(scopes=self.config["scope"]) # auth succeeded if "access_token" in result: self.setAccessToken(result["access_token"]) return self else: logging.error(result.get("error")) logging.error(result.get("error_description")) logging.error( result.get("correlation_id") ) # may need this when reporting a bug return None def setAccessToken(self, tok): """Docstring for setAccessToken. A setter method to set access_token for the object instance. :tok: the ms access token """ self.access_token = tok def getUsers(self): """Docstring for getUsers. Get all the users in the ms org. :returns: respose object containing ms response """ endpoint = "https://graph.microsoft.com/v1.0/users" headers = {"Authorization": "Bearer " + self.access_token} graph_data = requests.get(endpoint, headers=headers) return graph_data def getMyTeams(self): """Docstring for getUsers. Get all the teams for this user. :returns: respose object containing ms response """ endpoint = "https://graph.microsoft.com/beta/me/joinedTeams" headers = {"Authorization": "Bearer " + self.access_token} graph_data = requests.get(endpoint, headers=headers) return graph_data def getChatsByTeam(self, teamId): """Docstring for getChatsByTeam. Get all chats in given team id. :teamId: the team ID for the team to get the chats for :returns: respose object containing ms response """ endpoint = "https://graph.microsoft.com/beta/teams/%s/channels/" % (teamId) headers = {"Authorization": "Bearer " + self.access_token} graph_data = requests.get(endpoint, headers=headers) return graph_data def getChatByTeamAndId(self, teamId, channelId): """Docstring for getChatByTeamAndId. Get a specific channel given team and channel ids. :teamId: the team ID for the team to get the chats for :channelId: the channel ID for the channel to get :returns: respose object containing ms response """ endpoint = "https://graph.microsoft.com/beta/teams/%s/channels/%s" % ( teamId, channelId, ) headers = {"Authorization": "Bearer " + self.access_token} graph_data = requests.get(endpoint, headers=headers) return graph_data def writeChatMessageByTeamAndId(self, teamId, channelId, payload): """Docstring for writeChatMessageByTeamAndId. :teamId: the team ID for the team to post the chat for :channelId: the chat/channel ID for the channel to post to :payload: the exact POST dict payload to be used :returns: respose object containing ms response """ # other pathways # userId = "8a18e679-3603-4e10-a5e4-c75ad362fde0" # endpoint = "https://graph.microsoft.com/beta/users/%s/chats/%s/messages" % (userId, channelId) # endpoint = "https://graph.microsoft.com/beta/chats/%s/messages" % (channelId) # ms recommended pathway endpoint = "https://graph.microsoft.com/beta/teams/%s/channels/%s/messages/" % ( teamId, channelId, ) headers = { "Authorization": "Bearer " + self.access_token, "Content-Type": "application/json", } graph_data = requests.post(endpoint, headers=headers, data=json.dumps(payload)) return graph_data def getTeamByName(self, teamName): """TODO: Docstring for getTeamByName. :teamName: TODO :returns: TODO """ # get all my teams teamsLib = self.getMyTeams().json() if "value" in teamsLib: teams = {} for team in teamsLib["value"]: teams[team["id"]] = team["displayName"] # search lib for my team teamid = self.searchValuesinDict(teams, teamName) return teamid return "" def getChatByNameAndTeamId(self, teamId, channelName): """TODO: Docstring for getChatByName. :channelName: TODO :returns: TODO """ # get all channels in the Teams channelsLib = self.getChatsByTeam(teamId).json() if "value" in channelsLib: channels = {} for channel in channelsLib["value"]: channels[channel["id"]] = channel["displayName"] # search lib for my channel chatId = self.searchValuesinDict(channels, channelName) return chatId return "" @staticmethod def buildTeamsMessage(customMessage): """Docstring for buildTeamsMessage. Build a Teams compatible message as per spec: https://docs.microsoft.com/en-us/graph/api/resources/chatmessage?view=graph-rest-beta :customMessage: an dict containing the "name" and "text" fields :returns: a valid Teams message dict payload """ # schema teams_mssg = { "messageType": "message", "body": {"content": "", "contentType": "html"}, } # convert ``` snippets into block quotes ie. #
some code
rich_html_content = MSGraphApi.insertCodeBlocksInMessage(customMessage["text"]) html_content = ( "" + "

From: %s %s

" % (customMessage["name"], customMessage["ts"]) + "

%s

" % (rich_html_content) + "" ) split_content = html_content.split("\n") if len(split_content) > 1: for i, line in enumerate(split_content): split_content[i] = "

{}

".format(line) split_content_str = "".join(split_content) # TODO: sanitize string # teams_mssg["body"]["content"] = html_content teams_mssg["body"]["content"] = split_content_str return teams_mssg @staticmethod def insertCodeBlocksInMessage(text): """Docstring for insertCodeBlocksInMessage. Replace ``` code ``` to be html compliant, i.e. wrapped in blockquotes :text: the input string possibly containing ``` :returns: a string with html codeblocks instead of ``` """ toks = text.split("```") for i, tok in enumerate(toks): # IMPORTANT! this assumes that every ``` comes in pairs if (i % 2) != 0: toks[i] = "
{}
".format(tok) quoted_str = "".join(toks) return quoted_str @staticmethod def searchValuesinDict(lib, value): """TODO: Docstring for searchValuesinDict. :returns: the key of the dict where the value matches """ for key in lib.keys(): if lib[key] == value: return key return ""