<commit_before>/*jslint  node:true */
'use strict';

var express = require('express');
var router = express.Router();
var winston = require('winston');
var log = new (winston.Logger)({
    transports: [
        new (winston.transports.Console)({'timestamp': true})
    ]
});
var util = require('util');
var _ = require('lodash');
var redis = require('./../lib/redis.js');
var gatekeeper = require('./../lib/gatekeeper');
var storage = require('./../lib/azure-storage');
var async = require('async');
var azure = require('azure-storage');
var LZUTF8 = require('./../lib/lzutf8');

//var murmurhash = require('murmurhash');

var READ_CONCURRENCY = 10;
var WRITE_CONCURRENCY = 40;

var TABLE_BATCH_SIZE = 50;

require('moment-timezone');

var Agent = require('agentkeepalive').HttpsAgent;

var keepaliveAgent = new Agent({
    maxSockets: 1000,
    //maxFreeSockets: 100,
    timeout: 60000,
    keepAliveTimeout: 30000 // free socket keepalive for 30 seconds
});

var azureTable = require('azure-table-node');
var defaultClient = azureTable.createClient({
    agent: keepaliveAgent
});

//var SEED = 1234567;
//var TOTAL = 3000000;
//var BUCKET = TOTAL / 100;

var MAX_DAY = 99991231;
var MASK = '000000';


//var tableSvc = storage.tableService();

var createTable = function (callback) {
    // tableSvc.createTableIfNotExists(storage.viewerSummaryTable, function (error) {
    //     callback(error);
    // });

    defaultClient.createTable(storage.viewerSummaryTable, {ignoreIfExists: true}, function (error) {
        callback(error);
    });
};

var connectRedis = function (callback) {
    var client = redis.connect();
    client.on("error", function (err) {
        callback(err);
    });
    client.on("ready", function () {
        callback(null, client);
    });
};

//var connectTable = function (callback) {
//    storage.tableService();
//};

var getGames = function (redisClient, callback) {
    redisClient.hgetall('games', function (err, games) {
        callback(err, redisClient, games);
    });
};

var numFmt = function (num, mask) {
    return (mask + num).slice(-Math.max(mask.length, (num + '').length));
};

