import time import requests from bs4 import BeautifulSoup import traceback import backoff from functools import lru_cache class TimeOutException(Exception): pass ''' Wrapper around the RDRama API ''' class RDramaAPIInterface: def __init__(self, authorization_token, site, https: bool = True) -> None: self.headers={"Authorization": authorization_token} self.site = site self.protocol = "https" if https else "http" self.user_information = None self.badge_refresh = False def is_chudded(self): return "Chud" in self.get_my_badges_simple() def is_marsey_awarded(self): return "Marsey Award" in self.get_my_badges_simple() def is_bird_site(self): return "Bird Site Award" in self.get_my_badges_simple() def is_pizzashilled(self): return "Pizzashill Award" in self.get_my_badges_simple() def get_agendaposter_phrase(self): return self.get_me()['agendaposter_phrase'] def is_banned(self): return self.get_me()['is_banned'] @lru_cache(maxsize=None) def get_my_badges_simple(self): return [i['name'] for i in self.get_my_badges()] def get_my_badges(self): return self.get_me()['badges'] def make_post(self, title, submission_url, body, notify = True, hole=None): url=f"{self.protocol}://{self.site}/submit" return self.post(url, data={'title' : title, 'url': submission_url, 'body': body, 'notify': "on" if notify else "off", "sub":hole}) @lru_cache(maxsize=None) def get_me(self): self.get_front_page() url=f"{self.protocol}://{self.site}/@me" return self.get(url) ''' Sends a message to a user. ''' def send_message(self, username, message): url=f"{self.protocol}://{self.site}/@{username}/message" return self.post(url, data={'message':message}) ''' Replies to the comment with the given id. ''' def reply_to_comment(self,parent_fullname, parent_submission, message, file=None): url=f"{self.protocol}://{self.site}/comment" if file == None: return self.post(url, data={ 'parent_fullname':parent_fullname, 'submission': parent_submission, "body": message }) else: return self.post(url, data={ 'parent_fullname':parent_fullname, 'submission': parent_submission, "body": message }, files=file) ''' Replies to the comment with the given id. ''' def reply_to_comment_easy(self,comment_id, parent_submission, message, file=None): return self.reply_to_comment(f"c_{comment_id}", parent_submission, message, file=file) def reply_to_post(self, post_id, message): return self.reply_to_comment(f"p_{post_id}", post_id, message) def comment_on_wall(self, user_id, message): return self.reply_to_comment(f"u_{user_id}", None, message) ''' Gets "all" comments. ''' def get_comments(self, number_of_pages=1, user=None, sort="new", upper_bound = 0, lower_bound = 0, t="all"): return self.search("comments", number_of_pages, user, sort, upper_bound, lower_bound, t) ''' Gets "all" posts. ''' def get_posts(self, number_of_pages=1, user=None, sort="new", upper_bound = 0, lower_bound = 0, t="all"): return self.search("", number_of_pages, user, sort, upper_bound, lower_bound, t) def search(self, root, number_of_pages, user, sort, upper_bound, lower_bound, t): if (user == None): url=f"{self.protocol}://{self.site}/{root}" else: url=f"{self.protocol}://{self.site}/@{user}/{root}" params = f"?sort={sort}&t={t}&before={upper_bound}&after={lower_bound}" url+=params if number_of_pages == 1: return self.get(url) else: results = [] for i_ in range(number_of_pages): i = i_ + 1 full_url=f"{url}&page={i}" results += self.get(full_url)['data'] return { 'data': results } ''' Calls the notifications endpoint ''' def get_notifications(self, page : int): url=f"{self.protocol}://{self.site}/notifications?page={page}" return self.get(url) def reply_to_direct_message(self, message_id : int, message : str): url=f"{self.protocol}://{self.site}/reply" return self.post(url, data = { 'parent_id' : message_id, 'body': message }) def get_comment(self, id): url=f"{self.protocol}://{self.site}/comment/{id}" return self.get(url) def get_front_page(self): url=f"{self.protocol}://{self.site}" self.badge_refresh = True return self.get(url) def get_hole(self, hole: str): url = f"{self.protocol}://{self.site}/h/{hole}" return self.get(url) def has_url_been_posted(self, the_url): url=f"{self.protocol}://{self.site}/is_repost" return self.post(url, {'url': the_url})['permalink'] != '' def get_user_information(self, id): url=f"{self.protocol}://{self.site}/{id}/info" return self.get(url) ''' I have no clue what this is supposed to do, lol. ''' def clear_notifications(self): url=f"{self.protocol}://{self.site}/clear" return self.post(url, headers=self.headers) def get_unread_notifications(self): url=f"{self.protocol}://{self.site}/unread" return self.get(url) def give_coins(self, user, amount): url=f"{self.protocol}://{self.site}/@{user}/transfer_coins" return self.post(url, data={'amount':amount}) def get_post(self, id): url=f"{self.protocol}://{self.site}/post/{id}" return self.get(url) ''' Given a notification, returns whether or not the message is from Drama (ie, the messenger) ''' def is_message_from_drama(self,notification) -> bool: return notification['author_name'] == "Drama" or notification['author']['id'] == 1 ''' IMPLYING THAT THE MESSAGE IS FROM DRAMA, determines whether or not the notification is a gift transaction. ''' def is_message_is_a_gift_transaction(self,notification) -> bool: soup = BeautifulSoup(notification['body_html'], 'html.parser') p = soup.p #The first element is :marseycapitalistmanlet:. If not, we know this isn't a gift marsey_capitalist_manlet = p.contents[0] return (marsey_capitalist_manlet.name == "img" and marsey_capitalist_manlet['alt'] == ":marseycapitalistmanlet:") ''' Whether or not the message is a follow notification. ''' def is_message_a_follow_notification(self, notification): return "has followed you!" in notification['body_html'] ''' Whether or not the message is an unfollow notification. ''' def is_message_an_unfollow_notification(self, notification): return "has unfollowed you!" in notification['body_html'] ''' Parses a gift transaction ''' def parse_gift_transaction(self, notification): soup = BeautifulSoup(notification['body_html'], 'html.parser') p = soup.p #The third element is the username. It is a hyperlink tag containing the name of the user user_element = p.contents[2] user_id = user_element['href'].split("/")[2] #the hyperlink is formatted like so: /id/x, where x is the id user_name = user_element.contents[1].string[1:] #the username is within the image tag. we remove the first character, which is an @ #the fourth element is the "gift" string. It looks like this " has gifted you x coins". amount_string = p.contents[3] amount = amount_string.string.split(" ")[4] return { "type": "transfer", "user_name": user_name, "user_id": int(user_id), "amount": int(amount), "id": notification['id'] } ''' parses a post mention ''' def parse_post_mention(self, notification): soup = BeautifulSoup(notification['body_html'], 'html.parser') p = soup.p #The first element is the username. It is a hyperlink tag containing the name of the user user_element = p.contents[0] user_id = user_element['href'].split("/")[2] #the hyperlink is formatted like so: /id/x, where x is the id user_name = user_element.contents[1].string[1:] #the username is after the img tag. we remove the first character, which is an @ post_element = p.contents[-1] post_id = post_element['href'].split("/")[2] post_name = post_element.string return { "type": "post_mention", "user_name": user_name, "user_id": int(user_id), "id": notification['id'], "post_name": post_name, "post_id": post_id } ''' parses a follow notification ''' def parse_follow_notification(self, notification): soup = BeautifulSoup(notification['body_html'], 'html.parser') p = soup.p #The first element is the username. It is a hyperlink tag containing the name of the user user_element = p.contents[0] user_id = user_element['href'].split("/")[2] #the hyperlink is formatted like so: /id/x, where x is the id user_name = user_element.contents[1].string[1:] #the username is after the img tag. we remove the first character, which is an @ return { "type": "follow", "user_name": user_name, "user_id": int(user_id), "id": notification['id'] } ''' parses an unfollow notification ''' def parse_unfollow_notification(self, notification): soup = BeautifulSoup(notification['body_html'], 'html.parser') p = soup.p #The first element is the username. It is a hyperlink tag containing the name of the user user_element = p.contents[0] user_id = user_element['href'].split("/")[2] #the hyperlink is formatted like so: /id/x, where x is the id user_name = user_element.contents[1].string[1:] #the username is after the img tag. we remove the first character, which is an @ return { "type": "unfollow", "user_name": user_name, "user_id": int(user_id), "id": notification['id'] } ''' parses a dm ''' def parse_direct_message(self, notification): soup = BeautifulSoup(notification['body_html'], "html.parser") message = ''.join(soup.findAll(text=True)) return { "type": "direct_message", "user_name": notification['author_name'], "user_id": notification['author']['id'], "id": notification['id'], "message_html": notification['body_html'], "message": message } ''' parses a reply to a comment. ''' def parse_comment_reply(self, notification): #TODO: Lots of changes needed. return { "type": "comment_reply", "parent_id": notification["id"], "post_name": notification["post"]["title"], "post_id": notification["post"]["id"] } def parse_comment_mention(self, notification): #TODO: Work needs to be done here, I removed some stuff return { "type": "comment_mention", "user_name": notification['author_name'], "user_id": notification['author']['id'], "id": notification["id"], "message": notification['body'], "parent_comment_id": notification['parent_comment_id'] if notification['level'] != 1 else None, "post_id": notification['post_id'] } ''' Returns a list of notifications in an easy to process list. ''' def get_parsed_notification(self): to_return = [] notifications = self.get_unread_notifications()['data'] if (notifications == []): return [] for notification in notifications: parsed_notification = {} try: if self.is_message_from_drama(notification): if (self.is_message_is_a_gift_transaction(notification)): parsed_notification = self.parse_gift_transaction(notification) elif ("has mentioned you: " in notification['body_html']): parsed_notification = self.parse_post_mention(notification) elif (self.is_message_a_follow_notification(notification)): parsed_notification = self.parse_follow_notification(notification) elif (self.is_message_an_unfollow_notification(notification)): parsed_notification = self.parse_unfollow_notification(notification) elif (self.welcome_message in notification['body_html']): #Welcome message pass elif ("if you don't know what to do next" in notification['body_html']): #API approval message pass else: pass elif notification['post_id'] == 0: #Direct message parsed_notification = self.parse_direct_message(notification) else: #comment mention parsed_notification = self.parse_comment_mention(notification) to_return.append(parsed_notification) except BaseException as e: print(f"Exception {e}") print(f"Notification: {notification}") traceback.print_exc() return to_return @backoff.on_exception(backoff.expo, requests.exceptions.RequestException) def get(self, url): response = requests.get(url, headers=self.headers) print(f"GET {url} ({response.status_code})") if (response.status_code == 429): raise requests.exceptions.RequestException() if (response.status_code != 200): raise BaseException(f"GET {url} ({response.status_code}) {response.json()}") else: return response.json() @backoff.on_exception(backoff.expo, TimeOutException) def post(self, url, data, files = None): if files == None: response = requests.post(url, headers=self.headers, data=data) else: response = requests.post(url, headers=self.headers, data=data, files=files) print(f"POST {url} ({response.status_code}) {data}") if (response.status_code == 429): raise TimeOutException if (response.status_code != 200): raise BaseException(f"POST {url} ({response.status_code}) {data} => {response.json()}") else: return response.json()