Initial commit.

src/Retoorded.egg-info/PKG-INFO

@@ -0,0 +1,12 @@
+Metadata-Version: 2.1
+Name: Retoorded
+Version: 1.3.37
+Summary: A LLM to devRant mapper
+Author: Retoor
+Author-email: retoor@molodetz.nl
+License: MIT
+Requires-Python: >=3.7
+Description-Content-Type: text/markdown
+Requires-Dist: requests
+Requires-Dist: aiohttp
+Requires-Dist: dataset

src/Retoorded.egg-info/SOURCES.txt

@@ -0,0 +1,10 @@
+pyproject.toml
+setup.cfg
+src/Retoorded.egg-info/PKG-INFO
+src/Retoorded.egg-info/SOURCES.txt
+src/Retoorded.egg-info/dependency_links.txt
+src/Retoorded.egg-info/entry_points.txt
+src/Retoorded.egg-info/requires.txt
+src/Retoorded.egg-info/top_level.txt
+src/devranta/__init__.py
+src/devranta/__main__.py

src/Retoorded.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

src/Retoorded.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+retoorded.service = retoorded.bot:run

src/Retoorded.egg-info/requires.txt

@@ -0,0 +1,3 @@
+requests
+aiohttp
+dataset

src/Retoorded.egg-info/top_level.txt

@@ -0,0 +1 @@
+devranta

src/devranta.egg-info/PKG-INFO

@@ -0,0 +1,12 @@
+Metadata-Version: 2.1
+Name: devranta
+Version: 1.0.0
+Summary: Async devRant API client made with aiohttp.
+Author: retoor
+Author-email: retoor@molodetz.nl
+License: MIT
+Requires-Python: >=3.7
+Description-Content-Type: text/markdown
+Requires-Dist: requests
+Requires-Dist: aiohttp
+Requires-Dist: dataset

src/devranta.egg-info/SOURCES.txt

@@ -0,0 +1,12 @@
+pyproject.toml
+setup.cfg
+src/devranta/__init__.py
+src/devranta/__main__.py
+src/devranta/api.py
+src/devranta/tests.py
+src/devranta.egg-info/PKG-INFO
+src/devranta.egg-info/SOURCES.txt
+src/devranta.egg-info/dependency_links.txt
+src/devranta.egg-info/entry_points.txt
+src/devranta.egg-info/requires.txt
+src/devranta.egg-info/top_level.txt

src/devranta.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

src/devranta.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+devranta = devranta.__main__:main

src/devranta.egg-info/requires.txt

@@ -0,0 +1,3 @@
+requests
+aiohttp
+dataset

src/devranta.egg-info/top_level.txt

@@ -0,0 +1 @@
+devranta

src/devranta/__main__.py

@@ -0,0 +1,21 @@
+import asyncio
+
+from devranta.api import Api
+
+
+async def main_async():
+    api = Api()
+    async for rant in api.get_rants():
+        print(
+            "{}({}): {}".format(
+                rant["user_username"], rant["user_score"], rant["text"].split(".")[0]
+            )
+        )
+
+
+def main():
+    asyncio.run(main_async())
+
+
+if __name__ == "__main__":
+    main()

src/devranta/api.py

@@ -0,0 +1,167 @@
+import aiohttp
+
+
+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 = 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 patch_url(self, url: str):
+        return self.base_url.rstrip("/") + "/" + url.lstrip("/")
+
+    async def login(self):
+        if not self.username or not self.password:
+            raise Exception("No authentication defails supplied.")
+        async with self as session:
+            response = await session.post(
+                url=self.patch_url("users/auth-token"),
+                data={
+                    "username": self.username,
+                    "password": self.password,
+                    "app": self.app_id,
+                },
+            )
+            obj = await 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
+
+    async def ensure_login(self):
+        if not self.auth:
+            return await self.login()
+        return True
+
+    async def __aenter__(self):
+        self.session = aiohttp.ClientSession()
+        return self.session
+
+    async def __aexit__(self, *args, **kwargs):
+        await self.session.close()
+        self.session = None
+
+    async def post_comment(self, rant_id, comment):
+        response = None
+        if not await self.ensure_login():
+            return False
+        async with self as session:
+            response = await session.post(
+                url=self.patch_url(f"devrant/rants/{rant_id}/comments"),
+                data=self.patch_auth({"comment": comment, "plat": 2}),
+            )
+        obj = await response.json()
+        return obj.get("success", False)
+
+    async def get_comment(self, id_):
+        response = None
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url("comments/" + str(id_)), params=self.patch_auth()
+            )
+        obj = await response.json()
+
+        print(obj)
+        if not obj.get("success"):
+            return None
+
+        return obj.get("comment")
+
+    async def get_profile(self, id_):
+        response = None
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url(f"users/{id_}"), params=self.patch_auth()
+            )
+        obj = await response.json()
+        if not obj.get("success"):
+            return None
+        return obj.get("profile")
+
+    async def search(self, term):
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url("devrant/search"),
+                params=self.patch_auth({"term": term}),
+            )
+        obj = await response.json()
+        if not obj.get("success"):
+            return
+        for result in obj.get("results", []):
+            yield result
+
+    async def get_rant(self, id):
+        response = None
+        async with self as session:
+            response = await session.get(
+                self.patch_url(f"devrant/rants/{id}"),
+                params=self.patch_auth(),
+            )
+        return await response.json()
+
+    async def get_rants(self, sort="recent", limit=20, skip=0):
+        response = None
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url("devrant/rants"),
+                params=self.patch_auth({"sort": sort, "limit": limit, "skip": skip}),
+            )
+        obj = await response.json()
+        if not obj.get("success"):
+            return
+        for rant in obj.get("rants", []):
+            yield rant
+
+    async def get_user_id(self, username):
+        response = None
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url("get-user-id"),
+                params=self.patch_auth({"username": username}),
+            )
+        obj = await response.json()
+        if not obj.get("success"):
+            return None
+        return obj.get("user_id")
+
+    @property
+    async def mentions(self):
+        async for notif in self.notifs:
+            if notif["type"] == "comment_mention":
+                yield notif
+
+    @property
+    async def notifs(self):
+        response = None
+        if not await self.ensure_login():
+            return
+        async with self as session:
+            response = await session.get(
+                url=self.patch_url("users/me/notif-feed"), params=self.patch_auth()
+            )
+        for item in (await response.json()).get("data", {}).get("items", []):
+            yield item

src/devranta/tests.py

@@ -0,0 +1,31 @@
+import unittest
+
+from devranta.api import Api
+
+
+class ApiTestCase(unittest.IsolatedAsyncioTestCase):
+
+    def setUp(self):
+        self.api = Api()
+
+    async def async_list(self, generator):
+        list_ = []
+        async for v in generator:
+            list_.append(v)
+        return list_
+
+    async def async_len(self, generator):
+        return len(await self.async_list(generator))
+
+    async def test_get_rants(self):
+        self.assertTrue(await self.async_len(self.api.get_rants()))
+
+    async def test_search(self):
+        self.assertTrue(await self.async_len(self.api.search("retoor")))
+
+    async def test_get_user_id(self):
+        self.assertTrue(await self.api.get_user_id("retoor"))
+
+    async def test_get_profile(self):
+        user_id = await self.api.get_user_id("retoor")
+        self.assertTrue(await self.api.get_profile(user_id))