chris@160: #!/usr/bin/python
chris@160:
chris@160: # Copyright 2015 Section6. All Rights Reserved.
chris@160:
chris@160: import argparse
chris@160: import getpass
chris@160: import json
chris@160: import os
chris@160: import requests
chris@160: import shutil
chris@160: import stat
chris@160: import tempfile
chris@160: import time
chris@160: import urlparse
chris@160: import zipfile
chris@160: import sys
chris@160:
chris@160: class Colours:
chris@160: purple = "\033[95m"
chris@160: cyan = "\033[96m"
chris@160: dark_cyan = "\033[36m"
chris@160: blue = "\033[94m"
chris@160: green = "\033[92m"
chris@160: yellow = "\033[93m"
chris@160: red = "\033[91m"
chris@160: bold = "\033[1m"
chris@160: underline = "\033[4m"
chris@160: end = "\033[0m"
chris@160:
chris@160: def __zip_dir(in_dir, zip_path, file_filter=None):
chris@160: zf = zipfile.ZipFile(zip_path, mode="w", compression=zipfile.ZIP_DEFLATED)
chris@160: for subdir, dirs, files in os.walk(in_dir):
chris@160: for file in files:
chris@160: if (file_filter is None) or (len(file_filter) > 0 and file.lower().split(".")[-1] in file_filter):
chris@160: zf.write(
chris@160: filename=os.path.join(subdir,file),
chris@160: arcname=os.path.relpath(os.path.join(subdir,file), start=in_dir))
chris@160: return zip_path
chris@160:
chris@160: def __unzip(zip_path, target_dir):
chris@160: """Unzip a file to a given directory. All destination files are overwritten.
chris@160: """
chris@160: zipfile.ZipFile(zip_path).extractall(target_dir)
chris@160:
chris@160: def main():
chris@160: parser = argparse.ArgumentParser(
chris@160: description="Compiles a Pure Data file.")
chris@160: parser.add_argument(
chris@160: "input_dir",
chris@160: help="A directory containing _main.pd. The entire directory will be uploaded.")
chris@160: parser.add_argument(
chris@160: "-n", "--name",
chris@160: default="heavy",
chris@160: help="Patch name. If it doesn't exist, the uploader will fail. Make sure that it exists on the Heavy website.")
chris@160: parser.add_argument(
chris@160: "-g", "--gen",
chris@160: nargs="+",
chris@160: default=["c"],
chris@160: help="List of generator outputs. Currently supported generators are 'c' and 'js'.")
chris@160: parser.add_argument(
chris@160: "-b",
chris@160: help="All files will be placed in the output directory, placed in their own subdirectory corresonding to the generator name.",
chris@160: action="count")
chris@160: parser.add_argument(
chris@160: "-o", "--out",
chris@160: nargs="+",
chris@160: default=["./"], # by default
chris@160: help="List of destination directories for retrieved files. Order should be the same as for --gen.")
chris@160: parser.add_argument(
chris@160: "-d", "--domain",
chris@160: default="https://enzienaudio.com",
chris@160: help="Domain. Default is https://enzienaudio.com.")
chris@160: parser.add_argument(
chris@160: "-x",
chris@160: help="Don't save the returned token.",
chris@160: action="count")
chris@160: parser.add_argument(
chris@160: "-z",
chris@160: help="Force the use of a password, regardless of saved token.",
chris@160: action="count")
chris@160: parser.add_argument(
chris@160: "--noverify",
chris@160: help="Don't verify the SSL connection. Generally a bad idea.",
chris@160: action="count")
chris@160: parser.add_argument(
chris@160: "-v", "--verbose",
chris@160: help="Show debugging information.",
chris@160: action="count")
chris@160: args = parser.parse_args()
chris@160:
chris@160: domain = args.domain or "https://enzienaudio.com"
chris@160:
chris@160: post_data = {}
chris@160:
chris@160: # token should be stored in ~/.heavy/token
chris@160: token_path = os.path.expanduser(os.path.join("~/", ".heavy", "token"))
chris@160: if os.path.exists(token_path) and not args.z:
chris@160: with open(token_path, "r") as f:
chris@160: post_data["credentials"] = {
chris@160: "token": f.read()
chris@160: }
chris@160: else:
chris@160: # otherwise, get the username and password
chris@160: post_data["credentials"] = {
chris@160: "username": raw_input("Enter username: "),
chris@160: "password": getpass.getpass("Enter password: ")
chris@160: }
chris@160:
chris@160: tick = time.time()
chris@160:
chris@160: # make a temporary directory
chris@160: temp_dir = tempfile.mkdtemp(prefix="lroyal-")
chris@160:
chris@160: # zip up the pd directory into the temporary directory
chris@160: try:
chris@160: if not os.path.exists(os.path.join(args.input_dir, "_main.pd")):
chris@160: raise Exception("Root Pd directory does not contain a file named _main.pd.")
chris@160: zip_path = __zip_dir(
chris@160: args.input_dir,
chris@160: os.path.join(temp_dir, "archive.zip"),
chris@160: file_filter={"pd"})
chris@160: except Exception as e:
chris@160: print e
chris@160: shutil.rmtree(temp_dir) # clean up the temporary directory
chris@160: return
chris@160:
chris@160: post_data["name"] = args.name
chris@160:
chris@160: # the outputs to generate (always include c)
chris@160: __SUPPORTED_GENERATOR_SET = {"c", "js"}
chris@160: post_data["gen"] = list(({"c"} | set(args.gen)) & __SUPPORTED_GENERATOR_SET)
chris@160:
chris@160: # upload the job, get the response back
chris@160: # NOTE(mhroth): multipart-encoded file can only be sent as a flat dictionary,
chris@160: # but we want to send a json encoded deep dictionary. So we do a bit of a hack.
chris@160: r = requests.post(
chris@160: urlparse.urljoin(domain, "/a/heavy"),
chris@160: data={"json":json.dumps(post_data)},
chris@160: files={"file": (os.path.basename(zip_path), open(zip_path, "rb"), "application/zip")},
chris@160: verify=False if args.noverify else True)
chris@160:
chris@160: if r.status_code != requests.codes.ok:
chris@160: shutil.rmtree(temp_dir) # clean up the temporary directory
chris@160: r.raise_for_status() # raise an exception
chris@160:
chris@160: # decode the JSON API response
chris@160: r_json = r.json()
chris@160:
chris@160: """
chris@160: {
chris@160: "data": {
chris@160: "compileTime": 0.05078411102294922,
chris@160: "id": "mhroth/asdf/Edp2G",
chris@160: "slug": "Edp2G",
chris@160: "index": 3,
chris@160: "links": {
chris@160: "files": {
chris@160: "linkage": [
chris@160: {
chris@160: "id": "mhroth/asdf/Edp2G/c",
chris@160: "type": "file"
chris@160: }
chris@160: ],
chris@160: "self": "https://enzienaudio.com/h/mhroth/asdf/Edp2G/files"
chris@160: },
chris@160: "project": {
chris@160: "linkage": {
chris@160: "id": "mhroth/asdf",
chris@160: "type": "project"
chris@160: },
chris@160: "self": "https://enzienaudio.com/h/mhroth/asdf"
chris@160: },
chris@160: "self": "https://enzienaudio.com/h/mhroth/asdf/Edp2G",
chris@160: "user": {
chris@160: "linkage": {
chris@160: "id": "mhroth",
chris@160: "type": "user"
chris@160: },
chris@160: "self": "https://enzienaudio.com/h/mhroth"
chris@160: }
chris@160: },
chris@160: "type": "job"
chris@160: },
chris@160: "included": [
chris@160: {
chris@160: "filename": "file.c.zip",
chris@160: "generator": "c",
chris@160: "id": "mhroth/asdf/Edp2G/c",
chris@160: "links": {
chris@160: "self": "https://enzienaudio.com/h/mhroth/asdf/Edp2G/c/file.c.zip"
chris@160: },
chris@160: "mime": "application/zip",
chris@160: "type": "file"
chris@160: }
chris@160: ],
chris@160: "warnings": [],
chris@160: "meta": {
chris@160: "token": "11AS0qPRmjTUHEMSovPEvzjodnzB1xaz"
chris@160: }
chris@160: }
chris@160: """
chris@160: reply_json = r.json()
chris@160: if args.verbose:
chris@160: print json.dumps(
chris@160: reply_json,
chris@160: sort_keys=True,
chris@160: indent=2,
chris@160: separators=(",", ": "))
chris@160:
chris@160: # update the api token, if present
chris@160: if "token" in reply_json.get("meta",{}) and not args.x:
chris@160: if not os.path.exists(os.path.dirname(token_path)):
chris@160: os.makedirs(os.path.dirname(token_path)) # ensure that the .heavy directory exists
chris@160: with open(token_path, "w") as f:
chris@160: f.write(reply_json["meta"]["token"])
chris@160: os.chmod(token_path, stat.S_IRUSR | stat.S_IWUSR) # force rw------- permissions on the file
chris@160:
chris@160: # print any warnings
chris@160: for x in r_json["warnings"]:
chris@160: print "{0}Warning:{1} {2}".format(Colours.yellow, Colours.end, x["detail"])
chris@160:
chris@160: # check for errors
chris@160: if len(r_json.get("errors",[])) > 0:
chris@160: shutil.rmtree(temp_dir) # clean up the temporary directory
chris@160: for x in r_json["errors"]:
chris@160: print "{0}Error:{1} {2}".format(Colours.red, Colours.end, x["detail"])
chris@160: sys.exit(1)
chris@160: return
chris@160:
chris@160: # retrieve all requested files
chris@160: for i,g in enumerate(args.gen):
chris@160: file_url = __get_file_url_for_generator(reply_json, g)
chris@160: if file_url is not None and (len(args.out) > i or args.b):
chris@160: r = requests.get(
chris@160: file_url,
chris@160: cookies={"token": reply_json["meta"]["token"]},
chris@160: verify=False if args.noverify else True)
chris@160: r.raise_for_status()
chris@160:
chris@160: # write the reply to a temporary file
chris@160: c_zip_path = os.path.join(temp_dir, "archive.{0}.zip".format(g))
chris@160: with open(c_zip_path, "wb") as f:
chris@160: f.write(r.content)
chris@160:
chris@160: # unzip the files to where they belong
chris@160: if args.b:
chris@160: target_dir = os.path.join(os.path.abspath(os.path.expanduser(args.out[0])), g)
chris@160: else:
chris@160: target_dir = os.path.abspath(os.path.expanduser(args.out[i]))
chris@160: if not os.path.exists(target_dir):
chris@160: os.makedirs(target_dir) # ensure that the output directory exists
chris@160: __unzip(c_zip_path, target_dir)
chris@160:
chris@160: print "{0} files placed in {1}".format(g, target_dir)
chris@160: else:
chris@160: print "{0}Warning:{1} {2} files could not be retrieved.".format(
chris@160: Colours.yellow, Colours.end,
chris@160: g)
chris@160:
chris@160: # delete the temporary directory
chris@160: shutil.rmtree(temp_dir)
chris@160:
chris@160: print "Job URL", reply_json["data"]["links"]["self"]
chris@160: print "Total request time: {0}ms".format(int(1000.0*(time.time()-tick)))
chris@160:
chris@160: sys.exit(0)
chris@160:
chris@160: def __get_file_url_for_generator(json_api, g):
chris@160: """Returns the file link for a specific generator.
chris@160: Returns None if no link could be found.
chris@160: """
chris@160: for i in json_api["included"]:
chris@160: if g == i["generator"]:
chris@160: return i["links"]["self"]
chris@160: return None # by default, return None
chris@160:
chris@160:
chris@160:
chris@160: if __name__ == "__main__":
chris@160: main()