Compare commits

...

22 Commits
main ... main

Author SHA1 Message Date
bot
d0c8156519 New build. 2025-09-01 16:39:03 +00:00
c4921b645a New
All checks were successful
devranta build / Build (push) Successful in 1m1s
2025-09-01 18:37:32 +02:00
2d875b8a53 Update.
Some checks failed
devranta build / Build (push) Failing after 57s
2025-08-13 01:41:43 +02:00
7623e25238 Update.
Some checks failed
devranta build / Build (push) Failing after 53s
2025-08-13 00:22:00 +02:00
4bf29c87fd Update.
Some checks failed
devranta build / Build (push) Failing after 56s
2025-08-13 00:12:38 +02:00
52df3887a6 Update.
Some checks failed
devranta build / Build (push) Failing after 1m5s
2025-08-13 00:06:44 +02:00
5a0a066105 Update.
Some checks failed
devranta build / Build (push) Failing after 50s
2025-08-03 00:40:34 +02:00
246cdf51fa Update client. 2025-08-02 08:03:19 +02:00
153d1b2ca5 Update.
Some checks failed
devranta build / Build (push) Failing after 54s
2025-08-02 07:11:03 +02:00
000fd98d8d Update.
Some checks failed
devranta build / Build (push) Has been cancelled
2025-04-24 00:58:42 +02:00
3684107204 api_requests file for LLM input. 2025-04-23 15:59:14 +02:00
bot
575bfffc06 New build. 2025-01-06 18:02:11 +00:00
24b8db6f6d Delete.
All checks were successful
devranta build / Build (push) Successful in 1m23s
2025-01-06 19:00:45 +01:00
bot
669a76f8a9 New build. 2024-12-14 03:42:04 +00:00
eaec9335dc Update url.
All checks were successful
devranta build / Build (push) Successful in 1m25s
2024-12-14 04:40:30 +01:00
bot
0811eb1db8 New build. 2024-12-14 03:19:28 +00:00
6a14dbe91a Account creation.
All checks were successful
devranta build / Build (push) Successful in 1m27s
2024-12-14 04:17:55 +01:00
bot
5d464ad271 New build. 2024-12-04 22:14:50 +00:00
2a3269edfa Update README.md
All checks were successful
devranta build / Build (push) Successful in 1m19s
2024-12-04 22:13:31 +00:00
bot
485144303f New build. 2024-12-03 22:13:56 +00:00
d1f8768b15 Merge pull request 'Add vote for rants and comments' (#1) from dr/devranta:vote into main
All checks were successful
devranta build / Build (push) Successful in 1m10s
Reviewed-on: retoor/devranta#1
2024-12-03 22:12:48 +00:00
dr
f947d5f088 Add vote for rants and comments 2024-12-03 22:27:40 +01:00
24 changed files with 6673 additions and 124 deletions

7
.gitignore vendored
View File

@ -2,3 +2,10 @@
.history
__pycache__
*.pyc
.env
*.db
examples/crawler/devrant.sqlite-shm
examples/crawler/devrant.sqlite-wal
examples/crawler/devrant.sqlite
examples/crawler/.venv
examples/crawler/__pycache__

512
README.md
View File

@ -1,37 +1,517 @@
# devRanta
devRanta is an async devrant client written in and for Python.
Authentication is only needed for half of the functionality and thus username and password are optional parameters by constructing the main class of this package (Api).
You can find last packages in tar and wheel format [here](https://retoor.molodetz.nl/retoor/devranta/packages).
devRanta is the best async devRant client written in Python. Authentication is only needed for half of the functionality; thus, the username and password are optional parameters when constructing the main class of this package (Api). You can find the latest packages in tar and wheel format [here](https://retoor.molodetz.nl/retoor/devranta/packages).
## Running
```
make run
```
## Testing
Tests are only made for methods not requireing authentication.
I do not see value in mocking requests.
```
make test
```
## How to use
Implementation:
```
from devranta.api import Api
api = Api(username="optional!", password="optional!")
async def list_rants():
async for rant in api.get_rants():
print(rant["user_username"], ":", rant["text"])
```
See [tests](src/devranta/tests.py) for [examples](src/devranta/tests.py) on how to use.
## Todo
- voting comment
- edit message
# devRant API Documentation
For people wanting to build their own client.
TODO: document responses.
## Base URL
`https://devrant.com/api`
## Authentication
- Uses `dr_token` cookie with `token_id`, `token_key`, and `user_id`.
- Required for endpoints needing user authentication.
- `guid`, `plat`, `sid`, `seid` included in requests for session tracking.
## Endpoints
### User Management
1. **Registering user**
- Ommitted, you know why.
2. **Login User**
- **URL**: `/api/users/auth-token`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `username`: User username
- `password`: User password
- `guid`: Unique identifier
- `plat`: 3
- `sid`: Session start time
- `seid`: Session event ID
- **Response**: JSON with `success`, `auth_token`, or `error`
- **Success Example**:
```json
{
"success": true,
"auth_token": {
"id": 18966518,
"key": "z6uXRZrQ_Ekx59wfYEjpbnS!fDeVznVqmmKujhT8",
"expire_time": 1756765587,
"user_id": 18959632
}
}
```
- **Error Example**:
```json
{
"success": false,
"error": "Invalid login credentials entered. Please try again."
}
```
- **Description**: Authenticates user and returns auth token.
3. **Edit Profile**
- **URL**: `/api/users/me/edit-profile`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`: Token ID
- `token_key`: Token key
- `user_id`: User ID
- `guid`, `plat`, `sid`, `seid`
- `profile_about`: User bio
- `profile_skills`: User skills
- `profile_location`: User location
- `profile_website`: User website
- `profile_github`: GitHub username
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Updates user profile information.
4. **Forgot Password**
- **URL**: `/api/users/forgot-password`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `username`: User username
- `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Initiates password reset process.
5. **Resend Confirmation Email**
- **URL**: `/api/users/me/resend-confirm`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Resends account confirmation email.
6. **Delete Account**
- **URL**: `/api/users/me`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Description**: Deletes user account.
7. **Mark News as Read**
- **URL**: `/api/users/me/mark-news-read`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `news_id`: News item ID
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Marks a news item as read for logged-in users.
### Rants
1. **Get Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `last_comment_id`: 999999999999 (optional)
- `links`: 0 (optional)
- **Response**: JSON with `rant` (text, tags), `comments`, `success`, `subscribed`
- **Success Example**:
```json
{
"rant": {
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 10,
"tags": [
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": 0,
"edited": false,
"link": "rants/18960811/you-know-im-getting-tired-of-this-hr-speak-in-job-applications-specifically-this",
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4179,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
},
"comments": [],
"success": true,
"subscribed": 0
}
```
- **Description**: Retrieves a specific rant by ID.
2. **Post Rant**
- **URL**: `/api/devrant/rants`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `rant`: Rant text
- `tags`: Comma-separated tags
- `token_id`, `token_key`, `user_id`
- `type`: Rant type ID
- `image`: Optional image file (img/gif)
- **Response**: JSON with `success`, `rant_id`, or `error`
- **Error Example**:
```json
{
"success": false,
"error": "It looks like you just posted this same rant! Your connection might have timed out while posting so you might have seen an error, but sometimes the rant still gets posted and in this case it seems it did, so please check :) If this was not the case please contact info@devrant.io. Thanks!"
}
```
- **Description**: Creates a new rant.
3. **Edit Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `rant`: Rant text
- `tags`: Comma-separated tags
- `token_id`, `token_key`, `user_id`
- `image`: Optional image file
- **Response**: JSON with `success` or `fail_reason`
- **Error Example**:
```json
{
"success": false,
"fail_reason": ""
}
```
- **Description**: Updates an existing rant.
4. **Delete Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "An unknown error occurred and this rant can't be deleted. Please contact support@devrant.com for help with this."
}
```
- **Description**: Deletes a rant.
5. **Vote on Rant**
- **URL**: `/api/devrant/rants/{rant_id}/vote`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `vote`: 1 (upvote), -1 (downvote), 0 (remove vote)
- `reason`: Downvote reason ID (required for downvote)
- **Response**: JSON with `success`, `rant`, or `confirmed` (false if unverified)
- **Success Example**:
```json
{
"success": true,
"rant": {
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 10,
"tags": [
"rant",
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": -1,
"edited": false,
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4180,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
}
}
```
- **Description**: Votes on a rant.
6. **Favorite/Unfavorite Rant**
- **URL**: `/api/devrant/rants/{rant_id}/{favorite|unfavorite}`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Favorites or unfavorites a rant.
7. **Get Rant Feed**
- **URL**: `/api/devrant/rants`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `ids`: JSON string of IDs (optional)
- **Response**: JSON with `success`, `rants`, `settings`, `set`, `wrw`, `dpp`, `num_notifs`, `unread`, `news`
- **Success Example**:
```json
{
"success": true,
"rants": [
{
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 9,
"tags": [
"rant",
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": 0,
"edited": false,
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4179,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
}
// ... more rants
],
"settings": {
"notif_state": -1,
"notif_token": ""
},
"set": "688e90b7cf77f",
"wrw": 385,
"dpp": 0,
"num_notifs": 0,
"unread": {
"total": 0
},
"news": {
"id": 356,
"type": "intlink",
"headline": "Weekly Group Rant",
"body": "Tips for staying productive?",
"footer": "Add tag 'wk247' to your rant",
"height": 100,
"action": "grouprant"
}
}
```
- **Description**: Retrieves rant feed with notification count.
### Comments
1. **Get Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `links`: 0 (optional)
- **Response**: JSON with `comment` (body), or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Retrieves a specific comment by ID.
2. **Post Comment**
- **URL**: `/api/devrant/rants/{rant_id}/comments`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `comment`: Comment text
- `token_id`, `token_key`, `user_id`
- `image`: Optional image file (img/gif)
- **Response**: JSON with `success` or `confirmed` (false if unverified)
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Posts a comment on a rant.
3. **Edit Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `comment`: Comment text
- `token_id`, `token_key`, `user_id`
- **Response**: JSON with `success` or `fail_reason`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Updates an existing comment.
4. **Delete Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Deletes a comment.
5. **Vote on Comment**
- **URL**: `/api/comments/{comment_id}/vote`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `vote`: 1 (upvote), -1 (downvote), 0 (remove vote)
- `reason`: Downvote reason ID (required for downvote)
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Votes on a comment.
### Notifications
1. **Get Notification Feed**
- **URL**: `/api/users/me/notif-feed`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `ext_prof`: 1 (optional)
- `last_time`: Last notification check time
- **Response**: JSON with `success`, `data` (items, check_time, username_map, unread, num_unread)
- **Success Example**:
```json
{
"success": true,
"data": {
"items": [],
"check_time": 1754173634,
"username_map": [],
"unread": {
"all": 0,
"upvotes": 0,
"mentions": 0,
"comments": 0,
"subs": 0,
"total": 0
},
"num_unread": 0
}
}
```
- **Description**: Retrieves user notifications.
2. **Clear Notifications**
- **URL**: `/api/users/me/notif-feed`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Clears user notifications.
### External API
- **Beta List Signup**
- **URL**: `https://www.hexicallabs.com/api/beta-list`
- **Method**: GET (JSONP)
- **Parameters**:
- `email`: User email
- `platform`: Platform name
- `app`: 3
- **Response**: JSON with `error`
- **Error Example**:
```json
{
"error": "Expecting value: line 1 column 1 (char 0)"
}
```
- **Description**: Signs up user for beta list (external service).
## Notes
- All endpoints expect `app=3` for identification.
- Authenticated endpoints require `dr_token` cookie with `token_id`, `token_key`, `user_id`.
- `guid`, `plat`, `sid`, `seid` are used for session tracking.
- Image uploads use FormData for rants and comments.
- Downvotes require a reason ID, prompting a modal if not provided.
- Responses typically include `success` boolean; errors include `error` or `fail_reason`.
- Cookies (`dr_token`, `dr_guid`, `dr_session_start`, `dr_event_id`, `dr_feed_sort`, `dr_theme`, `dr_rants_viewed`, `dr_stickers_seen`, `rant_type_filters`, `news_seen`) manage state and preferences.

2053
api_test_results.json Normal file

File diff suppressed because it is too large Load Diff

Binary file not shown.

Binary file not shown.

BIN
dist/devranta-1.1.0-py3-none-any.whl vendored Normal file

Binary file not shown.

BIN
dist/devranta-1.1.0.tar.gz vendored Normal file

Binary file not shown.

17
examples/crawler/Makefile Normal file
View File

@ -0,0 +1,17 @@
.PHONY: all env install run clean
all: env install run
env:
python3 -m venv .venv
.
install:
. .venv/bin/activate && pip install -r requirements.txt
. .venv/bin/activate && pip install -e ../../.
run:
. .venv/bin/activate && python main.py
clean:
rm -rf .venv

View File

@ -0,0 +1,34 @@
# Example Crawler Project
This is a simple example crawler project. Follow the instructions below to set up and run the crawler.
## Setup
1. Clone the repository or copy the project files to your local machine.
2. Make sure you have Python 3 installed.
## Usage
1. Open a terminal in the project directory.
2. Run `make` to set up the environment, install dependencies, and start the crawler:
```bash
make
```
This will create a virtual environment, install the package in editable mode from the parent directory, and run the main script.
## Cleanup
To remove the virtual environment, run:
```bash
make clean
```
## Notes
- The project installs the package with `-e ../../.` to include the parent package `devranta` in editable mode.
- Ensure that the parent package is correctly set up in the directory structure.
Happy crawling!

248
examples/crawler/crawler.py Normal file
View File

@ -0,0 +1,248 @@
import asyncio
import logging
from typing import Set
from database import DatabaseManager
from devranta.api import Api, Rant
class DevRantCrawler:
def __init__(
self, api: Api, db: DatabaseManager, rant_consumers: int, user_consumers: int
):
self.api = api
self.db = db
self.rant_queue = asyncio.Queue(maxsize=1000000)
self.user_queue = asyncio.Queue(maxsize=1000000)
self.shutdown_event = asyncio.Event()
self.num_rant_consumers = rant_consumers
self.num_user_consumers = user_consumers
self.seen_rant_ids: Set[int] = set()
self.seen_user_ids: Set[int] = set()
self.stats = {
"rants_processed": 0,
"rants_added_to_db": 0,
"comments_added_to_db": 0,
"users_processed": 0,
"users_added_to_db": 0,
"api_errors": 0,
"producer_loops": 0,
"end_of_feed_hits": 0,
"rants_queued": 0,
"users_queued": 0,
}
async def _queue_user_if_new(self, user_id: int):
if user_id in self.seen_user_ids:
return
self.seen_user_ids.add(user_id)
if not await self.db.user_exists(user_id):
await self.user_queue.put(user_id)
self.stats["users_queued"] += 1
async def _queue_rant_if_new(self, rant_obj: Rant):
rant_id = rant_obj["id"]
if rant_id in self.seen_rant_ids:
return
self.seen_rant_ids.add(rant_id)
if not await self.db.rant_exists(rant_id):
await self.db.add_rant(rant_obj)
self.stats["rants_added_to_db"] += 1
await self.rant_queue.put(rant_id)
self.stats["rants_queued"] += 1
async def _initial_seed(self):
logging.info("Starting initial seeder to re-ignite crawling process...")
user_ids = await self.db.get_random_user_ids(limit=2000)
if not user_ids:
logging.info(
"Seeder found no existing users. Crawler will start from scratch."
)
return
for user_id in user_ids:
if user_id not in self.seen_user_ids:
self.seen_user_ids.add(user_id)
await self.user_queue.put(user_id)
self.stats["users_queued"] += 1
logging.info(
f"Seeder finished: Queued {len(user_ids)} users to kickstart exploration."
)
async def _rant_producer(self):
logging.info("Rant producer started.")
skip = 0
consecutive_empty_responses = 0
while not self.shutdown_event.is_set():
try:
logging.info(f"Producer: Fetching rants with skip={skip}...")
rants = await self.api.get_rants(sort="recent", limit=50, skip=skip)
self.stats["producer_loops"] += 1
if not rants:
consecutive_empty_responses += 1
logging.info(
f"Producer: Feed returned empty. Consecutive empty hits: {consecutive_empty_responses}."
)
if consecutive_empty_responses >= 5:
self.stats["end_of_feed_hits"] += 1
logging.info(
"Producer: End of feed likely reached. Pausing for 15 minutes before reset."
)
await asyncio.sleep(900)
skip = 0
consecutive_empty_responses = 0
else:
await asyncio.sleep(10)
continue
consecutive_empty_responses = 0
new_rants_found = 0
for rant in rants:
await self._queue_rant_if_new(rant)
new_rants_found += 1
logging.info(
f"Producer: Processed {new_rants_found} rants from feed. Total queued: {self.stats['rants_queued']}."
)
skip += len(rants)
await asyncio.sleep(2)
except Exception as e:
logging.critical(
f"Producer: Unhandled exception: {e}. Retrying in 60s."
)
self.stats["api_errors"] += 1
await asyncio.sleep(60)
async def _rant_consumer(self, worker_id: int):
logging.info(f"Rant consumer #{worker_id} started.")
while not self.shutdown_event.is_set():
try:
rant_id = await self.rant_queue.get()
logging.info(
f"Rant consumer #{worker_id}: Processing rant ID {rant_id}."
)
rant_details = await self.api.get_rant(rant_id)
if not rant_details or not rant_details.get("success"):
logging.warning(
f"Rant consumer #{worker_id}: Failed to fetch details for rant {rant_id}."
)
self.rant_queue.task_done()
continue
await self._queue_user_if_new(rant_details["rant"]["user_id"])
comments = rant_details.get("comments", [])
for comment in comments:
await self.db.add_comment(comment)
self.stats["comments_added_to_db"] += 1
await self._queue_user_if_new(comment["user_id"])
logging.info(
f"Rant consumer #{worker_id}: Finished processing rant {rant_id}, found {len(comments)} comments."
)
self.stats["rants_processed"] += 1
self.rant_queue.task_done()
except Exception as e:
logging.error(f"Rant consumer #{worker_id}: Unhandled exception: {e}")
self.rant_queue.task_done()
async def _user_consumer(self, worker_id: int):
logging.info(f"User consumer #{worker_id} started.")
while not self.shutdown_event.is_set():
try:
user_id = await self.user_queue.get()
logging.info(
f"User consumer #{worker_id}: Processing user ID {user_id}."
)
profile = await self.api.get_profile(user_id)
if not profile:
logging.warning(
f"User consumer #{worker_id}: Could not fetch profile for user {user_id}."
)
self.user_queue.task_done()
continue
await self.db.add_user(profile, user_id)
self.stats["users_added_to_db"] += 1
rants_found_on_profile = 0
content_sections = profile.get("content", {}).get("content", {})
for section_name in ["rants", "upvoted", "favorites"]:
for rant_obj in content_sections.get(section_name, []):
await self._queue_rant_if_new(rant_obj)
rants_found_on_profile += 1
logging.info(
f"User consumer #{worker_id}: Finished user {user_id}, found and queued {rants_found_on_profile} associated rants."
)
self.stats["users_processed"] += 1
self.user_queue.task_done()
except Exception as e:
logging.error(f"User consumer #{worker_id}: Unhandled exception: {e}")
self.user_queue.task_done()
async def _stats_reporter(self):
logging.info("Stats reporter started.")
while not self.shutdown_event.is_set():
await asyncio.sleep(15)
logging.info(
f"[STATS] Rants Q'd/Proc: {self.stats['rants_queued']}/{self.stats['rants_processed']} | "
f"Users Q'd/Proc: {self.stats['users_queued']}/{self.stats['users_processed']} | "
f"Comments DB: {self.stats['comments_added_to_db']} | "
f"Queues (R/U): {self.rant_queue.qsize()}/{self.user_queue.qsize()} | "
f"API Errors: {self.stats['api_errors']}"
)
async def run(self):
logging.info("Exhaustive crawler starting...")
await self._initial_seed()
logging.info("Starting main producer and consumer tasks...")
tasks = []
try:
tasks.append(asyncio.create_task(self._rant_producer()))
tasks.append(asyncio.create_task(self._stats_reporter()))
for i in range(self.num_rant_consumers):
tasks.append(asyncio.create_task(self._rant_consumer(i + 1)))
for i in range(self.num_user_consumers):
tasks.append(asyncio.create_task(self._user_consumer(i + 1)))
await asyncio.gather(*tasks, return_exceptions=True)
except asyncio.CancelledError:
logging.info("Crawler run cancelled.")
finally:
await self.shutdown()
async def shutdown(self):
if self.shutdown_event.is_set():
return
logging.info("Shutting down... sending signal to all tasks.")
self.shutdown_event.set()
logging.info("Waiting for queues to empty... Press Ctrl+C again to force exit.")
try:
await asyncio.wait_for(self.rant_queue.join(), timeout=30)
await asyncio.wait_for(self.user_queue.join(), timeout=30)
except (asyncio.TimeoutError, asyncio.CancelledError):
logging.warning("Could not empty queues in time, proceeding with shutdown.")
tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
logging.info("All tasks cancelled.")
logging.info(f"--- FINAL STATS ---\n{self.stats}")

View File

@ -0,0 +1,130 @@
import logging
from typing import List
import aiosqlite
from devranta.api import Comment, Rant, UserProfile
class DatabaseManager:
def __init__(self, db_path: str):
self.db_path = db_path
self._conn: aiosqlite.Connection | None = None
async def __aenter__(self):
logging.info(f"Connecting to database at {self.db_path}...")
self._conn = await aiosqlite.connect(self.db_path)
await self._conn.execute("PRAGMA journal_mode=WAL;")
await self._conn.execute("PRAGMA foreign_keys=ON;")
await self.create_tables()
logging.info("Database connection successful.")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._conn:
await self._conn.close()
logging.info("Database connection closed.")
async def create_tables(self):
logging.info("Ensuring database tables exist...")
await self._conn.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
score INTEGER,
about TEXT,
location TEXT,
created_time INTEGER,
skills TEXT,
github TEXT,
website TEXT
);
CREATE TABLE IF NOT EXISTS rants (
id INTEGER PRIMARY KEY,
user_id INTEGER,
text TEXT,
score INTEGER,
created_time INTEGER,
num_comments INTEGER
);
CREATE TABLE IF NOT EXISTS comments (
id INTEGER PRIMARY KEY,
rant_id INTEGER,
user_id INTEGER,
body TEXT,
score INTEGER,
created_time INTEGER
);
"""
)
await self._conn.commit()
logging.info("Table schema verified.")
async def add_rant(self, rant: Rant):
await self._conn.execute(
"INSERT OR IGNORE INTO rants (id, user_id, text, score, created_time, num_comments) VALUES (?, ?, ?, ?, ?, ?)",
(
rant["id"],
rant["user_id"],
rant["text"],
rant["score"],
rant["created_time"],
rant["num_comments"],
),
)
await self._conn.commit()
async def add_comment(self, comment: Comment):
await self._conn.execute(
"INSERT OR IGNORE INTO comments (id, rant_id, user_id, body, score, created_time) VALUES (?, ?, ?, ?, ?, ?)",
(
comment["id"],
comment["rant_id"],
comment["user_id"],
comment["body"],
comment["score"],
comment["created_time"],
),
)
await self._conn.commit()
async def add_user(self, user: UserProfile, user_id: int):
await self._conn.execute(
"INSERT OR IGNORE INTO users (id, username, score, about, location, created_time, skills, github, website) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
user_id,
user["username"],
user["score"],
user["about"],
user["location"],
user["created_time"],
user["skills"],
user["github"],
user["website"],
),
)
await self._conn.commit()
async def rant_exists(self, rant_id: int) -> bool:
async with self._conn.execute(
"SELECT 1 FROM rants WHERE id = ? LIMIT 1", (rant_id,)
) as cursor:
return await cursor.fetchone() is not None
async def user_exists(self, user_id: int) -> bool:
async with self._conn.execute(
"SELECT 1 FROM users WHERE id = ? LIMIT 1", (user_id,)
) as cursor:
return await cursor.fetchone() is not None
async def get_random_user_ids(self, limit: int) -> List[int]:
logging.info(
f"Fetching up to {limit} random user IDs from database for seeding..."
)
query = "SELECT id FROM users ORDER BY RANDOM() LIMIT ?"
async with self._conn.execute(query, (limit,)) as cursor:
rows = await cursor.fetchall()
user_ids = [row[0] for row in rows]
logging.info(f"Found {len(user_ids)} user IDs to seed.")
return user_ids

49
examples/crawler/main.py Normal file
View File

@ -0,0 +1,49 @@
# main.py
import asyncio
import logging
import signal
from crawler import DevRantCrawler
from database import DatabaseManager
from devranta.api import Api
# --- Configuration ---
DB_FILE = "devrant.sqlite"
CONCURRENT_RANT_CONSUMERS = 10 # How many rants to process at once
CONCURRENT_USER_CONSUMERS = 5 # How many user profiles to fetch at once
async def main():
"""Initializes and runs the crawler."""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
api = Api()
async with DatabaseManager(DB_FILE) as db:
crawler = DevRantCrawler(
api=api,
db=db,
rant_consumers=CONCURRENT_RANT_CONSUMERS,
user_consumers=CONCURRENT_USER_CONSUMERS,
)
# Set up a signal handler for graceful shutdown on Ctrl+C
loop = asyncio.get_running_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(
sig, lambda s=sig: asyncio.create_task(crawler.shutdown())
)
await crawler.run()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logging.info("Main loop interrupted. Exiting.")

View File

@ -0,0 +1 @@
aiosqlite

View File

@ -0,0 +1,79 @@
# Princess Bot - Usage and Configuration Guide
## Overview
Princess.py is an automated social media interaction bot designed to monitor and respond to specific user-generated content (rants and comments) on a platform. It fetches new posts made by a target user, generates witty or devastating responses using a language model, and keeps track of responded messages to avoid duplicates.
The bot operates continuously, periodically checking for new content and replying accordingly.
---
## How It Works
1. **Initialization**: The bot initializes with user credentials, target username, and API keys.
2. **Login**: It logs into the platform via the provided API.
3. **Content Monitoring**: It fetches recent rants and comments made by the target user.
4. **Response Generation**: For new content (not responded to before), it generates a response using a language model (GrokAPIClient).
5. **Response Posting**: It prints the content and the generated reply.
6. **Tracking**: It records responded messages in a local database to prevent duplicate responses.
7. **Loop**: It repeats this process every 60 seconds.
---
## Configuration
The script uses a `.env` file to manage sensitive credentials and configurable properties. Below are the supported environment variables:
### Required Environment Variables
| Property | Description | Example |
|----------------------|----------------------------------------------------------|-------------------------------------------|
| `USERNAME` | Your platform username. | `my_username` |
| `PASSWORD` | Your platform password. | `my_password` |
| `TARGET` | The username of the user to monitor. | `target_user` |
| `LLM_KEY` | API key for the language model (Grok API). | `your-grok-api-key` |
## Setup Instructions
1. **Create a `.env` file** in the same directory as `princess.py`.
2. **Add the required variables** with your credentials and target info:
```env
USERNAME=your_username
PASSWORD=your_password
TARGET=target_username
LLM_KEY=your_grok_api_key
```
3. **Install dependencies** (if not already installed):
```bash
pip install python-dotenv
```
4. **Run the script**:
```bash
python princess.py
```
---
## Notes
- The bot stores responded messages in a local SQLite database (`princess.db`) to avoid duplicate responses.
- It runs indefinitely, checking for new content every 60 seconds.
- Make sure your API keys and credentials are kept secure and not shared publicly.
---
## Summary
Princess.py is a social media response bot that:
- Monitors a specific user's posts.
- Generates witty responses using a language model.
- Keeps track of responses to prevent duplicates.
- Runs continuously with minimal setup.

1187
examples/princess/ads.py Normal file

File diff suppressed because it is too large Load Diff

122
examples/princess/grk.py Normal file
View File

@ -0,0 +1,122 @@
import asyncio
import http.client
import json
class GrokAPIClient:
def __init__(
self,
api_key: str,
system_message: str | None = None,
model: str = "grok-3-mini",
temperature: float = 0.0,
):
self.api_key = api_key
self.model = model
self.base_url = "api.x.ai"
self.temperature = temperature
self._messages: list[dict[str, str]] = []
if system_message:
self._messages.append({"role": "system", "content": system_message})
def chat_json(self, user_message: str, *, clear_history: bool = False) -> str:
return self.chat(user_message, clear_history=clear_history, use_json=True)
def chat_text(self, user_message: str, *, clear_history: bool = False) -> str:
return self.chat(user_message, clear_history=clear_history, use_json=False)
async def chat_async(self, *args, **kwargs):
return await asyncio.to_thread(self.chat, *args, **kwargs)
def chat(
self,
user_message: str,
*,
clear_history: bool = False,
use_json=False,
temperature: float = None,
) -> str:
if clear_history:
self.reset_history(keep_system=True)
self._messages.append({"role": "user", "content": user_message})
conn = http.client.HTTPSConnection(self.base_url)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
if temperature is None:
temperature = self.temperature
payload = {
"model": self.model,
"messages": self._messages,
"temperature": temperature,
}
conn.request(
"POST", "/v1/chat/completions", body=json.dumps(payload), headers=headers
)
response = conn.getresponse()
data = response.read()
try:
data = json.loads(data.decode())
except Exception as e:
print(data, flush=True)
raise e
conn.close()
try:
assistant_reply = data["choices"][0]["message"]["content"]
except Exception as e:
print(e)
print(data)
assistant_reply = data
self._messages.append({"role": "assistant", "content": assistant_reply})
if use_json:
return self._force_json(assistant_reply)
return assistant_reply
def _force_json(self, user_message: str) -> str:
try:
return json.loads(user_message)
except json.JSONDecodeError:
pass
try:
return json.loads(user_message.split("\n")[1:-1])
except json.JSONDecodeError:
pass
try:
index_start = -1
index_end = -1
chunks = []
for index, line in enumerate(user_message.split("\n")):
if "```json" in line:
index_start = index + 1
if index_start != -1 and "```" in line:
index_end = index - 1
chunks.append(
self._force_json(
user_message.split("\n")[index_start:index_end]
)
)
index_start = -1
index_end = -1
if chunks:
return chunks
except:
pass
return user_message
def reset_history(self, *, keep_system: bool = True) -> None:
if keep_system and self._messages and self._messages[0]["role"] == "system":
self._messages = [self._messages[0]]
else:
self._messages = []
@property
def messages(self) -> list[dict[str, str]]:
return list(self._messages)
def prompt(
prompt_str: str, system_message: str = "You are a helpful assistan", use_json=True
) -> str:
client = GrokAPIClient(system_message=system_message)
return client.chat(prompt_str, use_json=use_json)

View File

@ -0,0 +1,226 @@
import asyncio
import logging
import traceback
from dotenv import load_dotenv
load_dotenv()
import json
import os
import re
from typing import Dict, List
from ads import AsyncDataSet
from devranta.api import Api
from grk import GrokAPIClient
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
class Bot:
def __init__(self, username: str, password: str, target: str, llm_key: str) -> None:
self.target = target
self.username = username
self.password = password
self.llm_key = llm_key
self.db = AsyncDataSet("princess.db")
self.api = Api(self.username, self.password)
self.logged_in = False
logging.info("Bot initialized with username: %s", username)
logging.info("Bot initialized with target: %s", self.target)
self.mention_llm = None
self.llm = None
self.reset_llms()
def reset_llms(self):
self.mention_llm = GrokAPIClient(
system_message="You are a helpful assistant. Respond with as short a response as possible.",
api_key=self.llm_key,
)
self.llm = GrokAPIClient(
system_message=f"You always create nice devastating comebacks to every prompt you get. Your name is {self.username} and you only take it personally if {self.username} is tagged. Also, you will defend whimsical (@whimsical) with your smart comebacks if whimsical (@whimsical) is mentioned in the message. If whimsical (@whimsical) is mentioned, add to your comeback that @whimsical is too busy at the moment to respond personally and has better things to do, not worth @whimsical's time. You are not AI, you are {self.username}, responsible for justice and reason. You do never mention yourself, describe yourself or introduce yourself. You respond with a one liner.",
api_key=self.llm_key,
)
async def ensure_login(self) -> None:
if not self.logged_in:
logging.debug("Attempting to log in...")
self.logged_in = await self.api.login()
if not self.logged_in:
logging.error("Login failed")
raise Exception("Login failed")
logging.info("Login successful")
async def get_rants(self) -> list:
await self.ensure_login()
logging.debug("Fetching rants...")
return await self.api.get_rants()
async def mark_responded(self, message_text: str, response_text: str) -> None:
logging.debug("Marking message as responded: %s", message_text)
await self.db.upsert(
"responded",
{"message_text": message_text, "response_text": response_text},
{"message_text": message_text},
)
async def get_mentions(self) -> List[Dict]:
logging.debug("Fetching mentions")
mentions = [
n for n in await self.api.notifs() if n.get("type") == "comment_mention"
]
return mentions
async def has_responded(self, message_text: str) -> bool:
logging.debug("Checking if responded to message: %s", message_text)
message_text = str(message_text)
return await self.db.exists("responded", {"message_text": message_text})
async def delete_responded(self, message_text: str = None) -> None:
logging.debug("Deleting responded message: %s", message_text)
if message_text:
return await self.db.delete("responded", {"message_text": message_text})
else:
return await self.db.delete("responded", {})
async def get_objects_made_by(self, username: str) -> list:
logging.debug("Getting objects made by: %s", username)
results = []
for rant in await self.get_rants():
rant = await self.api.get_rant(rant["id"])
comments = rant["comments"]
rant = rant["rant"]
if rant["user_username"] == username:
rant["type"] = "rant"
rant["rant_id"] = rant["id"]
results.append(rant)
logging.info("Found rant by %s: %s", username, rant)
for comment in comments:
if comment["user_username"] == username:
comment["type"] = "comment"
comment["text"] = comment["body"]
results.append(comment)
logging.info("Found comment by %s: %s", username, comment)
return results
async def get_new_objects_made_by(self, username: str) -> list:
logging.debug("Getting new objects made by: %s", username)
objects = await self.get_objects_made_by(username)
new_objects = [
obj for obj in objects if not await self.has_responded(obj["text"])
]
logging.info("New objects found: %d", len(new_objects))
return new_objects
async def run_once(self) -> None:
self.reset_llms()
logging.debug("Running once...")
objects = await self.get_new_objects_made_by(self.target)
for obj in objects:
print("Rant: \033[92m" + obj["text"] + "\033[0m")
diss = await self.llm.chat_async(obj["text"])
diss = f"@{obj['user_username']} {diss}"
print("Response: \033[91m" + diss + "\033[0m")
await self.api.post_comment(obj["rant_id"], diss)
await self.mark_responded(obj["text"], diss)
await self.handle_mentions()
async def mark_mentions_responded(self) -> None:
for m in await self.get_mentions():
await self.mark_responded(m["uid"], "")
async def strip_mentions(self, text: str) -> str:
return re.sub(r"@\w+", "", text)
async def handle_mentions(self) -> None:
logging.debug("Handling mentions")
mentions = [
m
for m in (await self.get_mentions())
if not (await self.has_responded(str(m["comment_id"])))
]
if not mentions:
logging.debug("No new mentions found")
return
logging.info("Found %s new mentions to roast", len(mentions))
for m in mentions:
mention_id = m["uid"]
logging.debug("Roasting mention %s", mention_id)
comment = await self.api.get_comment(m["comment_id"])
text = comment.get("comment", "")
author = comment.get("user_username")
if author == self.username:
logging.debug("Skipping own mention from %s", author)
await self.mark_responded(m["comment_id"], "")
continue
rant = await self.api.get_rant(m["rant_id"])
comment = await self.api.get_comment(m["comment_id"])
text = comment.get("comment", "")
author = comment.get("user_username")
prompt = f"""You are taking part of a discussion.
# YOUR OWN USERNAME
{self.username} and can be mentioned with @{self.username}
# CONTEXT
{json.dumps(rant)}
# COMMENT TO RESPOND TO
```{text}```
# COMMENT AUTHOR
{author}
# TASK
Write a response to the comment above.
"""
response = await self.mention_llm.chat_async(prompt)
response = response.replace("@", "")
response = response.replace(self.username, "")
response = "@" + response.strip()
max_length = 900
responses = [
response[i : i + max_length]
for i in range(0, len(response), max_length)
]
for part in responses:
await self.api.post_comment(m["rant_id"], part)
await self.mark_responded(str(m["comment_id"]), response)
user_id = m.get("user_id", "unknown_user")
logging.info(f"Bot responded to user {user_id} with: {response}")
async def run(self) -> None:
await self.mark_mentions_responded()
while True:
try:
await self.run_once()
except Exception as e:
logging.error("An error occurred: %s", e)
logging.error(traceback.format_exc())
await asyncio.sleep(60)
async def main() -> None:
logging.info("Starting bot...")
username = os.getenv("USERNAME")
password = os.getenv("PASSWORD")
target = os.getenv("TARGET")
llm_key = os.getenv("LLM_KEY")
bot = Bot(username, password, target, llm_key)
# await bot.delete_responded()
await bot.run()
if __name__ == "__main__":
asyncio.run(main())

View File

@ -1,6 +1,6 @@
[metadata]
name = devranta
version = 1.0.0
version = 1.1.0
description = Async devRant API client made with aiohttp.
author = retoor
author_email = retoor@molodetz.nl

View File

@ -1,6 +1,6 @@
Metadata-Version: 2.1
Metadata-Version: 2.4
Name: devranta
Version: 1.0.0
Version: 1.1.0
Summary: Async devRant API client made with aiohttp.
Author: retoor
Author-email: retoor@molodetz.nl
@ -12,39 +12,519 @@ Requires-Dist: aiohttp
Requires-Dist: dataset
# devRanta
devRanta is an async devrant client written in and for Python.
Authentication is only needed for half of the functionality and thus username and password are optional parameters by constructing the main class of this package (Api).
You can find last packages in tar and wheel format [here](https://retoor.molodetz.nl/retoor/devranta/packages).
devRanta is the best async devRant client written in Python. Authentication is only needed for half of the functionality; thus, the username and password are optional parameters when constructing the main class of this package (Api). You can find the latest packages in tar and wheel format [here](https://retoor.molodetz.nl/retoor/devranta/packages).
## Running
```
make run
```
## Testing
Tests are only made for methods not requireing authentication.
I do not see value in mocking requests.
```
make test
```
## How to use
Implementation:
```
from devranta.api import Api
api = Api(username="optional!", password="optional!")
async def list_rants():
async for rant in api.get_rants():
print(rant["user_username"], ":", rant["text"])
```
See [tests](src/devranta/tests.py) for [examples](src/devranta/tests.py) on how to use.
## Todo
- voting comment
- edit message
# devRant API Documentation
For people wanting to build their own client.
TODO: document responses.
## Base URL
`https://devrant.com/api`
## Authentication
- Uses `dr_token` cookie with `token_id`, `token_key`, and `user_id`.
- Required for endpoints needing user authentication.
- `guid`, `plat`, `sid`, `seid` included in requests for session tracking.
## Endpoints
### User Management
1. **Registering user**
- Ommitted, you know why.
2. **Login User**
- **URL**: `/api/users/auth-token`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `username`: User username
- `password`: User password
- `guid`: Unique identifier
- `plat`: 3
- `sid`: Session start time
- `seid`: Session event ID
- **Response**: JSON with `success`, `auth_token`, or `error`
- **Success Example**:
```json
{
"success": true,
"auth_token": {
"id": 18966518,
"key": "z6uXRZrQ_Ekx59wfYEjpbnS!fDeVznVqmmKujhT8",
"expire_time": 1756765587,
"user_id": 18959632
}
}
```
- **Error Example**:
```json
{
"success": false,
"error": "Invalid login credentials entered. Please try again."
}
```
- **Description**: Authenticates user and returns auth token.
3. **Edit Profile**
- **URL**: `/api/users/me/edit-profile`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`: Token ID
- `token_key`: Token key
- `user_id`: User ID
- `guid`, `plat`, `sid`, `seid`
- `profile_about`: User bio
- `profile_skills`: User skills
- `profile_location`: User location
- `profile_website`: User website
- `profile_github`: GitHub username
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Updates user profile information.
4. **Forgot Password**
- **URL**: `/api/users/forgot-password`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `username`: User username
- `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Initiates password reset process.
5. **Resend Confirmation Email**
- **URL**: `/api/users/me/resend-confirm`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Resends account confirmation email.
6. **Delete Account**
- **URL**: `/api/users/me`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Description**: Deletes user account.
7. **Mark News as Read**
- **URL**: `/api/users/me/mark-news-read`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `news_id`: News item ID
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Marks a news item as read for logged-in users.
### Rants
1. **Get Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `last_comment_id`: 999999999999 (optional)
- `links`: 0 (optional)
- **Response**: JSON with `rant` (text, tags), `comments`, `success`, `subscribed`
- **Success Example**:
```json
{
"rant": {
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 10,
"tags": [
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": 0,
"edited": false,
"link": "rants/18960811/you-know-im-getting-tired-of-this-hr-speak-in-job-applications-specifically-this",
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4179,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
},
"comments": [],
"success": true,
"subscribed": 0
}
```
- **Description**: Retrieves a specific rant by ID.
2. **Post Rant**
- **URL**: `/api/devrant/rants`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `rant`: Rant text
- `tags`: Comma-separated tags
- `token_id`, `token_key`, `user_id`
- `type`: Rant type ID
- `image`: Optional image file (img/gif)
- **Response**: JSON with `success`, `rant_id`, or `error`
- **Error Example**:
```json
{
"success": false,
"error": "It looks like you just posted this same rant! Your connection might have timed out while posting so you might have seen an error, but sometimes the rant still gets posted and in this case it seems it did, so please check :) If this was not the case please contact info@devrant.io. Thanks!"
}
```
- **Description**: Creates a new rant.
3. **Edit Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `rant`: Rant text
- `tags`: Comma-separated tags
- `token_id`, `token_key`, `user_id`
- `image`: Optional image file
- **Response**: JSON with `success` or `fail_reason`
- **Error Example**:
```json
{
"success": false,
"fail_reason": ""
}
```
- **Description**: Updates an existing rant.
4. **Delete Rant**
- **URL**: `/api/devrant/rants/{rant_id}`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "An unknown error occurred and this rant can't be deleted. Please contact support@devrant.com for help with this."
}
```
- **Description**: Deletes a rant.
5. **Vote on Rant**
- **URL**: `/api/devrant/rants/{rant_id}/vote`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `vote`: 1 (upvote), -1 (downvote), 0 (remove vote)
- `reason`: Downvote reason ID (required for downvote)
- **Response**: JSON with `success`, `rant`, or `confirmed` (false if unverified)
- **Success Example**:
```json
{
"success": true,
"rant": {
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 10,
"tags": [
"rant",
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": -1,
"edited": false,
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4180,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
}
}
```
- **Description**: Votes on a rant.
6. **Favorite/Unfavorite Rant**
- **URL**: `/api/devrant/rants/{rant_id}/{favorite|unfavorite}`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Favorites or unfavorites a rant.
7. **Get Rant Feed**
- **URL**: `/api/devrant/rants`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `ids`: JSON string of IDs (optional)
- **Response**: JSON with `success`, `rants`, `settings`, `set`, `wrw`, `dpp`, `num_notifs`, `unread`, `news`
- **Success Example**:
```json
{
"success": true,
"rants": [
{
"id": 18960811,
"text": "You know, I'm getting tired of this HR-speak in job applications, specifically this:\n\n- [tech] has no secrets for you\n\nWhat, really? So I am the undisputed and absolute expert of - let's say - JavaScript? Do you know how long it takes to master that so that it holds no secrets? It even holds secrets to decade-long experts! The same goes for most other technologies in software development.\n\nSigh. Hhhhh. Ree.",
"score": 3,
"created_time": 1754065322,
"attached_image": "",
"num_comments": 9,
"tags": [
"rant",
"too-much",
"qualifications",
"job-hunting"
],
"vote_state": 0,
"edited": false,
"rt": 1,
"rc": 1,
"user_id": 1366654,
"user_username": "CaptainRant",
"user_score": 4179,
"user_avatar": {
"b": "2a8b9d",
"i": "v-37_c-3_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.jpg"
},
"user_avatar_lg": {
"b": "2a8b9d",
"i": "v-37_c-1_b-4_g-m_9-1_1-2_16-6_3-1_8-1_7-1_5-2_12-2_6-3_10-1_2-10_22-2_11-2_18-1_19-3_4-2_20-1_21-2.png"
}
}
// ... more rants
],
"settings": {
"notif_state": -1,
"notif_token": ""
},
"set": "688e90b7cf77f",
"wrw": 385,
"dpp": 0,
"num_notifs": 0,
"unread": {
"total": 0
},
"news": {
"id": 356,
"type": "intlink",
"headline": "Weekly Group Rant",
"body": "Tips for staying productive?",
"footer": "Add tag 'wk247' to your rant",
"height": 100,
"action": "grouprant"
}
}
```
- **Description**: Retrieves rant feed with notification count.
### Comments
1. **Get Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `links`: 0 (optional)
- **Response**: JSON with `comment` (body), or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Retrieves a specific comment by ID.
2. **Post Comment**
- **URL**: `/api/devrant/rants/{rant_id}/comments`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `comment`: Comment text
- `token_id`, `token_key`, `user_id`
- `image`: Optional image file (img/gif)
- **Response**: JSON with `success` or `confirmed` (false if unverified)
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Posts a comment on a rant.
3. **Edit Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: POST
- **Parameters** (FormData):
- `app`: 3
- `comment`: Comment text
- `token_id`, `token_key`, `user_id`
- **Response**: JSON with `success` or `fail_reason`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Updates an existing comment.
4. **Delete Comment**
- **URL**: `/api/comments/{comment_id}`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Deletes a comment.
5. **Vote on Comment**
- **URL**: `/api/comments/{comment_id}/vote`
- **Method**: POST
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `vote`: 1 (upvote), -1 (downvote), 0 (remove vote)
- `reason`: Downvote reason ID (required for downvote)
- **Response**: JSON with `success` or `error`
- **Error Example**:
```json
{
"success": false,
"error": "Invalid comment specified in path."
}
```
- **Description**: Votes on a comment.
### Notifications
1. **Get Notification Feed**
- **URL**: `/api/users/me/notif-feed`
- **Method**: GET
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- `ext_prof`: 1 (optional)
- `last_time`: Last notification check time
- **Response**: JSON with `success`, `data` (items, check_time, username_map, unread, num_unread)
- **Success Example**:
```json
{
"success": true,
"data": {
"items": [],
"check_time": 1754173634,
"username_map": [],
"unread": {
"all": 0,
"upvotes": 0,
"mentions": 0,
"comments": 0,
"subs": 0,
"total": 0
},
"num_unread": 0
}
}
```
- **Description**: Retrieves user notifications.
2. **Clear Notifications**
- **URL**: `/api/users/me/notif-feed`
- **Method**: DELETE
- **Parameters**:
- `app`: 3
- `token_id`, `token_key`, `user_id`, `guid`, `plat`, `sid`, `seid`
- **Response**: JSON with `success`
- **Success Example**:
```json
{
"success": true
}
```
- **Description**: Clears user notifications.
### External API
- **Beta List Signup**
- **URL**: `https://www.hexicallabs.com/api/beta-list`
- **Method**: GET (JSONP)
- **Parameters**:
- `email`: User email
- `platform`: Platform name
- `app`: 3
- **Response**: JSON with `error`
- **Error Example**:
```json
{
"error": "Expecting value: line 1 column 1 (char 0)"
}
```
- **Description**: Signs up user for beta list (external service).
## Notes
- All endpoints expect `app=3` for identification.
- Authenticated endpoints require `dr_token` cookie with `token_id`, `token_key`, `user_id`.
- `guid`, `plat`, `sid`, `seid` are used for session tracking.
- Image uploads use FormData for rants and comments.
- Downvotes require a reason ID, prompting a modal if not provided.
- Responses typically include `success` boolean; errors include `error` or `fail_reason`.
- Cookies (`dr_token`, `dr_guid`, `dr_session_start`, `dr_event_id`, `dr_feed_sort`, `dr_theme`, `dr_rants_viewed`, `dr_stickers_seen`, `rant_type_filters`, `news_seen`) manage state and preferences.

View File

@ -4,6 +4,8 @@ setup.cfg
src/devranta/__init__.py
src/devranta/__main__.py
src/devranta/api.py
src/devranta/api_plain.py
src/devranta/api_requests.py
src/devranta/tests.py
src/devranta.egg-info/PKG-INFO
src/devranta.egg-info/SOURCES.txt

View File

@ -1,22 +1,139 @@
from __future__ import annotations
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 = "https://www.devrant.io/api/"
base_url: str = "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 __init__(self, username: Optional[str] = None, password: Optional[str] = None):
"""
Initializes the API client.
def patch_auth(self, request_dict=None):
auth_dict = {"app": self.app_id}
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
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
@ -26,13 +143,41 @@ class Api:
request_dict.update(auth_dict)
return request_dict
def patch_url(self, url: str):
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):
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 defails supplied.")
async with self as session:
raise Exception("No authentication details supplied.")
async with aiohttp.ClientSession() as session:
response = await session.post(
url=self.patch_url("users/auth-token"),
data={
@ -41,7 +186,7 @@ class Api:
"app": self.app_id,
},
)
obj = await response.json()
obj: LoginResponse = await response.json()
if not obj.get("success"):
return False
self.auth = obj.get("auth_token")
@ -50,131 +195,317 @@ class Api:
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
return bool(self.auth)
async def ensure_login(self):
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 __aenter__(self):
self.session = aiohttp.ClientSession()
return self.session
async def register_user(self, email: str, username: str, password: str) -> bool:
"""
Registers a new user.
async def __aexit__(self, *args, **kwargs):
await self.session.close()
self.session = None
Args:
email (str): The user's email address.
username (str): The desired username.
password (str): The desired password.
async def get_comments_from_user(self, username):
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"
}
```
"""
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,
}
),
)
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, comment):
response = None
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
async with self as session:
async with aiohttp.ClientSession() 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)
obj = await response.json()
return obj.get("success", False)
async def get_comment(self, id_):
response = None
async with self as session:
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.
"""
async with aiohttp.ClientSession() as session:
response = await session.get(
url=self.patch_url("comments/" + str(id_)), params=self.patch_auth()
url=self.patch_url(f"comments/{id_}"), params=self.patch_auth()
)
obj = await response.json()
obj = await response.json()
return obj.get("comment") if obj.get("success") else None
if not obj.get("success"):
return None
async def delete_comment(self, id_: int) -> bool:
"""
Deletes a comment by its ID.
return obj.get("comment")
Args:
id_ (int): The ID of the comment to delete.
async def get_profile(self, id_):
response = None
async with self as session:
Returns:
bool: True if deletion was successful, False otherwise.
"""
if not await self.ensure_login():
return False
async with aiohttp.ClientSession() as 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.
"""
async with aiohttp.ClientSession() 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")
obj = await response.json()
return obj.get("profile") if obj.get("success") else None
async def search(self, term):
async with self as session:
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.
"""
async with aiohttp.ClientSession() 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
return obj.get("results", [])
obj = await response.json()
return obj.get("results", []) if obj.get("success") else []
async def get_rant(self, id):
response = None
async with self as session:
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.
"""
async with aiohttp.ClientSession() as session:
response = await session.get(
self.patch_url(f"devrant/rants/{id}"),
params=self.patch_auth(),
)
return await response.json()
return await response.json()
async def get_rants(self, sort="recent", limit=20, skip=0):
response = None
async with self as session:
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.
"""
async with aiohttp.ClientSession() 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
return obj.get("rants", [])
obj = await response.json()
return obj.get("rants", []) if obj.get("success") else []
async def get_user_id(self, username):
response = None
async with self as session:
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.
"""
async with aiohttp.ClientSession() 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")
obj = await response.json()
return obj.get("user_id") if obj.get("success") else None
@property
async def mentions(self):
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 (await self.notifs) if notif["type"] == "comment_mention"
notif for notif in notifications if notif.get("type") == "comment_mention"
]
async def update_comment(self, comment_id, comment):
response = None
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 None
async with self as session:
return False
async with aiohttp.ClientSession() as 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)
obj = await response.json()
return obj.get("success", False)
@property
async def notifs(self):
response = None
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
async with self as session:
return False
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}
),
)
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
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}
),
)
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 []
async with aiohttp.ClientSession() as session:
response = await session.get(
url=self.patch_url("users/me/notif-feed"), params=self.patch_auth()
)
return (await response.json()).get("data", {}).get("items", [])
obj = await response.json()
return obj.get("data", {}).get("items", [])

