<commit_before>var _ = require('lodash'),
    fs = require('fs'),
    moment = require('moment'),
    models  = require('./models'),
    parser = require('fec-parse'),
    through2 = require('through2');

var filing_id = 1132265;

var parse = parser();

var formModels = [
    models.fec_paper_filing,
    models.fec_paper_contribution,
    models.fec_paper_expenditure,
    models.fec_filing,
    models.fec_presidential_summary,
    models.fec_pac_summary,
    models.fec_group_summary,
    models.fec_lobbyist_bundler,
    models.fec_contribution,
    models.fec_expenditure,
    models.fec_debt,
    models.fec_loan,
    models.fec_ie
];

var tableName = null;

fs.createReadStream(__dirname + '/../data/downloaded/' + filing_id + '.fec',{
        encoding: 'utf8'
    })
    .pipe(parser())
    .pipe(through2.obj(function (row, enc, next) {
        row.filing_id = filing_id;

        var model = formModels.find(function (model) {
            return model.match(row);
        });

        if ('report_id' in row && row.report_id) {
            row.report_id = parseInt(row.report_id.replace('FEC-',''));
        }

        if (typeof model !== 'undefined') {
            if (tableName !== model.tableName) {
                if (tableName !== null) {
                    this.push('\\.\n');
                }

                tableName = model.tableName;

                if ('created_date' in model.attributes) {
                    row.created_date = moment().toISOString();
                }
                if ('updated_date' in model.attributes) {
                    row.updated_date = moment().toISOString();
                }

                this.push('COPY ' + tableName + ' (' +
                            '"' + Object.keys(row).join('", "') + '"' +
                            ') FROM STDIN;\n');
            }
/*
            if (row.transaction_id == 'SA11AI_64058890') {
                console.error(row);
            }*/

            var line = _.values(row).map(function (value) {
                if (value && typeof value == 'string') {
                    return value.replace(/\\/g,'').replace(/\t/g,' ');
                }
                else if (value) {
                    return value;
                }
                return '\\N';
            }).join('\t') + '\n';
/*
            if (row.transaction_id == 'SA11AI_64058890') {
                console.error(line);
            }*/

            this.push(line);
        }

        next();
    },
    function (next) {
        this.push('\\.\n');

        next();
    }))
    .pipe(process.stdout);<commit_msg>Cleaned up experimental PSQL streamer, added decent escaping<commit_after>var _ = require('lodash'),
    fs = require('fs'),
    moment = require('moment'),
    models  = require('./models'),
    parser = require('fec-parse'),
    through2 = require('through2');

var filing_id = 1132265;

var parse = parser();

var formModels = [
    models.fec_paper_filing,
    models.fec_paper_contribution,
    models.fec_paper_expenditure,
    models.fec_filing,
    models.fec_presidential_summary,
    models.fec_pac_summary,
    models.fec_group_summary,
    models.fec_lobbyist_bundler,
    models.fec_contribution,
    models.fec_expenditure,
    models.fec_debt,
    models.fec_loan,
    models.fec_ie
];

var tableName = null;

function processRow(row, enc, next) {
    row.filing_id = filing_id;

    var model = formModels.find(function (model) {
        return model.match(row);
    });

    if ('report_id' in row && row.report_id) {
        row.report_id = parseInt(row.report_id.replace('FEC-',''));
    }

    if (typeof model !== 'undefined') {
        if (tableName !== model.tableName) {
            if (tableName !== null) {
                this.push('\\.\n');
            }

            tableName = model.tableName;

            if ('created_date' in model.attributes) {
                row.created_date = moment().toISOString();
            }
            if ('updated_date' in model.attributes) {
                row.updated_date = moment().toISOString();
            }

            this.push('COPY ' + tableName + ' (' +
                        '"' + Object.keys(row).join('", "') + '"' +
                        ') FROM STDIN;\n');
        }

        var line = _.values(row).map(function (value) {
            if (value && typeof value == 'string') {
                return value
                        .replace(/\\/g,'\\\\')
                        .replace(/\n/g,'\\\n')
                        .replace(/\t/g,'\\\t')
                        .replace(/\\./g,'\\\\.');
            }
            else if (value) {
                return value;
            }
            return '\\N';
        }).join('\t') + '\n';

        this.push(line);
    }

    next();
}

fs.createReadStream(__dirname + '/../data/downloaded/' + filing_id + '.fec',{
        encoding: 'utf8'
    })
    .pipe(parser())
    .pipe(through2.obj(processRow,
    function (next) {
        this.push('\\.\n');

        next();
    }))
    .pipe(process.stdout);
{'fraction_non_alphanumeric': 0.15491559086395235, 'fraction_numerical': 0.008341608738828202, 'mean_word_length': 1.620187304890739, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6218236', 'n_tokens_mistral': 1601, 'n_tokens_neox': 1511, 'n_words': 257}