watch-finished/index.js

'use strict';

const fs = require('fs');
const path = require('path');
const minimist = require('minimist');
const low = require('lowdb');
const Queue = require('better-queue');
const FileSync = require('lowdb/adapters/FileSync');

const chokidar = require('chokidar');
const hbjs = require('handbrake-js');

const settings = require('./data/settings.json');

const MISSING_THRESHOLD = 10;
const MISSING_INTERVAL = 30 * 1000;

const args = minimist(process.argv.slice(2), {
  alias: {
    h: 'help',
    v: 'version',
    c: 'config',
    i: 'ignoreInitial',
    t: 'test',
  },
});

const version = () => {
  const pkg = require('./package.json');
  return console.log(pkg.version);
};

const help = () => {
  let message = `Usage:
\tnpm run movies
\tnpm run tvshows
\tnpm run pr0n
\tnpm run all
\tnpm run help
\tnpm run version\n`;
  return console.log(message);
};

String.prototype.toTitleCase = function() {
  var i, j, str, lowers, uppers;
  str = this.replace(/([^\W_]+[^\s-]*) */g, function(txt) {
    return txt.charAt(0).toUpperCase() + txt.substr(1).toLowerCase();
  });
  // Certain minor words should be left lowercase unless
  // they are the first or last words in the string
  lowers = (settings && settings.titlecase && settings.titlecase.lowers) || [];
  for (i = 0, j = lowers.length; i < j; i++)
    str = str.replace(new RegExp('\\s' + lowers[i] + '\\s', 'g'), function(txt) {
      return txt.toLowerCase();
    });
  // Certain words such as initialisms or acronyms should be left uppercase
  uppers = (settings && settings.titlecase && settings.titlecase.uppers) || [];
  for (i = 0, j = uppers.length; i < j; i++) str = str.replace(new RegExp('\\b' + uppers[i] + '\\b', 'g'), uppers[i].toUpperCase());
  return str;
};

// get the db from the path
const getDbForDir = (dir, opts) => {
  let options = opts || Object.assign({}, defaults, paths[dir]); // baseline the options
  let adapter = new FileSync(options.database); // init the db adapter
  let db = low(adapter); // connect the db
  return db;
};

// find a file in the db
const findFile = (db, file) => {
  return db
    .get('files')
    .find({
      input: file,
    })
    .value(); // does it already exist?
};

// set a file in the db
const setFile = (db, file, payload) => {
  if (!payload && typeof file === 'object')
    return db
      .get('files')
      .push(file)
      .write();
  return db
    .get('files')
    .find({
      input: file,
    })
    .assign(payload)
    .write();
};

// remove a file from the db
const removeFile = (db, file) => {
  return db
    .get('files')
    .remove({
      input: file,
    })
    .write(); // remove file from database
};

// write process output
const processOutput = str => {
  process.stdout.clearLine();
  process.stdout.cursorTo(0);
  process.stdout.write(str);
};

const getFilesize = async file => {
  let stats = fs.existsSync(file) ? fs.statSync(file) : false;
  let bytes = stats ? stats['size'] : 0;
  return bytes && bytes > 0 ? bytes / 1024 : 0;
};

// spawn a handbrake
const processWithHandbrake = (input, output, preset) => {
  return new Promise((resolve, reject) => {
    const inputName = path.basename(input);
    const outputName = path.basename(output);
    hbjs
      .spawn({
        input: input,
        output: output,
        preset: preset,
      })
      .on('start', err => {
        processOutput(` -> "${outputName}" [starting] (${new Date()} with "${preset}")`);
      })
      .on('error', err => {
        processOutput(` -> "${outputName}" [errored] (${new Date()}: ${err.message || err})\n`);
        reject(err);
      })
      .on('progress', progress => {
        processOutput(` -> "${outputName}" [transcoding] (${progress.percentComplete}%, ETA: ${progress.eta})`);
      })
      .on('cancelled', () => {
        processOutput(` -> "${outputName}" [cancelled] (${new Date()})\n`);
        reject(new Error(`Processing "${outputName}" cancelled`));
      })
      .on('complete', () => {
        processOutput(` -> "${outputName}" [completed] (${new Date()} with "${preset}")\n`);
        resolve(true);
      });
  });
};

