<commit_before>var trumpet = require('trumpet');
var Transform = require('readable-stream/transform');
var hyperglue = require('hyperglue');
var encode = require('ent').encode;
var through = require('through');

module.exports = function (html, opts, cb) {
    if (typeof opts === 'function') {
        cb = opts;
        opts = {};
    }
    if (!opts) opts = {};
    
    var tf = new Transform({ objectMode: true });
    tf._transform = function (line, _, next) {
        var row;
        if (typeof line === 'string' || Buffer.isBuffer(line)) {
            try { row = JSON.parse(line) }
            catch (err) { this.emit('error', err) }
        }
        else row = line;
        var res = cb(row);
        if (!res) return next();
        
        var tr = trumpet();
        if (opts.key && row.key) {
            tr.select('*').setAttribute(opts.key, row.key);
        }
        
        tr.on('data', function (buf) { tf.push(buf) });
        tr.on('end', function () { next() });
        
        Object.keys(res).forEach(function (key) {
            if (key === ':first') {
                each(key, tr.select('*'));
            }
            else if (/:first$/.test(key)) {<commit_msg>convert row.key to a string if it isn't one already<commit_after>var trumpet = require('trumpet');
var Transform = require('readable-stream/transform');
var hyperglue = require('hyperglue');
var encode = require('ent').encode;
var through = require('through');

module.exports = function (html, opts, cb) {
    if (typeof opts === 'function') {
        cb = opts;
        opts = {};
    }
    if (!opts) opts = {};
    
    var tf = new Transform({ objectMode: true });
    tf._transform = function (line, _, next) {
        var row;
        if (typeof line === 'string' || Buffer.isBuffer(line)) {
            try { row = JSON.parse(line) }
            catch (err) { this.emit('error', err) }
        }
        else row = line;
        var res = cb(row);
        if (!res) return next();
        
        var tr = trumpet();
        if (opts.key && row.key) {
            var rk = typeof row.key === 'string' ? row.key : String(row.key);
            tr.select('*').setAttribute(opts.key, rk);
        }
        
        tr.on('data', function (buf) { tf.push(buf) });
        tr.on('end', function () { next() });
        
        Object.keys(res).forEach(function (key) {
            if (key === ':first') {
                each(key, tr.select('*'));
            }
            else if (/:first$/.test(key)) {
{'fraction_non_alphanumeric': 0.17497010761259466, 'fraction_numerical': 0.0, 'mean_word_length': 1.9391100702576112, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5020103', 'n_tokens_mistral': 758, 'n_tokens_neox': 748, 'n_words': 194}