Bulk upload emotes to matrix room (absurd amount)

upload_emotes.py

#
# This scripts lets you bulk upload an absurd amount of emotes at once
# fluffychat (and nheko too) let you upload emotes in bulk but that is only
# reliable up to about 100 emotes (the json of the emote pack containing all
# the emote names and mxc uris has to be <65 kB, else it fails with
# "event too large")
#
# this script solves that:
#  - first uploads all emotes one by one, saving their mxc uris
#  - then publishes several emote packs (according to --batchsize)
#
# example usage:
# y upload_emotes.py \!xxxx:matrix.desktop31.eu my-emotes *.png --homeserver=matrix.vitapavlik.cz
#
# entirely vibe-coded with gemini 3.1 Pro, works reasonably well!
# date: 2026-04-17
# thorougly tested by uploading 1104 png emotes at once
# Python 3.13.5 requests 2.33.1
#
import argparse
import getpass
import mimetypes
import os
import sys
import time
import math
import requests

def main():
    parser = argparse.ArgumentParser(description="Upload an image pack (emotes/stickers) to a Matrix room.")
    parser.add_argument("room_id", help="The Matrix room ID (e.g. !abcdef:example.com)")
    parser.add_argument("pack_name", help="The internal ID for the pack (e.g. 'my_custom_emotes')")
    parser.add_argument("images", nargs='+', help="List of image files to upload")
    parser.add_argument("--homeserver", default="https://matrix-client.matrix.org",
                        help="Your homeserver Client API URL")
    parser.add_argument("--display-name", help="Display name for the pack (defaults to the pack_name)")
    parser.add_argument("--batch-size", type=int, default=50,
                        help="Number of emotes per state event to avoid size limits (default: 50)")

    args = parser.parse_args()

    # Add scheme if missing
    if not args.homeserver.startswith(("http://", "https://")):
        args.homeserver = f"https://{args.homeserver}"

    # Filter out invalid files first so our counts and alignments are accurate
    valid_images = [img for img in args.images if os.path.isfile(img)]
    if not valid_images:
        print("❌ No valid image files found.")
        sys.exit(1)

    total_images = len(valid_images)
    pad_count = len(str(total_images))

    # Calculate max lengths for perfectly aligned output (capped at 40 chars for sanity)
    max_filename_len = min(max((len(os.path.basename(p)) for p in valid_images), default=0), 40)
    max_shortcode_len = min(max((len(os.path.splitext(os.path.basename(p))[0]) for p in valid_images), default=0), 40)

    # 1. Prompt for credentials
    print("\n--- Matrix Authentication ---")
    username = input("Matrix Username (e.g. @user:example.com): ")
    password = getpass.getpass("Matrix Password: ")

    # 2. Login
    print("\nLogging in...")
    login_url = f"{args.homeserver}/_matrix/client/v3/login"
    login_payload = {
        "type": "m.login.password",
        "identifier": {"type": "m.id.user", "user": username},
        "password": password
    }

    login_req = requests.post(login_url, json=login_payload)
    if not login_req.ok:
        print(f"❌ Login failed: {login_req.json().get('error', login_req.text)}")
        sys.exit(1)

    access_token = login_req.json()["access_token"]
    headers = {"Authorization": f"Bearer {access_token}"}
    print("✅ Login successful.\n")

    # 3. Upload images one by one
    print("--- Uploading Images ---")
    images_dict = {}

    for idx, img_path in enumerate(valid_images, start=1):
        filename = os.path.basename(img_path)
        shortcode, _ = os.path.splitext(filename)

        # Truncate strings in output if they are absurdly long, just for display
        disp_filename = filename if len(filename) <= 40 else filename[:37] + "..."
        disp_shortcode = shortcode if len(shortcode) <= 40 else shortcode[:37] + "..."

        mime_type, _ = mimetypes.guess_type(img_path)
        if not mime_type:
            mime_type = "application/octet-stream"

        # Aligned progress output
        print(f"[{idx:>{pad_count}}/{total_images}] {disp_filename:<{max_filename_len}} -> :{disp_shortcode}:{' ' * (max_shortcode_len - len(disp_shortcode))} ... ", end="", flush=True)
        start_time = time.time()

        with open(img_path, 'rb') as f:
            data = f.read()

        upload_url = f"{args.homeserver}/_matrix/media/v3/upload?filename={filename}"
        upload_headers = headers.copy()
        upload_headers["Content-Type"] = mime_type

        upload_req = requests.post(upload_url, headers=upload_headers, data=data)
        elapsed_time = time.time() - start_time

        if upload_req.ok:
            mxc_url = upload_req.json()["content_uri"]
            print(f"✅ Success ({elapsed_time:.2f}s)")
            images_dict[shortcode] = {
                "url": mxc_url,
                "body": shortcode
            }
        else:
            print(f"❌ Failed! ({elapsed_time:.2f}s) - {upload_req.text}")

    if not images_dict:
        print("\n❌ No images were uploaded successfully. Exiting.")
        sys.exit(1)

    for item in images_dict:
        print(f"{item} {images_dict[item]['url']}")

    # 4. Update the room state in batches
    print("\n--- Publishing Emote Packs ---")
    base_display_name = args.display_name if args.display_name else args.pack_name

    items = list(images_dict.items())
    num_batches = math.ceil(len(items) / args.batch_size)

    for i in range(num_batches):

        # prevent TooManyRequests
        time.sleep(4.0)

        batch_items = items[i * args.batch_size : (i + 1) * args.batch_size]
        batch_dict = dict(batch_items)

        # If there's only one batch, keep the original name. Otherwise, append an index.
        current_pack_id = f"{args.pack_name}_{i+1}" if num_batches > 1 else args.pack_name
        current_display_name = f"{base_display_name} (Part {i+1:02d} of {num_batches})" if num_batches > 1 else base_display_name

        state_payload = {
            "images": batch_dict,
            "pack": {
                "display_name": current_display_name,
                "avatar_url": list(batch_dict.values())[0]["url"]
            }
        }

        state_url = f"{args.homeserver}/_matrix/client/v3/rooms/{args.room_id}/state/im.ponies.room_emotes/{current_pack_id}"
        state_req = requests.put(state_url, headers=headers, json=state_payload)

        if state_req.ok:
            print(f"✅ Published: {current_display_name} ({len(batch_dict)} emotes)")
        else:
            print(f"❌ Failed to publish {current_display_name}: {state_req.json().get('error', state_req.text)}")

        sys.stdout.flush()

    print("\n🎉 All done! Enjoy your new emotes.")

if __name__ == "__main__":
    main()