<commit_before>'''
Load data from /columns and /tables into elasticsearch using /mappings
'''

from luigi import Task, Parameter
from elasticsearch.helpers import bulk
import elasticsearch
import ujson as json
import os
import sys

from tasks.util import elastic_conn, get_logger


class Index(Task):

    destructive = Parameter(default=True)

    LOGGER = get_logger(__name__)
    ES_NAME = 'bigmetadata'
    ES_COLUMN = 'column'
    ES_TABLE = 'table'

    CACHE = {
        ES_COLUMN: {},
        'NAMES': set(),  # TODO cleaner way to not include duplicate names
        'CNT': 0
    }


    def index_column(self, path):
        '''
        Load json at path into idx.
        '''
        column_id = path
        #if econn.exists(ES_NAME, ES_COLUMN, id=column_id):
        #    return
        with open(path) as column_file:
            column = json.load(column_file)
        column_with_id = column.copy()
        column_with_id['id'] = column_id
        column_with_id['tables'] = []
        if int(column.get('weight', 0)) > 0 and column['name'] not in self.CACHE['NAMES']:
            self.CACHE['NAMES'].add(column['name'])
            self.CACHE[self.ES_COLUMN][column_id] = column_with_id
        #econn.index(ES_NAME, ES_COLUMN, id=column_id, body=body)
        #return [{
        #    '_op_type': 'index',
        #    '_index': ES_NAME,
        #    '_type': ES_COLUMN,
        #    '_id': column_id,
        #    '_source': column
        #}]


    def index_table(self, path):
        '''
        Add tables to columns.
        '''
        ops = []
        table_id = path
        with open(path) as table_file:
            table = json.load(table_file)

        # Replace real column with just an ID
        columns = table.pop('columns')
        table_for_column = table.copy()

        table['columns'] = [c['id'] for c in columns]
        table_for_column['id'] = table_id<commit_msg>prevent low-weight columns from overriding high-value ones<commit_after>'''
Load data from /columns and /tables into elasticsearch using /mappings
'''

from luigi import Task, Parameter
from elasticsearch.helpers import bulk
import elasticsearch
import ujson as json
import os
import sys

from tasks.util import elastic_conn, get_logger


class Index(Task):

    destructive = Parameter(default=True)

    LOGGER = get_logger(__name__)
    ES_NAME = 'bigmetadata'
    ES_COLUMN = 'column'
    ES_TABLE = 'table'

    CACHE = {
        ES_COLUMN: {},
        'NAMES': set(),  # TODO cleaner way to not include duplicate names
        'CNT': 0
    }


    def index_column(self, path):
        '''
        Load json at path into idx.
        '''
        column_id = path
        #if econn.exists(ES_NAME, ES_COLUMN, id=column_id):
        #    return
        with open(path) as column_file:
            column = json.load(column_file)
        column_with_id = column.copy()
        column_with_id['id'] = column_id
        column_with_id['tables'] = []
        if int(column.get('weight', 0)) > 2 and column['name'] not in self.CACHE['NAMES']:
            self.CACHE['NAMES'].add(column['name'])
            self.CACHE[self.ES_COLUMN][column_id] = column_with_id
        #econn.index(ES_NAME, ES_COLUMN, id=column_id, body=body)
        #return [{
        #    '_op_type': 'index',
        #    '_index': ES_NAME,
        #    '_type': ES_COLUMN,
        #    '_id': column_id,
        #    '_source': column
        #}]


    def index_table(self, path):
        '''
        Add tables to columns.
        '''
        ops = []
        table_id = path
        with open(path) as table_file:
            table = json.load(table_file)

        # Replace real column with just an ID
        columns = table.pop('columns')
        table_for_column = table.copy()

        table['columns'] = [c['id'] for c in columns]
        table_for_column['id'] = table_id
{'fraction_non_alphanumeric': 0.11655011655011654, 'fraction_numerical': 0.001554001554001554, 'mean_word_length': 2.253580454928391, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '7439835', 'n_tokens_mistral': 1203, 'n_tokens_neox': 1180, 'n_words': 305}