<commit_before>
var through = require('through2');
var esclient = require('pelias-esclient')();

var stream = through.obj( function( item, enc, done ) {

  // skip lookup if we already have geo info
  if( item.admin0 && item.admin1 && item.admin2 ) return done();

  esclient.search({
    index: 'pelias',
    type: 'geoname',
    body: {
      "fields": [ "admin0", "admin1", "admin2" ],
      "query": {
        "filtered": {
          "query": {
            "match_all": {}
          },
          "filter" : {
            "bool": {
              "must": [
                // {
                //   "geohash_cell": {
                //     "center_point": item.center_point,
                //     "precision": 12, // @note: setting this higher than 5 causes loads of misses!?
                //                      // ...and setting it around 5 is very very slow.
                //     "neighbors": true
                //   }
                // },
                {
                  "geo_distance" : {
                    "distance": "50km",
                    "distance_type": "plane",
                    "optimize_bbox": "indexed",
                    "center_point": {
                      "lat": Number( item.center_point.lat ).toFixed(2), // @note: make filter cachable
                      "lon": Number( item.center_point.lon ).toFixed(2)  // precision max ~1.113km off
                    },
                    "_cache": true
                  }
                }
              ]
            }
          }
        }
      },
      "sort": [{
        "_geo_distance": {
          "center_point": item.center_point,
          "order": "asc",
          "unit": "km"
        }
      }],
      "size": 1
    }
  }, function (error, resp) {

    if( error ) console.error( error );

    // console.log( resp.took + 'ms' );
    // console.log( JSON.stringify( resp, null, 2 ) );
    // process.exit(1);

    if( !error && 'object' == typeof resp && resp.hasOwnProperty('hits') && 
        Array.isArray( resp.hits.hits ) && resp.hits.hits.length ){

      var hit = resp.hits.hits[0].fields;
      if( hit.admin0 ) item.admin0 = hit.admin0[0];
      if( hit.admin1 ) item.admin1 = hit.admin1[0];
      if( hit.admin2 ) item.admin2 = hit.admin2[0];

      if( !hit.admin0 && !hit.admin1 && !hit.admin2 ){
        console.log( 'failed lookup' );
      }
    }

    else {
      console.log( 'failed lookup' );
    }
   
    this.push( item, enc );
    done();

  }.bind(this));

});

module.exports = stream;<commit_msg>improved error reporting and documentation<commit_after>
var through = require('through2'),
    esclient = require('pelias-esclient')(),
    DEBUG = false;

var stream = through.obj( function( item, enc, done ) {

  // Skip lookup if record already has geo info
  if( item.admin0 && item.admin1 && item.admin2 ){
    this.push( item, enc ); // Forward record down the pipe
    return done(); // ACK and take next record from the inbound stream
  }

  // Search ES for the nearest geonames record to this items center point
  esclient.search( buildSearchQuery( item.center_point ), function ( error, resp ) {

    // Response logger
    if( DEBUG ){
      console.error( 'response time:', resp.took + 'ms' );
      console.error( JSON.stringify( resp, null, 2 ) );
    }

    // Log errors from the es client
    if( error ){ console.error( 'eclient error: geonames lookup', error ); }

    // Check the response is valid ang contains at least one records
    else if( 'object' == typeof resp && resp.hasOwnProperty('hits') && 
        Array.isArray( resp.hits.hits ) && resp.hits.hits.length ){

      // We only need the document fields
      var hit = resp.hits.hits[0].fields;

      // This should never happen but was reported in the wild
      // please report this bug if you see it again
      if( !hit ){
        console.error( 'unexpected: invalid fields returned' );
        console.error( JSON.stringify( resp, null, 2 ) );
      }

      // It's possible (albeit very unlikely) that the record we found
      // contains no admin hierarchy at all
      else if( !hit.admin0 && !hit.admin1 && !hit.admin2 ){
        console.error( 'incomplete: geonames record contained no admin hierarchy' );
      }

      // Copy admin data from the geonames record to the osm record
      else {
        if( hit.admin0 ){ item.admin0 = hit.admin0[0]; }
        if( hit.admin1 ){ item.admin1 = hit.admin1[0]; }
        if( hit.admin2 ){ item.admin2 = hit.admin2[0]; }
      }
    }

    // The query returned 0 results
    else { console.error( 'miss: failed geonames lookup' ); }
   
    this.push( item, enc ); // Forward record down the pipe
    return done(); // ACK and take next record from the inbound stream

  }.bind(this));

});

// Build an elasticsearch query to augment the osm record with
// admin hierarchy info from the nearest geonames record
var buildSearchQuery = function( centroid ){
  return {
    index: 'pelias',
    type: 'geoname',
    body: {
      "fields": [ "admin0", "admin1", "admin2" ], // Only return fields related to admin hierarchy
      "query": {
        "filtered": {
          "query": {
            "match_all": {}
          },
          "filter" : {
            "bool": {
              "must": [
                // {
                //   "geohash_cell": {
                //     "center_point": item.center_point,
                //     "precision": 12, // @note: setting this higher than 5 causes loads of misses!?
                //                      // ...and setting it around 5 is very very slow.
                //     "neighbors": true
                //   }
                // },
                {
                  "geo_distance" : {
                    "distance": "50km",
                    "distance_type": "plane",
                    "optimize_bbox": "indexed",
                    "center_point": {
                      "lat": Number( centroid.lat ).toFixed(2), // @note: make filter cachable
                      "lon": Number( centroid.lon ).toFixed(2)  // precision max ~1.113km off
                    },
                    "_cache": true // Speed up duplicate queries. Memory impact?
                  }
                }
              ]
            }
          }
        }
      },
      "sort": [{
        "_geo_distance": {
          "center_point": centroid,
          "order": "asc",
          "unit": "km"
        }
      }],
      "size": 1
    }
  };
}

module.exports = stream;
{'fraction_non_alphanumeric': 0.1369313801079414, 'fraction_numerical': 0.01187355435620663, 'mean_word_length': 1.5485265225933202, 'pattern_counts': {'":': 48, '<': 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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6568635', 'n_tokens_mistral': 1933, 'n_tokens_neox': 1832, 'n_words': 520}