<commit_before>//  $$$$$$\ $$$$$$\$$$$$$$\         $$$$$$\          $$\                      $$\      $$\
// $$  __$$\\_$$  _$$  __$$\       $$  __$$\         $$ |                     $$ |     \__|
// $$ /  \__| $$ | $$ |  $$ |      $$ /  \__|$$$$$$\ $$ |$$$$$$\ $$$$$$\$$$$\ $$$$$$$\ $$\ $$$$$$\
// \$$$$$$\   $$ | $$$$$$$\ |      $$ |     $$  __$$\$$ $$  __$$\$$  _$$  _$$\$$  __$$\$$ |\____$$\
//  \____$$\  $$ | $$  __$$\       $$ |     $$ /  $$ $$ $$ /  $$ $$ / $$ / $$ $$ |  $$ $$ |$$$$$$$ |
// $$\   $$ | $$ | $$ |  $$ |      $$ |  $$\$$ |  $$ $$ $$ |  $$ $$ | $$ | $$ $$ |  $$ $$ $$  __$$ |
// \$$$$$$  $$$$$$\$$$$$$$  |      \$$$$$$  \$$$$$$  $$ \$$$$$$  $$ | $$ | $$ $$$$$$$  $$ \$$$$$$$ |
//  \______/\______\_______/        \______/ \______/\__|\______/\__| \__| \__\_______/\__|\_______|
//
//
//

var fs = require("fs");
var elasticsearch = require('elasticsearch');
var _ = require('underscore');
var ReadableSearch = require('elasticsearch-streams').ReadableSearch;
var JSONStream = require('JSONStream');
var es = require('event-stream');
var csv = require('fast-csv');
var zlib = require('zlib');
var archiver = require('archiver');
var nodemailer = require('nodemailer');
var moment = require('moment');
var winston = require('winston');

// create reusable transporter object using SMTP transport
var transporter = nodemailer.createTransport({
    service: 'Gmail',
    auth: {
      user: process.env.DOWNLOAD_GMAIL_USER,
      pass: process.env.DOWNLOAD_GMAIL_PASSWORD
    }
});

logger = new (winston.Logger)({
  transports: [
    new (winston.transports.Console)({ level: 'error' }),
    new (winston.transports.File)({ filename: 'logs/download-worker.log' })
    ]
});

var locationSaveFolder = process.env.DOWNLOAD_LOCAL_FOLDER;
var downloadURL = process.env.DOWNLOAD_WEB_FOLDER;

var kafka = require('kafka-node'),
    Consumer = kafka.Consumer,
    client = new kafka.Client(),
    consumer = new Consumer(
      client,
        [
          { topic: 'occurrencesDownload', partition: 0 }
        ],
        {
          groupId: 'kafka-node-group',
          autoCommit: true
        }
    );

consumer.on('message', processDownload);

