| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94 | 1x
1x
1x
1x
1x
1x
1x
1x
1x
1x
5x
5x
5x
5x
5x
8x
320x
2x
2x
1x
1x
1x
1x
2x
2x
6x
6x
2x
2x
2x
1x
1x
1x
2x
2x
2x
2x
2x
2x
2x
2x
8x
8x
8x
8x
122x
122x
122x
90548x
90436x
8x
8x
1x
1x
1x
1x
1x
1x
| const zlib = require("zlib")
const fs = require("fs")
const path = require("path")
const stream = require("stream")
const mkdirp = require("mkdirp")
const Multistream = require("multistream")
const pump = require('pump')
const async = require('async')
const split = require('split')
const mapS = require('map-stream')
class Unpack {
constructor(opts, config, logger) {
this.logger = logger
this.tableFilter = opts.filter
this.sourceLocation = path.resolve(process.cwd(), config.saveLocation)
this.outputLocation = path.resolve(process.cwd(), config.unpackLocation || './unpackedFiles')
this.schemaLocation = path.join(this.sourceLocation, 'schema.json')
}
buildTitlesHeader(table) {
return table.columns.map((c) => {
return c.name
}).join("\t") + '\n'
}
loadSchema(cb) {
fs.stat(this.schemaLocation, (err, stat) => {
if (err && !stat) {
this.logger.error('could not find schema, have you downloaded files yet?')
return cb(err)
}
const schema = require(this.schemaLocation)
cb(null, schema)
})
}
addTitleAndUnzip(schema, sourceDir, outputDir, cb) {
const toUnpack = []
for (let key in schema.schema) {
let table = schema.schema[key]
if (this.tableFilter.indexOf(table.tableName) >= 0 ) {
toUnpack.push(table)
}
}
this.logger.debug(`will unpack ${toUnpack.map((p) => p.tableName).join(',')}`)
if (toUnpack.length === 0) {
this.logger.warn('no files matched filter, nothing will be unpacked')
return cb()
}
async.each(toUnpack, (table, cb) => {
const inputDir = path.join(sourceDir, table.tableName)
const outputTableName = path.join(outputDir, table.tableName + '.txt')
const outputStream = fs.createWriteStream(outputTableName)
this.logger.info(`outputting ${table.tableName} to ${outputTableName}`)
this.processTable(table, inputDir, outputStream, (err) => {
Iif (err) return cb(err)
this.logger.info(`finished with ${table.tableName}`)
cb()
})
}, cb)
}
processTable(table, inputDir, outputStream, cb) {
outputStream.write(this.buildTitlesHeader(table))
fs.readdir(inputDir, (err, files) => {
Iif (err) return cb(err)
const streamCreators = files.map((f) => {
// return a function so that mulitstream
// lazily creates the streams
return function() {
const gunzip = zlib.createUnzip()
return fs.createReadStream(path.join(inputDir, f))
.pipe(gunzip)
.pipe(split())
.pipe(mapS((item, cb) => {
if (item.trim() === '') return cb()
// add newlines for each row
return cb(null, item + '\n')
}))
}
})
const multi = new Multistream(streamCreators)
pump(multi, outputStream, cb)
})
}
run(cb) {
this.loadSchema((err, schema) => {
Iif (err) return cb(err)
mkdirp(this.outputLocation, (err) => {
Iif (err) return cb(err)
this.addTitleAndUnzip(schema, this.sourceLocation, this.outputLocation, cb)
})
})
}
}
module.exports = Unpack
|