A downloader for moodle file submissions. Usage ./submission_downloader.py --token {token} --url {assignment_url}

submission_downloader.py

#!/usr/bin/env python3

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from functools import partial

MOODLE_URL = 'https://www.moodle.tum.de/'


def api_call(token, wsfunction, kwargs):
    params = {
        'wstoken': token,
        'moodlewsrestformat': 'json',
        'wsfunction': wsfunction
    }
    params.update(kwargs)
    query_url = MOODLE_URL + 'webservice/rest/server.php?' + urllib.parse.urlencode(params)
    res = urllib.request.urlopen(query_url).read()
    string = res.decode()
    try:
        return json.loads(string)
    except json.JSONDecodeError:
        print("Couldn't convert response to JSON: " + string)
        sys.exit(1)


def get_submissions_list(moduleid, token):
    bound_api = partial(api_call, token)
    res = bound_api('core_course_get_course_module', {'cmid': moduleid})
    instanceid = res['cm']['instance']

    assignments = bound_api('mod_assign_get_submissions', {'assignmentids[0]': instanceid})
    return assignments['assignments'][0]['submissions']


def handle_submission(token, parent_dir, submission):
    userid = submission['userid']
    user_info = get_user_info(token, userid)
    path = [parent_dir]
    path.append('%s - %s' % (user_info.get('idnumber', 'moodleid_%s' % user_info.get('id', 'unknown')), user_info['fullname']))

    for plugin in submission['plugins']:
        if plugin['type'] != 'file':
            continue
        for filearea in plugin['fileareas']:
            if 'files' not in filearea:
                continue
            for f in filearea['files']:
                filepath = path + f['filepath'].split('/') + f['filename'].split('/')
                download_file(token, f['fileurl'], filepath)


def get_user_info(token, userid):
    users = api_call(token, 'core_user_get_users_by_field', {"field": 'id', 'values[0]': userid}
                     )
    if not users:
        print(('\nCannot retrieve information about user %s, '
               'try openning https://www.moodle.tum.de/user/profile.php?id=%s manually') % (userid, userid))
        return {'fullname': 'Unknown', 'idnumber': 'moodle_%s' % userid}
    return users[0]


def create_path(path):
    dirs = path[:-1]
    for subpath_idx in range(1, len(dirs) + 1):
        subpath = os.path.join(*dirs[:subpath_idx])
        if not os.path.exists(subpath):
            os.makedirs(subpath)


def download_file(token, url, path):
    path_string = os.path.join(*path)
    try:
        filecontent = urllib.request.urlopen('%s?token=%s' % (url, token)).read()
        create_path(path)
        with open(path_string, 'wb') as target:
            target.write(filecontent)
    except urllib.error.HTTPError:
        print('\nCouldnot download file %s into path %s, skipping...' % (url, path_string))


def progress(list_like):
    list_length = len(list_like)
    bar_length = 20
    pos = 0
    for el in list_like:
        pos += 1
        bar_pos = int(pos / list_length * bar_length)
        print('\r[' + '#' * bar_pos + '-' * (bar_length - bar_pos) + ']' + ' %s/%s' % (pos, list_length), end='')
        yield el


def main():
    parser = argparse.ArgumentParser(description='Download all file submissions from a moodle assignment')
    parser.add_argument('--token', '-t', type=str, help='Moodle API token')
    parser.add_argument('--url', '-u', type=str, help='Assignment URL')
    args = parser.parse_args()
    token = args.token
    modulematch = re.match(re.escape(MOODLE_URL) + r'mod/assign/view\.php\?id\=(\d+)', args.url)
    if not modulematch:
        print('Not a moodle assignment url: ' + args.url)
        sys.exit(1)
    moduleid = modulematch.group(1)
    submissions_list = get_submissions_list(moduleid, token)
    dl_dir = 'submissions - %s' % moduleid

    for submission in progress(submissions_list):
        handle_submission(token, dl_dir, submission)


if __name__ == '__main__':
    main()