diff --git a/src/devranta/api_requests.py b/src/devranta/api_requests.py new file mode 100644 index 0000000..b714b4b --- /dev/null +++ b/src/devranta/api_requests.py @@ -0,0 +1,210 @@ +# THIS IS A REQUESTS VERSION THAT IS EASIER TO TRANSLATE TO C USING LLM. +# WHILE WORKING PERFECTLY, IT'S NOT MADE TO BE USED. USE THE ASYNC ONE. +# - retoor + +from typing import Literal, Optional +import requests +from enum import Enum + +class VoteReason(Enum): + NOT_FOR_ME = 0 + REPOST = 1 + OFFENSIVE_SPAM = 2 + +class Api: + + base_url = "https://www.devrant.io/api/" + + def __init__(self, username=None, password=None): + self.username = username + self.password = password + self.auth = None + self.app_id = 3 + self.user_id = None + self.token_id = None + self.token_key = None + self.session = requests.Session() + + def patch_auth(self, request_dict=None): + auth_dict = {"app": self.app_id} + if self.auth: + auth_dict.update( + user_id=self.user_id, token_id=self.token_id, token_key=self.token_key + ) + if not request_dict: + return auth_dict + request_dict.update(auth_dict) + return request_dict + + def patch_url(self, url: str): + return self.base_url.rstrip("/") + "/" + url.lstrip("/") + + def login(self): + if not self.username or not self.password: + raise Exception("No authentication details supplied.") + response = self.session.post( + url=self.patch_url("users/auth-token"), + data={ + "username": self.username, + "password": self.password, + "app": self.app_id, + }, + ) + obj = response.json() + if not obj.get("success"): + return False + self.auth = obj.get("auth_token") + if not self.auth: + return False + self.user_id = self.auth.get("user_id") + self.token_id = self.auth.get("id") + self.token_key = self.auth.get("key") + return self.auth and True or False + + def ensure_login(self): + if not self.auth: + return self.login() + return True + + def register_user(self, email, username, password): + response = self.session.post( + url=self.patch_url(f"users"), + data=self.patch_auth({ + "email": email, + "username": username, + "password": password, + "plat": 3 + }), + ) + if not response: + return False + obj = response.json() + return obj.get('success', False) + + def get_comments_from_user(self, username): + user_id = self.get_user_id(username) + profile = self.get_profile(user_id) + return profile.get("content", {}).get("content", {}).get("comments", []) + + def post_comment(self, rant_id, comment): + if not self.ensure_login(): + return False + response = self.session.post( + url=self.patch_url(f"devrant/rants/{rant_id}/comments"), + data=self.patch_auth({"comment": comment, "plat": 2}), + ) + obj = response.json() + return obj.get("success", False) + + def get_comment(self, id_): + response = self.session.get( + url=self.patch_url("comments/" + str(id_)), params=self.patch_auth() + ) + obj = response.json() + + if not obj.get("success"): + return None + + return obj.get("comment") + + def delete_comment(self, id_): + if not self.ensure_login(): + return False + response = self.session.delete( + url=self.patch_url("comments/" + str(id_)), params=self.patch_auth() + ) + obj = response.json() + return obj.get("success", False) + + def get_profile(self, id_): + response = self.session.get( + url=self.patch_url(f"users/{id_}"), params=self.patch_auth() + ) + obj = response.json() + if not obj.get("success"): + return None + return obj.get("profile") + + def search(self, term): + response = self.session.get( + url=self.patch_url("devrant/search"), + params=self.patch_auth({"term": term}), + ) + obj = response.json() + if not obj.get("success"): + return + return obj.get("results", []) + + def get_rant(self, id): + response = self.session.get( + self.patch_url(f"devrant/rants/{id}"), + params=self.patch_auth(), + ) + return response.json() + + def get_rants(self, sort="recent", limit=20, skip=0): + response = self.session.get( + url=self.patch_url("devrant/rants"), + params=self.patch_auth({"sort": sort, "limit": limit, "skip": skip}), + ) + obj = response.json() + if not obj.get("success"): + return + return obj.get("rants", []) + + def get_user_id(self, username): + response = self.session.get( + url=self.patch_url("get-user-id"), + params=self.patch_auth({"username": username}), + ) + obj = response.json() + if not obj.get("success"): + return None + return obj.get("user_id") + + @property + def mentions(self): + return [ + notif for notif in self.notifs if notif["type"] == "comment_mention" + ] + + def update_comment(self, comment_id, comment): + if not self.ensure_login(): + return None + response = self.session.post( + url=self.patch_url(f"comments/{comment_id}"), + data=self.patch_auth({"comment": comment}), + ) + obj = response.json() + return obj.get("success", False) + + def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None): + if not self.ensure_login(): + return None + response = self.session.post( + url=self.patch_url(f"devrant/rants/{rant_id}/vote"), + data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}), + ) + obj = response.json() + return obj.get("success", False) + + def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None): + if not self.ensure_login(): + return None + response = self.session.post( + url=self.patch_url(f"comments/{comment_id}/vote"), + data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}), + ) + obj = response.json() + return obj.get("success", False) + + @property + def notifs(self): + if not self.ensure_login(): + return + response = self.session.get( + url=self.patch_url("users/me/notif-feed"), params=self.patch_auth() + ) + return response.json().get("data", {}).get("items", []) + +