<commit_before>#!/usr/bin/env node
// parse2ncmb.js
(function(global) {
    "use strict";

    var Configfile = require('config')
      , fs = require('fs')
      , globule = require('globule')
      , JSONStream = require('JSONStream')
      , NCMB = require('ncmb')
      , program = require('commander')
      , Converter = require('./lib/converter')
      , throat = require('throat')
    ;

    // handling command-line
    program.version('0.0.1')
	.usage('[options] <directory>')
	.option('-c, --concurrency <number>', 'Set parallel concurrency',
		parseInt)
	.parse(process.argv);

    if (!program.args.length) {
	program.help();
	return;
    }
    var concurrency = 3; // default
    if (program.concurrency != undefined) {
	concurrency = program.concurrency;
    }

    let targetDir = program.args[0];
    // trim tailing '/'
    if (targetDir[targetDir.length - 1] == '/') {
	targetDir = targetDir.slice(0, -1);
    }
    let files = globule.find(targetDir + '/*');

    let getPathInfo = function(path) {
	let typemap = {
	    '_Installation.json': 'installation'
	    , '_Product.json': 'product'
	    , '_Role.json': 'role'
	    , '_User.json': 'user'
	};
	let joinPrefix = '_Join:';
	let type = null;
	let name = null;

	let file = path.replace(/.*\//, '');

	if (file.substr(0, joinPrefix.length) === joinPrefix) {
	    type = 'join';
	    name = file;
	} else if (typemap[file] !== undefined) {
	    type = typemap[file];
	    name = '';
	} else {
	    type = 'object';
	    let match = file.match(/(.*)\.json$/);
	    if (match === null) {
		type = null;
	    } else {
		name = match[1];
	    }
	}
	return {
	    path: path
	    , file: file
	    , type: type
	    , name: name
	};
    }

    var fileInfos = [];
    files.forEach(function(path) {
	let info = getPathInfo(path);
	if (info.type !== null) {
	    fileInfos.push(info);
	}
    });

    // sort for 'join' type will be end of array
    fileInfos.sort(function(a, b) {
	if (a.type == 'join') {
	    if (b.type == 'join') {
		return a.path - b.path;
	    } else {
		return 1;
	    }
	} else {
	    if (b.type == 'join') {
		return -1;
	    } else {
		return a.path - b.path;
	    }
	}
    });

    // initialize NCMB
    let app_key = Configfile.config.app_key;
    let client_key = Configfile.config.client_key;
    var ncmb = new NCMB(app_key, client_key);

    var Parallel = throat(Promise)(concurrency);

    fileInfos.forEach(function(info) {
	var converter = new Converter(ncmb, info.type, info.name);

	fs.createReadStream(info.path)
	    .pipe(JSONStream.parse('results.*'))
	    .on('data', function(data) {
		Parallel(function() {
		    return converter.convert(data)
		})
		.then(function(results) {
		    // something on success?
		})
		.catch(function(err) {
		    console.log(err);
		})
		;
	    });
    });
    return;

})((this || 0).self || global);<commit_msg>Add 2 phase process to import data.<commit_after>#!/usr/bin/env node
// parse2ncmb.js
(function(global) {
    "use strict";

    var Configfile = require('config')
      , fs = require('fs')
      , globule = require('globule')
      , JSONStream = require('JSONStream')
      , NCMB = require('ncmb')
      , program = require('commander')
      , Converter = require('./lib/converter')
      , ObjMapper = require('./lib/objmapper')
      , throat = require('throat')
    ;

    // handling command-line
    program.version('0.0.1')
	.usage('[options] <directory>')
	.option('-c, --concurrency <number>', 'Set parallel concurrency',
		parseInt)
	.option('-p, --phase <number>', 'Specify phase number', parseInt)
	.option('-s, --silent', 'Silent mode')
	.parse(process.argv);

    if (!program.args.length) {
	program.help();
	return;
    }
    var concurrency = 3; // default
    if (program.concurrency != undefined) {
	concurrency = program.concurrency;
    }
    if (program.phase === undefined) {
	console.error('Phase number required');
	process.exit(1);
    }
    if (program.phase != 1 && program.phase != 2) {
	console.error('Invalid phase number');
	process.exit(1);
    }

    let targetDir = program.args[0];
    // trim tailing '/'
    if (targetDir[targetDir.length - 1] == '/') {
	targetDir = targetDir.slice(0, -1);
    }
    let files = globule.find(targetDir + '/*');

    let getPathInfo = function(path) {
	let typemap = {
	    '_Installation.json': 'installation'
	    , '_Product.json': 'product'
	    , '_Role.json': 'role'
	    , '_User.json': 'user'
	};
	let joinPrefix = '_Join:';
	let type = null;
	let name = null;

	let file = path.replace(/.*\//, '');

	if (file.substr(0, joinPrefix.length) === joinPrefix) {
	    // join type
	    type = 'join';
	    let match = file.match(/^_Join:(.*)\.json$/);
	    name = match[1];
	} else if (typemap[file] !== undefined) {
	    // preset types
	    type = typemap[file];
	    let match = file.match(/(.*)\.json$/);
	    name = match[1];
	} else {
	    // other
	    type = 'object';
	    let match = file.match(/(.*)\.json$/);
	    if (match === null) {
		type = null;
	    } else {
		name = match[1];
	    }
	}
	return {
	    path: path
	    , file: file
	    , type: type
	    , name: name
	};
    }

    var objFiles = [], relFiles = [];
    files.forEach(function(path) {
	let info = getPathInfo(path);
	if (info.type === null) {
	    return;
	}
	if (info.type == 'join') {
	    relFiles.push(info);
	} else {
	    objFiles.push(info);
	}
    });

    // initialize NCMB
    let app_key = Configfile.config.app_key;
    let client_key = Configfile.config.client_key;
    var ncmb = new NCMB(app_key, client_key);
    var objMapper = new ObjMapper;

    var Parallel = throat(Promise)(concurrency);

    let targetFiles = null;
    if (program.phase == 1) {
	// Phase 1: store objects
	targetFiles = objFiles;
	objMapper.reset();
    } else {
	// Phase 2: store relations and others
	targetFiles = relFiles;
	objMapper.ensureIndex();
    }

    targetFiles.forEach(function(info) {
	var converter = new Converter(ncmb, info.type, info.name, objMapper);

	fs.createReadStream(info.path)
	    .pipe(JSONStream.parse('results.*'))
	    .on('data', function(data) {
		Parallel(function() {
		    return converter.convert(data)
		})
		.then(function(results) {
		})
		.catch(function(err) {
		    console.error(err);
		});
	    });
    });

})((this || 0).self || global);
{'fraction_non_alphanumeric': 0.17679733375654658, 'fraction_numerical': 0.00571337882875734, 'mean_word_length': 2.5929304446978336, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, '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': '1916768', 'n_tokens_mistral': 2300, 'n_tokens_neox': 2169, 'n_words': 462}