From 474fbe0850ba5dc6a7aa8eb06c60d229af674dcb Mon Sep 17 00:00:00 2001 From: kd Date: Tue, 9 Jun 2020 16:26:53 -0400 Subject: initial commit --- msgraph.py | 292 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 292 insertions(+) create mode 100644 msgraph.py (limited to 'msgraph.py') diff --git a/msgraph.py b/msgraph.py new file mode 100644 index 0000000..0ed77bc --- /dev/null +++ b/msgraph.py @@ -0,0 +1,292 @@ +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 "" -- cgit v1.2.3