path: root/msgraph.py

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. #<blockquote> some code </blockquote>
        rich_html_content = MSGraphApi.insertCodeBlocksInMessage(customMessage["text"])

        html_content = (
            "<HTML><BODY>"
            + "<p>From: <strong>%s</strong> <small>%s</small></p>"
            % (customMessage["name"], customMessage["ts"])
            + "<p>%s</p>" % (rich_html_content)
            + "</BODY></HTML>"
        )

        split_content = html_content.split("\n")
        if len(split_content) > 1:
            for i, line in enumerate(split_content):
                split_content[i] = "<p>{}</p>".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] = "<blockquote>{}</blockquote>".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 ""