Notes

if (parseInt(process().env.PORT) % 1000 !== 0) return;
var cronType = "document-processor";
var path = require('path');
var root = process().cwd();
var mspawn = require('spawn-module');
var Promise = require('bluebird');
var pubsub = require(path.join(root, 'lib/pubsub'));
var async = Promise.promisifyAll(require('async'));
var _ = require('underscore');
var s3class = require(path.join(root, 'lib/s3'));
var s3 = new s3class();
var cronID = false;
var defaults = {
username: process().env.DEFAULT_USER || 'Amazon S3'
};

var thisCron = {};

var start;

var processor = function (filepath, type) {
return new Promise(function (resolve, reject) {
mspawn(require, path.join(root, 'lib/processor.js'), function (err, proc) {
if (err) return reject(err);
proc.do(filepath, type, function (err, res) {
mspawn.kill(proc);
if (err) return reject(err);
resolve(res);
});
});
});
};

var exists = function () {
return new Promise(function (resolve, reject) {
dpd.cronlogs.get({
name: cronType
}).then(function (data) {
if (data.length !== 0) return reject({
exists: true
});
resolve();
}).catch(reject);
});
};

var create = function (documentId) {
return new Promise(function (resolve, reject) {
dpd.cronlogs.post({
name: cronType,
timestamp: Date.now(),
document: documentId
}).then(function (cron) {
cronID = cron.id;
resolve();
}).catch(reject);
});
};

var remove = function () {
var defer = Promise.defer();
if (thisCron.exists) {
defer.resolve();
} else {
dpd.cronlogs.get({
name: cronType
}).then(function (data) {
return async.eachAsync(data, function (item, callback) {
dpd.cronlogs.del({
id: item.id
}).then(function () {
callback();
}).catch(callback);
});
}).then(defer.resolve).catch(defer.reject);
}

return defer.promise;
};

var bulkInsert = function (data) {
var extratags = data.extratags;
var pagesData = data.text;
var docID = data.id;
data = _.omit(data, 'text', 'id', '_id');
return new Promise(function (resolve, reject) {
async.each(pagesData, function (page, cb) {
var entry = _.extend(data, {
text: page.text
}, {
thumbs: page.thumbs,
document: docID,
numberOfPages: pagesData.length,
status: 'new',
pageNo: page.pageNo,
portrait: page.portrait
});
//fix for pdfs taken from edgar
if (extratags && extratags.length) {
entry.status = 'tagged';
}
if (entry.pageNo !== 0) {
entry.extratags = [];
}
dpd.uploaderfileupload.post(_.omit(entry, 'id', '_id')).then(function () {
cb();
}).catch(cb);
}, function (err) {
if (err) return reject(err);
resolve(data);
});
});
};


exists().then(function () {
return dpd.uploaderfileupload.get({
text: {
$exists: false
},
error: {
$exists: false
},
status: {
$eq: 'processing'
},
$limit: 1
});
}).then(function (data) {
if (data.length === 0) return new Promise.reject({
nodata: true
});
thisCron.document = data[0];
return create(data[0].id);
}).then(function (data) {
thisCron.data = data;
start = Date.now();
return processor(path.join(root, 'public/upload', thisCron.document.subdir, thisCron.document.filename), thisCron.document.type);
}).then(function (data) {
thisCron.document.text = _.map(data.data, function (page) {
if (page.text) page.text = page.text.replace(/[^a-z0-9 ,.?@!$&]/gim, '');
return page;
});
thisCron.document.status = 'new';
thisCron.document.reprocessed = true;
thisCron.document.thumbfixed = true;
thisCron.document.duration = Date.now() - start;
thisCron.document.numberOfPages = thisCron.document.text.length;
thisCron.document.uniq = true;
thisCron.document.reviewed = true;
thisCron.document.researcher = "new";
return thisCron.document;
}).then(function () {
if (!thisCron.document.creator) return dpd.users.get({
username: defaults.username
});
return false;
}).then(function (data) {
if (data && data[0] && data[0].id) thisCron.document.creator = data[0].id;
return bulkInsert(thisCron.document);
}).then(function () {
return new Promise(function (resolve, reject) {
dpd.uploaderfileupload.del({
_id: thisCron.document.id
}).then(resolve).catch(reject);
});
}).then(function (data) {
pubsub.emit('cron:process-documents', data);
}).then(function () {
return s3.uploadFile(path.join(root, 'public/upload', thisCron.document.subdir, thisCron.document.filename));
}).catch(function (err) {
if (err.exists) thisCron.exists = true;
if (err.exists || err.nodata) return;
if (thisCron && thisCron.document && thisCron.document.document) {
dpd.uploaderfileupload.put(thisCron.id, {
status: "error",
error: JSON.stringify(err)
}).then(function (data) {
pubsub.emit('cron:process-documents', data);
}).catch(console.log);
}
}).finally(function () {
remove();
});