<commit_before>from functools import partial
import simplejson as json
import os
import tempfile

from celery.exceptions import RetryTaskError
from celery.task import task
import pandas as pd

from bamboo.lib.async import call_async
from bamboo.lib.datetools import recognize_dates
from bamboo.lib.schema_builder import filter_schema


@task(ignore_result=True)
def import_dataset(dataset, file_reader, delete=False):
    """For reading a URL and saving the corresponding dataset.

    Import a DataFrame using the provided `file_reader` function. All
    exceptions are caught and on exception the dataset is marked as failed and
    set for deletion after 24 hours.

    :param dataset: The dataset to import into.
    :param file_reader: Function for reading the dataset.
    :param delete: Delete filepath_or_buffer after import, default False.
    """
    try:
        dframe = file_reader()
        dataset.save_observations(dframe)
    except Exception as e:
        if isinstance(e, RetryTaskError):
            raise e
        else:
            dataset.failed(e.__str__())
            dataset.delete(countdown=86400)


def csv_file_reader(name, na_values=[], delete=False):
    try:
        return recognize_dates(
            pd.read_csv(name, encoding='utf-8', na_values=na_values))
    finally:
        if delete:
            os.unlink(name)


def json_file_reader(content):
    return recognize_dates(json.loads(content))

<commit_msg>wrap json load in df<commit_after>from functools import partial
import simplejson as json
import os
import tempfile

from celery.exceptions import RetryTaskError
from celery.task import task
import pandas as pd

from bamboo.lib.async import call_async
from bamboo.lib.datetools import recognize_dates
from bamboo.lib.schema_builder import filter_schema


@task(ignore_result=True)
def import_dataset(dataset, file_reader, delete=False):
    """For reading a URL and saving the corresponding dataset.

    Import a DataFrame using the provided `file_reader` function. All
    exceptions are caught and on exception the dataset is marked as failed and
    set for deletion after 24 hours.

    :param dataset: The dataset to import into.
    :param file_reader: Function for reading the dataset.
    :param delete: Delete filepath_or_buffer after import, default False.
    """
    try:
        dframe = file_reader()
        dataset.save_observations(dframe)
    except Exception as e:
        if isinstance(e, RetryTaskError):
            raise e
        else:
            dataset.failed(e.__str__())
            dataset.delete(countdown=86400)


def csv_file_reader(name, na_values=[], delete=False):
    try:
        return recognize_dates(
            pd.read_csv(name, encoding='utf-8', na_values=na_values))
    finally:
        if delete:
            os.unlink(name)


def json_file_reader(content):
    return recognize_dates(pd.DataFrame(json.loads(content)))


{'fraction_non_alphanumeric': 0.06959204662324306, 'fraction_numerical': 0.005485087418580734, 'mean_word_length': 3.394578313253012, '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': '5791984', 'n_tokens_mistral': 830, 'n_tokens_neox': 798, 'n_words': 272}