<commit_before># -*- coding: utf-8 -*-
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import division

import json

from celery import task, chord
from django.core.exceptions import ObjectDoesNotExist
from django.conf import settings
from django.db import transaction

from importer.models.base import GenericImportEvent, GenericImportRow
from importer.models.species import SpeciesImportEvent, SpeciesImportRow
from importer.models.trees import TreeImportEvent, TreeImportRow
from importer import errors
from importer.util import (clean_row_data, clean_field_name,
                           utf8_file_to_csv_dictreader)


def _create_rows_for_event(ie, csv_file):
    # Don't use a transaction for this possibly long-running operation
    # so we can show progress. Caller does manual cleanup if necessary.
    reader = utf8_file_to_csv_dictreader(csv_file)

    field_names = reader.fieldnames
    ie.field_order = json.dumps(field_names)
    ie.save()

    field_names = [clean_field_name(f) for f in field_names]
    file_valid = ie.validate_field_names(field_names)

    if file_valid:
        _create_rows(ie, reader)

        if ie.row_count == 0:
            file_valid = False
            ie.append_error(errors.EMPTY_FILE)

    if file_valid:
        return True
    else:
        ie.status = ie.FAILED_FILE_VERIFICATION
        ie.save()
        return False


def _create_rows(ie, reader):
    RowModel = get_import_row_model(ie.import_type)
    rows = []
    idx = 0

    for row in reader:
        data = json.dumps(clean_row_data(row))
        rows.append(RowModel(data=data, import_event=ie, idx=idx))

        idx += 1
        if ((int(idx / settings.IMPORT_BATCH_SIZE) *
             settings.IMPORT_BATCH_SIZE == idx)):
            RowModel.objects.bulk_create(rows)
            rows = []

    if rows:
        RowModel.objects.bulk_create(rows)  # create final partial block


@task()
def run_import_event_validation(import_type, import_event_id, file_obj):
    ie = _get_import_event(import_type, import_event_id)

    try:
        ie.status = GenericImportEvent.LOADING
        ie.save()
        success = _create_rows_for_event(ie, file_obj)
    except Exception as e:
        ie.append_error(errors.GENERIC_ERROR, data=[str(e)])
        ie.status = GenericImportEvent.FAILED_FILE_VERIFICATION
        ie.save()
        success = False

    if not success:
        try:
            ie.row_set().delete()
        except Exception:
            pass
        return

    ie.status = GenericImportEvent.PREPARING_VERIFICATION
    ie.save()

    try:
        row_set = ie.rows()
        validation_tasks = (
            _validate_rows.subtask(row_set[i:(i+settings.IMPORT_BATCH_SIZE)])
            for i in xrange(0, ie.row_count, settings.IMPORT_BATCH_SIZE))

        final_task = _finalize_validation.si(import_type, import_event_id)

        async_result = chord(validation_tasks, final_task).delay()
        group_result = async_result.parent
        if group_result:  # Has value None when run in unit tests
            group_result.save()
            ie.task_id = group_result.id

        ie.status = GenericImportEvent.VERIFIYING
        ie.save()
    except Exception as e:
        ie.status = GenericImportEvent.VERIFICATION_ERROR
        ie.save()
        try:
            ie.append_error(errors.GENERIC_ERROR, data=[str(e)])
            ie.save()
            ie.row_set().delete()
        except Exception:
            pass
        return


@task()
def _validate_rows(*rows):
    for row in rows:
        row.validate_row()


@task()
def _finalize_validation(import_type, import_event_id):
    ie = _get_import_event(import_type, import_event_id)

    ie.task_id = ''
    # There shouldn't be any rows left to verify, but it doesn't hurt to check
    if _get_waiting_row_count(ie) == 0:
        ie.status = GenericImportEvent.FINISHED_VERIFICATION

    ie.save()


@task()
def commit_import_event(import_type, import_event_id):
    ie = _get_import_event(import_type, import_event_id)

    commit_tasks = [
        _commit_rows.s(import_type, import_event_id, i)
        for i in xrange(0, ie.row_count, settings.IMPORT_BATCH_SIZE)]
<commit_msg>Fix race condition in celery subtasks<commit_after># -*- coding: utf-8 -*-
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import division

import json

from celery import task, chord
from django.core.exceptions import ObjectDoesNotExist
from django.conf import settings
from django.db import transaction

