# retoor <retoor@molodetz.nl>
from __future__ import annotations
import ssl
from enum import Enum
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
import aiohttp
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
score: int
created_time: int
attached_image: Union[Image, str]
num_comments: int
tags: List[str]
vote_state: int
edited: bool
link: str
rt: int
rc: int
user_id: int
user_username: str
user_score: int
user_avatar: UserAvatar
editable: bool
class Comment(TypedDict):
id: int
rant_id: int
body: str
score: int
created_time: int
vote_state: int
user_id: int
user_username: str
user_score: int
user_avatar: UserAvatar
class UserProfile(TypedDict):
username: str
score: int
about: str
location: str
created_time: int
skills: str
github: str
website: str
avatar: UserAvatar
content: Dict[str, Dict[str, Union[List[Rant], List[Comment]]]]
class Notification(TypedDict):
type: str
rant_id: int
comment_id: int
created_time: int
read: int
uid: int # User ID of the notifier
username: str
# --- API Class ---
class Api:
"""An asynchronous wrapper for the devRant API."""
base_url: str = "https://www.devrant.io/api/"
def __init__(self, username: Optional[str] = None, password: Optional[str] = None):
"""
Initializes the API client.
Args:
username (Optional[str]): The username for authentication.
password (Optional[str]): The password for authentication.
"""
self.username: Optional[str] = username
self.password: Optional[str] = password
self.auth: Optional[AuthToken] = None
self.app_id: int = 3
self.user_id: Optional[int] = None
self.token_id: Optional[int] = None
self.token_key: Optional[str] = None
self._session: Optional[aiohttp.ClientSession] = None
self._owns_session: bool = False
async def __aenter__(self):
"""Async context manager entry - creates shared HTTP session."""
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
connector = aiohttp.TCPConnector(ssl=ssl_context)
self._session = aiohttp.ClientSession(connector=connector)
self._owns_session = True
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit - closes shared HTTP session."""
await self.close()
async def _get_session(self) -> aiohttp.ClientSession:
"""Returns or creates a shared HTTP session for connection reuse."""
if self._session is None:
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
connector = aiohttp.TCPConnector(ssl=ssl_context)
self._session = aiohttp.ClientSession(connector=connector)
self._owns_session = True
return self._session
async def close(self):
"""Closes the HTTP session if owned by this instance."""
if self._session and self._owns_session:
await self._session.close()
self._session = None
self._owns_session = False
def patch_auth(
self, request_dict: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Adds authentication details to a request dictionary.
Args:
request_dict (Optional[Dict[str, Any]]): The dictionary to patch.
Returns:
Dict[str, Any]: The patched dictionary with auth details.
"""
auth_dict: Dict[str, Any] = {"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) -> str:
"""
Constructs the full API URL for an endpoint.
Args:
url (str): The endpoint path.
Returns:
str: The full API URL.
"""
return self.base_url.rstrip("/") + "/" + url.lstrip("/")
async def login(self) -> bool:
"""
Authenticates the user and stores the auth token.
Returns:
bool: True if login is successful, False otherwise.
Response Structure:
```json
{
"success": true,
"auth_token": {
"id": int, // Token ID
"key": "string", // Token key
"expire_time": int, // Unix timestamp of token expiration
"user_id": int // ID of the authenticated user
}
}
```
"""
if not self.username or not self.password:
raise Exception("No authentication details supplied.")
session = await self._get_session()
response = await session.post(
url=self.patch_url("users/auth-token"),
data={
"username": self.username,
"password": self.password,
"app": self.app_id,
},
)
obj: LoginResponse = 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 bool(self.auth)
async def ensure_login(self) -> bool:
"""Ensures the user is logged in before making a request."""
if not self.auth:
return await self.login()
return True
async def register_user(self, email: str, username: str, password: str) -> bool:
"""
Registers a new user.
Args:
email (str): The user's email address.
username (str): The desired username.
password (str): The desired password.
Returns:
bool: True on successful registration, False otherwise.
Failure Response Structure:
```json
{
"success": false,
"error": "Error message string.",
"error_field": "field_name" // e.g., "username" or "email"
}
```
"""
session = await self._get_session()
response = await session.post(
url=self.patch_url("users"),
data=self.patch_auth(
{
"email": email,
"username": username,
"password": password,
"plat": 3,
}
),
)
obj = await response.json()
return obj.get("success", False)
async def get_comments_from_user(self, username: str) -> List[Comment]:
"""
Fetches all comments posted by a specific user by first fetching their profile.
Args:
username (str): The username of the user.
Returns:
List[Comment]: A list of comment objects.
"""
user_id = await self.get_user_id(username)
if not user_id:
return []
profile = await self.get_profile(user_id)
if not profile:
return []
return profile.get("content", {}).get("content", {}).get("comments", [])
async def post_comment(self, rant_id: int, comment: str) -> bool:
"""
Posts a comment on a specific rant.
Args:
rant_id (int): The ID of the rant to comment on.
comment (str): The content of the comment.
Returns:
bool: True if the comment was posted successfully, False otherwise.
"""
if not await self.ensure_login():
return False
session = await self._get_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_: int) -> Optional[Comment]:
"""
Retrieves a single comment by its ID.
Args:
id_ (int): The ID of the comment.
Returns:
Optional[Comment]: A dictionary representing the comment, or None if not found.
"""
session = await self._get_session()
response = await session.get(
url=self.patch_url(f"comments/{id_}"), params=self.patch_auth()
)
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.
Args:
id_ (int): The ID of the comment to delete.
Returns:
bool: True if deletion was successful, False otherwise.
"""
if not await self.ensure_login():
return False
session = await self._get_session()
response = await session.delete(
url=self.patch_url(f"comments/{id_}"), params=self.patch_auth()
)
obj = await response.json()
return obj.get("success", False)
async def get_profile(self, id_: int) -> Optional[UserProfile]:
"""
Retrieves the profile of a user by their ID.
Args:
id_ (int): The user's ID.
Returns:
Optional[UserProfile]: A dictionary with the user's profile data.
"""
session = await self._get_session()
response = await session.get(
url=self.patch_url(f"users/{id_}"), params=self.patch_auth()
)
obj = await response.json()
return obj.get("profile") if obj.get("success") else None
async def search(self, term: str) -> List[Rant]:
"""
Searches for rants based on a search term.
Args:
term (str): The term to search for.
Returns:
List[Rant]: A list of rant objects from the search results.
"""
session = await self._get_session()
response = await session.get(
url=self.patch_url("devrant/search"),
params=self.patch_auth({"term": term}),
)
obj = await response.json()
return obj.get("results", []) if obj.get("success") else []
async def get_rant(self, id: int) -> Dict[str, Any]:
"""
Retrieves a single rant and its comments by ID.
Args:
id (int): The ID of the rant.
Returns:
Dict[str, Any]: The full API response object.
"""
session = await self._get_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: str = "recent", limit: int = 20, skip: int = 0
) -> List[Rant]:
"""
Fetches a list of rants.
Args:
sort (str): The sorting method ('recent', 'top', 'algo').
limit (int): The number of rants to return.
skip (int): The number of rants to skip for pagination.
Returns:
List[Rant]: A list of rant objects.
"""
session = await self._get_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()
return obj.get("rants", []) if obj.get("success") else []
async def get_user_id(self, username: str) -> Optional[int]:
"""
Retrieves a user's ID from their username.
Args:
username (str): The username to look up.
Returns:
Optional[int]: The user's ID, or None if not found.
"""
session = await self._get_session()
response = await session.get(
url=self.patch_url("get-user-id"),
params=self.patch_auth({"username": username}),
)
obj = await response.json()
return obj.get("user_id") if obj.get("success") else None
async def mentions(self) -> List[Notification]:
"""
Fetches notifications where the user was mentioned.
Returns:
List[Notification]: A list of mention notification objects.
"""
notifications = await self.notifs()
return [
notif for notif in notifications if notif.get("type") == "comment_mention"
]
async def update_comment(self, comment_id: int, comment: str) -> bool:
"""
Updates an existing comment.
Args:
comment_id (int): The ID of the comment to update.
comment (str): The new content of the comment.
Returns:
bool: True if the update was successful, False otherwise.
"""
if not await self.ensure_login():
return False
session = await self._get_session()
response = await session.post(
url=self.patch_url(f"comments/{comment_id}"),
data=self.patch_auth({"comment": comment}),
)
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:
"""
Casts a vote on a rant.
Args:
rant_id (int): The ID of the rant to vote on.
vote (Literal[-1, 0, 1]): -1 for downvote, 0 to unvote, 1 for upvote.
reason (Optional[VoteReason]): The reason for a downvote.
Returns:
bool: True if the vote was successful, False otherwise.
"""
if not await self.ensure_login():
return False
session = await self._get_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,
) -> bool:
"""
Casts a vote on a comment.
Args:
comment_id (int): The ID of the comment to vote on.
vote (Literal[-1, 0, 1]): -1 for downvote, 0 to unvote, 1 for upvote.
reason (Optional[VoteReason]): The reason for a downvote.
Returns:
bool: True if the vote was successful, False otherwise.
"""
if not await self.ensure_login():
return False
session = await self._get_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)
async def notifs(self) -> List[Notification]:
"""
Fetches the user's notification feed.
Returns:
List[Notification]: A list of notification items.
"""
if not await self.ensure_login():
return []
session = await self._get_session()
response = await session.get(
url=self.patch_url("users/me/notif-feed"), params=self.patch_auth()
)
obj = await response.json()
return obj.get("data", {}).get("items", [])
Reference in New Issue View Git Blame Copy Permalink