feat: add publish_collectible method for Limited publishing step

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "rbx-upload"
-version = "0.2.3"
+version = "0.2.8"
 description = "Roblox asset upload client"
 requires-python = ">=3.13"
 dependencies = [

src/rbx_upload/client.py

@@ -253,6 +253,59 @@ class RobloxClient:
         await asyncio.gather(*[_upload_one(item) for item in items])
         return result
 
+    async def publish_collectible(
+        self,
+        asset_id: int,
+        group_id: int,
+        name: str,
+        description: str,
+        price: int = 5,
+    ) -> str:
+        """Publish an asset as a Limited collectible. Returns the collectibleItemId."""
+        csrf = await self._get_csrf_token()
+        response = await self._http.post(
+            "https://itemconfiguration.roblox.com/v1/collectibles",
+            json={
+                "isRentalOptIn": False,
+                "idempotencyToken": str(uuid.uuid4()),
+                "targetId": asset_id,
+                "targetType": 0,
+                "publishingType": 2,
+                "agreedPublishingFee": 10,
+                "creatorGroupId": group_id,
+                "publisherUserId": self._publisher_user_id,
+                "quantity": 0,
+                "quantityLimitPerUser": 0,
+                "resaleRestriction": 2,
+                "priceInRobux": price,
+                "priceOffset": 0,
+                "optOutFromRegionalPricing": False,
+                "isFree": False,
+                "saleLocationConfiguration": {"saleLocationType": 1, "places": []},
+                "name": name,
+                "description": description,
+            },
+            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=self._csrf_cookies,
+        )
+
+        if response.status_code == 429:
+            raise RateLimitError("Rate limit hit during collectible publish.")
+        if response.status_code in (401, 403):
+            raise AuthError("Not authorized to publish this collectible.")
+
+        response.raise_for_status()
+        data = response.json()
+        collectible_item_id = data.get("collectibleItemId")
+        if not collectible_item_id:
+            raise UploadError(f"publish_collectible did not return a collectibleItemId: {data}")
+        return collectible_item_id
+
     async def onsale_asset(
         self,
         collectible_item_id: str,
@@ -289,16 +342,19 @@ class RobloxClient:
         response.raise_for_status()
         return response.json() if response.text else {}
 
-    async def get_collectible_item_id(self, asset_id: int) -> str | None:
+    async def get_collectible_item_id(self, asset_id: int, max_attempts: int = 10, poll_interval: float = 3.0) -> str:
-        """Look up the collectible item ID (UUID) for a given asset ID."""
+        """Look up the collectible item ID (UUID) for a given asset ID, retrying until available."""
-        response = await self._http.post(
+        for _ in range(max_attempts):
-            self._proxy_url("https://catalog.roblox.com/v1/catalog/items/details"),
+            response = await self._http.get(
-            json={"items": [{"itemType": "Asset", "id": asset_id}]},
+                f"https://itemconfiguration.roblox.com/v1/collectibles/0/{asset_id}",
-            cookies=self._csrf_cookies,
+                cookies=self._csrf_cookies,
-        )
+            )
-        response.raise_for_status()
+            response.raise_for_status()
-        items = response.json().get("data", [])
+            collectible_item_id = response.json().get("collectibleItemId")
-        return items[0].get("collectibleItemId") if items else None
+            if collectible_item_id:
+                return collectible_item_id
+            await asyncio.sleep(poll_interval)
+        raise UploadError(f"collectibleItemId not available for asset {asset_id} after {max_attempts} attempts.")
 
     async def close(self):
         """Close the underlying HTTP client."""