create a "tech support" HIT based on an existing supersubmiterator config

tech-support.py

#!/usr/bin/env python

import json, argparse, os, csv
import boto3
mturk = boto3.client('mturk')

def main():
    parser = argparse.ArgumentParser(description="Technical Support for MTurk workers")
    parser.add_argument(
        "--qualification_id",
        metavar="id",
        type=str,
        help="The QualificationTypeId to grant to workers who need support.  " +
        "If ommited, a new qualification type will be created",
        default=None)
    parser.add_argument(
        "label",
        metavar="label",
        help="supersubmiterator-style label.  workerIds will be loaded from {label}-tech-support-workerids.csv " +
        "and config from {label}.config.  A new config file will be created called {label}-tech-support.config",
        type=str)

    args = parser.parse_args()
    label = args.label
    qual_id = args.qualification_id

    config = parse_config(label)

    workers = load_workers(label)
    qual_id = check_qual_id(label, qual_id)
    grant_qual_to_workers(qual_id, workers)

    new_config = update_config(config, workers, qual_id)
    new_config_fn = label + "-tech-support.config"
    with open(new_config_fn, "w") as new_config_file:
        print("Writing updated config...")
        json.dump(new_config, new_config_file)
        print("✔ {}".format(new_config_fn))

    balance = mturk.get_account_balance()
    if float(balance["AvailableBalance"]) < float(config["reward"]) * 1.1 * len(workers):
        warn("⚠ available balance of {} is not sufficient to cover {} workers at {} each".format(
            balance["AvailableBalance"],
            len(workers),
            config["reward"]
        ))
    print("""

    Next steps: 
    1. use supersubmiterator to run HIT

    """)

def parse_config(label):
    config_fn = label + ".config"
    with open(config_fn, "r") as config_file:
        config = json.load(config_file)

    if config["liveHIT"] != "yes":
        warn("⚠ Trying to grant tech support based on sandbox config in {}".format(config_fn))

    return(config)
    
def load_workers(label):
    workers_fn = label + "-tech-support-workerids.csv"
    print("Loading workers needing tech support from {}...".format(workers_fn))
    try:
        with open(workers_fn, "r") as workers_file:
            contents = csv.DictReader(workers_file)
            workers = []
            for row in contents:
                workers.append(row["workerId"])
        print("✔ loaded {} workers".format(len(workers)))
    except:
        print("✘ does {} exist and have workerId column?".format(workers_fn))
    return(workers)

# create a qualification type
def check_qual_id(label, qual_id):
    if qual_id == None:
        qual_name = label + "-tech-support"
        print("Creating new qualification type: {}...".format(qual_name))
        try:
            response = mturk.create_qualification_type(
                Name=qual_name,
                Description="Technical support for LeAP Lab study",
                QualificationTypeStatus="Active",
                AutoGranted=False
            )
            qual_id = response["QualificationType"]["QualificationTypeId"]
        except:
            # possibly failure due to existing qual id with same name...
            print("✘ checking for existing type with same name...")
            response = mturk.list_qualification_types(
                Query=qual_name,
                MustBeRequestable=False,
                MustBeOwnedByCaller=True
            )
            if response["NumResults"] == 1:
                qual_id = response["QualificationTypes"][0]["QualificationTypeId"]
            else:
                error("Need exactly one qualification type matching name {}\n{}".format(qual_name, response))
        print("✔ {}".format(qual_id))
        return(qual_id)
    else:
        print("Checking existing qualification type: {}".format(qual_id))
        try:
            response = mturk.get_qualification_type(
                QualificationTypeId=qual_id
            )
            qual_type = response["QualificationType"]
            print("✔ Found existing qualification:\n  ", qual_type)
            qual_id = qual_type["QualificationTypeId"]
            return(qual_id)
        except Exception as err:
            print("✘", qual_id, err)
            raise(err)
        
def grant_qual_to_workers(qual_id, workers):
    print("Assigning qualification {} to {} workers".format(qual_id, len(workers)))
    for worker in workers:
        try:
            mturk.associate_qualification_with_worker(
                QualificationTypeId=qual_id,
                WorkerId=worker,
                IntegerValue=100,
                SendNotification=False
            )
            print("✔ {}".format(worker))
        except Exception as err:
            print("✘ {}: {}".format(worker, err))
            

# create modified HIT config
def update_config(config, workers, qualification_id):

    new_config = dict()
    new_config["liveHIT"] = config["liveHIT"]
    new_config["title"] = config["title"] + " TECH SUPPORT"
    new_config["description"] = "Technical support for workers in LeAP Lab studies"
    new_config["reward"] = config["reward"]
    new_config["hitlifetime"] = 60 * 60 * 24    # one day
    new_config["assignmentduration"] = 60 * 10  # 10 minutes
    new_config["autoapprovaldelay"] = 60 * 60   # one hours
    new_config["experimentURL"] = "https://experiments.leap-lab.org/tech-support/"
    new_config["frameheight"] = config["frameheight"]
    new_config["numberofassignments"] = len(workers)
    new_config["doesHaveQualification"] = qualification_id

    # these are mandatory keys that don't seem strictly necessary
    new_config["USonly?"] = config["USonly?"]
    new_config["minPercentPreviousHITsApproved"] = config["minPercentPreviousHITsApproved"]
    
    return new_config

if __name__ == '__main__':
    main()