<commit_before>"""
Utility functions used by API and other services
"""

# pylint: disable=no-member

import collections
import logging
import datetime
import itertools
from os.path import join

try:
    from cStringIO import StringIO
except:
    from StringIO import StringIO
import zipfile as zf
import csv
from flask import jsonify, request, Response, json

from google.appengine.api import memcache
from google.appengine.ext import ndb
from google.appengine.ext import deferred
import cloudstorage as gcs

from app import app
from app.constants import GRADES_BUCKET
from app.exceptions import BadValueError

# TODO Looks like this can be removed just by relocating parse_date
# To deal with circular imports
class ModelProxy(object):
    def __getattribute__(self, key):
        import app
        return app.models.__getattribute__(key)

ModelProxy = ModelProxy()

def parse_date(date):
    # TODO Describe what date translation is happening here. Probably needs
    # a rewrite to handle daylight savings and work with other time zones.
    try:
        date = datetime.datetime.strptime(
            date, app.config["GAE_DATETIME_FORMAT"])
    except ValueError:
        date = datetime.datetime.strptime(
            date, "%Y-%m-%d %H:%M:%S")

    delta = datetime.timedelta(hours=7)
    return datetime.datetime.combine(date.date(), date.time()) + delta

def coerce_to_json(data, fields):
    """
    Coerces |data| to json, using only the allowed |fields|
    """
    if hasattr(data, 'to_json'):
        return data.to_json(fields)
    elif isinstance(data, list):
        return [mdl.to_json(fields) if hasattr(mdl, 'to_json')
                else coerce_to_json(mdl, fields) for mdl in data]
    elif isinstance(data, dict):
        if hasattr(data, 'to_json'):
            return {
                k: mdl.to_json(fields.get(k, {}))
                for k, mdl in data.iteritems()}
        else:
            return {k: coerce_to_json(mdl, fields.get(k, {}))
                    for k, mdl in data.iteritems()}
    else:
        return data

#TODO(martinis) somehow having data be an empty list doesn't make it
# return an empty list, but an empty object.
def create_api_response(status, message, data=None):
    """Creates a JSON response that contains status code (HTTP),
    an arbitrary message string, and a dictionary or list of data"""
    if isinstance(data, dict) and 'results' in data:
        data['results'] = (
            coerce_to_json(data['results'], request.fields.get('fields', {})))
    else:
        data = coerce_to_json(data, request.fields.get('fields', {}))

    if request.args.get('format', 'default') == 'raw':
        response = Response(json.dumps(data))
    else:
        response = jsonify(**{
            'status': status,
            'message': message,
            'data': data
        })
    response.status_code = status
    return response


def create_zip(file_contents={}, dir=''):
    return finish_zip(*start_zip(file_contents, dir))


def start_zip(file_contents={}, dir=''):
    """
    Creates a file from the given dictionary of filenames to contents.
    Uses specified dir to store all files.
    """
    zipfile_str = StringIO()
    zipfile = zf.ZipFile(zipfile_str, 'w')
    zipfile = add_to_zip(zipfile, file_contents, dir)
    return zipfile_str, zipfile


def finish_zip(zipfile_str, zipfile):
    return zipfile_str.getvalue()


def add_to_zip(zipfile, file_contents, dir=''):
    """
    Adds files to a given zip file. Uses specified dir to store files.
    """
    for filename, contents in file_contents.items():
        zipfile.writestr(join(dir, filename), contents)
    return zipfile

def create_csv_content(content):
    """ 
    Return all contents in CSV file format. Content must be a list of lists.
    """
    scsv = StringIO()
    writer = csv.writer(scsv)
    try:
        writer.writerows(content)
    except csv.Error as e:
        scsv.close()
        sys.exit('Error creating CSV: {}'.format(e))
    contents = scsv.getvalue()
    scsv.close()
    return contents

def data_for_scores(assignment, user):
    """ 
    Returns a tuple of two values a list of lists of score info for assignment.
    Format: [['STUDENT', 'SCORE', 'MESSAGE', 'GRADER', 'TAG']] 
    """
    content = [['STUDENT', 'SCORE', 'MESSAGE', 'GRADER', 'TAG']]
    course = assignment.course.get()
    groups = ModelProxy.Group.lookup_by_assignment(assignment)
    seen_members = set()

    for group in groups:
        members = group.member
        seen_members |= set(members)
        content.extend(group.scores_for_assignment(assignment))

    students = [part.user.get() for part in course.get_students(user) if part.user not in seen_members]
    for student in students:
        content.extend(student.scores_for_assignment(assignment)[0])

    return content

