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

Add onsaling + better documentation.

Browse Source
This commit is contained in:
filoxenace
2025-12-21 21:05:20 -05:00
parent bb8d8325ad
commit ffd5b3b883
4 changed files with 100 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1
View File
@@ -1,3 +1,4 @@
TARGET_ID=
VALID_API_KEY=
ROBLOSECURITY_TOKEN=
PUBLISHER_USER_ID=
README.md
+3
View File
@@ -26,8 +26,11 @@ 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
```
**Important:** `PUBLISHER_USER_ID` must match the user ID of the account that owns the `ROBLOSECURITY_TOKEN`.
## ⚠️ 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.
src/main.py
+12 -4
View File
@@ -1,7 +1,7 @@
import os
from dotenv import load_dotenv
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import APIKeyHeader
import models
@@ -12,7 +12,7 @@ load_dotenv()
TARGET = os.getenv("TARGET_ID")
VALID_API_KEY = os.getenv("VALID_API_KEY")
if not TARGET:
if TARGET is None:
raise EnvironmentError("TARGET_ID is missing from environment.")
if not VALID_API_KEY:
raise EnvironmentError("VALID_API_KEY is missing from environment.")
@@ -25,8 +25,7 @@ api_key_header = APIKeyHeader(name="x-api-key")
async def verify_api_key(api_key: str = Depends(api_key_header)):
if api_key != VALID_API_KEY:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid API key"
status_code=status.HTTP_403_FORBIDDEN, detail="Invalid API key"
)
return api_key
@@ -56,4 +55,13 @@ async def reupload_asset(asset_id: int, _: str = Depends(verify_api_key)):
asset.asset_type,
models.RbxCreator(int(TARGET), "Upload_Group", "Group"),
)
new_asset_id = uploaded.get("asset_id")
if new_asset_id:
onsale = await roblox_service.onsale_asset(
new_asset_id,
asset.name,
asset.description,
int(TARGET),
)
return {"uploaded": uploaded}
return uploaded
src/utils/roblox_service.py
+84 -3
View File
@@ -1,5 +1,7 @@
import asyncio
import json
import os
import uuid
import xml.etree.ElementTree
import httpx
@@ -16,6 +18,12 @@ ROBLOSECURITY = os.getenv("ROBLOSECURITY_TOKEN")
if not ROBLOSECURITY:
raise EnvironmentError("ROBLOSECURITY_TOKEN is missing from environment.")
# Must match the user ID of the account that owns the ROBLOSECURITY_TOKEN
PUBLISHER_USER_ID = os.getenv("PUBLISHER_USER_ID")
if not PUBLISHER_USER_ID:
raise EnvironmentError("PUBLISHER_USER_ID is missing from environment.")
FETCH_HEADERS = {
"Cookie": f".ROBLOSECURITY={ROBLOSECURITY}",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
@@ -43,6 +51,7 @@ _client = None
def _get_client() -> httpx.AsyncClient:
"""Get or create the shared HTTP client for connection pooling."""
global _client
if _client is None:
_client = httpx.AsyncClient()
@@ -53,6 +62,7 @@ def _get_client() -> httpx.AsyncClient:
async def _economy_request(asset_id: int) -> httpx.Response:
"""Fetch asset details from the Roblox economy API."""
client = _get_client()
return await client.get(
ECONOMY_BASE_URL.format(asset_id=asset_id), headers=FETCH_HEADERS
@@ -60,13 +70,17 @@ async def _economy_request(asset_id: int) -> httpx.Response:
async def _asset_delivery_request(asset_id: int) -> httpx.Response:
"""Fetch an asset from the Roblox asset delivery API."""
client = _get_client()
return await client.get(
ASSET_DELIVERY_BASE_URL.format(asset_id=asset_id), headers=FETCH_HEADERS, follow_redirects=True
ASSET_DELIVERY_BASE_URL.format(asset_id=asset_id),
headers=FETCH_HEADERS,
follow_redirects=True,
)
async def _get_asset_xml(asset: models.RbxAsset) -> xml.etree.ElementTree.Element:
"""Fetch and parse asset XML content."""
response = await _asset_delivery_request(asset.asset_id)
response.raise_for_status()
content = response.content.decode("utf-8")
@@ -75,6 +89,7 @@ async def _get_asset_xml(asset: models.RbxAsset) -> xml.etree.ElementTree.Elemen
def _get_shirt_template_id_from_xml(root: xml.etree.ElementTree.Element) -> int:
"""Extract shirt template ID from asset XML."""
url_element = root.find(".//url")
if url_element is None:
raise ValueError("XML did not contain a <url> tag.")
@@ -85,9 +100,10 @@ def _get_shirt_template_id_from_xml(root: xml.etree.ElementTree.Element) -> int:
return int(template_id)
async def _get_csrf_token() -> str:
async def _get_csrf_token(url: str = CSRF_URL) -> str:
"""Retrieve CSRF token from the Roblox API."""
client = _get_client()
response = await client.post(CSRF_URL, cookies=CSRF_COOKIES, headers=CSRF_HEADERS)
response = await client.post(url, cookies=CSRF_COOKIES, headers=CSRF_HEADERS)
csrf = response.headers.get("X-CSRF-TOKEN")
if not csrf:
raise httpx.HTTPStatusError(
@@ -102,6 +118,7 @@ async def _get_csrf_token() -> str:
async def asset_from_id(id: int) -> models.RbxAsset:
"""Fetch asset information from Roblox by asset ID."""
response = await _economy_request(id)
response.raise_for_status()
asset_info = json.loads(response.content)
@@ -133,6 +150,7 @@ async def asset_from_id(id: int) -> models.RbxAsset:
async def fetch_clothing_image(asset: models.ClothingAsset) -> bytes:
"""Fetch the image data for a clothing asset."""
try:
xml = await _get_asset_xml(asset)
template_id = _get_shirt_template_id_from_xml(xml)
@@ -150,6 +168,7 @@ async def upload_clothing_image(
asset_type: models.RbxAssetType,
target: models.RbxCreator,
) -> dict:
"""Upload a clothing image to Roblox and return the asset ID."""
csrf = await _get_csrf_token()
meta = {
"displayName": name,
@@ -183,4 +202,66 @@ async def upload_clothing_image(
)
response.raise_for_status()
data = response.json()
operation_id = data.get("operationId")
if operation_id:
max_tries = 10
wait_time = 1
for attempt in range(max_tries):
await asyncio.sleep(wait_time)
op_response = await client.get(
f"https://apis.roblox.com/assets/user-auth/v1/operations/{operation_id}",
headers={"X-CSRF-TOKEN": csrf},
cookies=CSRF_COOKIES,
)
op_response.raise_for_status()
op_data = op_response.json()
if op_data.get("done"):
if op_data.get("response") and op_data["response"].get("assetId"):
return {"asset_id": op_data["response"]["assetId"]}
return op_data
return data
async def onsale_asset(
asset_id: int,
name: str,
description: str,
group_id: int,
price: int = 5,
):
"""Put an asset on sale."""
csrf = await _get_csrf_token()
data = {
"saleLocationConfiguration": {"saleLocationType": 1, "places": []},
"targetId": asset_id,
"priceInRobux": price,
"publishingType": 2,
"idempotencyToken": str(uuid.uuid4()),
"publisherUserId": PUBLISHER_USER_ID,
"creatorGroupId": group_id,
"name": name,
"description": description,
"isFree": False,
"agreedPublishingFee": 0,
"priceOffset": 0,
"quantity": 0,
"quantityLimitPerUser": 0,
"resaleRestriction": 2,
"targetType": 0,
}
client = _get_client()
response = await client.post(
"https://itemconfiguration.roblox.com/v1/collectibles",
json=data,
headers={
"X-CSRF-TOKEN": csrf,
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
"Referer": "https://create.roblox.com/",
"Origin": "https://create.roblox.com",
},
cookies=CSRF_COOKIES,
)
response.raise_for_status()