Roblox: Save inventory-owned assets to the Creator Store "Saved" list.

rbx_save_inv_assets.py

"""
Save inventory-owned assets to the Creator Store "Saved" list. (19/08/2025)
"""
import os
import time
import math
from dotenv import load_dotenv  # pip install python-dotenv
import requests  # pip install requests


# Path to .env file with `RBX_TOKEN`:
# https://github.com/exurd/DBR/blob/main/.example_env
ENV_FILEPATH = "/PATH/TO/.env"

# Save your own assets to the list?
SAVE_OWN_ASSETS = False

# Which asset types to save?
ASSET_TYPES = {
    3:  "Audio",
    10: "Model",
    13: "Decal",
    38: "Plugin",
    40: "MeshPart",
    62: "Video"
}


def validate_CSRF() -> str:
    """
    Gets X-CSRF-Token for Roblox.
    """
    requestSession.headers["X-CSRF-Token"] = None
    req = requestSession.post(url="https://auth.roblox.com/v2/logout", timeout=60)
    return req.headers["X-CSRF-Token"]


if not os.path.isfile(ENV_FILEPATH):
    raise Exception(f".env file \"{ENV_FILEPATH}\" was not found")

load_dotenv(ENV_FILEPATH)

requestSession = requests.Session()
adapter = requests.adapters.HTTPAdapter(max_retries=5)
requestSession.mount('https://', adapter)
requestSession.mount('http://', adapter)
requestSession.cookies[".ROBLOSECURITY"] = str(os.getenv("RBX_TOKEN"))
requestSession.headers['Referer'] = "https://www.roblox.com"
requestSession.headers["X-CSRF-Token"] = validate_CSRF()
print("Loaded .env file!")


def get_user_from_token() -> dict:
    """
    Gets user ID from .ROBLOSECURITY token.
    Results are not cached.
    """
    usercheck = requestSession.get("https://users.roblox.com/v1/users/authenticated")
    if usercheck is not None:
        if usercheck.ok:
            return usercheck.json()
    return None


MAIN_USER_ID = get_user_from_token()
if MAIN_USER_ID is None:
    raise Exception("User ID could not be requested.")
MAIN_USER_ID = MAIN_USER_ID["id"]


SAVEAPI_HEADERS = {
    "accept": "*/*",
    "cache-control": "no-cache",
    "content-type": "application/json-patch+json",
    "dnt": "1",
    "origin": "https://create.roblox.com",
    "pragma": "no-cache",
    "priority": "u=1, i",
    "referer": "https://create.roblox.com/",
    "sec-fetch-dest": "empty",
    "sec-fetch-mode": "cors",
    "sec-fetch-site": "same-site"
}


def request_save_asset(asset_id:int, asset_type:str):
    """
    Saves asset to saved list.

    HTTP codes I encountered:
    - 403 Forbidden
        - Fixed by updating the CSRF token
    - 415 Unsupported Media Type
        - Added headers for "content-type"
    - 400 Bad Request
        - Used Bruno to get as accurate to the original request (minus cookies and other bulk)
            - Didn't work...
        - `{"errors":{"":["Error parsing boolean value. Path '', line 1, position 1."],"request":["The request field is required."]},"type":"https://tools.ietf.org/html/rfc9110#section-15.5.1","title":"One or more validation errors occurred.","status":400,"traceId":"00-b1fc714eefb8ac3f2b79b892a37db2db-2620be73f64ea483-00"}`
            - hmm...
        - Fixed! requestSession.post: Switched from `data=payload` to `json=payload`
    - 409 Conflict
        - "Save already exists"
    """
    url = f"https://apis.roblox.com/toolbox-service/v1/saves"
    payload = {
        "targetType": asset_type,
        "targetId": str(asset_id)
    }

    for _ in range(5):
        response = requestSession.post(url, json=payload, headers=SAVEAPI_HEADERS)
        print(f"Response code is {response.status_code}!")
        if response.status_code == 201:  # Created
            print("Saved!")
            return True
        if response.status_code == 403:  # Conflict
            print("Already saved!")
            return True
        if response.status_code == 403:  # Forbidden
            print("Refreshing CSRF token...")
            requestSession.headers["X-CSRF-Token"] = validate_CSRF()
        print("Sleeping 5 secs...")
        time.sleep(5)
    print(f"Could not save asset {asset_id} as {asset_type}")
    return False