def create_gcs_file(gcs_filename, contents, content_type):
    """ 
    Creates a GCS csv file with contents CONTENTS. 
    """
    try:
        gcs_file = gcs.open(gcs_filename, 'w', content_type=content_type, options={'x-goog-acl': 'project-private'})
        gcs_file.write(contents)
        gcs_file.close()
    except Exception as e:
        logging.exception("ERROR: {}".format(e))
        try:
            gcs.delete(gcs_filename)
        except gcs.NotFoundError:
            logging.info("Could not delete file " + gcs_filename)
    logging.info("Created file " + gcs_filename)


def make_csv_filename(assignment, infotype):
    """ Returns filename of format INFOTYPE_COURSE_ASSIGNMENT.csv """
    course_name = assignment.course.get().offering
    assign_name = assignment.display_name
    filename = '{}_{}_{}.csv'.format(infotype, course_name, assign_name)
    return filename.replace('/', '_').replace(' ', '_')

def paginate(entries, page, num_per_page):
    """
    Added stuff from
    https://p.ota.to/blog/2013/4/pagination-with-cursors-
        in-the-app-engine-datastore/

    Support pagination for an NDB query.
    Arguments:
      |entries| - a query which returns the items to paginate over.
      |cursor|  - a cursor of where in the pagination the user is.
      |num_per_page| - the number of results to display per page.

    The return value will be different from a regular query:
    There will be 3 things returned:
    - results: a list of results
    - forward_curs: a urlsafe hash for the cursor. Use this to get
    the next page. To retrieve the cursor object, do Cursor(urlsafe=s)
    - more: a boolean for whether or not there is more content.

    For more documentation, look at:
    https://developers.google.com/appengine/docs/python/ndb/queryclass#Query_fetch_page
    """
    if num_per_page is None:
        return {
            'results': entries.fetch(),
            'page': 1,
            'more': False
        }

    query_serialized = (
        '_'.join(str(x) for x in (
            entries.kind, entries.filters, entries.orders)))
    query_serialized = query_serialized.replace(' ', '_')
    def get_mem_key(page):
        offset = (page - 1) * num_per_page
        return "cp_%s_%s" % (query_serialized, offset)
    this_page_key = get_mem_key(page)
    next_page_key = get_mem_key(page + 1)

    cursor = None
    store_cache = True
    if page > 1:
        cursor = memcache.get(this_page_key)
        if not cursor:
            page = 1 # Reset to the front, since memcached failed
            store_cache = False

    pages_to_fetch = int(num_per_page)
    if cursor is not None:
        results, forward_cursor, more = entries.fetch_page(
            pages_to_fetch, start_cursor=cursor)
    else:
        results, forward_cursor, more = entries.fetch_page(pages_to_fetch)

    if store_cache:
        memcache.set(next_page_key, forward_cursor)

    return {
        'results': results,
        'page': page,
        'more': more
    }


def _apply_filter(query, model, arg, value, op):
    """
    Applies a filter on |model| of |arg| |op| |value| to |query|.
    """
    if '.' in arg:
        arg = arg.split('.')
    else:
        arg = [arg]

    field = model
    while arg:
        field = getattr(field, arg.pop(0), None)
        if not field:
            # Silently swallow for now
            # TODO(martinis) cause an error
            return query

    if op == "==":
        filtered = field == value
    elif op == "<":
        filtered = field < value
    elif op == "<=":
        filtered = field <= value
    elif op == ">":
        filtered = field > value
    elif op == ">=":
        filtered = field >= value
    else:
        raise ValueError("Invalid filtering operator {}".format(op))

    return query.filter(filtered)

def filter_query(query, args, model):
    """
    Applies the filters in |args| to |query|.
    |args| is a dictionary of key to value, to be used to filter the query.
    |allowed| is an optional list of the allowed filters.

    Returns a modified query with the appropriate filters.
    """
    for arg, value in args.iteritems():
        if (isinstance(value, collections.Iterable)
                and not isinstance(value, str)):
            op, value = value
        else:
            value, op = value, '=='

        query = _apply_filter(query, model, arg, value, op)

    return query


