<commit_before>#!/usr/bin/env python
# coding: utf-8

# A Repeatable Extract-Tranform-Load Pipeline for NYPL Menus Data

# authors: <NAME> and <NAME>
# created: 17 October 2014
# updated: 28 October 2014, 19 November 2014, 3 December 2014,
#     14 December 2014, 8-9 January 2015

from collections import Counter
import datetime
import elasticsearch
from elasticsearch import helpers
import json
import logging
import logging.handlers
import os
import pandas as pd
import pytz
import re
import sys
import tarfile
import time


#TODO: Give option to set up or turn off at cmd line
LOG_FILENAME = os.path.join(os.environ['MENUS_LOG_HOME'],
                            'nypl_menus_data_transform.log')

PIPELINE_LOGGER = logging.getLogger('MenusDataTransformLogger')
PIPELINE_LOGGER.setLevel(logging.INFO)
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

handler = logging.handlers.RotatingFileHandler(
    LOG_FILENAME,
    maxBytes=500000,
    backupCount=10)
handler.setFormatter(formatter)
PIPELINE_LOGGER.addHandler(handler)
UTC = pytz.utc

DOC_TOTAL = 0


def normalize_names(obj):
    '''
    Take a name as a string, converts the string
    to lowercase, strips whitespace from beginning
    and end, normalizes multiple internal whitespace
    characters to a single space. E.g.:
    
    normalize_names('Chicken gumbo ') = 'chicken gumbo'
    
    '''
    tokens = obj.strip().lower().split()
    result = ' '.join(filter(None, tokens))
    return result


def fingerprint(obj):
    """
    A modified version of the fingerprint clustering algorithm implemented by Open Refine.
    See https://github.com/OpenRefine/OpenRefine/wiki/Clustering-In-Depth
    This does not normalize to ASCII characters since diacritics may be significant in this dataset
    """
    alphanumeric_tokens = filter(None, re.split('\W', obj))
    seen = set()
    seen_add = seen.add
    deduped = sorted([i for i in alphanumeric_tokens if i not in seen and not seen_add(i)])
    fingerprint = ' '.join(deduped)
    
    return fingerprint


# TOFIX: This code is very slow
def reformat_dates(obj):
    naive_date_in = datetime.datetime.strptime(obj, '%Y-%m-%d %H:%M:%S %Z')
    date_in = naive_date_in.replace(tzinfo=UTC)
    date_out = date_in.strftime('%Y%m%dT%H%M%S%z')
    return date_out


def reshape_data(obj):
    '''
    Takes JSON, loads it, and spits out a dictionary of slightly different
    shape needed for bulk import by Elasticsearch.
    
    Input is JSON rather than simply another dictionary so I can punt on
    properly serializing things like 'NaN' by leaving that to pandas own
    to_json() function.
    '''

    data = json.loads(obj)
    action = {
              "_index": "menus",
              "_type": "item",
              "_id": int(data['item_id']),
              "_source": data
              }
    return action