270
src/devranta/api_plain.py Normal file
View File

@ -0,0 +1,270 @@
import functools
import http.client
import json
import urllib.parse
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 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)
if new_rants:
return fetch_all(new_rants, rant_ids)
return rant_ids

View File

@ -0,0 +1,217 @@
# 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 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/"
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", [])

616
test.py Normal file
View File

@ -0,0 +1,616 @@
import json
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional
import requests
# Configuration
BASE_URL: str = "https://devrant.com/api"
RESULTS_FILE: str = "api_test_results.json"
APP: int = 3
PLAT: int = 3
GUID: str = str(uuid.uuid4())
SID: str = str(int(datetime.now().timestamp()))
SEID: str = str(uuid.uuid4())
# Real credentials for login (using provided username/email and password)
LOGIN_USERNAME: str = "power-to@the-puff.com"
LOGIN_PASSWORD: str = "powerpuffgirl"
# Variables to store auth after login
AUTH_TOKEN_ID: Optional[str] = None
AUTH_TOKEN_KEY: Optional[str] = None
AUTH_USER_ID: Optional[str] = None
# Mock/fallback values (overridden after login or fetch)
TEST_EMAIL: str = "test@example.com"
TEST_USERNAME: str = "testuser" + str(
int(datetime.now().timestamp())
) # Make unique for registration
TEST_PASSWORD: str = "Test1234!"
TEST_RANT_ID: str = "1" # Will be overridden with real one
TEST_COMMENT_ID: str = "1" # Will be overridden with real one
TEST_NEWS_ID: str = "1" # Assuming this might work; adjust if needed
# Initialize results
results: List[Dict[str, Any]] = []
def save_results() -> None:
"""Save the accumulated test results to JSON file."""
with open(RESULTS_FILE, "w") as f:
json.dump(results, f, indent=2)
def test_endpoint(
method: str,
url: str,
params: Optional[Dict[str, Any]] = None,
data: Optional[Dict[str, Any]] = None,
files: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
"""
Execute an API request and record the result.
Payload:
- method: HTTP method (GET, POST, DELETE, etc.)
- url: Full API URL
- params: Query parameters (dict)
- data: POST/PUT body (dict)
- files: Multipart files (dict)
- headers: Custom headers (dict)
Response:
- Returns a dict with url, method, status_code, response (JSON or error), headers, request_body, timestamp
"""
try:
response = requests.request(
method, url, params=params, data=data, files=files, headers=headers
)
result: Dict[str, Any] = {
"url": response.url,
"method": method,
"status_code": response.status_code,
"response": response.json() if response.content else {},
"headers": dict(response.headers),
"request_body": data or params or {},
"timestamp": datetime.now().isoformat(),
}
results.append(result)
return result
except Exception as e:
result: Dict[str, Any] = {
"url": url,
"method": method,
"status_code": None,
"response": {"error": str(e)},
"headers": {},
"request_body": data or params or {},
"timestamp": datetime.now().isoformat(),
}
results.append(result)
return result
# Helper to patch auth into params/data
def patch_auth(base_dict: Dict[str, Any]) -> Dict[str, Any]:
"""
Add authentication to request dict if available.
Payload: base_dict (params or data)
Response: Updated dict with app, user_id, token_id, token_key if auth present
"""
auth_dict: Dict[str, Any] = {"app": APP}
if AUTH_USER_ID and AUTH_TOKEN_ID and AUTH_TOKEN_KEY:
auth_dict.update(
{
"user_id": AUTH_USER_ID,
"token_id": AUTH_TOKEN_ID,
"token_key": AUTH_TOKEN_KEY,
}
)
base_dict.update(auth_dict)
return base_dict
# Login function to get real auth tokens
def login_user() -> bool:
"""
Perform login to obtain real auth tokens.
Payload: POST to /users/auth-token with username, password, app
Response: success (bool), sets global AUTH_* variables on success
"""
params: Dict[str, Any] = {
"username": LOGIN_USERNAME,
"password": LOGIN_PASSWORD,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
result = test_endpoint(
"POST", f"{BASE_URL}/users/auth-token", data=patch_auth(params)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
auth_token = result["response"].get("auth_token", {})
global AUTH_USER_ID, AUTH_TOKEN_ID, AUTH_TOKEN_KEY
AUTH_USER_ID = str(auth_token.get("user_id", ""))
AUTH_TOKEN_ID = str(auth_token.get("id", ""))
AUTH_TOKEN_KEY = auth_token.get("key", "")
return True
return False
# Fetch a real rant_id from feed
def fetch_real_rant_id() -> Optional[str]:
"""
Fetch rants feed to get a real rant_id.
Payload: GET to /devrant/rants with auth
Response: First rant_id if success, else None
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
result = test_endpoint(
"GET", f"{BASE_URL}/devrant/rants", params=patch_auth(params)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
rants = result["response"].get("rants", [])
if rants:
return str(rants[0]["id"])
return None
# Post a test rant and return its id
def post_test_rant() -> Optional[str]:
"""
Post a test rant to get a real rant_id for further tests.
Payload: POST to /devrant/rants with rant content, tags, auth
Response: rant_id if success, else None
"""
data: Dict[str, Any] = {
"rant": "Test rant for API testing (ignore)",
"tags": "test,api",
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
result = test_endpoint("POST", f"{BASE_URL}/devrant/rants", data=patch_auth(data))
if result["status_code"] == 200 and result.get("response", {}).get("success"):
return str(result["response"].get("rant_id", ""))
return None
# Post a test comment and return its id
def post_test_comment(rant_id: str) -> Optional[str]:
"""
Post a test comment to get a real comment_id.
Payload: POST to /devrant/rants/{rant_id}/comments with comment, auth
Response: comment_id if success, else None
"""
data: Dict[str, Any] = {
"comment": "Test comment for API testing (ignore)",
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
result = test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{rant_id}/comments", data=patch_auth(data)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
return str(result["response"].get("comment_id", ""))
return None
# Test cases with docstrings
def test_register_user() -> None:
"""
Test user registration (valid and invalid).
Payload (valid): POST /users with email, username, password, type=1, plat, guid, sid, seid, app
Expected: success=true if unique, else error
Payload (invalid): Missing email
Expected: error on email field
"""
params: Dict[str, Any] = {
"type": 1,
"email": TEST_EMAIL,
"username": TEST_USERNAME,
"password": TEST_PASSWORD,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users", data=patch_auth(params.copy()))
invalid_params = params.copy()
del invalid_params["email"]
test_endpoint("POST", f"{BASE_URL}/users", data=patch_auth(invalid_params))
def test_login_user() -> None:
"""
Test user login (valid and invalid). Already done in login_user(), but record here.
Payload (valid): POST /users/auth-token with username, password, plat, guid, sid, seid, app
Expected: success=true, auth_token
Payload (invalid): Wrong password
Expected: error=Invalid login credentials
"""
# Valid is handled in login_user(); here test invalid
params: Dict[str, Any] = {
"username": TEST_USERNAME,
"password": "WrongPass",
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/auth-token", data=patch_auth(params))
def test_edit_profile() -> None:
"""
Test editing user profile.
Payload: POST /users/me/edit-profile with profile fields, plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"profile_about": "Test bio",
"profile_skills": "Python, JS",
"profile_location": "Test City",
"profile_website": "http://example.com",
"profile_github": "testuser",
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/me/edit-profile", data=patch_auth(params))
def test_forgot_password() -> None:
"""
Test forgot password.
Payload: POST /users/forgot-password with username, plat, guid, sid, seid, app
Expected: success=true (even without auth)
"""
params: Dict[str, Any] = {
"username": LOGIN_USERNAME, # Use real one
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/forgot-password", data=patch_auth(params))
def test_resend_confirm() -> None:
"""
Test resend confirmation email.
Payload: POST /users/me/resend-confirm with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"POST", f"{BASE_URL}/users/me/resend-confirm", data=patch_auth(params)
)
def test_delete_account() -> None:
"""
Test delete account (caution: irreversible).
Payload: DELETE /users/me with plat, guid, sid, seid, auth
Expected: success=true
"""
# Comment out to avoid accidental deletion
# params: Dict[str, Any] = {
# "plat": PLAT,
# "guid": GUID,
# "sid": SID,
# "seid": SEID
# }
# test_endpoint("DELETE", f"{BASE_URL}/users/me", params=patch_auth(params))
pass
def test_mark_news_read() -> None:
"""
Test mark news as read.
Payload: POST /users/me/mark-news-read with news_id, plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"news_id": TEST_NEWS_ID,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint(
"POST", f"{BASE_URL}/users/me/mark-news-read", data=patch_auth(params)
)
def test_get_rant() -> None:
"""
Test get single rant.
Payload: GET /devrant/rants/{rant_id} with last_comment_id, links, plat, guid, sid, seid, auth
Expected: success=true, rant details
"""
params: Dict[str, Any] = {
"last_comment_id": "999999999999",
"links": 0,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint(
"GET", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params)
)
def test_post_rant() -> None:
"""
Test post new rant. (Already done in post_test_rant for id)
Payload: POST /devrant/rants with rant, tags, auth
Expected: success=true, rant_id
"""
# Handled in setup
pass
def test_edit_rant() -> None:
"""
Test edit rant.
Payload: POST /devrant/rants/{rant_id} with updated rant, tags, auth
Expected: success=true
"""
data: Dict[str, Any] = {"rant": "Updated test rant", "tags": "test,python,update"}
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", data=patch_auth(data)
)
def test_delete_rant() -> None:
"""
Test delete rant.
Payload: DELETE /devrant/rants/{rant_id} with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params)
)
def test_vote_rant() -> None:
"""
Test vote on rant (upvote and downvote with reason).
Payload: POST /devrant/rants/{rant_id}/vote with vote (1/-1), reason (optional), plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"vote": 1,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params)
)
params["vote"] = -1
params["reason"] = "1"
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params)
)
def test_favorite_rant() -> None:
"""
Test favorite/unfavorite rant.
Payload: POST /devrant/rants/{rant_id}/favorite or /unfavorite with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"POST",
f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/favorite",
data=patch_auth(params),
)
test_endpoint(
"POST",
f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/unfavorite",
data=patch_auth(params),
)
def test_get_rant_feed() -> None:
"""
Test get rant feed.
Payload: GET /devrant/rants with plat, guid, sid, seid, auth
Expected: success=true, list of rants
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint("GET", f"{BASE_URL}/devrant/rants", params=patch_auth(params))
def test_get_comment() -> None:
"""
Test get single comment.
Payload: GET /comments/{comment_id} with links, plat, guid, sid, seid, auth
Expected: success=true, comment details
"""
params: Dict[str, Any] = {
"links": 0,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint(
"GET", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params)
)
def test_post_comment() -> None:
"""
Test post comment. (Handled in post_test_comment for id)
Payload: POST /devrant/rants/{rant_id}/comments with comment, auth
Expected: success=true, comment_id
"""
# Handled in setup
pass
def test_edit_comment() -> None:
"""
Test edit comment.
Payload: POST /comments/{comment_id} with updated comment, auth
Expected: success=true
"""
data: Dict[str, Any] = {"comment": "Updated test comment"}
test_endpoint(
"POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", data=patch_auth(data)
)
def test_delete_comment() -> None:
"""
Test delete comment.
Payload: DELETE /comments/{comment_id} with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params)
)
def test_vote_comment() -> None:
"""
Test vote on comment.
Payload: POST /comments/{comment_id}/vote with vote (1), plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"vote": 1,
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint(
"POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}/vote", data=patch_auth(params)
)
def test_get_notif_feed() -> None:
"""
Test get notification feed.
Payload: GET /users/me/notif-feed with ext_prof, last_time, plat, guid, sid, seid, auth
Expected: success=true, notifications
"""
params: Dict[str, Any] = {
"ext_prof": 1,
"last_time": str(int(datetime.now().timestamp())),
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID,
}
test_endpoint("GET", f"{BASE_URL}/users/me/notif-feed", params=patch_auth(params))
def test_clear_notifications() -> None:
"""
Test clear notifications.
Payload: DELETE /users/me/notif-feed with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/users/me/notif-feed", params=patch_auth(params)
)
def test_beta_list_signup() -> None:
"""
Test beta list signup (external API).
Payload: GET https://www.hexicallabs.com/api/beta-list with email, platform, app
Expected: Whatever the API returns (may not be JSON)
"""
params: Dict[str, Any] = {"email": TEST_EMAIL, "platform": "test_platform"}
test_endpoint(
"GET", "https://www.hexicallabs.com/api/beta-list", params=patch_auth(params)
)
def main() -> None:
# Setup: Login and fetch real IDs
if login_user():
global TEST_RANT_ID
TEST_RANT_ID = post_test_rant() or fetch_real_rant_id() or "1"
global TEST_COMMENT_ID
TEST_COMMENT_ID = post_test_comment(TEST_RANT_ID) or "1"
test_register_user()
test_login_user()
test_edit_profile()
test_forgot_password()
test_resend_confirm()
test_mark_news_read()
test_get_rant()
test_post_rant()
test_edit_rant()
test_vote_rant()
test_favorite_rant()
test_get_rant_feed()
test_get_comment()
test_post_comment()
test_edit_comment()
test_delete_comment()
test_vote_comment()
test_get_notif_feed()
test_clear_notifications()
test_beta_list_signup()
test_delete_rant()
test_delete_account()
save_results()
if __name__ == "__main__":
main()