retoor/devranta
1
0
Fork 1
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add .env and *.db to gitignore and reformat imports in crawler examples

Browse Source
This commit is contained in:
retoor
2025-08-12 22:06:44 +00:00
parent e2369265d7
commit a551b80bc2
11 changed files with 1860 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/devranta/api.py
+47 -18
View File
@@ -1,37 +1,45 @@
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
import aiohttp
from enum import Enum
class VoteReason(Enum):
"""Enumeration for reasons when down-voting a rant or comment."""
NOT_FOR_ME = 0
REPOST = 1
OFFENSIVE_SPAM = 2
# --- TypedDicts for API Responses ---
class AuthToken(TypedDict):
id: int
key: str
expire_time: int
user_id: int
class LoginResponse(TypedDict):
success: bool
auth_token: AuthToken
class Image(TypedDict):
url: str
width: int
height: int
class UserAvatar(TypedDict):
b: str # background color
i: Optional[str] # image identifier
class Rant(TypedDict):
id: int
text: str
@@ -51,6 +59,7 @@ class Rant(TypedDict):
user_avatar: UserAvatar
editable: bool
class Comment(TypedDict):
id: int
rant_id: int
@@ -63,6 +72,7 @@ class Comment(TypedDict):
user_score: int
user_avatar: UserAvatar
class UserProfile(TypedDict):
username: str
score: int
@@ -75,6 +85,7 @@ class UserProfile(TypedDict):
avatar: UserAvatar
content: Dict[str, Dict[str, Union[List[Rant], List[Comment]]]]
class Notification(TypedDict):
type: str
rant_id: int
@@ -84,8 +95,10 @@ class Notification(TypedDict):
uid: int # User ID of the notifier
username: str
# --- API Class ---
class Api:
"""An asynchronous wrapper for the devRant API."""
@@ -108,7 +121,9 @@ class Api:
self.token_key: Optional[str] = None
self.session: Optional[aiohttp.ClientSession] = None
def patch_auth(self, request_dict: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
def patch_auth(
self, request_dict: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Adds authentication details to a request dictionary.
@@ -146,7 +161,7 @@ class Api:
Returns:
bool: True if login is successful, False otherwise.
Response Structure:
```json
{
@@ -199,7 +214,7 @@ class Api:
Returns:
bool: True on successful registration, False otherwise.
Failure Response Structure:
```json
{
@@ -212,15 +227,17 @@ class Api:
async with aiohttp.ClientSession() as session:
response = await session.post(
url=self.patch_url(f"users"),
data=self.patch_auth({
"email": email,
"username": username,
"password": password,
"plat": 3
}),
data=self.patch_auth(
{
"email": email,
"username": username,
"password": password,
"plat": 3,
}
),
)
obj = await response.json()
return obj.get('success', False)
return obj.get("success", False)
async def get_comments_from_user(self, username: str) -> List[Comment]:
"""
@@ -277,7 +294,7 @@ class Api:
)
obj = await response.json()
return obj.get("comment") if obj.get("success") else None
async def delete_comment(self, id_: int) -> bool:
"""
Deletes a comment by its ID.
@@ -349,7 +366,9 @@ class Api:
)
return await response.json()
async def get_rants(self, sort: str = "recent", limit: int = 20, skip: int = 0) -> List[Rant]:
async def get_rants(
self, sort: str = "recent", limit: int = 20, skip: int = 0
) -> List[Rant]:
"""
Fetches a list of rants.
@@ -420,7 +439,9 @@ 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) -> bool:
async def vote_rant(
self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None
) -> bool:
"""
Casts a vote on a rant.
@@ -437,12 +458,19 @@ class Api:
async with aiohttp.ClientSession() 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}),
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) -> bool:
async def vote_comment(
self,
comment_id: int,
vote: Literal[-1, 0, 1],
reason: Optional[VoteReason] = None,
) -> bool:
"""
Casts a vote on a comment.
@@ -459,7 +487,9 @@ class Api:
async with aiohttp.ClientSession() 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}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = await response.json()
return obj.get("success", False)
@@ -479,4 +509,3 @@ class Api:
)
obj = await response.json()
return obj.get("data", {}).get("items", [])
src/devranta/api_plain.py
+69 -48
View File
@@ -1,7 +1,8 @@
import functools
import http.client
import json
import urllib.parse
import http.client
import functools
class Api:
@@ -31,12 +32,14 @@ class Api:
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'}
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()
@@ -56,45 +59,46 @@ class Api:
return self.login()
return True
@functools.lru_cache()
@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'}
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)
return obj.get("success", False)
@functools.lru_cache()
@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()
@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'}
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()
@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()))
conn.request(
"GET", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth())
)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
@@ -102,21 +106,26 @@ class Api:
return None
return obj.get("comment")
@functools.lru_cache()
@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()))
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()
@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()))
conn.request(
"GET", f"/api/users/{id_}?" + urllib.parse.urlencode(self.patch_auth())
)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
@@ -124,7 +133,7 @@ class Api:
return None
return obj.get("profile")
@functools.lru_cache()
@functools.lru_cache
def search(self, term):
conn = http.client.HTTPSConnection(self.base_url)
params = urllib.parse.urlencode(self.patch_auth({"term": term}))
@@ -136,18 +145,23 @@ class Api:
return
return obj.get("results", [])
@functools.lru_cache()
@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()))
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()
@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}))
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()
@@ -156,7 +170,7 @@ class Api:
return
return obj.get("rants", [])
@functools.lru_cache()
@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}))
@@ -168,39 +182,39 @@ class Api:
return None
return obj.get("user_id")
@functools.lru_cache()
@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'}
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()
@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'}
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()
@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'}
headers = {"Content-Type": "application/json"}
conn.request("POST", f"/api/comments/{comment_id}/vote", payload, headers)
response = conn.getresponse()
data = response.read()
@@ -212,38 +226,45 @@ class Api:
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()))
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):
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):
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)
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_rant_ids = [
rant_id
for rant_id in filter_field("rant_id", profiles)
if rant_id not 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)
print(rant_id)
if new_rants:
return fetch_all(new_rants,rant_ids)
return fetch_all(new_rants, rant_ids)
return rant_ids
src/devranta/api_requests.py
+26 -19
View File
@@ -2,15 +2,18 @@
# 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
from typing import Literal, Optional
import requests
class VoteReason(Enum):
NOT_FOR_ME = 0
REPOST = 1
OFFENSIVE_SPAM = 2
class Api:
base_url = "https://www.devrant.io/api/"
@@ -69,17 +72,14 @@ class Api:
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
}),
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)
return obj.get("success", False)
def get_comments_from_user(self, username):
user_id = self.get_user_id(username)
@@ -106,7 +106,7 @@ class Api:
return None
return obj.get("comment")
def delete_comment(self, id_):
if not self.ensure_login():
return False
@@ -164,9 +164,7 @@ class Api:
@property
def mentions(self):
return [
notif for notif in self.notifs if notif["type"] == "comment_mention"
]
return [notif for notif in self.notifs if notif["type"] == "comment_mention"]
def update_comment(self, comment_id, comment):
if not self.ensure_login():
@@ -178,22 +176,33 @@ class Api:
obj = response.json()
return obj.get("success", False)
def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None):
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}),
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):
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}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = response.json()
return obj.get("success", False)
@@ -206,5 +215,3 @@ class Api:
url=self.patch_url("users/me/notif-feed"), params=self.patch_auth()
)
return response.json().get("data", {}).get("items", [])