re-add the pnut class
This commit is contained in:
parent
f313b2159f
commit
34e580284d
2 changed files with 271 additions and 0 deletions
|
@ -2,6 +2,7 @@ import json
|
|||
import requests
|
||||
import logging
|
||||
|
||||
from pnutlib import Pnut
|
||||
from flask import Flask, jsonify, request, abort
|
||||
from models import *
|
||||
|
||||
|
|
270
pnutlib.py
Normal file
270
pnutlib.py
Normal file
|
@ -0,0 +1,270 @@
|
|||
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
|
Loading…
Reference in a new issue