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) -> 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()

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