####################
# Deferred actions #
####################

ASSIGN_BATCH_SIZE = 20
def add_to_grading_queues(assign_key, cursor=None, num_updated=0):
    query = ModelProxy.FinalSubmission.query().filter(
        ModelProxy.FinalSubmission.assignment == assign_key)

    queues = list(ModelProxy.Queue.query(
        ModelProxy.Queue.assignment == assign_key))
    if not queues:
        logging.error("Tried to assign work, but no queues existed")
        return

    kwargs = {}

    if cursor:
        kwargs['start_cursor'] = cursor

    to_put = 0
    results, cursor, _ = query.fetch_page(ASSIGN_BATCH_SIZE, **kwargs)
    seen = set()
    for queue in queues:
        for subm in queue.submissions:
            if isinstance(subm, ndb.Key):
                seen.add(subm.get().submitter.id())
            else:
                seen.add(subm.submitter.id())

    for subm in results:
        user = subm.submitter.get()
        if not user.logged_in or user.key.id() in seen:
            continue
        queues.sort(key=lambda x: len(x.submissions))

        subm = user.get_selected_submission(assign_key, keys_only=True)
        if subm and user.is_final_submission(subm, assign_key):
            subm_got = subm.get()
            if subm_got.get_messages().get('file_contents'):
                queues[0].submissions.append(subm)
                seen.add(user.key.id())
                to_put += 1

    if to_put:
        num_updated += to_put
        ndb.put_multi(queues)
        logging.debug(
            'Put %d entities to Datastore for a total of %d',
            to_put, num_updated)
        deferred.defer(
            add_to_grading_queues, assign_key, cursor=cursor,
            num_updated=num_updated)
    else:
        logging.debug(
            'add_to_grading_queues complete with %d updates!', num_updated)

def assign_submission(backup_id, submit):
    """
    Create Submisson and FinalSubmission records for a submitted Backup.

    :param backup_id: ID of a Backup
    :param submit: Whether this backup is a submission to be graded
    """
    backup = ModelProxy.Backup.get_by_id(backup_id)
    if not backup.get_messages().get('file_contents'):
        logging.info("Submission had no file_contents; not processing")
        return

    if submit:
        assign = backup.assignment.get_async()
        subm = ModelProxy.Submission(backup=backup.key)
        subm.put()

        # Can only make a final submission before it's due
        if datetime.datetime.now() < assign.get_result().due_date:
            subm.mark_as_final()

def sort_by_assignment(key_func, entries):
    entries = sorted(entries, key=key_func)
    return itertools.groupby(entries, key_func)

@ndb.toplevel
def merge_user(user_key, dup_user_key):
    """
    Merges |dup_user| into |user|.
    """
    if isinstance(user_key, ModelProxy.Base):
        user = user_key
        user_key = user_key.key
        get_user = lambda: user
    else:
        user = user_key.get_async()
        def get_user():
            return user.get_result()

    if isinstance(dup_user_key, ModelProxy.Base):
        dup_user = dup_user_key
        get_dup_user = lambda: dup_user
        dup_user_key = dup_user_key.key
    else:
        dup_user = dup_user_key.get_async()
        def get_dup_user():
            return dup_user.get_result()

    # Leave all groups
    G = ModelProxy.Group
    groups = G.query(ndb.OR(
        G.member == dup_user_key,
        G.invited == dup_user_key)).fetch()
    for group in groups:
        group.exit(dup_user_key)
        
    # Deactivate all enrollments
    E = ModelProxy.Participant
    enrolls = E.query(E.user == dup_user_key).fetch()
    for enroll in enrolls:
        # enroll.status = 'inactive'
        enroll.put_async()

    # Re-submit submissions
    S = ModelProxy.Submission
    subms = S.query(S.submitter == dup_user_key).fetch()
    for subm in subms:
        subm.resubmit(user_key)

    dup_user = get_dup_user()
    # Change email

    user = get_user()
    lowered_emails = [email.lower() for email in user.email]
    for email in dup_user.email:
        if email.lower() not in lowered_emails:
            user.email.append(email.lower())

    # Invalidate emails
    dup_user.email = ['#'+email for email in dup_user.email]
    # dup_user.status = 'inactive'
    dup_user.put_async()
    user.put_async()

    log = ModelProxy.AuditLog()
    log.event_type = "Merge user"
    log.user = user_key
    log.description = "Merged user {} with {}. Merged emails {}".format(
        dup_user_key.id(), user_key.id(), dup_user.email)
    log.obj = dup_user_key
    log.put_async()

