<commit_before>var util = require('util')
, fs = require('fs')
, chalk = require('chalk')
, got = require('got')
, mkdirp = require('mkdirp')
, _ = require('lodash')
, ProgressBar = require('progress');

exports.downloadurlQueue = function(fullurlQueue, nextDlTaskcb) {
  var failed = [];
  var retries = 0;
  var missing = 0;

  urlQueue = fullurlQueue; //urlQueue needs to be global unless
                          //we put these other functions inside
                          //this one.

  //Setup ProgressBar
  var progmsg = 'Downloading files [:bar] :percent' +
              ' (:current/:total) [:elapseds elapsed, eta :eta]';
  var progopts = {
    total: fullurlQueue.length,
    width: 30,
    complete: chalk.green('=')
  };
  var dlprogress = new ProgressBar(progmsg, progopts);

  for(i=0; i<10; i++) {
  nextUrlTask(urlQueue); //spawn 10 workers
  }

function nextUrlTask() {
  if (urlQueue instanceof Array && urlQueue.length > 0) {
    var urlObj = urlQueue.splice(0,1)[0];
    downloadURL(urlObj);
  }
}

function downloadURL(urlObj) {
  var url = urlObj.url;
  var id = urlObj.id;
  var type = urlObj.type;
  var rename = urlObj.rename;
  var base = id + '/';
  log.debug('Creating directory: ' + base);
  mkdirp.sync(base);

  fs.readFile(base + rename, (err, data) => {
    if ((err)&&(err.code=='ENOENT')) {
      log.info('File of type: '+type+' and id: '+id+' already exists. Skipping.')
      nextUrlTask(urlQueue)
      return
    }
    else if (err) {
      throw err
    }
  })

  log.debug('Downloading ' + type + ': ' + url);
  var options = {
    timeout: 15000,
    encoding: null,
    retries: 3
  }

  var get = got(url, options, function(err, data, res) {
    dlprogress.tick();
    if (err) {
      if (err.code === 'ETIMEDOUT' || err.code === 'ESOCKETTIMEDOUT') {
        log.warn('Download timed out for URL ' + url);
      }
      if (!res) {
        failed.push(url);
      } else if ((res.statusCode == 404) && !(fourohfour === null)) {
        fourohfour();
      } else {
        failed.push(url);
      }
      done();
    } else {
      fs.writeFile(base + rename, data, done);
    }
      nextUrlTask(urlQueue);
    });
}

  var donefunc = function() {
    if (failed.length > 0) {
      log.warn(failed.length + ' downloads timed out on retry.');
    } else if (missing > 0) {
      var succeeded = urlQueue.length - missing;
      var suffix = missing > 1 ? 's' : ''
      log.info(succeeded + ' downloads succeeded. ' + missing +
               ' paper' + suffix + ' had urlQueue that could not be reached (404 error).');
    } else {
      log.info('All downloads succeeded!');
    }
    nextDlTaskcb();
  }

  var done = _.after(urlQueue.length, donefunc);

  var fourohfour = function() {
    missing ++;
  }
}<commit_msg>convert download.js to request retry<commit_after>var util = require('util')
, fs = require('fs')
, chalk = require('chalk')
, requestretry = require('requestretry')
, mkdirp = require('mkdirp')
, _ = require('lodash')
, ProgressBar = require('progress');

exports.downloadurlQueue = function(fullurlQueue, nextDlTaskcb) {
  var failed = [];
  var retries = 0;
  var missing = 0;

  var urlQueue = fullurlQueue;

  //Setup ProgressBar
  var progmsg = 'Downloading files [:bar] :percent' +
              ' (:current/:total) [:elapseds elapsed, eta :eta]';
  var progopts = {
    total: fullurlQueue.length,
    width: 30,
    complete: chalk.green('=')
  };
  var dlprogress = new ProgressBar(progmsg, progopts);

  for(i=0; i<10; i++) {
  nextUrlTask(urlQueue); //spawn 10 workers
  }

function nextUrlTask() {
  if (urlQueue instanceof Array && urlQueue.length > 0) {
    var urlObj = urlQueue.splice(0,1)[0];
    testIfFileExists(urlObj, downloadURL);
  }
}

// Run callback if file doesn't exist
function testIfFileExists(urlObj, cb) {
  dlprogress.tick();
  var url = urlObj.url;
  var id = urlObj.id;
  var type = urlObj.type;
  var rename = urlObj.rename;
  var base = id + '/';
  fs.readFile(base + rename, (err, data) => {
    if ((err)&&(err.code=='ENOENT')) {
      cb(urlObj)
      //File doesn't exist so start download procedure
    }
    else if (err) {
      throw err
    }
    else {
      log.info('File of type: '+type+' and id: '+id+' already exists. Skipping.')
      nextUrlTask(urlQueue)
      return
    }
  })
}

function downloadURL(urlObj) {
  var url = urlObj.url;
  var id = urlObj.id;
  var type = urlObj.type;
  var rename = urlObj.rename;
  var base = id + '/';
  log.debug('Creating directory: ' + base);
  mkdirp.sync(base);

  log.debug('Downloading ' + type + ': ' + url);
  var options = {
    timeout: 15000,
    encoding: null,
    retries: 3
  }

  function handleDownload(data) {
      fs.writeFile(base + rename, data, done);
      nextUrlTask(urlQueue);
    }

  function throwErr(err){
    if (err) throw err
  }

  rq = requestretry.get({url: url, fullResponse: false} );
  rq.then(handleDownload)
  rq.catch(throwErr)
}

  var donefunc = function() {
    if (failed.length > 0) {
      log.warn(failed.length + ' downloads timed out on retry.');
    } else if (missing > 0) {
      var succeeded = urlQueue.length - missing;
      var suffix = missing > 1 ? 's' : ''
      log.info(succeeded + ' downloads succeeded. ' + missing +
               ' paper' + suffix + ' had urlQueue that could not be reached (404 error).');
    } else {
      log.info('All downloads succeeded!');
    }
    nextDlTaskcb();
  }

  var done = _.after(fullurlQueue.length, donefunc);

  var fourohfour = function() {
    missing ++;
  }
}
{'fraction_non_alphanumeric': 0.15144404332129963, 'fraction_numerical': 0.009566787003610109, 'mean_word_length': 2.696464309539693, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4390832', 'n_tokens_mistral': 1947, 'n_tokens_neox': 1848, 'n_words': 474}