-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
0 parents
commit 5cdaefb
Showing
5 changed files
with
158 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,5 @@ | ||
.vscode/ | ||
__pycache__/ | ||
caches/ | ||
|
||
*ignored.* |
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,36 @@ | ||
import time | ||
import urllib | ||
|
||
import requests | ||
|
||
|
||
class SnooNotesAuth: | ||
def __init__(self, username, user_key): | ||
self.username = username | ||
self.user_key = user_key | ||
self.base_url = "https://snoonotes.com/" | ||
|
||
def get_access_token(self): | ||
|
||
url = urllib.parse.urljoin(self.base_url, "/auth/connect/token") | ||
r = requests.post(url, data={ | ||
"grant_type": "password", | ||
"username": self.username, | ||
"password": self.user_key, | ||
"client_id": "bots" | ||
}) | ||
if r.ok: | ||
print(f"Authenticated as user: {self.username}") | ||
response = r.json() | ||
self.access_token = response['access_token'] | ||
self.expire_time = round(time.time()) + response['expires_in'] | ||
self.token_type = response['token_type'] | ||
else: | ||
print(f"Failed with code: {r.status_code}") | ||
|
||
def refresh_access_token(self): | ||
if round(time.time()) > self.expire_time: | ||
print("Token expired, refreshing.") | ||
self.get_access_token() | ||
else: | ||
return |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,101 @@ | ||
import os | ||
import pickle | ||
import time | ||
import urllib | ||
from pprint import pprint | ||
|
||
import requests | ||
|
||
from .auth import SnooNotesAuth | ||
|
||
|
||
class SnooNotes(SnooNotesAuth): | ||
def __init__(self, username, user_key): | ||
super().__init__(username, user_key) | ||
# Create a cache directory | ||
if not os.path.exists("./caches"): | ||
os.mkdir("./caches/") | ||
|
||
def _endpoint_url(self, endpoint): | ||
"""Internal helper function to generate url from an endpoint""" | ||
return urllib.parse.urljoin(self.base_url, endpoint) | ||
|
||
def _authed_request(self, request_type, endpoint, data=None): | ||
"""Internal helper function that performs the request after authenticating as the user""" | ||
try: | ||
if self.access_token is not None: | ||
pass | ||
except: | ||
self.get_access_token() | ||
# Optionally refresh_access_tokens. | ||
self.refresh_access_token() | ||
|
||
if request_type == "POST" and data == None: | ||
print("No data provided for POST") | ||
return | ||
|
||
if request_type == "POST": | ||
r = requests.post(url=self._endpoint_url(endpoint), headers={ | ||
"Authorization": f"{self.token_type} {self.access_token}" | ||
}, json=data) | ||
elif request_type == "GET": | ||
r = requests.get(url=self._endpoint_url(endpoint), headers={ | ||
"Authorization": f"{self.token_type} {self.access_token}" | ||
}) | ||
if r.ok: | ||
return r.json() | ||
else: | ||
print(f"Failed with Error: {r.status_code}") | ||
exit() | ||
|
||
def add_note_for_user(self, username, note_type, subreddit, message, url): | ||
"""Add a Snoonote to the `username` under the specific `subreddit`. | ||
:param username: User to add note for | ||
:param subreddit: Subreddit under which the note should be added | ||
:param message: The content of the note | ||
:param url: The url to the reddit comment/post | ||
""" | ||
self._authed_request("POST", "/api/note", data={ | ||
"NoteTypeID": note_type, | ||
"SubName": subreddit, | ||
"Message": message, | ||
"AppliesToUsername": username, | ||
"Url": url | ||
}) | ||
pass | ||
|
||
def get_notes_for_user(self, usernames): | ||
"""Returns a dict with the usernames passed and the Snoonotes they have. | ||
:param usernames: `String` or a list of `Strings` that represent the Reddit usernames to query for. | ||
""" | ||
if isinstance(usernames, str): | ||
usernames = [usernames] | ||
return self._authed_request("POST", "/api/Note/GetNotes", data=usernames) | ||
|
||
def get_notes_for_subreddit(self, subreddit, use_cache=True): | ||
"""Return usernotes supported by a subreddit. Also caches the data for a day to prevent repeated calls to the API. Should therefore be very fast for repeated access. | ||
:param use_cache: Set to `False` to manually prevent cache use. | ||
""" | ||
if use_cache: | ||
if os.path.exists(f"./caches/{subreddit}"): | ||
with open(f"./caches/{subreddit}", "rb") as sub_notes_cache: | ||
cache = pickle.load(sub_notes_cache) | ||
if time.time() - int(cache.pop("LastCache")) > 86400: | ||
print("Last cache older than a day. Not using cache.") | ||
data = self._authed_request("GET", f"restapi/Subreddit/{subreddit}")[0] | ||
data["LastCache"] = round(time.time()) | ||
with open(f"./caches/{subreddit}", "wb") as sub_notes_cache: | ||
pickle.dump(data, sub_notes_cache) | ||
del data["LastCache"] | ||
return data | ||
else: | ||
return cache | ||
data = self._authed_request("GET", f"restapi/Subreddit/{subreddit}")[0] | ||
data["LastCache"] = round(time.time()) | ||
with open(f"./caches/{subreddit}", "wb") as sub_notes_cache: | ||
pickle.dump(data, sub_notes_cache) | ||
del data["LastCache"] | ||
return data |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
from SnooNotes.core import SnooNotes | ||
|
||
if __name__ == "__main__": | ||
|
||
# Authenticate with a reddit username and an user_key that can be generated from "https://snoonotes.com/#!/userkey". The generated key is sent via PM to your reddit account. | ||
sn = SnooNotes("username", "user_key") | ||
|
||
# Returns a dictionary with usernames and their associated usernotes. | ||
notes_for_username = sn.get_notes_for_user("username") | ||
notes_for_usernames12 = sn.get_notes_for_user(["username1", "username2"]) | ||
|
||
# Returns a dictionary to get NoteTypes supported in that particular subreddit | ||
subreddit_notes = sn.get_notes_for_subreddit("subreddit") | ||
|
||
# Add a new usernote to the user under the subreddit with a note with a link to the comment/post. | ||
sn.add_note_for_user("username","note_type", "subreddit","Reason for note", "www.reddit.com/r/subreddit/123jks/.../jkas12") |