def unique_email_address(user):
    U = ModelProxy.User

    dups = []
    for email in user.email:
        users = U.query(U.email == email).fetch()
        for found_user in users:
            if found_user.key != user.key:
                dups.append((user, found_user))

    for usera, userb in dups:
        if usera.email[0].lower() != usera.email[0]:
            user, dup_user = usera, userb
        else:
            user, dup_user = userb, usera

        merge_user(user, dup_user)

def unique_final_submission(user):
    FS = ModelProxy.FinalSubmission

    key_func = lambda subm: subm.assignment
    submissions = FS.query(FS.submitter == user.key).fetch()
    for lst in sort_by_assignment(key_func, submissions):
        if len(lst) > 1:
            lst = sorted(lst, key=lambda subm: subm.server_time)[1:]
            for subm in lst:
                subm.key.delete()

def unique_group(user):
    G = ModelProxy.Group
    key_func = lambda group: group.assignment
    groups = G.query(G.member == user.key).fetch()
    for lst in sort_by_assignment(key_func, groups):
        if len(lst) > 1:
            # TODO(martinis, denero) figure out what to do
            pass

def deferred_check_user(user_id):
    user = ModelProxy.User.get_by_id(user_id)
    if not user:
        raise deferred.PermanentTaskFailure("User id {} is invalid.".format(user_id))

    unique_email_address(user)
    unique_final_submission(user)
    unique_group(user)


def check_user(user_key):
    if isinstance(user_key, ModelProxy.User):
        user_key = user_key.key.id()

    if isinstance(user_key, ndb.Key):
        user_key = user_key.id()

    deferred.defer(deferred_check_user, user_key)


def scores_to_gcs(assignment, user):
    """ Writes all final submission scores 
    for the given assignment to GCS csv file. """
    content = data_for_scores(assignment, user)
    csv_contents = create_csv_content(content)
    csv_filename = '/{}/{}'.format(GRADES_BUCKET, make_csv_filename(assignment, 'scores'))
    create_gcs_file(csv_filename, csv_contents, 'text/csv')


def add_subm_to_zip(subm, Submission, zipfile, result):
    """ Adds submission contents to a zipfile in-place, returns zipfile """
    try:
        if isinstance(result, Submission):
            result = result.backup.get()
        name, file_contents = subm.data_for_zip(result)
        return add_to_zip(zipfile, file_contents, name)
    except BadValueError as e:
        if str(e) != 'Submission has no contents to download':
            raise e


def make_zip_filename(user, now):
    """ Makes zip filename: query_USER EMAIL_DATETIME.zip """
    outlawed = [' ', '.', ':', '@']
    filename = '/{}/{}'.format(
        GRADES_BUCKET, 
        '%s_%s_%s' % (
            'query', 
            user.email[0], 
            str(now)))
    for outlaw in outlawed:
        filename = filename.replace(outlaw, '-')
    return filename+'.zip'


def subms_to_gcs(SearchAPI, SubmissionAPI, Submission, user, data, datetime):
    """
    Writes all submissions for a given search query
    to a GCS zip file.
    """
    results = SearchAPI.results(data)
    zipfile_str, zipfile = start_zip()
    subm = SubmissionAPI()
    for result in results:
        zipfile = add_subm_to_zip(subm, Submission, zipfile, result)
    zip_contents = finish_zip(zipfile_str, zipfile)
    zip_filename = make_zip_filename(user, datetime)
    create_gcs_file(zip_filename, zip_contents, 'application/zip')<commit_msg>all this... forgot to close<commit_after>"""
Utility functions used by API and other services
"""

# pylint: disable=no-member

import collections
import logging
import datetime
import itertools
from os.path import join

