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

Make async and add clothing upload

Browse Source
This commit is contained in:
Ivy C
2025-12-21 17:39:27 -05:00
parent 18eb2d611f
commit 97380f3160
5 changed files with 142 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.py
+45 -4
View File
@@ -1,18 +1,59 @@
from fastapi import FastAPI
import os
from dotenv import load_dotenv
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import APIKeyHeader
import models
from utils import roblox_service
load_dotenv()
TARGET = os.getenv("TARGET_ID")
VALID_API_KEY = os.getenv("VALID_API_KEY")
if not TARGET:
raise EnvironmentError("TARGET_ID is missing from environment.")
if not VALID_API_KEY:
raise EnvironmentError("VALID_API_KEY is missing from environment.")
app = FastAPI()
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"
)
return api_key
@app.get("/asset/{asset_id}")
def get_asset_info(asset_id: int):
asset = roblox_service.asset_from_id(asset_id)
async def get_asset_info(asset_id: int, _: str = Depends(verify_api_key)):
asset = await roblox_service.asset_from_id(asset_id)
final_dict = {asset}
if isinstance(asset, models.ClothingAsset):
print("clothing asset found")
image = roblox_service.fetch_clothing_image(asset)
image = await roblox_service.fetch_clothing_image(asset)
with open(f"tests/{asset.asset_id}.png", "wb") as output:
output.write(image)
return final_dict
@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")
image = await roblox_service.fetch_clothing_image(asset)
uploaded = await roblox_service.upload_clothing_image(
image,
asset.name,
asset.description,
asset.asset_type,
models.RbxCreator(int(TARGET), "Upload_Group", "Group"),
)
return uploaded
src/models.py
+2 -2
View File
@@ -67,7 +67,7 @@ class ClothingAsset(RbxAsset):
asset_type=asset_type,
)
def get_image(self) -> bytes:
async def get_image(self) -> bytes:
from utils import roblox_service
return roblox_service.fetch_clothing_image(self)
return await roblox_service.fetch_clothing_image(self)
src/utils/roblox_service.py
+92 -19
View File
@@ -2,7 +2,7 @@ import json
import os
import xml.etree.ElementTree
import requests
import httpx
from dotenv import load_dotenv
import models
@@ -21,35 +21,55 @@ FETCH_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
# Economy is an unauthed API; we should minimize cookie use where possible
ANONYMIZED_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
CSRF_HEADERS = {
"X-CSRF-TOKEN": "",
"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",
}
CSRF_COOKIES = {".ROBLOSECURITY": ROBLOSECURITY}
CSRF_URL = "https://apis.roblox.com/assets/user-auth/v1/assets"
ASSET_DELIVERY_BASE_URL = "https://assetdelivery.roblox.com/v1/asset/?id={asset_id}"
ECONOMY_BASE_URL = "https://economy.roblox.com/v2/assets/{asset_id}/details"
UPLOAD_URL = "https://apis.roblox.com/assets/user-auth/v1/assets"
# Client instance for connection pooling
_client = None
def _get_client() -> httpx.AsyncClient:
global _client
if _client is None:
_client = httpx.AsyncClient()
return _client
# Internal methods
def _economy_request(asset_id: int) -> requests.Response:
return requests.get(
async def _economy_request(asset_id: int) -> httpx.Response:
client = _get_client()
return await client.get(
ECONOMY_BASE_URL.format(asset_id=asset_id), headers=FETCH_HEADERS
)
def _asset_delivery_request(asset_id: int) -> requests.Response:
return requests.get(
ASSET_DELIVERY_BASE_URL.format(asset_id=asset_id), headers=FETCH_HEADERS
async def _asset_delivery_request(asset_id: int) -> httpx.Response:
client = _get_client()
return await client.get(
ASSET_DELIVERY_BASE_URL.format(asset_id=asset_id), headers=FETCH_HEADERS, follow_redirects=True
)
def _get_asset_xml(asset: models.RbxAsset) -> xml.etree.ElementTree.Element:
response = _asset_delivery_request(asset.asset_id)
content = response.content.decode("utf-8")
async def _get_asset_xml(asset: models.RbxAsset) -> xml.etree.ElementTree.Element:
response = await _asset_delivery_request(asset.asset_id)
response.raise_for_status()
content = response.content.decode("utf-8")
xml_root = xml.etree.ElementTree.fromstring(content)
return xml_root
@@ -65,11 +85,24 @@ def _get_shirt_template_id_from_xml(root: xml.etree.ElementTree.Element) -> int:
return int(template_id)
async def _get_csrf_token() -> str:
client = _get_client()
response = await client.post(CSRF_URL, cookies=CSRF_COOKIES, headers=CSRF_HEADERS)
csrf = response.headers.get("X-CSRF-TOKEN")
if not csrf:
raise httpx.HTTPStatusError(
"Failed to retrieve X-CSRF-TOKEN.",
request=response.request,
response=response,
)
return csrf
# External methods
def asset_from_id(id: int) -> models.RbxAsset:
response = _economy_request(id)
async def asset_from_id(id: int) -> models.RbxAsset:
response = await _economy_request(id)
response.raise_for_status()
asset_info = json.loads(response.content)
creator_info = asset_info["Creator"]
@@ -99,15 +132,55 @@ def asset_from_id(id: int) -> models.RbxAsset:
)
def fetch_clothing_image(asset: models.ClothingAsset) -> bytes:
async def fetch_clothing_image(asset: models.ClothingAsset) -> bytes:
try:
xml = _get_asset_xml(asset)
xml = await _get_asset_xml(asset)
template_id = _get_shirt_template_id_from_xml(xml)
image = _asset_delivery_request(template_id)
image = await _asset_delivery_request(template_id)
image.raise_for_status()
return image.content
except Exception:
raise # TODO add logging
# def upload_clothing_image(image: bytes, target: models.RbxCreator) -> models.RbxAsset:
async def upload_clothing_image(
image: bytes,
name: str,
description: str,
asset_type: models.RbxAssetType,
target: models.RbxCreator,
) -> dict:
csrf = await _get_csrf_token()
meta = {
"displayName": name,
"description": description,
"assetType": asset_type,
# TODO add support for user creation context
"creationContext": {
"creator": {"groupId": target.creator_id},
"expectedPrice": 10,
},
}
client = _get_client()
response = await client.post(
UPLOAD_URL,
files={
"request": (None, json.dumps(meta), "application/json"),
"fileContent": ("clothing_upload", image, "image/png"),
},
headers={
"X-CSRF-TOKEN": csrf,
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
"Accept": "*/*",
"Accept-Language": "en-US,en;q=0.5",
"Referer": "https://create.roblox.com/",
"Origin": "https://create.roblox.com",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
},
cookies=CSRF_COOKIES,
)
response.raise_for_status()
data = response.json()
return data