var processGroup = function (task, callback) {

    console.time('hgetall>' + task.key);
    console.time('processGroup > ' + task.key);
    task.redisClient.hgetall(task.key, function (err, pairs) {
        console.timeEnd('hgetall>' + task.key);
        if (err) {
            return callback(err);
        }

        var //entGen = azure.TableUtilities.entityGenerator,
            keyParts = task.key.split(':'),
            day = MAX_DAY - parseInt(keyParts[2], 10),
            partitionKey = util.format("%d:%s", day, numFmt(keyParts[3], MASK));

        var totals = _.reduce(pairs, function (result, dur, key) {
            var parts = key.split(':'),
                channel = parts[1],
                game = task.games[parts[2]] || parts[2],
                viewer = parts[0],
                duration = parseInt(dur, 10);

            result[viewer] = result[viewer] || {views: [], total: 0};
            result[viewer].total = (result[viewer].total || 0) + duration;
            result[viewer].views.push({game: game, channel: channel, duration: duration});

            return result;
        }, {});

        log.info('Processing Group', task.key, _.size(totals));


        var cargo = async.cargo(function (jobs, cargocb) {

            var batch = defaultClient.startBatch();

            _.each(jobs, function (job) {
                batch.insertOrReplaceEntity(storage.viewerSummaryTable, job);
            });

            batch.commit(function (err, data) {
                if (!err) {
                    task.stats.batches += 1;
                    task.stats.records += jobs.length;
                } else {
                    log.error(err);
                    task.errors += 1;
                }
                cargocb(null, data);
            });

        }, TABLE_BATCH_SIZE);

        var complete = function (err) {
            console.timeEnd('executeBatch > ' + task.key);
            console.timeEnd('processGroup > ' + task.key);
            if (err) {
                return callback(err);
            }

            if (task.errors > 0) {
                return callback(err);
            }

            // remove from cache
            //task.redisClient.del(task.key, function (err) {
            //    log.info('Removed key: ', task.key);
            //    callback(err);
            //});
        };

        var done = false;
        cargo.drain = function () {<commit_msg>disable cleanup + callback<commit_after>/*jslint  node:true */
'use strict';

var express = require('express');
var router = express.Router();
var winston = require('winston');
var log = new (winston.Logger)({
    transports: [
        new (winston.transports.Console)({'timestamp': true})
    ]
});
var util = require('util');
var _ = require('lodash');
var redis = require('./../lib/redis.js');
var gatekeeper = require('./../lib/gatekeeper');
var storage = require('./../lib/azure-storage');
var async = require('async');
var azure = require('azure-storage');
var LZUTF8 = require('./../lib/lzutf8');

//var murmurhash = require('murmurhash');

var READ_CONCURRENCY = 10;
var WRITE_CONCURRENCY = 40;

var TABLE_BATCH_SIZE = 50;

require('moment-timezone');

var Agent = require('agentkeepalive').HttpsAgent;

var keepaliveAgent = new Agent({
    maxSockets: 1000,
    //maxFreeSockets: 100,
    timeout: 60000,
    keepAliveTimeout: 30000 // free socket keepalive for 30 seconds
});

var azureTable = require('azure-table-node');
var defaultClient = azureTable.createClient({
    agent: keepaliveAgent
});

//var SEED = 1234567;
//var TOTAL = 3000000;
//var BUCKET = TOTAL / 100;

var MAX_DAY = 99991231;
var MASK = '000000';


//var tableSvc = storage.tableService();

var createTable = function (callback) {
    // tableSvc.createTableIfNotExists(storage.viewerSummaryTable, function (error) {
    //     callback(error);
    // });

    defaultClient.createTable(storage.viewerSummaryTable, {ignoreIfExists: true}, function (error) {
        callback(error);
    });
};

var connectRedis = function (callback) {
    var client = redis.connect();
    client.on("error", function (err) {
        callback(err);
    });
    client.on("ready", function () {
        callback(null, client);
    });
};

//var connectTable = function (callback) {
//    storage.tableService();
//};

var getGames = function (redisClient, callback) {
    redisClient.hgetall('games', function (err, games) {
        callback(err, redisClient, games);
    });
};

var numFmt = function (num, mask) {
    return (mask + num).slice(-Math.max(mask.length, (num + '').length));
};

var processGroup = function (task, callback) {

    console.time('hgetall>' + task.key);
    console.time('processGroup > ' + task.key);
    task.redisClient.hgetall(task.key, function (err, pairs) {
        console.timeEnd('hgetall>' + task.key);
        if (err) {
            return callback(err);
        }

        var //entGen = azure.TableUtilities.entityGenerator,
            keyParts = task.key.split(':'),
            day = MAX_DAY - parseInt(keyParts[2], 10),
            partitionKey = util.format("%d:%s", day, numFmt(keyParts[3], MASK));

        var totals = _.reduce(pairs, function (result, dur, key) {
            var parts = key.split(':'),
                channel = parts[1],
                game = task.games[parts[2]] || parts[2],
                viewer = parts[0],
                duration = parseInt(dur, 10);

            result[viewer] = result[viewer] || {views: [], total: 0};
            result[viewer].total = (result[viewer].total || 0) + duration;
            result[viewer].views.push({game: game, channel: channel, duration: duration});

            return result;
        }, {});

        log.info('Processing Group', task.key, _.size(totals));


        var cargo = async.cargo(function (jobs, cargocb) {

            var batch = defaultClient.startBatch();

            _.each(jobs, function (job) {
                batch.insertOrReplaceEntity(storage.viewerSummaryTable, job);
            });

            batch.commit(function (err, data) {
                if (!err) {
                    task.stats.batches += 1;
                    task.stats.records += jobs.length;
                } else {
                    log.error(err);
                    task.errors += 1;
                }
                cargocb(null, data);
            });

        }, TABLE_BATCH_SIZE);

        var complete = function (err) {
            console.timeEnd('executeBatch > ' + task.key);
            console.timeEnd('processGroup > ' + task.key);
            if (err) {
                return callback(err);
            }

            if (task.errors > 0) {
                return callback(err);
            }

            // remove from cache
            //task.redisClient.del(task.key, function (err) {
            //    log.info('Removed key: ', task.key);
                callback(err);
            //});
        };

        var done = false;
        cargo.drain = function () {
{'fraction_non_alphanumeric': 0.15344959964900734, 'fraction_numerical': 0.016014039706043655, 'mean_word_length': 2.2196327683615817, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, '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': '7350878', 'n_tokens_mistral': 2891, 'n_tokens_neox': 2697, 'n_words': 594}