upload_planets_json.py

#!/usr/bin/env python3

# Usage python3 upload_planets_json.py -d <display_name> -w <world_id> <path_to_planet_json_file>
#
# The Display Name OR the world ID should be provided. One or the other are required.
# If both are provided, World ID will take priority
#
# Example:
#     python3 upload_planets_json.py -w 23 planet.json
#     python3 upload_planets_json.py -d Dand planet.json
#
# IMPORTANT: The <display_name> of the planet MUST match the planet name exactly.
# Otherwise the block colors will not be linked propery to the actually planet when
# it is pulled from the API

import argparse
import json
from urllib import request
import os
from urllib.error import HTTPError

BASE_URL = "https://boundlexx.app"
TEST_BASE_URL = "https://testing.boundlexx.app"


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("json_file", type=argparse.FileType("r"))
    parser.add_argument("-d", "--display_name", type=str)
    parser.add_argument("-w", "--world_id", type=int)
    parser.add_argument("-t", "--is_test", action="store_true")
    parser.add_argument("-c", "--auth_cache_path", default=".auth_token")

    args = parser.parse_args()

    if args.display_name is None and args.world_id is None:
        print("Display name (-d) or World ID (-w) are required")
        exit(1)

    if args.is_test:
        base_url = TEST_BASE_URL
    else:
        base_url = BASE_URL

    world_config_json = args.json_file.read()
    world_config = json.loads(world_config_json)

    if args.world_id is not None:
        world_config["world_id"] = args.world_id
    else:
        world_config["display_name"] = args.display_name.strip()
    world_config_json = json.dumps(world_config)

    auth_valid = False
    auth_token_from_file = False
    while not auth_valid:
        auth_token = None
        if os.path.exists(args.auth_cache_path):
            print("Found existing auth token!")

            auth_token_from_file = True
            with open(args.auth_cache_path, "r") as f:
                auth_token = f.read().strip()

        if auth_token is None:
            auth_token = input("Enter auth token: ")  # nosec
            auth_token = auth_token.strip()

        auth_valid = True
        try:
            r = request.Request(
                url=f"{base_url}/api/ingest-ws-data/",
                data=world_config_json.encode("utf8"),
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Token {auth_token}",
                },
            )
            response = request.urlopen(r)  # nosec
        except HTTPError as ex:
            if ex.code == 401:
                auth_valid = False
                if auth_token_from_file:
                    print("Invalid cached auth token, removing...")
                    auth_token_from_file = False
                    os.remove(args.auth_cache_path)
            elif ex.code == 425:
                print(
                    "Unable to process planet JSON via Display Name now\n"
                    "The world must be created first. Try again later"
                )
            else:
                print(f"Unknown error: {str(ex)}")
                print(world_config_json)
        else:
            print(response.read().decode("utf8"))

    if not auth_token_from_file:
        with open(args.auth_cache_path, "w") as f:
            f.write(auth_token)


if __name__ == "__main__":
    main()