try:
    from cStringIO import StringIO
except:
    from StringIO import StringIO
import zipfile as zf
import csv
from flask import jsonify, request, Response, json

from google.appengine.api import memcache
from google.appengine.ext import ndb
from google.appengine.ext import deferred
import cloudstorage as gcs

from app import app
from app.constants import GRADES_BUCKET
from app.exceptions import BadValueError

# TODO Looks like this can be removed just by relocating parse_date
# To deal with circular imports
class ModelProxy(object):
    def __getattribute__(self, key):
        import app
        return app.models.__getattribute__(key)

ModelProxy = ModelProxy()

def parse_date(date):
    # TODO Describe what date translation is happening here. Probably needs
    # a rewrite to handle daylight savings and work with other time zones.
    try:
        date = datetime.datetime.strptime(
            date, app.config["GAE_DATETIME_FORMAT"])
    except ValueError:
        date = datetime.datetime.strptime(
            date, "%Y-%m-%d %H:%M:%S")

    delta = datetime.timedelta(hours=7)
    return datetime.datetime.combine(date.date(), date.time()) + delta

def coerce_to_json(data, fields):
    """
    Coerces |data| to json, using only the allowed |fields|
    """
    if hasattr(data, 'to_json'):
        return data.to_json(fields)
    elif isinstance(data, list):
        return [mdl.to_json(fields) if hasattr(mdl, 'to_json')
                else coerce_to_json(mdl, fields) for mdl in data]
    elif isinstance(data, dict):
        if hasattr(data, 'to_json'):
            return {
                k: mdl.to_json(fields.get(k, {}))
                for k, mdl in data.iteritems()}
        else:
            return {k: coerce_to_json(mdl, fields.get(k, {}))
                    for k, mdl in data.iteritems()}
    else:
        return data

#TODO(martinis) somehow having data be an empty list doesn't make it
# return an empty list, but an empty object.
def create_api_response(status, message, data=None):
    """Creates a JSON response that contains status code (HTTP),
    an arbitrary message string, and a dictionary or list of data"""
    if isinstance(data, dict) and 'results' in data:
        data['results'] = (
            coerce_to_json(data['results'], request.fields.get('fields', {})))
    else:
        data = coerce_to_json(data, request.fields.get('fields', {}))

    if request.args.get('format', 'default') == 'raw':
        response = Response(json.dumps(data))
    else:
        response = jsonify(**{
            'status': status,
            'message': message,
            'data': data
        })
    response.status_code = status
    return response


def create_zip(file_contents={}, dir=''):
    return finish_zip(*start_zip(file_contents, dir))


def start_zip(file_contents={}, dir=''):
    """
    Creates a file from the given dictionary of filenames to contents.
    Uses specified dir to store all files.
    """
    zipfile_str = StringIO()
    zipfile = zf.ZipFile(zipfile_str, 'w')
    zipfile = add_to_zip(zipfile, file_contents, dir)
    return zipfile_str, zipfile


def finish_zip(zipfile_str, zipfile):
    zipfile.close()
    return zipfile_str.getvalue()


def add_to_zip(zipfile, file_contents, dir=''):
    """
    Adds files to a given zip file. Uses specified dir to store files.
    """
    for filename, contents in file_contents.items():
        zipfile.writestr(join(dir, filename), contents)
    return zipfile

def create_csv_content(content):
    """ 
    Return all contents in CSV file format. Content must be a list of lists.
    """
    scsv = StringIO()
    writer = csv.writer(scsv)
    try:
        writer.writerows(content)
    except csv.Error as e:
        scsv.close()
        sys.exit('Error creating CSV: {}'.format(e))
    contents = scsv.getvalue()
    scsv.close()
    return contents

def data_for_scores(assignment, user):
    """ 
    Returns a tuple of two values a list of lists of score info for assignment.
    Format: [['STUDENT', 'SCORE', 'MESSAGE', 'GRADER', 'TAG']] 
    """
    content = [['STUDENT', 'SCORE', 'MESSAGE', 'GRADER', 'TAG']]
    course = assignment.course.get()
    groups = ModelProxy.Group.lookup_by_assignment(assignment)
    seen_members = set()

    for group in groups:
        members = group.member
        seen_members |= set(members)
        content.extend(group.scores_for_assignment(assignment))

    students = [part.user.get() for part in course.get_students(user) if part.user not in seen_members]
    for student in students:
        content.extend(student.scores_for_assignment(assignment)[0])

    return content