from importer.models.base import GenericImportEvent, GenericImportRow
from importer.models.species import SpeciesImportEvent, SpeciesImportRow
from importer.models.trees import TreeImportEvent, TreeImportRow
from importer import errors
from importer.util import (clean_row_data, clean_field_name,
                           utf8_file_to_csv_dictreader)


def _create_rows_for_event(ie, csv_file):
    # Don't use a transaction for this possibly long-running operation
    # so we can show progress. Caller does manual cleanup if necessary.
    reader = utf8_file_to_csv_dictreader(csv_file)

    field_names = reader.fieldnames
    ie.field_order = json.dumps(field_names)
    ie.save()

    field_names = [clean_field_name(f) for f in field_names]
    file_valid = ie.validate_field_names(field_names)

    if file_valid:
        _create_rows(ie, reader)

        if ie.row_count == 0:
            file_valid = False
            ie.append_error(errors.EMPTY_FILE)

    if file_valid:
        return True
    else:
        ie.status = ie.FAILED_FILE_VERIFICATION
        ie.save()
        return False


def _create_rows(ie, reader):
    RowModel = get_import_row_model(ie.import_type)
    rows = []
    idx = 0

    for row in reader:
        data = json.dumps(clean_row_data(row))
        rows.append(RowModel(data=data, import_event=ie, idx=idx))

        idx += 1
        if ((int(idx / settings.IMPORT_BATCH_SIZE) *
             settings.IMPORT_BATCH_SIZE == idx)):
            RowModel.objects.bulk_create(rows)
            rows = []

    if rows:
        RowModel.objects.bulk_create(rows)  # create final partial block


@task()
def run_import_event_validation(import_type, import_event_id, file_obj):
    ie = _get_import_event(import_type, import_event_id)

    try:
        ie.status = GenericImportEvent.LOADING
        ie.save()
        success = _create_rows_for_event(ie, file_obj)
    except Exception as e:
        ie.append_error(errors.GENERIC_ERROR, data=[str(e)])
        ie.status = GenericImportEvent.FAILED_FILE_VERIFICATION
        ie.save()
        success = False

    if not success:
        try:
            ie.row_set().delete()
        except Exception:
            pass
        return

    ie.status = GenericImportEvent.PREPARING_VERIFICATION
    ie.save()

    try:
        validation_tasks = []
        for i in xrange(0, ie.row_count, settings.IMPORT_BATCH_SIZE):
            validation_tasks.append(_validate_rows.s(import_type, ie.id, i))

        final_task = _finalize_validation.si(import_type, import_event_id)

        async_result = chord(validation_tasks, final_task).delay()
        group_result = async_result.parent
        if group_result:  # Has value None when run in unit tests
            group_result.save()
            ie.task_id = group_result.id

        ie.status = GenericImportEvent.VERIFIYING
        ie.save()
    except Exception as e:
        ie.status = GenericImportEvent.VERIFICATION_ERROR
        ie.save()
        try:
            ie.append_error(errors.GENERIC_ERROR, data=[str(e)])
            ie.save()
            ie.row_set().delete()
        except Exception:
            pass
        return


@task()
def _validate_rows(import_type, import_event_id, start_row_id):
    ie = _get_import_event(import_type, import_event_id)
    rows = ie.rows()[start_row_id:(start_row_id+settings.IMPORT_BATCH_SIZE)]
    for row in rows:
        row.validate_row()


@task()
def _finalize_validation(import_type, import_event_id):
    ie = _get_import_event(import_type, import_event_id)

    ie.task_id = ''
    # There shouldn't be any rows left to verify, but it doesn't hurt to check
    if _get_waiting_row_count(ie) == 0:
        ie.status = GenericImportEvent.FINISHED_VERIFICATION

    ie.save()


@task()
def commit_import_event(import_type, import_event_id):
    ie = _get_import_event(import_type, import_event_id)

    commit_tasks = [
        _commit_rows.s(import_type, import_event_id, i)
        for i in xrange(0, ie.row_count, settings.IMPORT_BATCH_SIZE)]

{'fraction_non_alphanumeric': 0.0811156304474143, 'fraction_numerical': 0.002091807088901801, 'mean_word_length': 2.8941176470588235, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1438214', 'n_tokens_mistral': 2704, 'n_tokens_neox': 2634, 'n_words': 596}