Bulk upload to Cloudflare images

cf_images_upload.py

from tqdm import tqdm
from glob import glob
from requests import post
from argparse import ArgumentParser


"""
requires
requests>=2.28.2
tqdm>=4.64.1

note
source path must end with trailing slash

example
python upload.py \
  --token=<api token> \
  --account_id=<account id> \
  --source=<path to images>/ \
  --types=JPG,PNG,png,jpg,jpeg,gif
"""


class Upload:
    def __init__(self, file_id, file_name, timestamp, variants):
        self.file_id = file_id
        self.file_name = file_name
        self.timestamp = timestamp
        self.variants = variants

    def __str__(self):
        return f"{self.file_name} ({self.file_id})"

    def to_tsv(self):
        columns = [
            self.file_name,
            self.file_id,
            self.timestamp,
            "\t".join(self.variants)
        ]

        return "\t".join(columns)


def make_endpoint(account_id):
    # see docs:
    # https://developers.cloudflare.com/images/cloudflare-images/
    return f"https://api.cloudflare.com/client/v4/accounts/{account_id}/images/v1"


def upload_file(endpoint, api_token, file_path):
    # see docs:
    # https://developers.cloudflare.com/images/cloudflare-images/api-request/
    with open(file_path, "rb") as f:
        # do the post request
        r = post(url=endpoint,
                 files={
                     "file": f
                 },
                 headers={
                     "User-Agent": "william >:)",
                     "Authorization": f"Bearer {api_token}"
                 })

        # check the response
        if not r.ok:
            raise Exception(r.text)
        else:
            json = r.json()
            return json["result"]


if __name__ == "__main__":
    # create arg parser
    parser = ArgumentParser(
        description="bulk upload images to cloudflare")

    # add the arguments
    parser.add_argument(
        "--token", help="cloudflare images api token")
    parser.add_argument(
        "--account_id", help="cloudflare account id")
    parser.add_argument(
        "--source", help="source directory")
    parser.add_argument(
        "--types", help="image file types, comma separated")

    # parse the arguments
    args = parser.parse_args()

    path = args.source
    token = args.token
    account = args.account_id
    file_types = args.types.split(",")

    # make the upload endpoint
    endpoint = make_endpoint(account)

    # find all images
    file_paths = [glob(f"{path}*.{file_type}")
                  for file_type in file_types]

    # merge list of lists
    file_paths = sum(file_paths, [])

    # put successful uploads here
    uploads = []

    print("---")
    print(f"uploading {len(file_paths)} images to {endpoint}")

    for file_path in tqdm(file_paths):
        print("---")

        try:
            # try to upload the file
            result = upload_file(endpoint, token, file_path)

            up = Upload(result["id"],
                        result["filename"],
                        result["uploaded"],
                        result["variants"])

            uploads.append(up)

            print(f"successfully uploaded {up}")

        except Exception as e:
            # the upload failed
            print(f"error uploading {file_path}: {e}")

    # log the successful uploads to a file
    with open("uploads.tsv", "a+") as f:
        for upload in uploads:
            f.write(f"{upload.to_tsv()}\n")

    print(f"wrote {len(uploads)} uploads to uploads.tsv")