+rbx_upload
+ + + + + + +1from .client import RobloxClient + 2from .models import ( + 3 AssetNotFoundError, + 4 AuthError, + 5 BatchResult, + 6 BatchUploadItem, + 7 ClothingAsset, + 8 RateLimitError, + 9 RbxAsset, +10 RbxAssetType, +11 RbxCreator, +12 RbxError, +13 UploadError, +14) +15 +16__all__ = [ +17 "RobloxClient", +18 "RbxError", +19 "AuthError", +20 "RateLimitError", +21 "UploadError", +22 "AssetNotFoundError", +23 "BatchUploadItem", +24 "BatchResult", +25 "RbxAsset", +26 "ClothingAsset", +27 "RbxCreator", +28 "RbxAssetType", +29] +
23class RobloxClient: + 24 def __init__( + 25 self, + 26 roblosecurity: str, + 27 publisher_user_id: int, + 28 proxy: str | None = None, + 29 ): + 30 self._roblosecurity = roblosecurity + 31 self._publisher_user_id = publisher_user_id + 32 self._proxy = proxy + 33 self._http = httpx.AsyncClient() + 34 + 35 self._fetch_headers = { + 36 "Cookie": f".ROBLOSECURITY={roblosecurity}", + 37 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + 38 } + 39 self._csrf_headers = { + 40 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", + 41 "Referer": "https://create.roblox.com/", + 42 "Origin": "https://create.roblox.com", + 43 } + 44 self._csrf_cookies = {".ROBLOSECURITY": roblosecurity} + 45 + 46 def _proxy_url(self, url: str) -> str: + 47 if not self._proxy: + 48 return url + 49 return url.replace("roblox.com", self._proxy) + 50 + 51 async def _get_csrf_token(self) -> str: + 52 url = self._proxy_url("https://apis.roblox.com/assets/user-auth/v1/assets") + 53 response = await self._http.post( + 54 url, cookies=self._csrf_cookies, headers=self._csrf_headers + 55 ) + 56 csrf = response.headers.get("X-CSRF-TOKEN") + 57 if not csrf: + 58 if response.status_code in (401, 403): + 59 raise AuthError("Invalid or expired ROBLOSECURITY token.") + 60 raise AuthError("Failed to retrieve X-CSRF-TOKEN.") + 61 return csrf + 62 + 63 async def _economy_request(self, asset_id: int) -> httpx.Response: + 64 url = self._proxy_url( + 65 f"https://economy.roblox.com/v2/assets/{asset_id}/details" + 66 ) + 67 return await self._http.get(url, headers=self._fetch_headers) + 68 + 69 async def _asset_delivery_request(self, asset_id: int) -> httpx.Response: + 70 url = self._proxy_url( + 71 f"https://assetdelivery.roblox.com/v1/asset/?id={asset_id}" + 72 ) + 73 return await self._http.get( + 74 url, headers=self._fetch_headers, follow_redirects=True + 75 ) + 76 + 77 async def _get_asset_xml(self, asset: RbxAsset) -> xml.etree.ElementTree.Element: + 78 response = await self._asset_delivery_request(asset.asset_id) + 79 response.raise_for_status() + 80 content = response.content.decode("utf-8") + 81 return xml.etree.ElementTree.fromstring(content) + 82 + 83 @staticmethod + 84 def _get_shirt_template_id_from_xml(root: xml.etree.ElementTree.Element) -> int: + 85 url_element = root.find(".//url") + 86 if url_element is None: + 87 raise ValueError("XML did not contain a <url> tag.") + 88 url = url_element.text + 89 if not url: + 90 raise ValueError("<url> tag did not contain any text.") + 91 return int(url.split("id=")[1]) + 92 + 93 async def asset_from_id(self, asset_id: int) -> RbxAsset: + 94 """Fetch asset information from Roblox by asset ID.""" + 95 response = await self._economy_request(asset_id) + 96 if response.status_code == 404: + 97 raise AssetNotFoundError(f"Asset {asset_id} not found.") + 98 if response.status_code in (401, 403): + 99 raise AuthError("Not authorized to fetch this asset.") +100 response.raise_for_status() +101 asset_info = response.json() +102 creator_info = asset_info["Creator"] +103 creator = RbxCreator( +104 creator_id=creator_info["Id"], +105 username=creator_info["Name"], +106 creator_type=creator_info["CreatorType"], +107 ) +108 asset_type_id = asset_info["AssetTypeId"] +109 if asset_type_id in (RbxAssetType.SHIRT, RbxAssetType.PANTS): +110 return ClothingAsset( +111 asset_id=asset_info["AssetId"], +112 creator=creator, +113 name=asset_info["Name"], +114 description=asset_info["Description"], +115 asset_type=asset_type_id, +116 ) +117 return RbxAsset( +118 asset_id=asset_info["AssetId"], +119 creator=creator, +120 name=asset_info["Name"], +121 description=asset_info["Description"], +122 asset_type=asset_type_id, +123 ) +124 +125 async def fetch_clothing_image(self, asset: ClothingAsset) -> bytes: +126 """Fetch the image data for a clothing asset.""" +127 xml_root = await self._get_asset_xml(asset) +128 template_id = self._get_shirt_template_id_from_xml(xml_root) +129 image = await self._asset_delivery_request(template_id) +130 image.raise_for_status() +131 return image.content +132 +133 async def upload_clothing_image( +134 self, +135 image: bytes, +136 name: str, +137 description: str, +138 asset_type: RbxAssetType, +139 group_id: int, +140 max_attempts: int = 10, +141 poll_interval: float = 1.0, +142 ) -> dict: +143 """Upload a clothing image to Roblox and return the operation result. +144 +145 Args: +146 image: Raw PNG bytes of the clothing image. +147 name: Display name for the asset. +148 description: Description for the asset. +149 asset_type: RbxAssetType.SHIRT or RbxAssetType.PANTS. +150 group_id: ID of the group to upload the asset to. +151 max_attempts: Number of times to poll the operation status. Defaults to 10. +152 poll_interval: Seconds to wait between polls. Defaults to 1.0. +153 """ +154 csrf = await self._get_csrf_token() +155 upload_url = self._proxy_url( +156 "https://apis.roblox.com/assets/user-auth/v1/assets" +157 ) +158 meta = { +159 "displayName": name, +160 "description": description, +161 "assetType": asset_type, +162 "creationContext": { +163 "creator": {"groupId": group_id}, +164 "expectedPrice": 10, +165 }, +166 } +167 upload_headers = { +168 "X-CSRF-TOKEN": csrf, +169 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", +170 "Accept": "*/*", +171 "Accept-Language": "en-US,en;q=0.5", +172 "Referer": "https://create.roblox.com/", +173 "Origin": "https://create.roblox.com", +174 "Sec-Fetch-Dest": "empty", +175 "Sec-Fetch-Mode": "cors", +176 "Sec-Fetch-Site": "same-site", +177 } +178 response = await self._http.post( +179 upload_url, +180 files={ +181 "request": (None, json.dumps(meta), "application/json"), +182 "fileContent": ("clothing_upload", image, "image/png"), +183 }, +184 headers=upload_headers, +185 cookies=self._csrf_cookies, +186 ) +187 +188 if response.status_code == 429: +189 raise RateLimitError("Rate limit hit during upload.") +190 if response.status_code in (401, 403): +191 raise AuthError("Not authorized to upload assets.") +192 +193 response.raise_for_status() +194 data = response.json() +195 +196 operation_id = data.get("operationId") +197 if operation_id: +198 for _ in range(max_attempts): +199 await asyncio.sleep(poll_interval) +200 op_response = await self._http.get( +201 self._proxy_url( +202 f"https://apis.roblox.com/assets/user-auth/v1/operations/{operation_id}" +203 ), +204 headers={"X-CSRF-TOKEN": csrf}, +205 cookies=self._csrf_cookies, +206 ) +207 op_response.raise_for_status() +208 op_data = op_response.json() +209 if op_data.get("done"): +210 if op_data.get("response", {}).get("assetId"): +211 return {"asset_id": op_data["response"]["assetId"]} +212 return op_data +213 raise UploadError( +214 f"Upload operation did not complete after {max_attempts} attempts." +215 ) +216 +217 return data +218 +219 async def batch_upload( +220 self, +221 items: list[BatchUploadItem], +222 max_attempts: int = 10, +223 poll_interval: float = 1.0, +224 ) -> BatchResult: +225 """Upload multiple clothing images with limited concurrency. +226 +227 Processes items 2 at a time. Continues on failure and reports all +228 failures in the returned BatchResult. +229 +230 Args: +231 items: List of BatchUploadItem to upload. +232 max_attempts: Passed to each upload_clothing_image call. +233 poll_interval: Passed to each upload_clothing_image call. +234 """ +235 result = BatchResult() +236 semaphore = asyncio.Semaphore(2) +237 +238 async def _upload_one(item: BatchUploadItem): +239 async with semaphore: +240 try: +241 upload_result = await self.upload_clothing_image( +242 image=item.image, +243 name=item.name, +244 description=item.description, +245 asset_type=item.asset_type, +246 group_id=item.group_id, +247 max_attempts=max_attempts, +248 poll_interval=poll_interval, +249 ) +250 result.succeeded.append((item, upload_result)) +251 except Exception as e: +252 result.failed.append((item, e)) +253 +254 await asyncio.gather(*[_upload_one(item) for item in items]) +255 return result +256 +257 async def onsale_asset( +258 self, +259 asset_id: int, +260 name: str, +261 description: str, +262 group_id: int, +263 price: int = 5, +264 ) -> dict: +265 """Put an asset on sale.""" +266 csrf = await self._get_csrf_token() +267 data = { +268 "saleLocationConfiguration": {"saleLocationType": 1, "places": []}, +269 "targetId": asset_id, +270 "priceInRobux": price, +271 "publishingType": 2, +272 "idempotencyToken": str(uuid.uuid4()), +273 "publisherUserId": self._publisher_user_id, +274 "creatorGroupId": group_id, +275 "name": name, +276 "description": description, +277 "isFree": False, +278 "agreedPublishingFee": 0, +279 "priceOffset": 0, +280 "quantity": 0, +281 "quantityLimitPerUser": 0, +282 "resaleRestriction": 2, +283 "targetType": 0, +284 } +285 response = await self._http.post( +286 self._proxy_url("https://itemconfiguration.roblox.com/v1/collectibles"), +287 json=data, +288 headers={ +289 "X-CSRF-TOKEN": csrf, +290 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", +291 "Referer": "https://create.roblox.com/", +292 "Origin": "https://create.roblox.com", +293 }, +294 cookies=self._csrf_cookies, +295 ) +296 +297 if response.status_code == 429: +298 raise RateLimitError("Rate limit hit during onsale.") +299 if response.status_code in (401, 403): +300 raise AuthError("Not authorized to put this asset on sale.") +301 +302 response.raise_for_status() +303 return response.json() +304 +305 async def close(self): +306 """Close the underlying HTTP client.""" +307 await self._http.aclose() +308 +309 async def __aenter__(self): +310 return self +311 +312 async def __aexit__(self, *args): +313 await self.close() +
24 def __init__( +25 self, +26 roblosecurity: str, +27 publisher_user_id: int, +28 proxy: str | None = None, +29 ): +30 self._roblosecurity = roblosecurity +31 self._publisher_user_id = publisher_user_id +32 self._proxy = proxy +33 self._http = httpx.AsyncClient() +34 +35 self._fetch_headers = { +36 "Cookie": f".ROBLOSECURITY={roblosecurity}", +37 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", +38 } +39 self._csrf_headers = { +40 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", +41 "Referer": "https://create.roblox.com/", +42 "Origin": "https://create.roblox.com", +43 } +44 self._csrf_cookies = {".ROBLOSECURITY": roblosecurity} +
93 async def asset_from_id(self, asset_id: int) -> RbxAsset: + 94 """Fetch asset information from Roblox by asset ID.""" + 95 response = await self._economy_request(asset_id) + 96 if response.status_code == 404: + 97 raise AssetNotFoundError(f"Asset {asset_id} not found.") + 98 if response.status_code in (401, 403): + 99 raise AuthError("Not authorized to fetch this asset.") +100 response.raise_for_status() +101 asset_info = response.json() +102 creator_info = asset_info["Creator"] +103 creator = RbxCreator( +104 creator_id=creator_info["Id"], +105 username=creator_info["Name"], +106 creator_type=creator_info["CreatorType"], +107 ) +108 asset_type_id = asset_info["AssetTypeId"] +109 if asset_type_id in (RbxAssetType.SHIRT, RbxAssetType.PANTS): +110 return ClothingAsset( +111 asset_id=asset_info["AssetId"], +112 creator=creator, +113 name=asset_info["Name"], +114 description=asset_info["Description"], +115 asset_type=asset_type_id, +116 ) +117 return RbxAsset( +118 asset_id=asset_info["AssetId"], +119 creator=creator, +120 name=asset_info["Name"], +121 description=asset_info["Description"], +122 asset_type=asset_type_id, +123 ) +
Fetch asset information from Roblox by asset ID.
+125 async def fetch_clothing_image(self, asset: ClothingAsset) -> bytes: +126 """Fetch the image data for a clothing asset.""" +127 xml_root = await self._get_asset_xml(asset) +128 template_id = self._get_shirt_template_id_from_xml(xml_root) +129 image = await self._asset_delivery_request(template_id) +130 image.raise_for_status() +131 return image.content +
Fetch the image data for a clothing asset.
+133 async def upload_clothing_image( +134 self, +135 image: bytes, +136 name: str, +137 description: str, +138 asset_type: RbxAssetType, +139 group_id: int, +140 max_attempts: int = 10, +141 poll_interval: float = 1.0, +142 ) -> dict: +143 """Upload a clothing image to Roblox and return the operation result. +144 +145 Args: +146 image: Raw PNG bytes of the clothing image. +147 name: Display name for the asset. +148 description: Description for the asset. +149 asset_type: RbxAssetType.SHIRT or RbxAssetType.PANTS. +150 group_id: ID of the group to upload the asset to. +151 max_attempts: Number of times to poll the operation status. Defaults to 10. +152 poll_interval: Seconds to wait between polls. Defaults to 1.0. +153 """ +154 csrf = await self._get_csrf_token() +155 upload_url = self._proxy_url( +156 "https://apis.roblox.com/assets/user-auth/v1/assets" +157 ) +158 meta = { +159 "displayName": name, +160 "description": description, +161 "assetType": asset_type, +162 "creationContext": { +163 "creator": {"groupId": group_id}, +164 "expectedPrice": 10, +165 }, +166 } +167 upload_headers = { +168 "X-CSRF-TOKEN": csrf, +169 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", +170 "Accept": "*/*", +171 "Accept-Language": "en-US,en;q=0.5", +172 "Referer": "https://create.roblox.com/", +173 "Origin": "https://create.roblox.com", +174 "Sec-Fetch-Dest": "empty", +175 "Sec-Fetch-Mode": "cors", +176 "Sec-Fetch-Site": "same-site", +177 } +178 response = await self._http.post( +179 upload_url, +180 files={ +181 "request": (None, json.dumps(meta), "application/json"), +182 "fileContent": ("clothing_upload", image, "image/png"), +183 }, +184 headers=upload_headers, +185 cookies=self._csrf_cookies, +186 ) +187 +188 if response.status_code == 429: +189 raise RateLimitError("Rate limit hit during upload.") +190 if response.status_code in (401, 403): +191 raise AuthError("Not authorized to upload assets.") +192 +193 response.raise_for_status() +194 data = response.json() +195 +196 operation_id = data.get("operationId") +197 if operation_id: +198 for _ in range(max_attempts): +199 await asyncio.sleep(poll_interval) +200 op_response = await self._http.get( +201 self._proxy_url( +202 f"https://apis.roblox.com/assets/user-auth/v1/operations/{operation_id}" +203 ), +204 headers={"X-CSRF-TOKEN": csrf}, +205 cookies=self._csrf_cookies, +206 ) +207 op_response.raise_for_status() +208 op_data = op_response.json() +209 if op_data.get("done"): +210 if op_data.get("response", {}).get("assetId"): +211 return {"asset_id": op_data["response"]["assetId"]} +212 return op_data +213 raise UploadError( +214 f"Upload operation did not complete after {max_attempts} attempts." +215 ) +216 +217 return data +
Upload a clothing image to Roblox and return the operation result.
+ +Args: + image: Raw PNG bytes of the clothing image. + name: Display name for the asset. + description: Description for the asset. + asset_type: RbxAssetType.SHIRT or RbxAssetType.PANTS. + group_id: ID of the group to upload the asset to. + max_attempts: Number of times to poll the operation status. Defaults to 10. + poll_interval: Seconds to wait between polls. Defaults to 1.0.
+219 async def batch_upload( +220 self, +221 items: list[BatchUploadItem], +222 max_attempts: int = 10, +223 poll_interval: float = 1.0, +224 ) -> BatchResult: +225 """Upload multiple clothing images with limited concurrency. +226 +227 Processes items 2 at a time. Continues on failure and reports all +228 failures in the returned BatchResult. +229 +230 Args: +231 items: List of BatchUploadItem to upload. +232 max_attempts: Passed to each upload_clothing_image call. +233 poll_interval: Passed to each upload_clothing_image call. +234 """ +235 result = BatchResult() +236 semaphore = asyncio.Semaphore(2) +237 +238 async def _upload_one(item: BatchUploadItem): +239 async with semaphore: +240 try: +241 upload_result = await self.upload_clothing_image( +242 image=item.image, +243 name=item.name, +244 description=item.description, +245 asset_type=item.asset_type, +246 group_id=item.group_id, +247 max_attempts=max_attempts, +248 poll_interval=poll_interval, +249 ) +250 result.succeeded.append((item, upload_result)) +251 except Exception as e: +252 result.failed.append((item, e)) +253 +254 await asyncio.gather(*[_upload_one(item) for item in items]) +255 return result +
Upload multiple clothing images with limited concurrency.
+ +Processes items 2 at a time. Continues on failure and reports all +failures in the returned BatchResult.
+ +Args: + items: List of BatchUploadItem to upload. + max_attempts: Passed to each upload_clothing_image call. + poll_interval: Passed to each upload_clothing_image call.
+257 async def onsale_asset( +258 self, +259 asset_id: int, +260 name: str, +261 description: str, +262 group_id: int, +263 price: int = 5, +264 ) -> dict: +265 """Put an asset on sale.""" +266 csrf = await self._get_csrf_token() +267 data = { +268 "saleLocationConfiguration": {"saleLocationType": 1, "places": []}, +269 "targetId": asset_id, +270 "priceInRobux": price, +271 "publishingType": 2, +272 "idempotencyToken": str(uuid.uuid4()), +273 "publisherUserId": self._publisher_user_id, +274 "creatorGroupId": group_id, +275 "name": name, +276 "description": description, +277 "isFree": False, +278 "agreedPublishingFee": 0, +279 "priceOffset": 0, +280 "quantity": 0, +281 "quantityLimitPerUser": 0, +282 "resaleRestriction": 2, +283 "targetType": 0, +284 } +285 response = await self._http.post( +286 self._proxy_url("https://itemconfiguration.roblox.com/v1/collectibles"), +287 json=data, +288 headers={ +289 "X-CSRF-TOKEN": csrf, +290 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0", +291 "Referer": "https://create.roblox.com/", +292 "Origin": "https://create.roblox.com", +293 }, +294 cookies=self._csrf_cookies, +295 ) +296 +297 if response.status_code == 429: +298 raise RateLimitError("Rate limit hit during onsale.") +299 if response.status_code in (401, 403): +300 raise AuthError("Not authorized to put this asset on sale.") +301 +302 response.raise_for_status() +303 return response.json() +
Put an asset on sale.
+Base exception for all rbx-upload errors.
+25class AuthError(RbxError): +26 """Raised when authentication fails or the ROBLOSECURITY token is invalid.""" +27 pass +
Raised when authentication fails or the ROBLOSECURITY token is invalid.
+30class RateLimitError(RbxError): +31 """Raised when hitting Roblox rate limits (HTTP 429).""" +32 pass +
Raised when hitting Roblox rate limits (HTTP 429).
+Raised when an asset upload fails.
+Raised when an asset cannot be found.
+86@dataclass +87class BatchUploadItem: +88 image: bytes +89 name: str +90 asset_type: RbxAssetType +91 group_id: int +92 description: str = "" +
95@dataclass + 96class BatchResult: + 97 succeeded: list[tuple[BatchUploadItem, dict]] = field(default_factory=list) + 98 failed: list[tuple[BatchUploadItem, Exception]] = field(default_factory=list) + 99 +100 @property +101 def all_succeeded(self) -> bool: +102 return len(self.failed) == 0 +
52class RbxAsset: +53 def __init__( +54 self, +55 asset_id: int, +56 creator: RbxCreator, +57 name: str, +58 description: str, +59 asset_type: RbxAssetType, +60 ) -> None: +61 self.asset_id = asset_id +62 self.name = name +63 self.description = description +64 self.creator = creator +65 self.asset_type = asset_type +
68class ClothingAsset(RbxAsset): +69 def __init__( +70 self, +71 asset_id: int, +72 creator: RbxCreator, +73 name: str, +74 description: str, +75 asset_type: ClothingAssetType, +76 ) -> None: +77 super().__init__( +78 asset_id=asset_id, +79 creator=creator, +80 name=name, +81 description=description, +82 asset_type=asset_type, +83 ) +
69 def __init__( +70 self, +71 asset_id: int, +72 creator: RbxCreator, +73 name: str, +74 description: str, +75 asset_type: ClothingAssetType, +76 ) -> None: +77 super().__init__( +78 asset_id=asset_id, +79 creator=creator, +80 name=name, +81 description=description, +82 asset_type=asset_type, +83 ) +
45class RbxCreator: +46 def __init__(self, creator_id: int, username: str, creator_type: CreatorType): +47 self.creator_id = creator_id +48 self.username = username +49 self.creator_type = creator_type +