def check_if_saved(asset_id:int, asset_type:str):
    """
    Response JSON:
    {
        "totalCount": 0,
        "saves": []
    }
    """
    url = f"https://apis.roblox.com/toolbox-service/v1/saves?targetType={asset_type}&targetId={str(asset_id)}"
    response = requestSession.get(url, headers=SAVEAPI_HEADERS)
    if response.ok:
        response = response.json()
        if response["totalCount"] != 0:
            return True
    return False


def get_economy_info(asset_id:int):
    """
    Gets the info of an asset.

    Response JSON of Economy details API:
    {
        "TargetId": 136176178143668,
        "ProductType": "User Product",
        "AssetId": 136176178143668,
        "ProductId": 3330191709,
        "Name": "CrazyTracer-OPEN",
        "Description": "Open source raytracing project!",
        "AssetTypeId": 10,
        "Creator": {
            "Id": 247535,
            "Name": "Crazyblox",
            "CreatorType": "User",
            "CreatorTargetId": 247535,
            "HasVerifiedBadge": true
        },
        "IconImageAssetId": 0,
        "Created": "2025-07-10T17:40:45.157Z",
        "Updated": "2025-07-10T17:40:45.16Z",
        "PriceInRobux": null,
        "PriceInTickets": null,
        "Sales": 0,
        "IsNew": false,
        "IsForSale": false,
        "IsPublicDomain": true,
        "IsLimited": false,
        "IsLimitedUnique": false,
        "Remaining": null,
        "MinimumMembershipLevel": 0,
        "ContentRatingTypeId": 0,
        "SaleAvailabilityLocations": null,
        "SaleLocation": null,
        "CollectibleItemId": null,
        "CollectibleProductId": null,
        "CollectiblesItemDetails": null
    }
    """
    url = f"https://economy.roblox.com/v2/assets/{str(asset_id)}/details"
    response = requestSession.get(url)
    if response.ok:
        response = response.json()
        return response
    print(f"Could not get asset info! Status code: {response.status_code}")
    return False


def save_asset(asset_id:int, asset_type=""):
    """Saves assets."""
    if asset_id and asset_type != "":
        return request_save_asset(asset_id, asset_type)
    if asset_details := get_economy_info(asset_id):
        if not SAVE_OWN_ASSETS and int(asset_details["Creator"]["Id"]) == MAIN_USER_ID:
            print("Owner is account in session; skipping...")
            return False

        at_id = asset_details["AssetTypeId"]
        asset_type = ASSET_TYPES[at_id] if at_id in ASSET_TYPES else False
        if asset_type:
            if check_if_saved(asset_id, asset_type):
                print("Already saved!")
                return False
            return request_save_asset(asset_id, asset_type)
    return False


def loop_through_inv(asset_type_id, user_id=MAIN_USER_ID):
    page_cursor = ""
    while True:
        url = f"https://inventory.roblox.com/v2/users/{str(user_id)}/inventory/{str(asset_type_id)}?cursor=&limit=100&sortOrder=Desc{page_cursor}"
        response = requestSession.get(url)
        if response.ok:
            response = response.json()
            for asset in response["data"]:
                asset_id = asset["assetId"]
                print(f"Attempting to save asset {asset_id} to profile...")
                save_asset(asset_id)
            if response["nextPageCursor"] is None:
                print("No more pages to search!")
                return True
            page_cursor = f"&cursor={response["nextPageCursor"]}"


if __name__ == "__main__":
    for at_id in ASSET_TYPES:
        print("=" * 30)
        print(f"{' ' * math.floor((30 - len(ASSET_TYPES[at_id])) / 2)}{ASSET_TYPES[at_id]}")
        print("=" * 30)
        loop_through_inv(at_id)