rbx_upload

 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, force_direct: bool = False) -> str:
 47        if not self._proxy or force_direct:
 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 publish_collectible(
258        self,
259        asset_id: int,
260        group_id: int,
261        name: str,
262        description: str,
263        price: int = 5,
264    ) -> str:
265        """Publish an asset as a Limited collectible. Returns the collectibleItemId."""
266        csrf = await self._get_csrf_token()
267        response = await self._http.post(
268            "https://itemconfiguration.roblox.com/v1/collectibles",
269            json={
270                "isRentalOptIn": False,
271                "idempotencyToken": str(uuid.uuid4()),
272                "targetId": asset_id,
273                "targetType": 0,
274                "publishingType": 2,
275                "agreedPublishingFee": 10,
276                "creatorGroupId": group_id,
277                "publisherUserId": self._publisher_user_id,
278                "quantity": 0,
279                "quantityLimitPerUser": 0,
280                "resaleRestriction": 2,
281                "priceInRobux": price,
282                "priceOffset": 0,
283                "optOutFromRegionalPricing": False,
284                "isFree": False,
285                "saleLocationConfiguration": {"saleLocationType": 1, "places": []},
286                "name": name,
287                "description": description,
288            },
289            headers={
290                "X-CSRF-TOKEN": csrf,
291                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
292                "Referer": "https://create.roblox.com/",
293                "Origin": "https://create.roblox.com",
294            },
295            cookies=self._csrf_cookies,
296        )
297
298        if response.status_code == 429:
299            raise RateLimitError("Rate limit hit during collectible publish.")
300        if response.status_code in (401, 403):
301            raise AuthError(f"Not authorized to publish this collectible ({response.status_code}): {response.text}")
302
303        response.raise_for_status()
304        data = response.json()
305        collectible_item_id = data.get("collectibleItemId")
306        if not collectible_item_id:
307            # status=0 means already published — look up the ID
308            collectible_item_id = await self.get_collectible_item_id(asset_id)
309        return collectible_item_id
310
311    async def onsale_asset(
312        self,
313        collectible_item_id: str,
314        price: int = 5,
315    ) -> dict:
316        """Put an asset on sale."""
317        csrf = await self._get_csrf_token()
318        response = await self._http.patch(
319            f"https://itemconfiguration.roblox.com/v1/collectibles/{collectible_item_id}",
320            json={
321                "saleLocationConfiguration": {"saleLocationType": 1, "places": []},
322                "saleStatus": 0,
323                "quantityLimitPerUser": 0,
324                "resaleRestriction": 2,
325                "priceInRobux": price,
326                "priceOffset": 0,
327                "optOutFromRegionalPricing": False,
328                "isFree": False,
329            },
330            headers={
331                "X-CSRF-TOKEN": csrf,
332                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
333                "Referer": "https://create.roblox.com/",
334                "Origin": "https://create.roblox.com",
335            },
336            cookies=self._csrf_cookies,
337        )
338
339        if response.status_code == 429:
340            raise RateLimitError("Rate limit hit during onsale.")
341        if response.status_code in (401, 403):
342            raise AuthError("Not authorized to put this asset on sale.")
343
344        response.raise_for_status()
345        return response.json() if response.text else {}
346
347    async def get_collectible_item_id(self, asset_id: int, max_attempts: int = 10, poll_interval: float = 3.0) -> str:
348        """Look up the collectible item ID (UUID) for a given asset ID, retrying until available."""
349        for _ in range(max_attempts):
350            response = await self._http.get(
351                f"https://itemconfiguration.roblox.com/v1/collectibles/0/{asset_id}",
352                cookies=self._csrf_cookies,
353            )
354            response.raise_for_status()
355            collectible_item_id = response.json().get("collectibleItemId")
356            if collectible_item_id:
357                return collectible_item_id
358            await asyncio.sleep(poll_interval)
359        raise UploadError(f"collectibleItemId not available for asset {asset_id} after {max_attempts} attempts.")
360
361    async def close(self):
362        """Close the underlying HTTP client."""
363        await self._http.aclose()
364
365    async def __aenter__(self):
366        return self
367
368    async def __aexit__(self, *args):
369        await self.close()

20class RbxError(Exception):
21    """Base exception for all rbx-upload errors."""
22    pass

25class AuthError(RbxError):
26    """Raised when authentication fails or the ROBLOSECURITY token is invalid."""
27    pass

30class RateLimitError(RbxError):
31    """Raised when hitting Roblox rate limits (HTTP 429)."""
32    pass

35class UploadError(RbxError):
36    """Raised when an asset upload fails."""
37    pass

40class AssetNotFoundError(RbxError):
41    """Raised when an asset cannot be found."""
42    pass

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        )

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

 7class RbxAssetType(IntEnum):
 8    IMAGE = 1
 9    SHIRT = 11
10    PANTS = 12