270 lines
9.2 KiB
Python
270 lines
9.2 KiB
Python
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 }
|
|
headers = {}
|
|
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):
|
|
url = "https://api.pnut.io/v0/channels/" + str(chan_id) + "/messages"
|
|
headers = {
|
|
"Authorization": "Bearer " + self.token,
|
|
"Content-Type": "application/json"
|
|
}
|
|
body = json.dumps({"text": text })
|
|
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
|