<commit_before>var utility = require('./utility');
var async = require('async');
var db = utility.db;
var logger = utility.logger;
var fs = require('fs');
var moment = require('moment');
var getData = utility.getData;
var request = require('request');
var insertPlayer = utility.insertPlayer;
var insertMatch = utility.insertMatch;
var spawn = require('child_process').spawn;
var replay_dir = process.env.REPLAY_DIR || "./replays/";
var domain = require('domain');

function processParse(job, cb) {
    var t1 = new Date();
    var match_id = job.data.payload.match_id;
    var retries = job.toJSON().attempts.remaining;
    var noRetry = retries <= 1;
    async.waterfall([
        async.apply(checkLocal, job),
        getReplayUrl,
        streamReplay,
    ], function(err, job2) {
        logger.info("[PARSER] match_id %s, parse time: %s", match_id, (new Date() - t1) / 1000);
        if (err === "replay expired" || noRetry) {
            logger.info("match_id %s, error %s, not retrying", match_id, err);
            return db.matches.update({
                match_id: match_id
            }, {
                $set: {
                    parse_status: 1
                }
            }, function(err2) {
                //nullify error if replay expired
                err = err === "replay expired" ? null : err;
                cb(err || err2);
            });
        }
        else if (err) {
            logger.info("match_id %s, error %s, retries %s", match_id, err, retries);
            return cb(err);
        }
        else {
            return cb(err);
        }
    });
}

function checkLocal(job, cb) {
    var match_id = job.data.payload.match_id;
    var fileName = job.data.fileName || replay_dir + match_id + ".dem";
    if (fs.existsSync(fileName)) {
        logger.info("[PARSER] %s, found local replay", match_id);
        job.data.fileName = fileName;
        job.update();
    }
    cb(null, job);
}

function getReplayUrl(job, cb) {
    if (job.data.url || job.data.fileName) {
        logger.info("has url or fileName");
        return cb(null, job);
    }
    var match = job.data.payload;
    if (match.start_time > moment().subtract(7, 'days').format('X')) {
        getData("http://retriever?match_id=" + job.data.payload.match_id, function(err, body) {
            if (err || !body || !body.match) {
                return cb("invalid body or error");
            }
            var url = "http://replay" + body.match.cluster + ".valve.net/570/" + match.match_id + "_" + body.match.replaySalt + ".dem.bz2";
            job.data.url = url;
            job.update();
            return cb(null, job);
        });
    }
    /*
    else if (process.env.AWS_S3_BUCKET) {
        getS3Url(match.match_id, function(err, url) {
            if (!url) {
                return cb("replay expired");
            }
            job.data.url = url;
            job.update();
            return cb(err, job);
        });
    }
    */
    else {
        cb("replay expired");
    }
}

function streamReplay(job, cb) {
    //var fileName = replay_dir + match_id + ".dem";
    //var archiveName = fileName + ".bz2";
    var match_id = job.data.payload.match_id;
    logger.info("[PARSER] streaming from %s", job.data.url || job.data.fileName);
    var d = domain.create();
    d.on('error', function(err) {
        logger.info(err);
        cb(err);
    });
    d.run(function() {
        var parser = utility.runParse(function(err, output) {
            if (err) {
                return cb(err);
            }
            db.matches.update({
                match_id: match_id
            }, {
                $set: {
                    parsed_data: output,
                    parse_status: 2
                }
            }, function(err) {
                cb(err, job);
            });
        });
        if (job.data.fileName) {
            fs.createReadStream(job.data.fileName).pipe(parser.stdin);
        }
        else {
            var bz = spawn("bzcat");
            var downStream = request.get({
                url: job.data.url,
                encoding: null,
                timeout: 180000
            });
            downStream.pipe(bz.stdin);
            bz.stdout.pipe(parser.stdin);
        }
    });
}