def create_gcs_file(gcs_filename, contents, content_type):
    """ 
    Creates a GCS csv file with contents CONTENTS. 
    """
    try:
        gcs_file = gcs.open(gcs_filename, 'w', content_type=content_type, options={'x-goog-acl': 'project-private'})
        gcs_file.write(contents)
        gcs_file.close()
    except Exception as e:
        logging.exception("ERROR: {}".format(e))
        try:
            gcs.delete(gcs_filename)
        except gcs.NotFoundError:
            logging.info("Could not delete file " + gcs_filename)
    logging.info("Created file " + gcs_filename)


def make_csv_filename(assignment, infotype):
    """ Returns filename of format INFOTYPE_COURSE_ASSIGNMENT.csv """
    course_name = assignment.course.get().offering
    assign_name = assignment.display_name
    filename = '{}_{}_{}.csv'.format(infotype, course_name, assign_name)
    return filename.replace('/', '_').replace(' ', '_')

def paginate(entries, page, num_per_page):
    """
    Added stuff from
    https://p.ota.to/blog/2013/4/pagination-with-cursors-
        in-the-app-engine-datastore/

    Support pagination for an NDB query.
    Arguments:
      |entries| - a query which returns the items to paginate over.
      |cursor|  - a cursor of where in the pagination the user is.
      |num_per_page| - the number of results to display per page.

    The return value will be different from a regular query:
    There will be 3 things returned:
    - results: a list of results
    - forward_curs: a urlsafe hash for the cursor. Use this to get
    the next page. To retrieve the cursor object, do Cursor(urlsafe=s)
    - more: a boolean for whether or not there is more content.

    For more documentation, look at:
    https://developers.google.com/appengine/docs/python/ndb/queryclass#Query_fetch_page
    """
    if num_per_page is None:
        return {
            'results': entries.fetch(),
            'page': 1,
            'more': False
        }

    query_serialized = (
        '_'.join(str(x) for x in (
            entries.kind, entries.filters, entries.orders)))
    query_serialized = query_serialized.replace(' ', '_')
    def get_mem_key(page):
        offset = (page - 1) * num_per_page
        return "cp_%s_%s" % (query_serialized, offset)
    this_page_key = get_mem_key(page)
    next_page_key = get_mem_key(page + 1)

    cursor = None
    store_cache = True
    if page > 1:
        cursor = memcache.get(this_page_key)
        if not cursor:
            page = 1 # Reset to the front, since memcached failed
            store_cache = False

    pages_to_fetch = int(num_per_page)
    if cursor is not None:
        results, forward_cursor, more = entries.fetch_page(
            pages_to_fetch, start_cursor=cursor)
    else:
        results, forward_cursor, more = entries.fetch_page(pages_to_fetch)

    if store_cache:
        memcache.set(next_page_key, forward_cursor)

    return {
        'results': results,
        'page': page,
        'more': more
    }


def _apply_filter(query, model, arg, value, op):
    """
    Applies a filter on |model| of |arg| |op| |value| to |query|.
    """
    if '.' in arg:
        arg = arg.split('.')
    else:
        arg = [arg]

    field = model
    while arg:
        field = getattr(field, arg.pop(0), None)
        if not field:
            # Silently swallow for now
            # TODO(martinis) cause an error
            return query

    if op == "==":
        filtered = field == value
    elif op == "<":
        filtered = field < value
    elif op == "<=":
        filtered = field <= value
    elif op == ">":
        filtered = field > value
    elif op == ">=":
        filtered = field >= value
    else:
        raise ValueError("Invalid filtering operator {}".format(op))

    return query.filter(filtered)

def filter_query(query, args, model):
    """
    Applies the filters in |args| to |query|.
    |args| is a dictionary of key to value, to be used to filter the query.
    |allowed| is an optional list of the allowed filters.

    Returns a modified query with the appropriate filters.
    """
    for arg, value in args.iteritems():
        if (isinstance(value, collections.Iterable)
                and not isinstance(value, str)):
            op, value = value
        else:
            value, op = value, '=='

        query = _apply_filter(query, model, arg, value, op)

    return query


