Python script to upload 3D Tiles.

upload_3dtiles.py

import os
import argparse
from minio import Minio
from minio.error import S3Error
from dotenv import load_dotenv
import urllib3
from tqdm import tqdm

# Disable SSL verification warnings (internal use only)
urllib3.disable_warnings()
load_dotenv()

# --------------------------------------
# ✅ ONLY MODIFY THESE TWO
LOCAL_DIR = "/Users/ashwin.nair/Desktop/Projects/minio-nginx-3d-serve/AUH_Untextured_Filtered"
REMOTE_DIR = "3dtiles/AUH_Untextured_Filtered"
# --------------------------------------

# Connect to MinIO
minio_client = Minio(
    endpoint=os.environ["MINIO_ENDPOINT"],
    access_key=os.environ["MINIO_ACCESS_KEY"],
    secret_key=os.environ["MINIO_SECRET_KEY"],
    secure=False,
    http_client=urllib3.PoolManager(cert_reqs="CERT_NONE"),
)


def folder_exists_in_minio(bucket, remote_prefix):
    """Check if any object exists under this folder prefix"""
    objects = minio_client.list_objects(bucket, prefix=remote_prefix, recursive=True)
    return any(True for _ in objects)


def upload_folder(
    bucket, local_data_root, remote_data_root, folder_name, dry_run=False, verbose=False
):
    """Upload all files in a folder to MinIO bucket"""
    local_folder = os.path.join(local_data_root, folder_name)

    # Count files first to show progress
    file_list = []
    for root, _, files in os.walk(local_folder):
        for file in files:
            full_path = os.path.join(root, file)
            rel_path = os.path.relpath(full_path, local_data_root)
            dst_path = os.path.join(remote_data_root, rel_path).replace("\\", "/")
            file_list.append((full_path, rel_path, dst_path))

    with tqdm(
        total=len(file_list),
        desc=f"{'DRY RUN' if dry_run else 'Uploading'} {folder_name}",
        unit="file",
    ) as pbar:
        for full_path, rel_path, dst_path in file_list:
            if dry_run:
                if verbose:
                    print(f"[DRY RUN] Would upload {rel_path} to {dst_path}")
            else:
                try:
                    minio_client.fput_object(bucket, dst_path, full_path)
                    if verbose:
                        print(f"Uploaded {rel_path} to {dst_path}")
                except S3Error as e:
                    print(f"Error uploading to {dst_path}: {e}")
            pbar.update(1)


def main(bucket, dry_run=False, verbose=False):
    if not minio_client.bucket_exists(bucket):
        print(f"❌ Error: Bucket '{bucket}' does not exist.")
        return

    local_data_path = os.path.join(LOCAL_DIR, "data")
    remote_data_path = os.path.join(REMOTE_DIR, "data").replace("\\", "/")

    if not os.path.isdir(local_data_path):
        print(f"❌ Error: '{local_data_path}' does not exist or is not a directory.")
        return

    for folder_name in os.listdir(local_data_path):
        local_folder_path = os.path.join(local_data_path, folder_name)
        if not os.path.isdir(local_folder_path):
            continue

        remote_prefix = f"{remote_data_path}/{folder_name}/"
        if folder_exists_in_minio(bucket, remote_prefix):
            if verbose:
                print(f"⏭️ Skipping existing folder: {folder_name}")
            continue

        if verbose:
            print(
                f"{'[DRY RUN] Would upload' if dry_run else 'Uploading'} folder: {folder_name}"
            )
        upload_folder(
            bucket,
            local_data_path,
            remote_data_path,
            folder_name,
            dry_run=dry_run,
            verbose=verbose,
        )
        #break


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Upload 3D tiles to MinIO with dry run and verbosity support"
    )
    parser.add_argument(
        "--bucket", default="digitaltwin-public", help="MinIO bucket name"
    )
    parser.add_argument(
        "--dry-run", action="store_true", help="Simulate upload without making changes"
    )
    parser.add_argument(
        "--verbose", action="store_true", help="Enable detailed logging"
    )
    args = parser.parse_args()

    main(bucket=args.bucket, dry_run=args.dry_run, verbose=args.verbose)