builds.py

Last Updated on April 5, 2025 by chase

This utility enables listing and downloading of client and server builds from AWS S3.

Listing: Uses a predefined search prefix and bucket ID to identify and iterate through available prefixes for downloadable builds.

python builds.py -p linux -k ******************** -s **************************************** list -t server
collecting builds...
  available builds:
    => Headless_linux_415 : 2196.18 mb

python builds.py -p linux -k ******************** -s **************************************** list -t server

collecting builds...

available builds:

=> Headless_linux_415 : 2196.18 mb

Downloading: Requires specifying the exact prefix corresponding to the full object path in the bucket and utilizes the platform-specific bucket ID for retrieval.

python builds.py -p linux -k ******************** -s **************************************** download -t server -id Headless_linux_415
attempting to download build id: Headless_linux_415
Builds: Headless_linux_415
  => downloading to: ./builds/Headless_linux_415 (2196.18 mb)
 59%|████████████████████████████▏                   | 1.29G/2.20G [01:01<00:13, 68.7MiB/s,                           level8]

python builds.py -p linux -k ******************** -s **************************************** download -t server -id Headless_linux_415

attempting to download build id: Headless_linux_415

Builds: Headless_linux_415

=> downloading to: ./builds/Headless_linux_415 (2196.18 mb)

59%|████████████████████████████▏ | 1.29G/2.20G [01:01<00:13, 68.7MiB/s, level8]

A more generic approach would be to enable specifying bucket id and search prefix through arguments, to allow listing and downloading any available S3 contents.

I’ve also incorporated typings to enable clear understanding of function and parameter purpose.

from typing import Dict, List
import argparse
from argparse import ArgumentParser
import boto3
from mypy_boto3_s3 import S3Client
from mypy_boto3_s3.type_defs import ObjectTypeDef
from tqdm import tqdm
import os

class S3Build:
    def __init__(self, buildId:str, contents:ObjectTypeDef) -> None:
        self.build_id = buildId
        self.contents = contents
        self.total_bytes = 0
        for c in contents:
            self.total_bytes += c['Size']

    def SizeInMb(self) -> float:
        return round(self.total_bytes * 1e-6, 2)

def get_search_prefix(type:str, platform:str) -> str:
    if type == 'client':
        return 'Client_windows'
    # server
    return 'Headless_windows' if platform == 'windows' else 'Headless_linux'

def get_bucket_id(type:str) -> str:
    if type == 'client':
        return 'game-client-builds'
    # server
    return 'game-server-builds'

def collect_s3_builds_by_prefix(s3:S3Client, prefix:str, type:str) -> List[S3Build]:
    bucket_id = get_bucket_id(type)
    available_builds:List[S3Build] = []
    builds = s3.list_objects_v2(Bucket=bucket_id, Prefix=prefix)
    if builds:
        key_count = builds['KeyCount']
        if key_count > 0:
            contents = []
            contents.append(builds['Contents'])
            continue_token = builds['NextContinuationToken'] if 'NextContinuationToken' in builds else None
            while continue_token:
                build_contents = builds['Contents']
                last_key = build_contents[len(build_contents) - 1]['Key']
                # next iteration
                builds = s3.list_objects_v2(Bucket=bucket_id, Prefix=prefix, ContinuationToken=continue_token, StartAfter=last_key)
                contents.append(builds['Contents'])
                continue_token = builds['NextContinuationToken'] if 'NextContinuationToken' in builds else None

            server_builds:Dict[str, List[ObjectTypeDef]] = {}
            for c in contents:
                for obj in c:
                    build_id = obj['Key']
                    build_id = build_id[:build_id.find('/')]
                    if not build_id in server_builds:
                        server_builds[build_id] = []
                    server_builds[build_id].append(obj)
            for k in server_builds:
                available_builds.append(S3Build(k, server_builds[k]))

    return available_builds

def download_build(s3:S3Client, build:S3Build, dest:str, type:str) -> None:
    bucketId = get_bucket_id(type)
    t = tqdm(total=build.total_bytes, unit='iB', unit_scale=True, )
    for c in build.contents:
        content_key = c['Key']
        file_name = content_key[content_key.rfind('/') + 1:]
        dest_dir = content_key[:content_key.rfind('/')]
        file_path = f'{dest}/{dest_dir}/{file_name}'

        os.makedirs(f'{dest}/{dest_dir}', exist_ok=True)
        on_downloaded = lambda bytes : t.update(bytes)

        progress_str = file_name
        strlen = len(progress_str)
        if strlen < 32:
            pad = ''
            for _ in range(32 - strlen): pad += ' '
            progress_str = pad + progress_str
        else:
            progress_str = '...' + progress_str[strlen - 29:]

        t.set_postfix_str(progress_str)

        s3.download_file(Bucket=bucketId,
            Key=content_key,
            Filename=file_path,
            Callback=on_downloaded)
    t.close()

