#!/usr/bin/env node

const fs = require('fs');
const parse = require('csv-parse');
const _ = require('underscore');
const underscored = require('underscore.string/underscored');
const moment = require('moment');
const db = require('./lib/db');
const {fixAmount, fixDate, normalizeNameAndAddress} = require('./lib/util');
const csvOptions = {columns: true};
const currentCommittees = {};
const batchSize = 10;

db.createTables().then(readCommittees);

function readCommittees() {
    const parser = parse(csvOptions);
    const input = fs.createReadStream(__dirname + '/' + db.committeeTableName + '.csv');
    const records = [];
    parser.on('readable', function () {
        let checked = false;
        let record;
        while ((record = parser.read())) {
            record = transformRecord(record);
            if (!checked) {
                if (!arraysAreEqual(db.committeeColumns, Object.keys(record))) {
                    console.warn(db.committeeColumns, Object.keys(record));
                    throw new Error('Committee columns have changed');
                }
                checked = true;
            }
            currentCommittees[record.committee_name] = true;
            records.push(record);
        }
    });
    parser.on('error', function (err) {
        console.error(err.message);
        throw err;
    });
    parser.on('finish', function () {
        console.log('Finished reading committees');
        console.log(Object.keys(currentCommittees).sort());
        db.batchInsertCommittees(records, batchSize)
            .then(readContributions);
    });
    input.pipe(parser);
}

function readContributions() {
    const parser = parse(csvOptions);
    const input = fs.createReadStream(__dirname + '/' + db.contributionTableName + '.csv');
    const seen = {};
    const unrecognized = [];
    let totalCount = 0;
    let currentCount = 0;
    let batch = [];
    parser.on('readable', function () {
        let record;
        while ((record = parser.read())) {
            totalCount++;
            record = transformRecord(record);
            if (totalCount === 1) {
                const expected = db.contributionColumns.filter(v => v !== 'normalized');
                if (!arraysAreEqual(expected, Object.keys(record))) {
                    console.warn(expected, Object.keys(record));
                    throw new Error('Contribution columns have changed');
                }
            }
            const name = record.committee_name;
            if (!seen[name]) {
                seen[name] = true;
                console.log(name);
                if (!currentCommittees[name]) {
                    unrecognized.push(name);
                }
            }
            if (!currentCommittees[name]) {
                continue;
            }
            record.normalized = normalizeNameAndAddress(record);
            batch.push(record);
            if (batch.length >= 10000) {
                db.batchInsertContributions(batch, batchSize)
                    .then(function () {});
                batch = [];
            }
            currentCount++;
        }
    });
    parser.on('error', function (err) {
        console.error(err.message);
        throw err;
    });
    parser.on('finish', function () {
        if (batch.length) {
            db.batchInsertContributions(batch, batchSize)
                .then(function () {});
        }
        console.log('Finished reading %d contributions', totalCount);
        console.log('Inserted %d contributions', currentCount);
        console.log('Unrecognized committees:\n', unrecognized.sort());
        readExpenditures();
    });
    input.pipe(parser);
}

function readExpenditures() {
    const parser = parse(csvOptions);
    const input = fs.createReadStream(__dirname + '/' + db.expenditureTableName + '.csv');
    const seen = {};
    const unrecognized = [];
    let totalCount = 0;
    let currentCount = 0;
    let batch = [];
    parser.on('readable', function () {
        let record;
        while ((record = parser.read())) {
            totalCount++;
            record = transformRecord(record);
            if (totalCount === 1) {
                const expected = db.expenditureColumns.filter(v => v !== 'normalized');
                if (!arraysAreEqual(expected, Object.keys(record))) {
                    console.warn(expected, Object.keys(record));
                    throw new Error('Expenditure columns have changed');
                }
            }
            const name = record.committee_name;
            if (!seen[name]) {
                seen[name] = true;
                console.log(name);
                if (!currentCommittees[name]) {
                    unrecognized.push(name);
                }
            }
            if (!currentCommittees[name]) {
                continue;
            }
            record.normalized = normalizeNameAndAddress(record);
            batch.push(record);
            if (batch.length >= 10000) {
                db.batchInsertExpenditures(batch, batchSize)
                    .then(function () {});
                batch = [];
            }
            currentCount++;
        }
    });
    parser.on('error', function (err) {
        console.error(err.message);
        throw err;
    });
    parser.on('finish', function () {
        if (batch.length) {
            db.batchInsertExpenditures(batch, batchSize)
                .then(finish);
        }
        else {
            finish();
        }

        function finish() {
            console.log('Finished reading %d expenditures', totalCount);
            console.log('Inserted %d expenditures', currentCount);
            if (unrecognized.length) {
                console.log('Unrecognized committees:\n', unrecognized.sort());
            }
            db.addDummyContributions();
        }
    });
    input.pipe(parser);
}

function transformRecord(record) {
    const newRecord = {};
    _.each(record, function (value, key) {
        value = trim(value);
        key = underscored(key);
        if (/date/.test(key) && /^\d\d?\/\d\d?\/\d{4}$/.test(value)) {
            value = fixDate(value);
        }
        else if (/amount|total/.test(key) && /^[$.,\d]+$/.test(value)) {
            value = fixAmount(value);
        }
        newRecord[key] = value;
    });
    return newRecord;
}

function trim(s) {
    return s ? s.replace(/\s+/g, ' ').trim() : '';
}

function arraysAreEqual(array1, array2) {
    return array1.length === array2.length && array1.every((value, index) => value === array2[index]);
}

@@ -175,42 +175,8 @@
 ');%0A
-const moment = require('moment');%0A
 cons
@@ -199,24 +199,24 @@
 ./lib/db');%0A
+
 const %7BfixAm
@@ -1349,35 +1349,36 @@
         console.
-log
+warn
 ('Finished readi
@@ -1380,16 +1380,19 @@
 reading 
+%25d 
 committe
@@ -1398,68 +1398,24 @@
 ees'
-);%0A        console.log(Object.keys(currentCommittees).sort()
+, records.length
 );%0A 
@@ -2479,43 +2479,8 @@
 ue;%0A
-                console.log(name);%0A
     
@@ -3299,35 +3299,36 @@
         console.
-log
+warn
 ('Finished readi
@@ -3370,35 +3370,36 @@
         console.
-log
+warn
 ('Inserted %25d co
@@ -3435,35 +3435,36 @@
         console.
-log
+warn
 ('Unrecognized c
@@ -4452,32 +4452,32 @@
 (!seen%5Bname%5D) %7B%0A
+
                 
@@ -4499,43 +4499,8 @@
 ue;%0A
-                console.log(name);%0A
     

tweak output of load-data.js