####################
# Deferred actions #
####################

ASSIGN_BATCH_SIZE = 20
def add_to_grading_queues(assign_key, cursor=None, num_updated=0):
    query = ModelProxy.FinalSubmission.query().filter(
        ModelProxy.FinalSubmission.assignment == assign_key)

    queues = list(ModelProxy.Queue.query(
        ModelProxy.Queue.assignment == assign_key))
    if not queues:
        logging.error("Tried to assign work, but no queues existed")
        return

    kwargs = {}

    if cursor:
        kwargs['start_cursor'] = cursor

    to_put = 0
    results, cursor, _ = query.fetch_page(ASSIGN_BATCH_SIZE, **kwargs)
    seen = set()
    for queue in queues:
        for subm in queue.submissions:
            if isinstance(subm, ndb.Key):
                seen.add(subm.get().submitter.id())
            else:
                seen.add(subm.submitter.id())

    for subm in results:
        user = subm.submitter.get()
        if not user.logged_in or user.key.id() in seen:
            continue
        queues.sort(key=lambda x: len(x.submissions))

        subm = user.get_selected_submission(assign_key, keys_only=True)
        if subm and user.is_final_submission(subm, assign_key):
            subm_got = subm.get()
            if subm_got.get_messages().get('file_contents'):
                queues[0].submissions.append(subm)
                seen.add(user.key.id())
                to_put += 1

    if to_put:
        num_updated += to_put
        ndb.put_multi(queues)
        logging.debug(
            'Put %d entities to Datastore for a total of %d',
            to_put, num_updated)
        deferred.defer(
            add_to_grading_queues, assign_key, cursor=cursor,
            num_updated=num_updated)
    else:
        logging.debug(
            'add_to_grading_queues complete with %d updates!', num_updated)

def assign_submission(backup_id, submit):
    """
    Create Submisson and FinalSubmission records for a submitted Backup.

    :param backup_id: ID of a Backup
    :param submit: Whether this backup is a submission to be graded
    """
    backup = ModelProxy.Backup.get_by_id(backup_id)
    if not backup.get_messages().get('file_contents'):
        logging.info("Submission had no file_contents; not processing")
        return

    if submit:
        assign = backup.assignment.get_async()
        subm = ModelProxy.Submission(backup=backup.key)
        subm.put()

        # Can only make a final submission before it's due
        if datetime.datetime.now() < assign.get_result().due_date:
            subm.mark_as_final()

def sort_by_assignment(key_func, entries):
    entries = sorted(entries, key=key_func)
    return itertools.groupby(entries, key_func)

@ndb.toplevel
def merge_user(user_key, dup_user_key):
    """
    Merges |dup_user| into |user|.
    """
    if isinstance(user_key, ModelProxy.Base):
        user = user_key
        user_key = user_key.key
        get_user = lambda: user
    else:
        user = user_key.get_async()
        def get_user():
            return user.get_result()

    if isinstance(dup_user_key, ModelProxy.Base):
        dup_user = dup_user_key
        get_dup_user = lambda: dup_user
        dup_user_key = dup_user_key.key
    else:
        dup_user = dup_user_key.get_async()
        def get_dup_user():
            return dup_user.get_result()

    # Leave all groups
    G = ModelProxy.Group
    groups = G.query(ndb.OR(
        G.member == dup_user_key,
        G.invited == dup_user_key)).fetch()
    for group in groups:
        group.exit(dup_user_key)
        
    # Deactivate all enrollments
    E = ModelProxy.Participant
    enrolls = E.query(E.user == dup_user_key).fetch()
    for enroll in enrolls:
        # enroll.status = 'inactive'
        enroll.put_async()

    # Re-submit submissions
    S = ModelProxy.Submission
    subms = S.query(S.submitter == dup_user_key).fetch()
    for subm in subms:
        subm.resubmit(user_key)

    dup_user = get_dup_user()
    # Change email

    user = get_user()
    lowered_emails = [email.lower() for email in user.email]
    for email in dup_user.email:
        if email.lower() not in lowered_emails:
            user.email.append(email.lower())

    # Invalidate emails
    dup_user.email = ['#'+email for email in dup_user.email]
    # dup_user.status = 'inactive'
    dup_user.put_async()
    user.put_async()

    log = ModelProxy.AuditLog()
    log.event_type = "Merge user"
    log.user = user_key
    log.description = "Merged user {} with {}. Merged emails {}".format(
        dup_user_key.id(), user_key.id(), dup_user.email)
    log.obj = dup_user_key
    log.put_async()