def create_s3_client(keyid:str, secret:str, region:str = 'us-west-1') -> S3Client:
    s3 = boto3.client('s3',
        aws_access_key_id=keyid,
        aws_secret_access_key=secret,
        region_name=region)
    return s3

def collect_args() -> ArgumentParser:
    args = argparse.ArgumentParser(description='Collect & display information for uploaded RAWMEN builds')
    args.add_argument('--platform', '-p', required=True, help='which platform to target', choices=['windows', 'linux'])
    args.add_argument('--keyid', '-k', required=True, help='aws access key id')
    args.add_argument('--secret', '-s', required=True, help='aws secret access key')
    args.add_argument('--region', '-r', help='aws region to connect to', default='us-west-1')

    functions = args.add_subparsers(description='Build retrieval choices')
    list_parser = functions.add_parser('list', help='list available builds')
    list_parser.add_argument('--type', '-t', required=True, help='which build type to collect', choices=['client', 'server'])
    list_parser.set_defaults(func=perform_list_builds)

    download_parser = functions.add_parser('download', help='download the desired build')
    download_parser.add_argument('--type', '-t', required=True, help='which build type to collect', choices=['client', 'server'])
    download_parser.add_argument('--buildid', '-id', required=True, help='the build id to download')
    download_parser.add_argument('--dest', '-d', help='destination relative directory path to download to', default='./builds')
    download_parser.set_defaults(func=perform_download_build)

    return args.parse_args()

def perform_list_builds(args:ArgumentParser) -> None:
    s3 = create_s3_client(args.keyid, args.secret, args.region)
    print('collecting builds...')
    search_prefix = get_search_prefix(args.type, args.platform)
    builds = collect_s3_builds_by_prefix(s3, search_prefix, args.type)
    print('  available builds:')
    for b in builds:
        print(f'    => {b.build_id} : {b.SizeInMb()} mb')

def perform_download_build(args:ArgumentParser) -> None:
    s3 = create_s3_client(args.keyid, args.secret, args.region)
    print(f'attempting to download build id: {args.buildid}')
    builds = collect_s3_builds_by_prefix(s3, args.buildid, args.type)
    if builds:
        print(f'Builds: {"".join([b.build_id for b in builds])}')
        b = builds[0]
        print(f'  => downloading to: {args.dest}/{b.build_id} ({b.SizeInMb()} mb)')
        download_build(s3, b, args.dest, args.type)
    else:
        print(f'  => no build found with id: {args.buildid}')
    pass

# how to compile changes to .exe:
# python 3.8.*
# python -m pip venv ./
# ./Scripts/activate.bat
# python -m pip install -r ./requirements.txt
# {python}/Scripts/pyinstaller.exe --onefile ./builds.py
# => outputs new builds.exe here: ./dist/
# ./Scripts/deactivate.bat
if __name__ == '__main__':
    args = collect_args()
    args.func(args)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

from typing import Dict, List

import argparse

from argparse import ArgumentParser

import boto3

from mypy_boto3_s3 import S3Client

from mypy_boto3_s3.type_defs import ObjectTypeDef

from tqdm import tqdm

import os

class S3Build:

def __init__(self, buildId:str, contents:ObjectTypeDef) -> None:

self.build_id = buildId

self.contents = contents

self.total_bytes = 0

for c in contents:

self.total_bytes += c['Size']

def SizeInMb(self) -> float:

return round(self.total_bytes * 1e-6, 2)

def get_search_prefix(type:str, platform:str) -> str:

if type == 'client':

return 'Client_windows'

# server

return 'Headless_windows' if platform == 'windows' else 'Headless_linux'

def get_bucket_id(type:str) -> str:

if type == 'client':

return 'game-client-builds'

# server

return 'game-server-builds'

def collect_s3_builds_by_prefix(s3:S3Client, prefix:str, type:str) -> List[S3Build]:

bucket_id = get_bucket_id(type)

available_builds:List[S3Build] = []

builds = s3.list_objects_v2(Bucket=bucket_id, Prefix=prefix)

if builds:

key_count = builds['KeyCount']

if key_count > 0:

contents = []

contents.append(builds['Contents'])

continue_token = builds['NextContinuationToken'] if 'NextContinuationToken' in builds else None

while continue_token:

build_contents = builds['Contents']