function processApi(job, cb) {
    //process an api request
    var payload = job.data.payload;
    getData(job.data.url, function(err, data) {
        if (err) {
            return cb(err);
        }
        if (data.response) {
            logger.info("summaries response");
            async.map(data.response.players, insertPlayer, function(err) {<commit_msg>trying direct callback on not 200 download<commit_after>var utility = require('./utility');
var async = require('async');
var db = utility.db;
var logger = utility.logger;
var fs = require('fs');
var moment = require('moment');
var getData = utility.getData;
var request = require('request');
var insertPlayer = utility.insertPlayer;
var insertMatch = utility.insertMatch;
var spawn = require('child_process').spawn;
var replay_dir = process.env.REPLAY_DIR || "./replays/";
var domain = require('domain');

function processParse(job, cb) {
    var t1 = new Date();
    var match_id = job.data.payload.match_id;
    var retries = job.toJSON().attempts.remaining;
    var noRetry = retries <= 1;
    async.waterfall([
        async.apply(checkLocal, job),
        getReplayUrl,
        streamReplay,
    ], function(err, job2) {
        logger.info("[PARSER] match_id %s, parse time: %s", match_id, (new Date() - t1) / 1000);
        if (err === "replay expired" || noRetry) {
            logger.info("match_id %s, error %s, not retrying", match_id, err);
            return db.matches.update({
                match_id: match_id
            }, {
                $set: {
                    parse_status: 1
                }
            }, function(err2) {
                //nullify error if replay expired
                err = err === "replay expired" ? null : err;
                cb(err || err2);
            });
        }
        else if (err) {
            logger.info("match_id %s, error %s, retries %s", match_id, err, retries);
            return cb(err);
        }
        else {
            return cb(err);
        }
    });
}

function checkLocal(job, cb) {
    var match_id = job.data.payload.match_id;
    var fileName = job.data.fileName || replay_dir + match_id + ".dem";
    if (fs.existsSync(fileName)) {
        logger.info("[PARSER] %s, found local replay", match_id);
        job.data.fileName = fileName;
        job.update();
    }
    cb(null, job);
}

function getReplayUrl(job, cb) {
    if (job.data.url || job.data.fileName) {
        logger.info("has url or fileName");
        return cb(null, job);
    }
    var match = job.data.payload;
    if (match.start_time > moment().subtract(7, 'days').format('X')) {
        getData("http://retriever?match_id=" + job.data.payload.match_id, function(err, body) {
            if (err || !body || !body.match) {
                return cb("invalid body or error");
            }
            var url = "http://replay" + body.match.cluster + ".valve.net/570/" + match.match_id + "_" + body.match.replaySalt + ".dem.bz2";
            job.data.url = url;
            job.update();
            return cb(null, job);
        });
    }
    /*
    else if (process.env.AWS_S3_BUCKET) {
        getS3Url(match.match_id, function(err, url) {
            if (!url) {
                return cb("replay expired");
            }
            job.data.url = url;
            job.update();
            return cb(err, job);
        });
    }
    */
    else {
        cb("replay expired");
    }
}

function streamReplay(job, cb) {
    //var fileName = replay_dir + match_id + ".dem";
    //var archiveName = fileName + ".bz2";
    var match_id = job.data.payload.match_id;
    logger.info("[PARSER] streaming from %s", job.data.url || job.data.fileName);
    var d = domain.create();
    d.on('error', function(err) {
        logger.info(err);
        cb(err);
    });
    d.run(function() {
        var parser = utility.runParse(function(err, output) {
            if (err) {
                return cb(err);
            }
            db.matches.update({
                match_id: match_id
            }, {
                $set: {
                    parsed_data: output,
                    parse_status: 2
                }
            }, function(err) {
                cb(err, job);
            });
        });
        if (job.data.fileName) {
            fs.createReadStream(job.data.fileName).pipe(parser.stdin);
        }
        else {
            var bz = spawn("bzcat");
            var downStream = request.get({
                url: job.data.url,
                encoding: null,
                timeout: 180000
            });
            downStream.on('response', function(resp) {
                if (resp.statusCode !== 200) {
                    cb("download error");
                }
            });
            downStream.pipe(bz.stdin);
            bz.stdout.pipe(parser.stdin);
        }
    });
}

function processApi(job, cb) {
    //process an api request
    var payload = job.data.payload;
    getData(job.data.url, function(err, data) {
        if (err) {
            return cb(err);
        }
        if (data.response) {
            logger.info("summaries response");
            async.map(data.response.players, insertPlayer, function(err) {
{'fraction_non_alphanumeric': 0.14266030736618973, 'fraction_numerical': 0.0061473237943826175, 'mean_word_length': 1.7232323232323232, 'pattern_counts': {'":': 0, '<': 5, '<?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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2041109', 'n_tokens_mistral': 2926, 'n_tokens_neox': 2860, 'n_words': 629}