def unique_email_address(user):
    U = ModelProxy.User

    dups = []
    for email in user.email:
        users = U.query(U.email == email).fetch()
        for found_user in users:
            if found_user.key != user.key:
                dups.append((user, found_user))

    for usera, userb in dups:
        if usera.email[0].lower() != usera.email[0]:
            user, dup_user = usera, userb
        else:
            user, dup_user = userb, usera

        merge_user(user, dup_user)

def unique_final_submission(user):
    FS = ModelProxy.FinalSubmission

    key_func = lambda subm: subm.assignment
    submissions = FS.query(FS.submitter == user.key).fetch()
    for lst in sort_by_assignment(key_func, submissions):
        if len(lst) > 1:
            lst = sorted(lst, key=lambda subm: subm.server_time)[1:]
            for subm in lst:
                subm.key.delete()

def unique_group(user):
    G = ModelProxy.Group
    key_func = lambda group: group.assignment
    groups = G.query(G.member == user.key).fetch()
    for lst in sort_by_assignment(key_func, groups):
        if len(lst) > 1:
            # TODO(martinis, denero) figure out what to do
            pass

def deferred_check_user(user_id):
    user = ModelProxy.User.get_by_id(user_id)
    if not user:
        raise deferred.PermanentTaskFailure("User id {} is invalid.".format(user_id))

    unique_email_address(user)
    unique_final_submission(user)
    unique_group(user)


def check_user(user_key):
    if isinstance(user_key, ModelProxy.User):
        user_key = user_key.key.id()

    if isinstance(user_key, ndb.Key):
        user_key = user_key.id()

    deferred.defer(deferred_check_user, user_key)


def scores_to_gcs(assignment, user):
    """ Writes all final submission scores 
    for the given assignment to GCS csv file. """
    content = data_for_scores(assignment, user)
    csv_contents = create_csv_content(content)
    csv_filename = '/{}/{}'.format(GRADES_BUCKET, make_csv_filename(assignment, 'scores'))
    create_gcs_file(csv_filename, csv_contents, 'text/csv')


def add_subm_to_zip(subm, Submission, zipfile, result):
    """ Adds submission contents to a zipfile in-place, returns zipfile """
    try:
        if isinstance(result, Submission):
            result = result.backup.get()
        name, file_contents = subm.data_for_zip(result)
        return add_to_zip(zipfile, file_contents, name)
    except BadValueError as e:
        if str(e) != 'Submission has no contents to download':
            raise e


def make_zip_filename(user, now):
    """ Makes zip filename: query_USER EMAIL_DATETIME.zip """
    outlawed = [' ', '.', ':', '@']
    filename = '/{}/{}'.format(
        GRADES_BUCKET, 
        '%s_%s_%s' % (
            'query', 
            user.email[0], 
            str(now)))
    for outlaw in outlawed:
        filename = filename.replace(outlaw, '-')
    return filename+'.zip'


def subms_to_gcs(SearchAPI, SubmissionAPI, Submission, user, data, datetime):
    """
    Writes all submissions for a given search query
    to a GCS zip file.
    """
    results = SearchAPI.results(data)
    zipfile_str, zipfile = start_zip()
    subm = SubmissionAPI()
    for result in results:
        zipfile = add_subm_to_zip(subm, Submission, zipfile, result)
    zip_contents = finish_zip(zipfile_str, zipfile)
    zip_filename = make_zip_filename(user, datetime)
    create_gcs_file(zip_filename, zip_contents, 'application/zip')
{'fraction_non_alphanumeric': 0.10863927151172843, 'fraction_numerical': 0.0014984727105065991, 'mean_word_length': 2.6606540084388186, 'pattern_counts': {'":': 10, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2715126', 'n_tokens_mistral': 10887, 'n_tokens_neox': 10674, 'n_words': 2882}