fxn/threads
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add QoL features

Browse Source
This commit is contained in:
Ivy C
2026-01-05 23:49:55 -05:00
parent 6147390a25
commit 420d9435da
7 changed files with 166 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1
View File
@@ -2,3 +2,4 @@ TARGET_ID=
VALID_API_KEY=
ROBLOSECURITY_TOKEN=
PUBLISHER_USER_ID=
DISCORD_WEBHOOK_URL=
.gitignore
+2 -1
View File
@@ -1,4 +1,4 @@
# Byte-compiled / optimized / DLL files
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
@@ -160,3 +160,4 @@ cython_debug/
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
src/database.db
README.md
+16 -4
View File
@@ -25,12 +25,20 @@ Create a `.env` file in the project root with the following variables:
```
TARGET_ID=<group_id> # The Roblox group ID to upload clothing to
VALID_API_KEY=<your_api_key> # API key for authorizing requests to this service
ROBLOSECURITY_TOKEN=<cookie> # Your Roblox roblosecurity cookie (used only for Roblox API calls)
PUBLISHER_USER_ID=<user_id> # The Roblox user ID associated with the ROBLOSECURITY_TOKEN
ROBLOSECURITY_TOKEN=<cookie> # Your Roblox roblosecurity cookie
PUBLISHER_USER_ID=<user_id> # The Roblox user ID for the cookie
DISCORD_WEBHOOK_URL=<url> # (Optional) Discord webhook for upload notifications
```
**Important:** `PUBLISHER_USER_ID` must match the user ID of the account that owns the `ROBLOSECURITY_TOKEN`.
## Features
- **Duplicate Prevention**: Uses SHA256 image hashing and a local SQLite database to prevent re-uploading the same item twice.
- **Race Condition Protection**: Uses asynchronous locking to ensure simultaneous requests for the same image don't trigger multiple uploads.
- **Discord Notifications**: Sends a rich embed to Discord whenever an asset is successfully uploaded.
- **Metadata Preservation**: Automatically appends a link to the original Roblox asset in the new asset's description.
## ⚠️ Disclaimer
This tool uses Roblox's APIs in a way that violates their Terms of Service. Roblox may moderate or ban accounts that use this. Use at your own risk.
@@ -58,6 +66,10 @@ The server will run on port 8000 by default.
```
src/
├── main.py # FastAPI application and endpoints
├── models.py # Data models for Roblox assets and creators
└── utils/ # Utility modules for Roblox API interactions
├── models.py # Data models for Roblox assets
├── database.py # SQLite database layer for duplicate tracking
└── utils/ # Utility modules
├── roblox_service.py
├── hashing.py
└── discord.py
```
src/database.py
+47
View File
@@ -0,0 +1,47 @@
import sqlite3
import os
from datetime import datetime
from typing import Optional
DB_PATH = os.path.join(os.path.dirname(__file__), "database.db")
def init_db():
"""Initializes the database and creates the necessary table."""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS uploaded_assets (
image_hash TEXT PRIMARY KEY,
original_asset_id INTEGER NOT NULL,
new_asset_id INTEGER NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
def get_uploaded_asset(image_hash: str) -> Optional[int]:
"""
Checks if an image hash already exists in the database.
Returns the new_asset_id if found, else None.
"""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT new_asset_id FROM uploaded_assets WHERE image_hash = ?",
(image_hash,)
)
row = cursor.fetchone()
return row[0] if row else None
def save_uploaded_asset(image_hash: str, original_asset_id: int, new_asset_id: int):
"""Saves a new upload record to the database."""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO uploaded_assets (image_hash, original_asset_id, new_asset_id) VALUES (?, ?, ?)",
(image_hash, original_asset_id, new_asset_id)
)
conn.commit()
# Initialize the DB on module import
init_db()
src/main.py
+39 -4
View File
@@ -5,7 +5,8 @@ from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import APIKeyHeader
import models
from utils import roblox_service
from utils import roblox_service, hashing, discord
import database
load_dotenv()
@@ -42,26 +43,60 @@ async def get_asset_info(asset_id: int, _: str = Depends(verify_api_key)):
return final_dict
# Global dictionary to store locks per image hash
upload_locks: dict[str, asyncio.Lock] = {}
@app.post("/create/")
async def reupload_asset(asset_id: int, _: str = Depends(verify_api_key)):
asset = await roblox_service.asset_from_id(asset_id)
if isinstance(asset, models.ClothingAsset):
print("clothing asset found")
print(f"Clothing asset found: {asset.name}")
image = await roblox_service.fetch_clothing_image(asset)
# Check for duplicates using hash
image_hash = hashing.get_image_hash(image)
# Get or create a lock for this specific hash
if image_hash not in upload_locks:
upload_locks[image_hash] = asyncio.Lock()
async with upload_locks[image_hash]:
# Double-check database inside the lock
existing_new_id = database.get_uploaded_asset(image_hash)
if existing_new_id:
print(f"Asset already uploaded (hash match): {existing_new_id}")
return {"uploaded": {"asset_id": existing_new_id}}
# Prepare description with original URL
original_url = f"https://www.roblox.com/catalog/{asset_id}"
new_description = f"{asset.description}\n\nOriginal: {original_url}"
uploaded = await roblox_service.upload_clothing_image(
image,
asset.name,
asset.description,
new_description,
asset.asset_type,
models.RbxCreator(int(TARGET), "Upload_Group", "Group"),
)
new_asset_id = uploaded.get("asset_id")
if new_asset_id:
# Save to database
database.save_uploaded_asset(image_hash, asset_id, new_asset_id)
onsale = await roblox_service.onsale_asset(
new_asset_id,
asset.name,
asset.description,
new_description,
int(TARGET),
)
# Send Discord notification
asset_type_name = "Shirt" if asset.asset_type == models.RbxAssetType.SHIRT else "Pants"
await discord.send_upload_webhook(
asset.name, asset_id, new_asset_id, asset_type_name
)
return {"uploaded": uploaded}
return uploaded
raise HTTPException(status_code=400, detail="Asset is not a clothing asset.")
src/utils/discord.py
+39
View File
@@ -0,0 +1,39 @@
import os
import httpx
from dotenv import load_dotenv
load_dotenv()
DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
async def send_upload_webhook(name: str, original_id: int, new_id: int, asset_type: str):
"""
Sends a Discord webhook notification with an embed for the successful upload.
"""
if not DISCORD_WEBHOOK_URL:
print("Discord Webhook URL not configured, skipping notification.")
return
original_url = f"https://www.roblox.com/catalog/{original_id}"
new_url = f"https://www.roblox.com/catalog/{new_id}"
embed = {
"title": "New Clothing Uploaded",
"color": 3066993, # Green
"fields": [
{"name": "Name", "value": name, "inline": False},
{"name": "Type", "value": asset_type, "inline": True},
{"name": "Original Asset", "value": f"[Link]({original_url})", "inline": True},
{"name": "New Asset", "value": f"[Link]({new_url})", "inline": True},
],
"footer": {"text": "Filoxen Labs"},
}
payload = {"embeds": [embed]}
async with httpx.AsyncClient() as client:
try:
response = await client.post(DISCORD_WEBHOOK_URL, json=payload)
response.raise_for_status()
except Exception as e:
print(f"Failed to send Discord webhook: {e}")
src/utils/hashing.py
+8
View File
@@ -0,0 +1,8 @@
import hashlib
def get_image_hash(image_bytes: bytes) -> str:
"""
Computes the SHA256 hash of the given image bytes.
Returns the hex digest.
"""
return hashlib.sha256(image_bytes).hexdigest()