NewsBlur-viq/vendor/appdotnet.py

import json

import requests

# To add
# - Identity Delegation
# - Streams (in dev by app.net)
# - Filters (in dev by app.net)


class Appdotnet:
    """Once access has been given, you don't have to pass through the
    client_id, client_secret, redirect_uri, or scope. These are just
    to get the authentication token.

    Once authenticated, you can initialise appdotnet with only the
    access token: ie

    api = Appdotnet(access_token='<insert token here>')
    """

    def __init__(self, client_id=None, client_secret=None, redirect_uri=None, scope=None, access_token=None):
        # for server authentication flow
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
        self.scope = scope

        self.access_token = access_token

        self.api_anchor = "alpha.app.net"  # for when the versions change
        # anchors currently different
        self.public_api_anchor = "alpha-api.app.net"

        # scopes provided by app.net API
        self.allowed_scopes = ["stream", "email", "write_post", "follow", "messages", "export"]

    def generateAuthUrl(self):
        url = (
            "https://"
            + self.api_anchor
            + "/oauth/authenticate?client_id="
            + self.client_id
            + "&response_type=code&adnview=appstore&redirect_uri="
            + self.redirect_uri
            + "&scope="
        )

        for scope in self.scope:
            if scope in self.allowed_scopes:
                url += scope + " "

        return url

    def getAuthResponse(self, code):
        # generate POST request
        url = "https://alpha.app.net/oauth/access_token"
        post_data = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "grant_type": "authorization_code",
            "redirect_uri": self.redirect_uri,
            "code": code,
        }

        r = requests.post(url, data=post_data)

        return r.text

    """
    API Calls
    """

    # GET REQUESTS
    def getRequest(self, url, getParameters=None):
        if not getParameters:
            getParameters = {}
        # access token
        url = url + "?access_token=" + self.access_token

        # if there are any extra get parameters aside from the access_token, append to the url
        if getParameters != {}:
            for key, value in getParameters.items():
                if not value:
                    continue
                url = url + "&" + key + "=" + str(value)
        print(url)
        r = requests.get(url)
        if r.status_code == requests.codes.ok:
            return r.text
        else:
            j = json.loads(r.text)
            resp = {"error_code": r.status_code, "message": j["error"]["message"]}
            return json.dumps(resp)

    def getUser(self, user_id):
        url = "https://%s/stream/0/users/%s" % (self.public_api_anchor, user_id)
        return self.getRequest(url)

    def getUserPosts(self, user_id):
        url = "https://%s/stream/0/users/%s/posts" % (self.public_api_anchor, user_id)
        return self.getRequest(url)

    def getUserStars(self, user_id):
        url = "https://%s/stream/0/users/%s/stars" % (self.public_api_anchor, user_id)
        return self.getRequest(url)

    def getGlobalStream(self):
        url = "https://%s/stream/0/posts/stream/global" % self.public_api_anchor
        return self.getRequest(url)

    def getUserStream(self):
        url = "https://%s/stream/0/posts/stream" % self.public_api_anchor
        return self.getRequest(url)

    def getUserMentions(self, user_id):
        url = "https://%s/stream/0/users/%s/mentions" % (self.public_api_anchor, user_id)
        return self.getRequest(url)

    def getPost(self, post_id):
        url = "https://%s/stream/0/posts/%s" % (self.public_api_anchor, post_id)
        return self.getRequest(url)

    def getReposters(self, post_id):
        url = "https://%s/stream/0/posts/%s/reposters" % (self.public_api_anchor, post_id)
        return self.getRequest(url)

    def getStars(self, post_id):
        url = "https://%s/stream/0/posts/%s/stars" % (self.public_api_anchor, post_id)
        return self.getRequest(url)

    def getPostReplies(self, post_id):
        url = "https://%s/stream/0/posts/%s/replies" % (self.public_api_anchor, post_id)
        return self.getRequest(url)

    def getPostsByTag(self, tag):
        url = "https://%s/stream/0/posts/tag/%s" % (self.public_api_anchor, tag)
        return self.getRequest(url)

    def getUserFollowing(self, user_id, since_id=None, before_id=None):
        url = "https://%s/stream/0/users/%s/following" % (self.public_api_anchor, user_id)
        return self.getRequest(
            url,
            getParameters={
                "since_id": since_id,
                "before_id": before_id,
            },
        )

    def getUserFollowingIds(self, user_id, since_id=None, before_id=None):
        url = "https://%s/stream/0/users/%s/following/ids" % (self.public_api_anchor, user_id)
        return self.getRequest(
            url,
            getParameters={
                "since_id": since_id,
                "before_id": before_id,
            },
        )

    def getUserFollowers(self, user_id):
        url = "https://%s/stream/0/users/%s/followers" % (self.public_api_anchor, user_id)
        return self.getRequest(url)

    def getMutedUsers(self):
        url = "https://%s/stream/0/users/me/muted" % self.public_api_anchor
        return self.getRequest(url)

    def searchUsers(self, q):
        url = "https://%s/stream/0/users/search" % (self.public_api_anchor)
        return self.getRequest(url, getParameters={"q": q})

    def getCurrentToken(self):
        url = "https://%s/stream/0/token" % self.public_api_anchor
        return self.getRequest(url)

    # POST REQUESTS
    def postRequest(self, url, data=None, headers=None):
        if not data:
            data = {}

        if not headers:
            headers = {}

        headers["Authorization"] = "Bearer %s" % self.access_token
        url = url
        r = requests.post(url, data=json.dumps(data), headers=headers)
        if r.status_code == requests.codes.ok:
            return r.text
        else:
            try:
                j = json.loads(r.text)
                resp = {"error_code": r.status_code, "message": j["error"]["message"]}
                return resp
            except:  # generic error
                print(r.text)
                return "{'error':'There was an error'}"

    def followUser(self, user_id):
        url = "https://%s/stream/0/users/%s/follow" % (self.public_api_anchor, user_id)
        return self.postRequest(url)

    def repostPost(self, post_id):
        url = "https://%s/stream/0/posts/%s/repost" % (self.public_api_anchor, post_id)
        return self.postRequest(url)

    def starPost(self, post_id):
        url = "https://%s/stream/0/posts/%s/star" % (self.public_api_anchor, post_id)
        return self.postRequest(url)

    def muteUser(self, user_id):
        url = "https://%s/stream/0/users/%s/mute" % (self.public_api_anchor, user_id)
        return self.postRequest(url)

    # requires: text
    # optional: reply_to, annotations, links
    def createPost(self, text, reply_to=None, annotations=None, links=None):
        url = "https://%s/stream/0/posts" % self.public_api_anchor
        if annotations != None:
            url = url + "?include_annotations=1"

        data = {"text": text}
        if reply_to != None:
            data["reply_to"] = reply_to
        if annotations != None:
            data["annotations"] = annotations
        if links != None:
            data["links"] = links

        return self.postRequest(url, data, headers={"content-type": "application/json"})

    # DELETE request
    def deleteRequest(self, url):
        url = url + "?access_token=" + self.access_token
        r = requests.delete(url)
        if r.status_code == requests.codes.ok:
            return r.text
        else:
            try:
                j = json.loads(r.text)
                resp = {"error_code": r.status_code, "message": j["error"]["message"]}
                return resp
            except:  # generic error
                print(r.text)
                return "{'error':'There was an error'}"

    def deletePost(self, post_id):
        url = "https://%s/stream/0/posts/%s" % (self.public_api_anchor, post_id)
        return self.deleteRequest(url)

    def unrepostPost(self, post_id):
        url = "https://%s/stream/0/posts/%s/repost" % (self.public_api_anchor, post_id)
        return self.deleteRequest(url)

    def unstarPost(self, post_id):
        url = "https://%s/stream/0/posts/%s/star" % (self.public_api_anchor, post_id)
        return self.deleteRequest(url)

    def unfollowUser(self, user_id):
        url = "https://%s/stream/0/users/%s/follow" % (self.public_api_anchor, user_id)
        return self.deleteRequest(url)

    def unmuteUser(self, user_id):
        url = "https://%s/stream/0/users/%s/mute" % (self.public_api_anchor, user_id)
        return self.deleteRequest(url)