last_key = build_contents[len(build_contents) - 1]['Key']

# next iteration

builds = s3.list_objects_v2(Bucket=bucket_id, Prefix=prefix, ContinuationToken=continue_token, StartAfter=last_key)

contents.append(builds['Contents'])

continue_token = builds['NextContinuationToken'] if 'NextContinuationToken' in builds else None

server_builds:Dict[str, List[ObjectTypeDef]] = {}

for c in contents:

for obj in c:

build_id = obj['Key']

build_id = build_id[:build_id.find('/')]

if not build_id in server_builds:

server_builds[build_id] = []

server_builds[build_id].append(obj)

for k in server_builds:

available_builds.append(S3Build(k, server_builds[k]))

return available_builds

def download_build(s3:S3Client, build:S3Build, dest:str, type:str) -> None:

bucketId = get_bucket_id(type)

t = tqdm(total=build.total_bytes, unit='iB', unit_scale=True, )

for c in build.contents:

content_key = c['Key']

file_name = content_key[content_key.rfind('/') + 1:]

dest_dir = content_key[:content_key.rfind('/')]

file_path = f'{dest}/{dest_dir}/{file_name}'

os.makedirs(f'{dest}/{dest_dir}', exist_ok=True)

on_downloaded = lambda bytes : t.update(bytes)

progress_str = file_name

strlen = len(progress_str)

if strlen < 32:

pad = ''

for _ in range(32 - strlen): pad += ' '

progress_str = pad + progress_str

else:

progress_str = '...' + progress_str[strlen - 29:]

t.set_postfix_str(progress_str)

s3.download_file(Bucket=bucketId,

Key=content_key,

Filename=file_path,

Callback=on_downloaded)

t.close()

def create_s3_client(keyid:str, secret:str, region:str = 'us-west-1') -> S3Client:

s3 = boto3.client('s3',

aws_access_key_id=keyid,

aws_secret_access_key=secret,

region_name=region)

return s3

def collect_args() -> ArgumentParser:

args = argparse.ArgumentParser(description='Collect & display information for uploaded RAWMEN builds')

args.add_argument('--platform', '-p', required=True, help='which platform to target', choices=['windows', 'linux'])

args.add_argument('--keyid', '-k', required=True, help='aws access key id')

args.add_argument('--secret', '-s', required=True, help='aws secret access key')

args.add_argument('--region', '-r', help='aws region to connect to', default='us-west-1')

functions = args.add_subparsers(description='Build retrieval choices')

list_parser = functions.add_parser('list', help='list available builds')

list_parser.add_argument('--type', '-t', required=True, help='which build type to collect', choices=['client', 'server'])

list_parser.set_defaults(func=perform_list_builds)

download_parser = functions.add_parser('download', help='download the desired build')

download_parser.add_argument('--type', '-t', required=True, help='which build type to collect', choices=['client', 'server'])

download_parser.add_argument('--buildid', '-id', required=True, help='the build id to download')

download_parser.add_argument('--dest', '-d', help='destination relative directory path to download to', default='./builds')

download_parser.set_defaults(func=perform_download_build)

return args.parse_args()

def perform_list_builds(args:ArgumentParser) -> None:

s3 = create_s3_client(args.keyid, args.secret, args.region)

print('collecting builds...')

search_prefix = get_search_prefix(args.type, args.platform)

builds = collect_s3_builds_by_prefix(s3, search_prefix, args.type)

print(' available builds:')

for b in builds:

print(f' => {b.build_id} : {b.SizeInMb()} mb')

def perform_download_build(args:ArgumentParser) -> None:

s3 = create_s3_client(args.keyid, args.secret, args.region)

print(f'attempting to download build id: {args.buildid}')

builds = collect_s3_builds_by_prefix(s3, args.buildid, args.type)

if builds:

print(f'Builds: {"".join([b.build_id for b in builds])}')

b = builds[0]

print(f' => downloading to: {args.dest}/{b.build_id} ({b.SizeInMb()} mb)')

download_build(s3, b, args.dest, args.type)

else:

print(f' => no build found with id: {args.buildid}')

pass

# how to compile changes to .exe:

# python 3.8.*

# python -m pip venv ./

# ./Scripts/activate.bat

# python -m pip install -r ./requirements.txt

# {python}/Scripts/pyinstaller.exe --onefile ./builds.py

# => outputs new builds.exe here: ./dist/

# ./Scripts/deactivate.bat

if __name__ == '__main__':

args = collect_args()

args.func(args)

Chase Hutchens

Software Engineer

builds.py

Leave a Reply Cancel reply