<commit_before>import json
import requests
import itertools
import tempfile
import datetime

import ckanserviceprovider.job as job
import ckanserviceprovider.util as util
import dataconverters as dc


TYPE_MAPPING = {
    'String': 'text',
    'Integer': 'numeric',  # 'int' may not be big enough,
                    # and type detection may not realize it needs to be big
    'Float': 'float',
    'Decimal': 'numeric',
    'DateTime': 'timestamp'
}


def check_response(response, datastore_create_request_url):
    if not response.status_code:
        raise util.JobError('Datastore is not reponding at %s with '
                'response %s' % (datastore_create_request_url, response))

    if response.status_code not in (201, 200):
        try:
            response.content = json.loads(response.content)
            raise util.JobError('Datastorer bad response code (%s) on %s. Response was %s' %
                (response.status_code, datastore_create_request_url, response.content))
        except:
            raise util.JobError('Datastorer bad response code (%s) on %s.' %
                (response.status_code, datastore_create_request_url))


# generates chunks of data that can be loaded into ckan
# n is the maximum size of a chunk
def chunky(iterable, n):
    it = iter(iterable)
    while True:
        chunk = list(
            itertools.imap(
                dict, itertools.islice(it, n)))
        if not chunk:
            return
        yield chunk


class DatastoreEncoder(json.JSONEncoder):

    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()

        return json.JSONEncoder.default(self, obj)


def check_provided_data(data):
    if not 'url' in data:
        raise util.JobError("Did not provide URL to resource.")


@job.async
def import_into_datastore(task_id, input):
    """
    Expected input dictionary with keys:
        'url'
    """

    data = input['data']
    check_provided_data(data)

    ckan_url = data['ckan_url'].rstrip('/')
    datastore_create_request_url = '%s/api/action/datastore_create' % (ckan_url)

    excel_types = ['xls', 'application/ms-excel', 'application/xls',
                   'application/vnd.ms-excel']
    excel_xml_types = ['xlsx']
    tsv_types = ['tsv', 'text/tsv', 'text/tab-separated-values']
    csv_types = ['csv', 'text/csv', 'text/comma-separated-values']

    requested_resource = requests.get(data['url'])

    content_type = requested_resource.headers['content-type']\
                                    .split(';', 1)[0]  # remove parameters

    def is_of_type(types):
        return content_type in types or data['format'] in types

    with tempfile.NamedTemporaryFile() as f:
        f.write(requested_resource.text)
        f.flush()
        f.seek(0)

        if is_of_type(excel_types):
            result, metadata = dc.xls.xls_parse(f, header_type=1)
        elif is_of_type(excel_xml_types):
            pass
        elif is_of_type(csv_types):
            result, metadata = dc.csv.csv_parse(f, header_type=1)
        elif is_of_type(tsv_types):
            pass

        headers = [dict(id=field['id'], type=TYPE_MAPPING.get(field['type'])) for field in metadata['fields']]
        print 'headers', headers
        print 'result', result

        def send_request(records):
            request = {'resource_id': data['resource_id'],
                       'fields': headers,
                       'records': records}
            response = requests.post(datastore_create_request_url,
                             data=json.dumps(request, cls = DatastoreEncoder),
                             headers={'Content-Type': 'application/json',
                                      'Authorization': input['apikey']},
                             )
            check_response(response, datastore_create_request_url)

        #logger.info('Creating: {0}.'.format(resource['id']))

        count = 0
        for records in chunky(result, 100):
            count += len(records)
            send_request(records)

        #from nose.tools import set_trace; set_trace()

        #logger.info("There should be {n} entries in {res_id}.".format(n=count, res_id=resource['id']))<commit_msg>code simplification and adaptions to latest data converters<commit_after>import json
import requests
import itertools
import tempfile
import datetime

import ckanserviceprovider.job as job
import ckanserviceprovider.util as util
import dataconverters as dc


TYPE_MAPPING = {
    'String': 'text',
    'Integer': 'numeric',  # 'int' may not be big enough,
                    # and type detection may not realize it needs to be big
    'Float': 'float',
    'Decimal': 'numeric',
    'DateTime': 'timestamp'
}


def check_response(response, datastore_create_request_url):
    if not response.status_code:
        raise util.JobError('Datastore is not reponding at %s with '
                'response %s' % (datastore_create_request_url, response))

    if response.status_code not in (201, 200):
        try:
            response.content = json.loads(response.content)
            raise util.JobError('Datastorer bad response code (%s) on %s. Response was %s' %
                (response.status_code, datastore_create_request_url, response.content))
        except:
            raise util.JobError('Datastorer bad response code (%s) on %s.' %
                (response.status_code, datastore_create_request_url))


# generates chunks of data that can be loaded into ckan
# n is the maximum size of a chunk
def chunky(iterable, n):
    it = iter(iterable)
    while True:
        chunk = list(
            itertools.imap(
                dict, itertools.islice(it, n)))
        if not chunk:
            return
        yield chunk


class DatastoreEncoder(json.JSONEncoder):

    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()

        return json.JSONEncoder.default(self, obj)


def check_provided_data(data):
    if not 'url' in data:
        raise util.JobError("Did not provide URL to resource.")


@job.async
def import_into_datastore(task_id, input):
    """
    Expected input dictionary with keys:
        'url'
    """

    data = input['data']
    check_provided_data(data)

    ckan_url = data['ckan_url'].rstrip('/')
    datastore_create_request_url = '%s/api/action/datastore_create' % (ckan_url)

    excel_types = ['xls', 'application/ms-excel', 'application/xls',
                   'application/vnd.ms-excel']
    excel_xml_types = ['xlsx']
    tsv_types = ['tsv', 'text/tsv', 'text/tab-separated-values']
    csv_types = ['csv', 'text/csv', 'text/comma-separated-values']

    requested_resource = requests.get(data['url'])

    content_type = requested_resource.headers['content-type']\
                                    .split(';', 1)[0]  # remove parameters

    def is_of_type(types):
        return content_type in types or data['format'] in types

    with tempfile.NamedTemporaryFile() as f:
        f.write(requested_resource.text)
        f.flush()
        f.seek(0)

        parser = None
        if is_of_type(excel_types):
            parser = dc.xls
        elif is_of_type(excel_xml_types):
            pass
        elif is_of_type(csv_types):
            parser = dc.csv
        elif is_of_type(tsv_types):
            pass

        if parser:
            result, metadata = parser.parse(f, header_type=1)
        else:
            raise util.JobError('No parser for {} found.'.format(content_type))

        headers = [dict(id=field['id'], type=TYPE_MAPPING.get(field['type'])) for field in metadata['fields']]
        print 'headers', headers
        print 'result', result

        def send_request(records):
            request = {'resource_id': data['resource_id'],
                       'fields': headers,
                       'records': records}
            response = requests.post(datastore_create_request_url,
                             data=json.dumps(request, cls=DatastoreEncoder),
                             headers={'Content-Type': 'application/json',
                                      'Authorization': input['apikey']},
                             )
            check_response(response, datastore_create_request_url)

        #logger.info('Creating: {0}.'.format(resource['id']))

        count = 0
        for records in chunky(result, 100):
            count += len(records)
            send_request(records)

        #from nose.tools import set_trace; set_trace()

        #logger.info("There should be {n} entries in {res_id}.".format(n=count, res_id=resource['id']))
{'fraction_non_alphanumeric': 0.11078886310904873, 'fraction_numerical': 0.0035962877030162412, 'mean_word_length': 2.3623244929797194, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4985156', 'n_tokens_mistral': 2572, 'n_tokens_neox': 2488, 'n_words': 615}