Update.

Reviewed-on: retoor/devranta#1

README.md

@@ -30,8 +30,4 @@ async def list_rants():
 ```
 See [tests](src/devranta/tests.py) for [examples](src/devranta/tests.py) on how to use. 
 
-## Todo 
-
- - voting comment 
- - edit message 
 

src/devranta.egg-info/PKG-INFO

@@ -43,8 +43,4 @@ async def list_rants():
 ```
 See [tests](src/devranta/tests.py) for [examples](src/devranta/tests.py) on how to use. 
 
-## Todo 
-
- - voting comment 
- - edit message 
 

src/devranta/api.py

@@ -1,5 +1,11 @@
+from typing import Literal, Optional
 import aiohttp
+from enum import Enum
 
+class VoteReason(Enum):
+    NOT_FOR_ME = 0
+    REPOST = 1
+    OFFENSIVE_SPAM = 2
 
 class Api:
 
@@ -65,6 +71,23 @@ class Api:
         await self.session.close()
         self.session = None
 
+    async def register_user(self, email, username, password):
+        response = None
+        async with self as session:
+            response = await 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 = await response.json()
+        return obj.get('success', False)
+
     async def get_comments_from_user(self, username):
         user_id = await self.get_user_id(username)
         profile = await self.get_profile(user_id)
@@ -94,6 +117,17 @@ class Api:
             return None
 
         return obj.get("comment")
+    
+    async def delete_comment(self, id_):
+        response = None
+        if not await self.ensure_login():
+            return False
+        async with self as session:
+            response = await session.delete(
+                url=self.patch_url("comments/" + str(id_)), params=self.patch_auth()
+            )
+        obj = await response.json()
+        return obj.get("success", False)
 
     async def get_profile(self, id_):
         response = None
@@ -168,6 +202,28 @@ class Api:
         obj = await response.json()
         return obj.get("success", False)
 
+    async def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None):
+        if not await self.ensure_login():
+            return None
+        async with self as session:
+            response = await 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 = await response.json()
+        return obj.get("success", False)
+
+    async def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None):
+        if not await self.ensure_login():
+            return None
+        async with self as session:
+            response = await 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 = await response.json()
+        return obj.get("success", False)
+
     @property
     async def notifs(self):
         response = None

src/devranta/api_plain.py