const processFile = async file => {
  let db,
    options,
    preset = 'Fast 1080p30',
    clean = {
      '.': ' ',
    },
    ext = 'm4v',
    filename = path.parse(file).name,
    input = file,
    output = filename,
    titlecase = false,
    folder = false,
    destination = '';
  // determine the presets and clean object
  for (let i = 0, l = dirs.length; i < l; i++) {
    let dir = dirs[i]; // pointer to the dir
    if (file && dir && file.indexOf(dir) > -1) {
      // is this in this path?
      options = Object.assign({}, defaults, paths[dir]); // baseline the options
      db = getDbForDir(dir, options); // init the db connection
      let found = findFile(db, file); // does it already exist?
      if (found && found.status && found.status === 'success') {
        // was it already processed?
        return false; // break this loop
      } else if (!found) {
        // was it found? .. nope
        processOutput(` -> "${path.basename(file)}" [processing]`);
        setFile(db, {
          input: file,
          output: '',
          status: '',
          date: new Date(),
        }); // push onto the list an entry
      } else {
        processOutput(` -> "${path.basename(file)}" [re-processing]`);
        setFile(db, file, {
          status: '',
          date: new Date(),
        }); // set the status to blank
      }
      preset = options.preset; // apply the specifics ...
      clean = options.clean; // ...
      ext = options.ext; // ...
      titlecase = !!options.titlecase; // ...
      folder = !!options.folder; // ...
      destination = options.destination; // ...
      break; // break the loop
    }
  }
  // clean the output name
  let cleanKeys = Object.keys(clean);
  for (let c in cleanKeys) {
    let key = cleanKeys[c];
    let val = clean[key];
    let re = new RegExp(key, 'gi');
    output = output.replace(re, val);
  }
  // baseline the output name
  output = output.trim(); // trim the whitespace
  output = output.replace(/(\d{4})$/, `($1)`); // if there is a date at the end wrap it
  // do we have a sub folder option?
  if (folder) {
    let match = output.match(/^(.*?)?\./gm); // get the name for the file before the first .
    folder = match && match.length > 0 ? match[0].slice(0, -1) : false; // get just the stuff before the dot ... or false
  }
  // do we have title case enabled?
  if (options.titlecase) output = output.toTitleCase(); // titlecase that string
  // baseline the target
  let target = destination + (folder ? '/' + folder : ''); // setup the target location
  output = target + '/' + output + '.' + ext; // update the new name
  // do we already have an existing output that matches?
  if (fs.existsSync(output)) {
    let outputSize = await getFilesize(output); //get output filesize
    if (outputSize > 100) {
      //make sure its bigger than 100k
      processOutput(` -> "${path.basename(file)}" [skipping] (already processed)\n`);
      setFile(db, file, {
        output: output,
        status: 'success',
        date: new Date(),
      }); // update database with status
      return false;
    }
  }
  //process.stdout.write('\n'); // send a new line
  // update database with output name
  setFile(db, file, {
    output: output,
  });
  // create parent if required
  if (target && !fs.existsSync(target)) {
    processOutput(` -> "${path.basename(file)}" [creating parent directory] ("${target}")`);
    fs.mkdirSync(target, {
      recursive: true,
    });
  }
  // spawn handbrake
  if (!test) {
    try {
      await processWithHandbrake(input, output, preset);
      setFile(db, file, {
        status: 'success',
        date: new Date(),
      }); // update database with status
    } catch (err) {
      setFile(db, file, {
        status: 'failure',
        date: new Date(),
      }); // update database with status
    }
  }
  return true; // when complete return true
};

// cleanup removes from the db
const cleanup = file => {
  let db;
  for (let i = 0, l = dirs.length; i < l; i++) {
    let dir = dirs[i]; // pointer to the dir
    if (file && dir && file.indexOf(dir) > -1) {
      // is this in this path?
      db = getDbForDir(dir); // init the db connection
      let exists = fs.existsSync(file); // check if the file actually exists
      let entry = findFile(db, file); // get the existing data record
      // if the file esists and the missing counder is less than the treshold update the counter
      if (exists) {
        console.log(` -> "${file}" [exists] (${new Date()})`);
        setFile(db, file, { missing: 0 }); // update the missing count to 0
      } else if (entry && (!entry.missing || entry.missing < MISSING_THRESHOLD)) {
        console.log(` -> "${file}" [missing] ${entry.missing || 1} (${new Date()})`);
        setFile(db, file, { missing: (entry.missing || 0) + 1 }); // update the missing count
        setTimeout(() => {
          cleanup(file);
        }, MISSING_INTERVAL); // wait for the missing interval to check again
      } else {
        removeFile(db, file); // remove file form database
      }
    }
  }
};

// handle args
if (args.version) return version(); // show version
if (args.help || !args.config) return help(); // show help
const ignoreInitial = args.hasOwnProperty('ignoreInitial') ? args.ignoreInitial : false; // ignore initial files
const test = args.hasOwnProperty('test') ? args.test : false; // do a dry run ... just log

// get our paths
const paths = require(args.config);

// init our defaults
const defaults = {
  preset: 'Fast 1080p30',
  clean: {},
  titlecase: false,
  folder: false,
  database: 'data/db.json',
};

// setup watcher options
const opts = {
  ignored: /(^|[\/\\])\..|([s|S]ample\.*)/,
  ignoreInitial: ignoreInitial,
  persistent: true,
  usePolling: true,
  interval: 10*1000,
  depth: 1,
  awaitWriteFinish: {
    stabilityThreshold: 3000,
    pollInterval: 1000,
  },
  ignorePermissionErrors: false,
  atomic: true,
};

// parse the paths to dirs
const dirs = Object.keys(paths);

// initialize watches and db then start the watcher
const main = async () => {
  var queue = new Queue(
    async (input, cb) => {
      // init the queue
      let result = await processFile(input); // process the queue
      cb(null, result);
    },
    {
      batchSize: 1,
      concurrent: 1,
      maxRetries: 3,
      retryDelay: 5000,
    }
  );
  var watches = []; // array of things to watch
  for (let d in dirs) {
    // loop the dirs
    let dir = dirs[d]; // pointer
    let options = Object.assign({}, defaults, paths[dir]); // baseline the options
    let db = getDbForDir(dir);
    db.defaults({
      files: [],
    }).write(); // init the database
    for (let e in options.exts) {
      // loop the exts to watch
      let ext = options.exts[e]; // alias the ext
      watches.push(`${dir}/**/*.${ext}`); // push the watch
    }
  }
  const watcher = chokidar.watch(watches, opts); // init our watcher
  console.log('Watching:', watches);
  watcher
    .on('add', async file => {
      // when a file is added ...
      queue.push(file); // push the file onto the queue to be processed
    })
    .on('change', file => {
      // when a file changes ...
      console.log(` -> "${file}" [changed] (${new Date()})`);
    })
    .on('unlink', async file => {
      // when a file is removed ...
      console.log(` -> "${file}" [removed] (${new Date()})`);
      cleanup(file);
    })
    .on('error', error => {
      console.error(` -> "${file}" [errored] (${new Date()}: ${error.message || error})`);
    });
};

(async () => {
  main();
})().catch(err => {
  console.log(` -> Error: ${err.message || err}`);
});