def server(hostname, host_port):
    """
    Makes sure that an elasticsearch server is running.
    Sets up the appropriate index and mapping.
    """
    
    PIPELINE_LOGGER.info('Verifying Elasticsearch server is ready to receive data …')
    
    try:
        es = elasticsearch.Elasticsearch([{
            'host': hostname,
            'port': host_port}])
    except Exception as e:
        PIPELINE_LOGGER.error('Check to make sure the elasticsearch server is running at the specified host/port and try again. Exiting.')

    # TODO: Make these customizable
    index_name = 'menus'
    type_name = 'item'

    # Check if the index already exists and, if so, delete it to keep things idempotent
    if es.indices.exists(index_name):
        PIPELINE_LOGGER.info("deleting '{0}' index...".format(index_name))
        res = es.indices.delete(index = index_name)
        PIPELINE_LOGGER.info(" response: '{0}'".format(res))

    PIPELINE_LOGGER.info("creating '{0}' index...".format(index_name))
    res = es.indices.create(index = index_name)
    PIPELINE_LOGGER.info(" response: '{0}'".format(res))

    item_mapping = {
    "dynamic":      "strict",
    "date_detection": false,
    "properties": {
        'item_updated_at': {
            "type": "date",
            "format": "basic_date_time_no_millis"
        },
        'menu_page_id': {
            "type": "string",
        } ,<commit_msg>Was missing mapping for menu_date; fix dynamic date detection mapping<commit_after>#!/usr/bin/env python
# coding: utf-8

# A Repeatable Extract-Tranform-Load Pipeline for NYPL Menus Data

# authors: <NAME> and <NAME>
# created: 17 October 2014
# updated: 28 October 2014, 19 November 2014, 3 December 2014,
#     14 December 2014, 8-9 January 2015

from collections import Counter
import datetime
import elasticsearch
from elasticsearch import helpers
import json
import logging
import logging.handlers
import os
import pandas as pd
import pytz
import re
import sys
import tarfile
import time


#TODO: Give option to set up or turn off at cmd line
LOG_FILENAME = os.path.join(os.environ['MENUS_LOG_HOME'],
                            'nypl_menus_data_transform.log')

PIPELINE_LOGGER = logging.getLogger('MenusDataTransformLogger')
PIPELINE_LOGGER.setLevel(logging.INFO)
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')

handler = logging.handlers.RotatingFileHandler(
    LOG_FILENAME,
    maxBytes=500000,
    backupCount=10)
handler.setFormatter(formatter)
PIPELINE_LOGGER.addHandler(handler)
UTC = pytz.utc

DOC_TOTAL = 0


def normalize_names(obj):
    '''
    Take a name as a string, converts the string
    to lowercase, strips whitespace from beginning
    and end, normalizes multiple internal whitespace
    characters to a single space. E.g.:
    
    normalize_names('<NAME> ') = 'chicken gumbo'
    
    '''
    tokens = obj.strip().lower().split()
    result = ' '.join(filter(None, tokens))
    return result


def fingerprint(obj):
    """
    A modified version of the fingerprint clustering algorithm implemented by Open Refine.
    See https://github.com/OpenRefine/OpenRefine/wiki/Clustering-In-Depth
    This does not normalize to ASCII characters since diacritics may be significant in this dataset
    """
    alphanumeric_tokens = filter(None, re.split('\W', obj))
    seen = set()
    seen_add = seen.add
    deduped = sorted([i for i in alphanumeric_tokens if i not in seen and not seen_add(i)])
    fingerprint = ' '.join(deduped)
    
    return fingerprint


# TOFIX: This code is very slow
def reformat_dates(obj):
    naive_date_in = datetime.datetime.strptime(obj, '%Y-%m-%d %H:%M:%S %Z')
    date_in = naive_date_in.replace(tzinfo=UTC)
    date_out = date_in.strftime('%Y%m%dT%H%M%S%z')
    return date_out


def reshape_data(obj):
    '''
    Takes JSON, loads it, and spits out a dictionary of slightly different
    shape needed for bulk import by Elasticsearch.
    
    Input is JSON rather than simply another dictionary so I can punt on
    properly serializing things like 'NaN' by leaving that to pandas own
    to_json() function.
    '''

    data = json.loads(obj)
    action = {
              "_index": "menus",
              "_type": "item",
              "_id": int(data['item_id']),
              "_source": data
              }
    return action


def server(hostname, host_port):
    """
    Makes sure that an elasticsearch server is running.
    Sets up the appropriate index and mapping.
    """
    
    PIPELINE_LOGGER.info('Verifying Elasticsearch server is ready to receive data …')
    
    try:
        es = elasticsearch.Elasticsearch([{
            'host': hostname,
            'port': host_port}])
    except Exception as e:
        PIPELINE_LOGGER.error('Check to make sure the elasticsearch server is running at the specified host/port and try again. Exiting.')

    # TODO: Make these customizable
    index_name = 'menus'
    type_name = 'item'

    # Check if the index already exists and, if so, delete it to keep things idempotent
    if es.indices.exists(index_name):
        PIPELINE_LOGGER.info("deleting '{0}' index...".format(index_name))
        res = es.indices.delete(index = index_name)
        PIPELINE_LOGGER.info(" response: '{0}'".format(res))

    PIPELINE_LOGGER.info("creating '{0}' index...".format(index_name))
    res = es.indices.create(index = index_name)
    PIPELINE_LOGGER.info(" response: '{0}'".format(res))

    item_mapping = {
    "dynamic":      "strict",
    "date_detection": "false",
    "properties": {
        'item_updated_at': {
            "type": "date",
            "format": "basic_date_time_no_millis"
        },
        'menu_date': {
            "type": "date",
            "format": "date"
        },
        'menu_page_id': {
            "type": "string",
        } ,
{'fraction_non_alphanumeric': 0.10580674776222171, 'fraction_numerical': 0.011246270369520313, 'mean_word_length': 3.0161290322580645, 'pattern_counts': {'":': 22, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 2, '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': '2188421', 'n_tokens_mistral': 2796, 'n_tokens_neox': 2688, 'n_words': 799}