@@ -0,0 +1,249 @@
+import json
+import urllib.parse
+import http.client
+import functools 
+
+class Api:
+
+    base_url = "www.devrant.io"
+
+    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
+
+    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 login(self):
+        if not self.username or not self.password:
+            raise Exception("No authentication details supplied.")
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps({
+            "username": self.username,
+            "password": self.password,
+            "app": self.app_id,
+        })
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", "/api/users/auth-token", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        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 True
+
+    def ensure_login(self):
+        if not self.auth:
+            return self.login()
+        return True
+
+    @functools.lru_cache()
+    def register_user(self, email, username, password):
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps(self.patch_auth({
+            "email": email,
+            "username": username,
+            "password": password,
+            "plat": 3
+        }))
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", "/api/users", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get('success', False)
+
+    @functools.lru_cache()
+    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", [])
+
+    @functools.lru_cache()
+    def post_comment(self, rant_id, comment):
+        if not self.ensure_login():
+            return False
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps(self.patch_auth({"comment": comment, "plat": 2}))
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", f"/api/devrant/rants/{rant_id}/comments", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get("success", False)
+
+    @functools.lru_cache()
+    def get_comment(self, id_):
+        conn = http.client.HTTPSConnection(self.base_url)
+        conn.request("GET", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        if not obj.get("success"):
+            return None
+        return obj.get("comment")
+
+    @functools.lru_cache()
+    def delete_comment(self, id_):
+        if not self.ensure_login():
+            return False
+        conn = http.client.HTTPSConnection(self.base_url)
+        conn.request("DELETE", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get("success", False)
+
+    @functools.lru_cache()
+    def get_profile(self, id_):
+        conn = http.client.HTTPSConnection(self.base_url)
+        conn.request("GET", f"/api/users/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        if not obj.get("success"):
+            return None
+        return obj.get("profile")
+
+    @functools.lru_cache()
+    def search(self, term):
+        conn = http.client.HTTPSConnection(self.base_url)
+        params = urllib.parse.urlencode(self.patch_auth({"term": term}))
+        conn.request("GET", f"/api/devrant/search?{params}")
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        if not obj.get("success"):
+            return
+        return obj.get("results", [])
+
+    @functools.lru_cache()
+    def get_rant(self, id):
+        conn = http.client.HTTPSConnection(self.base_url)
+        conn.request("GET", f"/api/devrant/rants/{id}?"+urllib.parse.urlencode(self.patch_auth()))
+        response = conn.getresponse()
+        data = response.read()
+        return json.loads(data)
+
+    @functools.lru_cache()
+    def get_rants(self, sort="recent", limit=20, skip=0):
+        conn = http.client.HTTPSConnection(self.base_url)
+        params = urllib.parse.urlencode(self.patch_auth({"sort": sort, "limit": limit, "skip": skip}))
+        conn.request("GET", f"/api/devrant/rants?{params}")
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        if not obj.get("success"):
+            return
+        return obj.get("rants", [])
+
+    @functools.lru_cache()
+    def get_user_id(self, username):
+        conn = http.client.HTTPSConnection(self.base_url)
+        params = urllib.parse.urlencode(self.patch_auth({"username": username}))
+        conn.request("GET", f"/api/get-user-id?{params}")
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        if not obj.get("success"):
+            return None
+        return obj.get("user_id")
+
+    @functools.lru_cache()
+    def update_comment(self, comment_id, comment):
+        if not self.ensure_login():
+            return None
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps(self.patch_auth({"comment": comment}))
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", f"/api/comments/{comment_id}", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get("success", False)
+
+    @functools.lru_cache()
+    def vote_rant(self, rant_id, vote, reason=None):
+        if not self.ensure_login():
+            return None
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps(self.patch_auth({"vote": vote, "reason": reason}))
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", f"/api/devrant/rants/{rant_id}/vote", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get("success", False)
+
+    @functools.lru_cache()
+    def vote_comment(self, comment_id, vote, reason=None):
+        if not self.ensure_login():
+            return None
+        conn = http.client.HTTPSConnection(self.base_url)
+        payload = json.dumps(self.patch_auth({"vote": vote, "reason": reason}))
+        headers = {'Content-Type': 'application/json'}
+        conn.request("POST", f"/api/comments/{comment_id}/vote", payload, headers)
+        response = conn.getresponse()
+        data = response.read()
+        obj = json.loads(data)
+        return obj.get("success", False)
+
+    @property
+    def notifs(self):
+        if not self.ensure_login():
+            return
+        conn = http.client.HTTPSConnection(self.base_url)
+        conn.request("GET", "/api/users/me/notif-feed?" + urllib.parse.urlencode(self.patch_auth()))
+        response = conn.getresponse()
+        data = response.read()
+        return json.loads(data).get("data", {}).get("items", [])
+
+def filter_field(name, obj):
+    results = []
+    if type(obj) in (list,tuple):
+        for value in obj:
+            results += filter_field(name, value)
+    elif type(obj) == dict:
+        for key, value in obj.items():
+            if key == name:
+                results.append(value)
+            if type(value) in (list,dict,tuple):
+                results += filter_field(name, value)
+    return results
+       
+
+def fetch_all(rants, rant_ids):
+    usernames = filter_field("user_username",rants)
+    user_ids = [api.get_user_id(username) for username in usernames]
+    profiles = [api.get_profile(user_id) for user_id in user_ids]
+    new_rant_ids = [rant_id for rant_id in filter_field("rant_id", profiles) if not rant_id in rant_ids]
+    new_rants = []
+    for rant_id in set(new_rant_ids):
+        rant_ids.append(rant_id)
+        new_rants.append(api.get_rant(rant_id))
+        print(rant_id)    
+    
+    if new_rants:
+        return fetch_all(new_rants,rant_ids)
+
+    return rant_ids
+    

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", [])
+
+