soracom_harvest_downloader.py

import argparse
import json
import re
import requests
import logging

logging.basicConfig(level=logging.DEBUG)

API = "https://api.soracom.io/v1"

def post_json(url, body_dict):
    return requests.post(
        url, headers={"content-type": "application/json"}, data=json.dumps(body_dict)
    )

def get_with_auth(url, token):
    return requests.get(
        url,
        headers={
            "x-soracom-api-key": token["apiKey"],
            "x-soracom-token": token["token"],
        },
    )

def auth(auth_key_id, auth_key):
    params = {"authKeyId": auth_key_id, "authKey": auth_key}
    return post_json(API + "/auth", params).json()

def list_file(path, search, num_to_list, list_offset, token):
    url = API + "/files/private" + path + "?limit=" + str(num_to_list)
    result = get_with_auth(url, token)
    if result.status_code == 404:
        return []

    data = result.json()
    if len(data) == 0:
        return []

    files = []
    if search:
        for entry in data:
            if re.search(search, entry["filename"]):
                files.append(entry["filename"])
    else:
        files = map(lambda entry: entry["filename"], data)
    return files

def get_and_save_images(file_names, base_path, save_path, token):
    for file_name in file_names:
        response = get_with_auth(API + "/files/private" + base_path + file_name, token)
        with open(save_path + file_name, "wb") as handle:
            print("Writing {}".format(file_name))
            handle.write(response.content)

def main(args):
    token = auth(args.auth_key_id, args.auth_key)
    file_names = list_file(
        args.base_path, args.search, args.num_to_list, args.list_offset, token
    )
    get_and_save_images(file_names, args.base_path, args.save_path, token)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--base_path", type=str, required=True)
    parser.add_argument("--save_path", type=str, default="./")
    parser.add_argument("--search", type=str, default=None)
    parser.add_argument("--num_to_list", type=int, default=300)
    parser.add_argument("--list_offset", type=int, default=0)
    parser.add_argument("--auth_key_id", type=str, required=True)
    parser.add_argument("--auth_key", type=str, required=True)
    args = parser.parse_args()

    main(args)