function processDownload(message) {
  var request = JSON.parse(message.value);

  // Connect the client, requests will be
  // load-balanced between them using round-robin
  var client = elasticsearch.Client({
    hosts: [
      'localhost:9200'
    ],
    requestTimeout: 60000
  });

  var totalRegisters = 0;
  var offset = 0;
  var limit  = 10000000;
  var page   = 2000;
  var dataSets = {};

  var searchExec = function searchExec(from, callback) {
    client.search({
      index: 'sibexplorer',
      from: from + offset,
      size: (offset + from + page) > limit ? (limit - offset - from) : page,
      body: createQuery(request.type, request.query)
    }, callback);
  };

  var fileName = "occurrence-search-"+message.offset+"-"+Date.now();
  var destinationFile = locationSaveFolder+'/'+fileName;
  var writableStream = fs.createWriteStream(destinationFile+".txt", {encoding: "utf8"});
  var rs = new ReadableSearch(searchExec);

  var formatStream = csv
    .createWriteStream({headers: true, delimiter: ((process.env.DOWNLOAD_CSV_DELIMITER == 'tab') ? '\t' : ',')})
    .transform(function(obj){
      return {
        "Publicador de datos": (typeof(obj._source.provider.name) != "undefined") ? obj._source.provider.name : "",
        "Conjunto de datos": (typeof(obj._source.resource.name) != "undefined") ? obj._source.resource.name : "",
        "Dataset Rights": (typeof(obj._source.resource.rights) != "undefined") ? obj._source.resource.rights : "",
        "Fecha del evento": (typeof(obj._source.occurrence_date) != "undefined") ? obj._source.occurrence_date : "",
        "Código de la institución": (typeof(obj._source.institution.code) != "undefined") ? obj._source.institution.code : "",
        "Código de la colección": (typeof(obj._source.collection.code) != "undefined") ? obj._source.collection.code : "",
        "Base del registro": (typeof(obj._source.basis_of_record.name_spanish) != "undefined") ? obj._source.basis_of_record.name_spanish : "Registro Biológico",
        "Occurrence_id": obj._source.id,
        "Nombre científico": (typeof(obj._source.resource.rights) != "undefined") ? obj._source.canonical : "",
        "Autor": (typeof(obj._source.taxon_name_author) != "undefined") ? obj._source.taxon_name_author : "",
        "Reino": (typeof(obj._source.taxonomy.kingdom_name) != "undefined") ? obj._source.taxonomy.kingdom_name : "",
        "Filo": (typeof(obj._source.taxonomy.phylum_name) != "undefined") ? obj._source.taxonomy.phylum_name : "",
        "Clase": (typeof(obj._source.taxonomy.class_name) != "undefined") ? obj._source.taxonomy.class_name : "",
        "Orden": (typeof(obj._source.taxonomy.order_name) != "undefined") ? obj._source.taxonomy.order_name : "",
        "Familia": (typeof(obj._source.taxonomy.family_name) != "undefined") ? obj._source.taxonomy.family_name : "",
        "Género": (typeof(obj._source.taxonomy.genus_name) != "undefined") ? obj._source.taxonomy.genus_name : "",
        "Categoría taxonómica": (typeof(obj._source.taxon_rank) != "undefined") ? obj._source.taxon_rank : "",
        "País": (typeof(obj._source.country_name) != "undefined") ? obj._source.country_name : "",
        "País (calculado)": (typeof(obj._source.country_name_calculated) != "undefined") ? obj._source.country_name_calculated : "",
        "Localidad": (typeof(obj._source.locality) != "undefined") ? obj._source.locality : "",
        "Departamento": (typeof(obj._source.department_name) != "undefined") ? obj._source.department_name : "",
        "Municipio": (typeof(obj._source.county_name) != "undefined") ? obj._source.county_name : "",
        "Páramo": (typeof(obj._source.paramo_name) != "undefined") ? obj._source.paramo_name : "",
        "Área marítima": (typeof(obj._source.marine_zone_name) != "undefined") ? obj._source.marine_zone_name : "",
        "Área protegida": (typeof(obj._source.protected_area) != "undefined") ? obj._source.protected_area : "",
        "País del publicador": (typeof(obj._source.provider.country_name) != "undefined") ? obj._source.provider.country_name : "",
        "Latitud": ((typeof(obj._source.location.lat) != "undefined") && obj._source.location.lat != "1000") ? obj._source.location.lat : "",
        "Longitud": ((typeof(obj._source.location.lon) != "undefined") && obj._source.location.lon != "1000") ? obj._source.location.lon : "",
        "Profundidad": (typeof(obj._source.depth_centimeters) != "undefined") ? obj._source.depth_centimeters : "",
        "Altitud": (typeof(obj._source.altitude_meters) != "undefined") ? obj._source.altitude_meters : "",
        "URL del portal de SIB Colombia": "http://data.sibcolombia.net/occurrences/"+obj._source.id,
        "URL del servicio web de SIB Colombia": "http://data.sibcolombia.net/ws/rest/occurrence/get?key="+obj._source.id
      };
    });

  writableStream.on("finish", function() {
    var rightsText = "Por favor respete los derechos y permisos declarados para cada conjunto de datos en esta descarga:\n";
    for (var property in dataSets) {
      if(dataSets[property].rights) {
        rightsText = rightsText + "\nConjunto de datos: "+property+", http://data.sibcolombia.net/datasets/resource/"+dataSets[property].id+"\n"+"Licencia de uso: "+dataSets[property].rights+"\n";
      }
    }
    var citationText = "";
    for (var property in dataSets) {
      if(dataSets[property].citation) {
        citationText = citationText + "\nConjunto de datos: "+property+"\n\nPor favor citar estos datos de la siguiente manera:: "+dataSets[property].citation+"\n\n\n";
      }
    }
    var archive = archiver('zip');
    var output = fs.createWriteStream(destinationFile+'.zip');

    output.on("finish", function() {
      fs.unlinkSync(destinationFile+'.txt');

      moment.locale('es'); // change the global locale to Spanish
      var stats = fs.statSync(destinationFile+'.zip')
      var fileSizeInBytes = stats["size"]
      //Convert the file size to megabytes (optional)
      var fileSizeInMegabytes = fileSizeInBytes / 1000000.0

      logger.info('Download request success', {email: request.email, totalRegisters: totalRegisters, requestDate: request.date, processFinishDate: Date.now()/1000, pid: process.pid, fileSize: Number((fileSizeInBytes / 1000.0).toFixed(2)), type: request.type, query: request.query, reason: request.reason, remoteip: request.remoteip});

      if(totalRegisters == 0) {
        // setup e-mail data with unicode symbols
        var mailOptions = {
          from: 'SIB Colombia <<EMAIL>>', // sender address
          to: request.email+' <'+request.email+'>', // list of receivers
          subject: 'Su consulta en la red del SIB esta disponible para descarga', // Subject line
          html: '<p>La solicitud que realizó mediante la plataformaa de SIB Colombia no arrojó ningún registro que cumpla con sus condiciones de búsqueda.</p><p>Le invitamos a cambiar los parámetros de búsqueda en: <a href="http://data.sibcolombia.net">http://data.sibcolombia.net</a> o <a href="http://maps.sibcolombia.net">http://maps.sibcolombia.net</a></p><p>Gracias por usar nuestro sistema de información</p>' // html body
        };<commit_msg>Include fileSize existance validation<commit_after>//  $$$$$$\ $$$$$$\$$$$$$$\         $$$$$$\          $$\                      $$\      $$\
// $$  __$$\\_$$  _$$  __$$\       $$  __$$\         $$ |                     $$ |     \__|
// $$ /  \__| $$ | $$ |  $$ |      $$ /  \__|$$$$$$\ $$ |$$$$$$\ $$$$$$\$$$$\ $$$$$$$\ $$\ $$$$$$\
// \$$$$$$\   $$ | $$$$$$$\ |      $$ |     $$  __$$\$$ $$  __$$\$$  _$$  _$$\$$  __$$\$$ |\____$$\
//  \____$$\  $$ | $$  __$$\       $$ |     $$ /  $$ $$ $$ /  $$ $$ / $$ / $$ $$ |  $$ $$ |$$$$$$$ |
// $$\   $$ | $$ | $$ |  $$ |      $$ |  $$\$$ |  $$ $$ $$ |  $$ $$ | $$ | $$ $$ |  $$ $$ $$  __$$ |
// \$$$$$$  $$$$$$\$$$$$$$  |      \$$$$$$  \$$$$$$  $$ \$$$$$$  $$ | $$ | $$ $$$$$$$  $$ \$$$$$$$ |
//  \______/\______\_______/        \______/ \______/\__|\______/\__| \__| \__\_______/\__|\_______|
//
//
//

var fs = require("fs");
var elasticsearch = require('elasticsearch');
var _ = require('underscore');
var ReadableSearch = require('elasticsearch-streams').ReadableSearch;
var JSONStream = require('JSONStream');
var es = require('event-stream');
var csv = require('fast-csv');
var zlib = require('zlib');
var archiver = require('archiver');
var nodemailer = require('nodemailer');
var moment = require('moment');
var winston = require('winston');

// create reusable transporter object using SMTP transport
var transporter = nodemailer.createTransport({
    service: 'Gmail',
    auth: {
      user: process.env.DOWNLOAD_GMAIL_USER,
      pass: <PASSWORD>
    }
});

logger = new (winston.Logger)({
  transports: [
    new (winston.transports.Console)({ level: 'error' }),
    new (winston.transports.File)({ filename: 'logs/download-worker.log' })
    ]
});

var locationSaveFolder = process.env.DOWNLOAD_LOCAL_FOLDER;
var downloadURL = process.env.DOWNLOAD_WEB_FOLDER;

var kafka = require('kafka-node'),
    Consumer = kafka.Consumer,
    client = new kafka.Client(),
    consumer = new Consumer(
      client,
        [
          { topic: 'occurrencesDownload', partition: 0 }
        ],
        {
          groupId: 'kafka-node-group',
          autoCommit: true
        }
    );

consumer.on('message', processDownload);

function processDownload(message) {
  var request = JSON.parse(message.value);

  // Connect the client, requests will be
  // load-balanced between them using round-robin
  var client = elasticsearch.Client({
    hosts: [
      'localhost:9200'
    ],
    requestTimeout: 60000
  });

  var totalRegisters = 0;
  var offset = 0;
  var limit  = 10000000;
  var page   = 2000;
  var dataSets = {};

  var searchExec = function searchExec(from, callback) {
    client.search({
      index: 'sibexplorer',
      from: from + offset,
      size: (offset + from + page) > limit ? (limit - offset - from) : page,
      body: createQuery(request.type, request.query)
    }, callback);
  };

  var fileName = "occurrence-search-"+message.offset+"-"+Date.now();
  var destinationFile = locationSaveFolder+'/'+fileName;
  var writableStream = fs.createWriteStream(destinationFile+".txt", {encoding: "utf8"});
  var rs = new ReadableSearch(searchExec);

  var formatStream = csv
    .createWriteStream({headers: true, delimiter: ((process.env.DOWNLOAD_CSV_DELIMITER == 'tab') ? '\t' : ',')})
    .transform(function(obj){
      return {
        "Publicador de datos": (typeof(obj._source.provider.name) != "undefined") ? obj._source.provider.name : "",
        "Conjunto de datos": (typeof(obj._source.resource.name) != "undefined") ? obj._source.resource.name : "",
        "Dataset Rights": (typeof(obj._source.resource.rights) != "undefined") ? obj._source.resource.rights : "",
        "Fecha del evento": (typeof(obj._source.occurrence_date) != "undefined") ? obj._source.occurrence_date : "",
        "Código de la institución": (typeof(obj._source.institution.code) != "undefined") ? obj._source.institution.code : "",
        "Código de la colección": (typeof(obj._source.collection.code) != "undefined") ? obj._source.collection.code : "",
        "Base del registro": (typeof(obj._source.basis_of_record.name_spanish) != "undefined") ? obj._source.basis_of_record.name_spanish : "Registro Biológico",
        "Occurrence_id": obj._source.id,
        "Nombre científico": (typeof(obj._source.resource.rights) != "undefined") ? obj._source.canonical : "",
        "Autor": (typeof(obj._source.taxon_name_author) != "undefined") ? obj._source.taxon_name_author : "",
        "Reino": (typeof(obj._source.taxonomy.kingdom_name) != "undefined") ? obj._source.taxonomy.kingdom_name : "",
        "Filo": (typeof(obj._source.taxonomy.phylum_name) != "undefined") ? obj._source.taxonomy.phylum_name : "",
        "Clase": (typeof(obj._source.taxonomy.class_name) != "undefined") ? obj._source.taxonomy.class_name : "",
        "Orden": (typeof(obj._source.taxonomy.order_name) != "undefined") ? obj._source.taxonomy.order_name : "",
        "Familia": (typeof(obj._source.taxonomy.family_name) != "undefined") ? obj._source.taxonomy.family_name : "",
        "Género": (typeof(obj._source.taxonomy.genus_name) != "undefined") ? obj._source.taxonomy.genus_name : "",
        "Categoría taxonómica": (typeof(obj._source.taxon_rank) != "undefined") ? obj._source.taxon_rank : "",
        "País": (typeof(obj._source.country_name) != "undefined") ? obj._source.country_name : "",
        "País (calculado)": (typeof(obj._source.country_name_calculated) != "undefined") ? obj._source.country_name_calculated : "",
        "Localidad": (typeof(obj._source.locality) != "undefined") ? obj._source.locality : "",
        "Departamento": (typeof(obj._source.department_name) != "undefined") ? obj._source.department_name : "",
        "Municipio": (typeof(obj._source.county_name) != "undefined") ? obj._source.county_name : "",
        "Páramo": (typeof(obj._source.paramo_name) != "undefined") ? obj._source.paramo_name : "",
        "Área marítima": (typeof(obj._source.marine_zone_name) != "undefined") ? obj._source.marine_zone_name : "",
        "Área protegida": (typeof(obj._source.protected_area) != "undefined") ? obj._source.protected_area : "",
        "País del publicador": (typeof(obj._source.provider.country_name) != "undefined") ? obj._source.provider.country_name : "",
        "Latitud": ((typeof(obj._source.location.lat) != "undefined") && obj._source.location.lat != "1000") ? obj._source.location.lat : "",
        "Longitud": ((typeof(obj._source.location.lon) != "undefined") && obj._source.location.lon != "1000") ? obj._source.location.lon : "",
        "Profundidad": (typeof(obj._source.depth_centimeters) != "undefined") ? obj._source.depth_centimeters : "",
        "Altitud": (typeof(obj._source.altitude_meters) != "undefined") ? obj._source.altitude_meters : "",
        "URL del portal de SIB Colombia": "http://data.sibcolombia.net/occurrences/"+obj._source.id,
        "URL del servicio web de SIB Colombia": "http://data.sibcolombia.net/ws/rest/occurrence/get?key="+obj._source.id
      };
    });

  writableStream.on("finish", function() {
    var rightsText = "Por favor respete los derechos y permisos declarados para cada conjunto de datos en esta descarga:\n";
    for (var property in dataSets) {
      if(dataSets[property].rights) {
        rightsText = rightsText + "\nConjunto de datos: "+property+", http://data.sibcolombia.net/datasets/resource/"+dataSets[property].id+"\n"+"Licencia de uso: "+dataSets[property].rights+"\n";
      }
    }
    var citationText = "";
    for (var property in dataSets) {
      if(dataSets[property].citation) {
        citationText = citationText + "\nConjunto de datos: "+property+"\n\nPor favor citar estos datos de la siguiente manera:: "+dataSets[property].citation+"\n\n\n";
      }
    }
    var archive = archiver('zip');
    var output = fs.createWriteStream(destinationFile+'.zip');

    output.on("finish", function() {
      fs.unlinkSync(destinationFile+'.txt');

      moment.locale('es'); // change the global locale to Spanish
      var stats = fs.statSync(destinationFile+'.zip')
      var fileSizeInBytes = 0;
      if(typeof stats["size"] !== "undefined") {
        fileSizeInBytes = stats["size"];
      }
      //Convert the file size to megabytes (optional)
      var fileSizeInMegabytes = fileSizeInBytes / 1000000.0

      logger.info('Download request success', {email: request.email, totalRegisters: totalRegisters, requestDate: request.date, processFinishDate: Date.now()/1000, pid: process.pid, fileSize: Number((fileSizeInBytes / 1000.0).toFixed(2)), type: request.type, query: request.query, reason: request.reason, remoteip: request.remoteip});

      if(totalRegisters == 0) {
        // setup e-mail data with unicode symbols
        var mailOptions = {
          from: 'SIB Colombia <<EMAIL>>', // sender address
          to: request.email+' <'+request.email+'>', // list of receivers
          subject: 'Su consulta en la red del SIB esta disponible para descarga', // Subject line
          html: '<p>La solicitud que realizó mediante la plataformaa de SIB Colombia no arrojó ningún registro que cumpla con sus condiciones de búsqueda.</p><p>Le invitamos a cambiar los parámetros de búsqueda en: <a href="http://data.sibcolombia.net">http://data.sibcolombia.net</a> o <a href="http://maps.sibcolombia.net">http://maps.sibcolombia.net</a></p><p>Gracias por usar nuestro sistema de información</p>' // html body
        };
{'fraction_non_alphanumeric': 0.18804789394911267, 'fraction_numerical': 0.005612572161642078, 'mean_word_length': 3.3078517154040985, 'pattern_counts': {'":': 64, '<': 30, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 8, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3538106', 'n_tokens_mistral': 6177, 'n_tokens_neox': 5966, 'n_words': 1080}