pnut-matrix/pnutlib.py

import requests
import json

# API root:  https://api.pnut.io/v0

class Pnut:

    def __init__(self, token=""):
        self.token = token
        self.username = self._getme()

    def logout(self):
        url = "https://api.pnut.io/v0/token";
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.delete(url, headers=headers)
        return r

    def get_stream(self,since_id=None):
        url = "https://api.pnut.io/v0/posts/streams/me"
        headers = { "Authorization": "Bearer " + self.token }
        parameters = {}
        if since_id:
            parameters = { "since_id": since_id }
        r = requests.get(url, headers=headers, params=parameters)
        return r

    def get_unified_stream(self,since_id=None):
        url = "https://api.pnut.io/v0/posts/streams/unified"
        headers = { "Authorization": "Bearer " + self.token }
        parameters = {}
        if since_id:
            parameters = { "since_id": since_id }
        r = requests.get(url, headers=headers, params=parameters)
        return r

    def get_global_stream(self,since_id=None):
        url = "https://api.pnut.io/v0/posts/streams/global?include_raw=1"
        headers = { "Authorization": "Bearer " + self.token }
        parameters = {}
        if since_id:
            parameters = { "since_id": since_id }
        try:
            r = requests.get(url, headers=headers, params=parameters)
        except requests.exceptions.ConnectionError:
            r = None
        except:
            r = None
            raise
        return r

    def get_channel_stream(self,chan_id,since_id=None):
        url = "https://api.pnut.io/v0/channels/" + str(chan_id) + "/messages?include_raw=1"
        headers = { "Authorization": "Bearer " + self.token }
        parameters = {}
        if since_id:
            parameters = { "since_id": since_id }
        try:
            r = requests.get(url, headers=headers, params=parameters)
        except requests.exceptions.ConnectionError:
            r = None
        except:
            r = None
            raise
        return r

    def get_user(self, uid):
        url = "https://api.pnut.io/v0/users/" + str(uid)
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def get_post(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid) + "?include_raw=1"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def get_bookmarks(self):
        url = "https://api.pnut.io/v0/users/me/bookmarks"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def add_bookmark(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid) + "/bookmark"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.put(url, headers=headers)
        return r

    def delete_bookmark(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid) + "/bookmark"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.delete(url, headers=headers)
        return r

    def add_repost(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid) + "/repost"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.put(url, headers=headers)
        return r

    def delete_repost(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid) + "/repost"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.delete(url, headers=headers)
        return r

    def follow_user(self, uid):
        url = "https://api.pnut.io/v0/users/" + str(uid) + "/follow"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.put(url, headers=headers)
        return r

    def unfollow_user(self, uid):
        url = "https://api.pnut.io/v0/users/" + str(uid) + "/follow"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.delete(url, headers=headers)
        return r

    def get_followers(self):
        url = "https://api.pnut.io/v0/users/me/followers"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def get_following(self):
        url = "https://api.pnut.io/v0/users/me/following"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def get_mentions(self):
        url = "https://api.pnut.io/v0/users/me/mentions"
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        return r

    def post(self, text):
        url = "https://api.pnut.io/v0/posts"
        headers = { "Authorization": "Bearer " + self.token }
        body = {
            "text": text
        }
        r = requests.post(url, data=body, headers=headers)
        return r

    def channel_post(self, chan_id, text, raw):
        url = "https://api.pnut.io/v0/channels/" + str(chan_id) + "/messages"
        headers = {
            "Authorization": "Bearer " + self.token,
            "Content-Type": "application/json"
        }
        data = {"text": text }
        if raw:
            url += "?include_raw=1"
            data["raw"] = raw
        body = json.dumps(data)
        r = requests.post(url, data=body, headers=headers)
        return r

    def delete_channel_post(self, chan_id, msg_id):
        url = "https://api.pnut.io/v0/channels/" + str(chan_id) + "/messages/" + str(msg_id)
        headers = {
            "Authorization": "Bearer " + self.token,
            "Content-Type": "application/json"
        }
        r = requests.delete(url, data=json.dumps({}), headers=headers)
        return r

    def global_post_oembed(self, text, embed):
        url = "https://api.pnut.io/v0/posts?include_raw=1"
        headers = {
            "Authorization": "Bearer " + self.token,
            "Content-Type": "application/json"
        }
        body = json.dumps({"text": text, "raw": embed})
        r = requests.post(url, data=body, headers=headers)
        return r

    def process_text(self, text):
        url = "https://api.pnut.io/v0/text/process"
        headers = { "Authorization": "Bearer " + self.token }
        body = {
            "text": text
        }
        r = requests.post(url, json=body, headers=headers)
        return r

    def reply(self, postid, text):
        url = "https://api.pnut.io/v0/posts"
        headers = { "Authorization": "Bearer " + self.token }
        me = self.username
        author = self._getauthor(postid)
        if author == me or author == None:
            mtext = text
        else:
            mtext = "@%s %s" % (author, text)
        body = {}
        body["text"] = mtext
        if len(postid) > 0:
            body["reply_to"] = postid
        r = requests.post(url, data=body, headers=headers)
        return r

    def replyall(self, postid, text):
        url = "https://api.pnut.io/v0/posts"
        headers = { "Authorization": "Bearer " + self.token }
        me = self.username
        author = self._getauthor(postid)
        mentions = self._getmentions(postid)
        mtext = "@%s " % (author)
        mentions = [m for m in mentions if m != me]
        mentions = [m for m in mentions if m != author]
        if mentions:
            mtext += " "
            for nick in mentions:
                mtext += "@%s " % nick
            mtext += ""
        mtext += text
        body = {}
        body["text"] = mtext
        if len(str(postid)) > 0:
            body["reply_to"] = postid
        r = requests.post(url, data=body, headers=headers)
        return r

    def _getme(self):
        if len(self.token) > 0:
            url = "https://api.pnut.io/v0/users/me"
            headers = { "Authorization": "Bearer " + self.token }
            r = requests.get(url, headers=headers)
            if r.status_code == 200:
                rdata = rdata = json.loads(r.text)
                return rdata["data"]["username"]
            else:
                return None
        else:
            return None

    def _getpost(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid)
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        if r.status_code == 200:
            rdata = json.loads(r.text)
        else:
            return None
        return rdata["data"]

    def _getauthor(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid)
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        if r.status_code == 200:
            rdata = json.loads(r.text)
        else:
            return None
        return rdata["data"]["user"]["username"]

    def _getmentions(self, postid):
        url = "https://api.pnut.io/v0/posts/" + str(postid)
        headers = { "Authorization": "Bearer " + self.token }
        r = requests.get(url, headers=headers)
        if r.status_code == 200:
            rdata = json.loads(r.text)
        else:
            return None
        mentions = []
        entities = rdata["data"]["content"]["entities"]
        if "mentions" in entities:
            for m in entities["mentions"]:
                mentions.append(m["text"])
        return mentions