Skip to content

Instantly share code, notes, and snippets.

@gthole

gthole/progress_matchbox_assignments.py

Last active December 15, 2015 19:29
Download ZIP

Revisions

  1. gthole revised this gist Apr 16, 2013. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -221,7 +221,7 @@ def progress_assignments(from_name, to_name, who, when, since=None):
    ACTIVITY_HASH = activity_hash()

    for progression in SETTINGS['activities']:
    from_, to, who, when = [progression[k] for k
    from_, to, who, when = [progression.get(k) for k
    in ['from', 'to', 'assignee', 'date_due']]

    # Make sure the date is understood and formatted correctly.
  2. gthole revised this gist Apr 15, 2013. 1 changed file with 29 additions and 18 deletions.
    47 changes: 29 additions & 18 deletions progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -17,17 +17,25 @@
    "program": "4ea32279a54ea7c732000002", # ID of program
    "username": "<api username>",
    "password": "<api password>",
    "last_run": "2013-04-05 12:00:00",
    "activities": [
    ["Reader I", "Reader II", "<user id>"],
    {
    "from": Reader I",
    "to": Reader II",
    "assignee": "<user id>",
    "date_due": "<new due date>"
    },
    ]
    }
    The above configuration will find all unexpired and completed
    'Reader I' assignments, and create a new 'Reader II' with
    the same date_due and application, but assigned to the user
    for "<user id>". The "<user id>" may also be `null`, whereupon
    the assignee will be the same as the one on the completed
    assignment. The start_date will be the date the script is run.
    The above configuration will find all 'Reader I' assignments, that
    have been completed since April 5th, 2013, and create a new 'Reader II'
    for the same application, but assigned to the user listed for "assignee"
    and with the given "date_due". The "assignee" may also be `null`,
    whereupon the assignee will be the same as the one on the completed
    assignment. If "date_due" is `null`, then the date_due will similarly
    be the same as the completed assignment. The start_date will be the
    date the script is run.
    More activity pairs may be listed.
    @@ -134,18 +142,11 @@ def already_assigned(assign, search_list):
    assignee = assign['assignee']
    app = assign['application']

    # Check that the base assignment hasn't expired.
    date_due = parser.parse(assign['date_due'])
    if date_due < datetime.utcnow():
    return True

    found = False
    for upstream_assign in search_list:
    if ((upstream_assign['assignee'] == assignee) and
    (upstream_assign['application'] == app)):
    found = True
    break
    return found
    return True
    return False


    def setup_activity(name, since, completed):
    @@ -155,7 +156,7 @@ def setup_activity(name, since, completed):
    return id_, list_


    def progress_assignments(from_name, to_name, who, since=None):
    def progress_assignments(from_name, to_name, who, when, since=None):
    from_id, from_list = setup_activity(from_name, since, '1')
    to_id, to_list = setup_activity(to_name, since, '0')

    @@ -176,6 +177,9 @@ def progress_assignments(from_name, to_name, who, since=None):
    else:
    new['assignee'] = assign['assignee']

    if when:
    new['date_due'] = when

    new['activity'] = '/api/v2/activity/%s/' % to_id
    new['start_date'] = START_DATE.isoformat()

    @@ -216,11 +220,18 @@ def progress_assignments(from_name, to_name, who, since=None):

    ACTIVITY_HASH = activity_hash()

    for from_, to, who in SETTINGS['activities']:
    for progression in SETTINGS['activities']:
    from_, to, who, when = [progression[k] for k
    in ['from', 'to', 'assignee', 'date_due']]

    # Make sure the date is understood and formatted correctly.
    when = parser.parse(when).isoformat()

    progress_assignments(
    from_name=from_,
    to_name=to,
    who=who,
    when=when,
    since=SETTINGS.get('last_run'))

    SETTINGS['last_run'] = START_DATE.strftime('%Y-%m-%d %H:%M:%S')
  3. gthole revised this gist Apr 8, 2013. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -45,7 +45,7 @@
    from os import path
    from shutil import copyfile

    DOMAIN = 'http://app.admitpad.com'
    DOMAIN = 'https://app.admitpad.com'


    def login(user, pswd, program):
  4. gthole revised this gist Apr 8, 2013. 1 changed file with 8 additions and 7 deletions.
    15 changes: 8 additions & 7 deletions progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,5 +1,5 @@
    """
    Progress Matchbox Assignments - Automatically move completed
    Progress Matchbox Assignments - Automatically move completed
    assignments from one role to another for the same assignee and
    due date.
    @@ -148,17 +148,16 @@ def already_assigned(assign, search_list):
    return found


    def setup_activity(name, hashed, since, completed):
    def setup_activity(name, since, completed):
    "Helper function to set up activity variables."
    id_ = hashed[name]
    id_ = ACTIVITY_HASH[name]
    list_ = assignment_list(id_, since, completed)
    return id_, list_


    def progress_assignments(from_name, to_name, who, since=None):
    hashed = activity_hash()
    from_id, from_list = setup_activity(from_name, hashed, since, '1')
    to_id, to_list = setup_activity(to_name, hashed, since, '0')
    from_id, from_list = setup_activity(from_name, since, '1')
    to_id, to_list = setup_activity(to_name, since, '0')

    assignee = '/api/v2/user/%s/' % who if who else None

    @@ -215,11 +214,13 @@ def progress_assignments(from_name, to_name, who, since=None):
    SETTINGS['password'],
    SETTINGS['program'])

    ACTIVITY_HASH = activity_hash()

    for from_, to, who in SETTINGS['activities']:
    progress_assignments(
    from_name=from_,
    to_name=to,
    base_assignee=who,
    who=who,
    since=SETTINGS.get('last_run'))

    SETTINGS['last_run'] = START_DATE.strftime('%Y-%m-%d %H:%M:%S')
  5. gthole revised this gist Apr 4, 2013. 1 changed file with 17 additions and 6 deletions.
    23 changes: 17 additions & 6 deletions progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -18,14 +18,16 @@
    "username": "<api username>",
    "password": "<api password>",
    "activities": [
    ["Reader I", "Reader II"],
    ["Reader I", "Reader II", "<user id>"],
    ]
    }
    The above configuration will find all unexpired and completed
    'Reader I' assignments, and create a new 'Reader II' with
    the same date_due, assignee, and application. The
    start_date will be the date the script is run.
    the same date_due and application, but assigned to the user
    for "<user id>". The "<user id>" may also be `null`, whereupon
    the assignee will be the same as the one on the completed
    assignment. The start_date will be the date the script is run.
    More activity pairs may be listed.
    @@ -153,20 +155,28 @@ def setup_activity(name, hashed, since, completed):
    return id_, list_


    def progress_assignments(from_name, to_name, since=None):
    def progress_assignments(from_name, to_name, who, since=None):
    hashed = activity_hash()
    from_id, from_list = setup_activity(from_name, hashed, since, '1')
    to_id, to_list = setup_activity(to_name, hashed, since, '0')

    assignee = '/api/v2/user/%s/' % who if who else None

    count = 0
    failed = []
    for assign in from_list:
    if not already_assigned(assign, to_list):
    # Create a new assignment
    base_fields = ['assignee', 'application', 'date_due']
    base_fields = ['application', 'date_due']
    new = {}
    for field in base_fields:
    new[field] = assign[field]

    if assignee:
    new['assignee'] = assignee
    else:
    new['assignee'] = assign['assignee']

    new['activity'] = '/api/v2/activity/%s/' % to_id
    new['start_date'] = START_DATE.isoformat()

    @@ -205,10 +215,11 @@ def progress_assignments(from_name, to_name, since=None):
    SETTINGS['password'],
    SETTINGS['program'])

    for from_, to in SETTINGS['activities']:
    for from_, to, who in SETTINGS['activities']:
    progress_assignments(
    from_name=from_,
    to_name=to,
    base_assignee=who,
    since=SETTINGS.get('last_run'))

    SETTINGS['last_run'] = START_DATE.strftime('%Y-%m-%d %H:%M:%S')
  6. gthole created this gist Apr 4, 2013.
    220 changes: 220 additions & 0 deletions progress_matchbox_assignments.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,220 @@
    """
    Progress Matchbox Assignments - Automatically move completed
    assignments from one role to another for the same assignee and
    due date.
    progress (intransitive verb):
    1: to move forward, proceed
    2: to develop to a higher, better, or more advanced stage
    author - [email protected], 4/3/2013
    usage - To be run as a command line script with one argument pointing
    to a json configuration file.
    Example config:
    {
    "program": "4ea32279a54ea7c732000002", # ID of program
    "username": "<api username>",
    "password": "<api password>",
    "activities": [
    ["Reader I", "Reader II"],
    ]
    }
    The above configuration will find all unexpired and completed
    'Reader I' assignments, and create a new 'Reader II' with
    the same date_due, assignee, and application. The
    start_date will be the date the script is run.
    More activity pairs may be listed.
    The output of the script is a count of the number of
    assignments successfully progressed for each activity pair.
    """


    import requests
    from urllib import urlencode
    from datetime import datetime
    from dateutil import parser
    import json
    from sys import argv
    from os import path
    from shutil import copyfile

    DOMAIN = 'http://app.admitpad.com'


    def login(user, pswd, program):
    "Log in with given credentials, return requests session object"

    # Set up session.
    c = requests.session()
    r = c.get('%s/login/' % DOMAIN)

    # Compile login data.
    post_data = {'username': user, 'password': pswd}
    csrf = [
    cookie for cookie in c.cookies
    if cookie.name == 'csrftoken'
    ][0].value

    post_data['csrfmiddlewaretoken'] = csrf
    c.headers.update({'Referer': '%s/login/' % DOMAIN})

    # Log in.
    r = c.post('%s/login/' % DOMAIN,
    data=post_data,
    allow_redirects=True)
    assert (r.status_code == 200)

    # Choose program.
    r = c.get(
    (('%s/dashboard/choose_program?id=' + program + '&next=/') %
    DOMAIN))
    assert (r.status_code == 200)

    return c


    def resource(uri, parameters={}):
    "GET a resource from the api."
    if isinstance(parameters, dict):
    parameters.update({'format': 'json'})
    params = '?' + urlencode(parameters)
    else:
    params = ''

    r = SESSION.get('%s%s%s' % (DOMAIN, uri, params))
    if (r.status_code != 200):
    # Fail if a request fails at any point.
    raise AssertionError(
    "Failed to get resource (%d): %s%s" % (
    r.status_code, uri, params))
    return r.json()


    def activity_hash():
    "Name to id look-up so config activities can be stored by name."
    activity_hash = {}
    r = resource('/api/v2/activity/')
    for activity in r['objects']:
    activity_hash[activity['name']] = activity['id']
    return activity_hash


    def assignment_list(id_, since, completed):
    "Work through assignment pagination."
    params = {'completed': completed, 'activity': id_}

    # Filter out assignments that we've handled
    # already in a previous run of the script.
    if since and (completed == '1'):
    params['date_completed__gt'] = since

    r = resource('/api/v2/assignment/', params)
    assigns = r['objects']

    # Paginate through.
    while r['meta']['next']:
    r = resource(r['meta']['next'], '')
    assigns += r['objects']

    return assigns


    def already_assigned(assign, search_list):
    """
    Returns True if there is an assignment in search_list with
    the same application and assignee as the given assignment.
    """
    assignee = assign['assignee']
    app = assign['application']

    # Check that the base assignment hasn't expired.
    date_due = parser.parse(assign['date_due'])
    if date_due < datetime.utcnow():
    return True

    found = False
    for upstream_assign in search_list:
    if ((upstream_assign['assignee'] == assignee) and
    (upstream_assign['application'] == app)):
    found = True
    break
    return found


    def setup_activity(name, hashed, since, completed):
    "Helper function to set up activity variables."
    id_ = hashed[name]
    list_ = assignment_list(id_, since, completed)
    return id_, list_


    def progress_assignments(from_name, to_name, since=None):
    hashed = activity_hash()
    from_id, from_list = setup_activity(from_name, hashed, since, '1')
    to_id, to_list = setup_activity(to_name, hashed, since, '0')

    count = 0
    failed = []
    for assign in from_list:
    if not already_assigned(assign, to_list):
    # Create a new assignment
    base_fields = ['assignee', 'application', 'date_due']
    new = {}
    for field in base_fields:
    new[field] = assign[field]
    new['activity'] = '/api/v2/activity/%s/' % to_id
    new['start_date'] = START_DATE.isoformat()

    r = SESSION.post(
    '%s/api/v2/assignment/' % DOMAIN, data=json.dumps(new),
    headers={'content-type': 'application/json',
    'accept': 'application/json'}
    )
    if r.status_code == 201:
    count += 1
    else:
    failed.append(assign['id'])

    # Print results per activity.
    print "Progressed %d applications from '%s' to '%s'." % (
    count, from_name, to_name)
    if failed:
    print "Unable to progress the following assignments:"
    for id_ in failed:
    print id_


    if __name__ == '__main__':
    # Back up and get config file.
    if len(argv) == 2 and path.exists(argv[1]):
    copyfile(argv[1], '%s.BAK' % argv[1])
    with open(argv[1]) as file_:
    SETTINGS = json.loads(file_.read())

    else:
    raise ValueError('Could not find config file.')

    START_DATE = datetime.utcnow()
    SESSION = login(
    SETTINGS['username'],
    SETTINGS['password'],
    SETTINGS['program'])

    for from_, to in SETTINGS['activities']:
    progress_assignments(
    from_name=from_,
    to_name=to,
    since=SETTINGS.get('last_run'))

    SETTINGS['last_run'] = START_DATE.strftime('%Y-%m-%d %H:%M:%S')

    with open(argv[1], 'w') as file_:
    file_.write(json.dumps(SETTINGS,
    sort_keys=True,
    indent=4,
    separators=(',', ': ')))