如何处理大体积 XLSX/CSV/TXT 文件？

  

• 大体积文件支持切片上传  
  
• 可以断点续传  
  
• 可以得知上传进度
  

function sliceInPiece(file, piece = 1024 * 1024 * 5) {
  let totalSize = file.size; // 文件总大小
  let start = 0; // 每次上传的开始字节
  let end = start + piece; // 每次上传的结尾字节
  let chunks = []
  while (start < totalSize) {
  // 根据长度截取每次需要上传的数据
  // File对象继承自Blob对象，因此包含slice方法
  let blob = file.slice(start, end); 
  chunks.push(blob)
  start = end;
  end = start + piece;
  }
  return chunks
}

let file =  document.querySelector("[name=file]").files[0];
const LENGTH = 1024 * 1024 * 0.1;
let chunks = sliceInPiece(file, LENGTH); // 首先拆分切片
chunks.forEach(chunk=>{
  let fd = new FormData();
  fd.append("file", chunk);
  post('/upload', fd)
})

  

• 文件分块：直接切分二进制文件成小块。分块大小固定为 1M。最后一个分块除外。  
  
• 上传阶段：使用 x-upyun-multi-stage 参数来指示断点续传的阶段。分为以下三个阶段: initate(上传初始化), upload(上传中), complete(上传结束)。各阶段依次进行。  
  
• 分片序号：使用 x-upyun-part-id 参数来指示当前的分片序号，序号从 0 起算。  
  
• 顺序上传：对于同一个断点续传任务，只支持顺序上传。  
  
• 上传标识：使用 x-upyun-multi-uuid 参数来唯一标识一次上传任务, 类型为字符串, 长度为 36 位。  
  
• 上传清理：断点续传未完成的文件，会保存 24 小时，超过后，文件会被删除。
  

export function createElement (tagName: string, vnode: VNode): Element {
  const elm = document.createElement(tagName)
  if (tagName !== 'select') {
  return elm
  }
  // false or null will remove the attribute but undefined will not
  if (vnode.data && vnode.data.attrs && vnode.data.attrs.multiple !== undefined) {
  elm.setAttribute('multiple', 'multiple')
  }
  return elm
}

export function createElement (tagName: string): WeexElement {
  return document.createElement(tagName)
}

import * as nodeOps from 'web/runtime![]()de-ops'\
import { createPatchFunction } from 'core![]()dom/patch'\
import baseModules from 'core![]()dom/modules/index'\
import platformModules from 'web/runtime/modules/index'\
\
// the directive module should be applied last, after all\
// built-in modules have been applied.\
const modules = platformModules.concat(baseModules)\
\
// nodeops 封装了一系列DOM操作方法。modules定义了一些模块的钩子函数的实现\
export const patch: Function = createPatchFunction({ nodeOps, modules })

import * as nodeOps from 'weex/runtime![]()de-ops'\
import { createPatchFunction } from 'core![]()dom/patch'\
import baseModules from 'core![]()dom/modules/index'\
import platformModules from 'weex/runtime/modules/index'\
\
// the directive module should be applied last, after all\
// built-in modules have been applied.\
const modules = platformModules.concat(baseModules)\
\
export const patch: Function = createPatchFunction({\
  nodeOps,\
  modules,\
  LONG_LIST_THRESHOLD: 10\
})

  

• fs.read(fd, buffer, offset, length, position, callback)读取文件数据。要操作文件，得先打开文件，这个方法的fd，就是调用 fs.open 返回的文件描述符。  
  
• fs.readFile(path[, options], callback) 异步地读取文件的全部内容。可以看做是fs.read的进一步封装。
  

import { readFile } from 'fs';
readFile('/etc/passwd','utf-8', (err, data) => {
  if (err) throw err;
  console.log(data);
});

// pipe 消费
readable.pipe(writable)
// 或者
// 事件消费
readable.on('data', (chunk) => {
  writable.write(chunk);
});
readable.on('end', () => {
  writable.end();
});

async readOneByOne () {
   try {
  for (let i = 0; i < needParsePathArr.length; i++) {
    const filePath = needParsePathArr[i]
    console.log(`解析到第${i}个文件，文件名：${filePath}`)
    await streamInsertDB(filePath)
  }
  } catch (err) {
  }
}

async function streamInsertDB (filePath) {
  return new Promise((resolve, reject) => {
  const ext = path.extname(filePath)
  // 判断了下文件类型
  if (ext === '.csv') {
    // 解析csv
    parseAndInsertFromCSV(filePath, resolve, reject)
  } else if (ext === '.xlsx') {
    // 自执行函数
    (async function getName () {
    try {
      // 先转换成csv。也可以不转换，直接解析xlsx，后文会详细解释。
      const csvFileName = await convertXlsx2Csv(filePath)
      // 复用解析csv的逻辑
      parseAndInsertFromCSV(csvFileName, resolve, reject)
    } catch (error) {
      reject(`error: ${error.message || error}`)
    }
    })()
  }
  })
}

  

• chardet：这个函数的作用是监测 CSV 文件的编码格式的，毕竟不是每个 CSV 都是 UTF-8 编码，带中文的 CSV 编码类型可能是 GBK 或者 GB18030、GB18031 等等，这种格式不经过处理直接读取，中文会显示为乱码。所以需要执行转换的函数 iconv 转换一下。  
  
• pipe：可以用来建立管道链，可以理解为 pipe 的作用就像一个管道，可以对目标流边读边写，这里我们是一边解码一边重新编码。  
  
• insertInBlock：这个函数是获取到一定数量的数据后（本例中是从 CSV 中解析出 3 万条左右数据的时候），暂停一下来执行一些操作，比如写入数据库或者对里面的数据进行过滤、处理等等，根据实际需要来定。  
  
• csv：这个函数的作用就是读出流中的具体数据的。
  

const chardet = require('chardet');
const csv = require('fast-csv'); // 比较快解析csv的速度的工具
const iconv = require('iconv-lite');
const arrayFromParseCSV = []  // 存放解析出来的一行行csv数据的
let count = 0 // 计数
// resolve, reject 是外部函数传进来的，用以判断函数执行的状态，以便正确的进行后续逻辑处理
function parseAndInsertFromCSV (filePath, resolve, reject) {
  const rs = fs.createReadStream(filePath)  // 创建可读流
  // 这里的防抖和柯里化
  const delayInsert = debounce((isEnd, cb = () => {}) => insertInBlock(isEnd, cb, rs, resolve, reject), 300)
  /// sampleSize: 5120 表示值读取文件前5120个字节的数据，就可以判断出文件的编码类型了，不需要全部读取
  chardet.detectFile(filePath, { sampleSize: 5120 }).then(encoding => {
  // 如果不是UTF-8编码，转换为utf8编码
  if (encoding !== 'UTF-8') {
    rs.pipe(iconv.decodeStream(encoding))
    .pipe(iconv.encodeStream('UTF-8'))
    .pipe(csv.parse({ header: false, ignoreEmpty: true, trim: true })) // 解析csv
    .on('error', error => {
      reject(`解析csv error: ${error}`)
    })
    .on('data', rows => {
      count++ // 计数，因为我们要分块读取和操作
      arrayFromParseCSV.push(rows) // 读到就推送到数组中
      if (count > 30000) { // 已经读了30000行，我们就要先把这3w行处理掉，避免占用过多内存。
      rs.pause() // 暂停可读流
      delayInsert(false) // false 还没有结束。注意：即使rs.pause, 流的读取也不是立即暂停的，所以需要防抖。
      }      
    }).on('end', rowCount => {
      console.log(`解析完${filePath}文件一共${rowCount}行`)
      delayInsert(true, () => {
      rs.destroy() // 销毁流
      resolve('ok') // 一个文件读取完毕了
      })
    })
  }
  })
}

function insertInBlock (isEnd, cb, filePath, resolve, reject) {
  const arr = doSomethingWithData() // 可能会有一些清洗数据的操作
  // 假如我们后续的需求是将数据写入数据库
  const batchInsert = () => {
  batchInsertDatabasePromise().then(() => {
    if (cb && typeof cb === 'function') cb()
    !isEnd && rs.resume() // 这一个片段的数据写入完毕，可以恢复流继续读了
  })
  }
  
  const truely = schemaHasTable() // 比如判断数据库中有没有某个表，有就写入。没有先建表再写入。
  if (truely) { //
   batchInsert()
   } else {
   // 建表或者其他操作，然后再写入
   doSomething().then(() => batchInsert())
  }
}

const { XLSX } = require('xlsx-extract')
new XLSX().convert('path/to/file.xlsx', 'path/to/destfile.tsv')
  .on('error', function (err) {
    console.error(err);
  })
  .on('end', function () {
    console.log('written');
  })

import { parse } from '@fast-csv/parse';
const stream = parse({ delimiter: '\t' })
  .on('error', error => console.error(error))
  .on('data', row => console.log(row))
  .on('end', (rowCount: number) => console.log(`Parsed ${rowCount} rows`));

const { xslx } = require('xlsx-extract') // 流式解析xlsx文件工具
// parser: expat, 需要额外安装node-expat,可以提高解析速度。
new XLSX().extract(filePath, { sheet_nr: 1, parser: 'expat' })
  .on('row', function (row) {
    // 每一行数据获取到时都可以触发
    }).on('error', function (err) {
    // error
   });

const readStream = fs.createReadableStream('path/to/xlsx.xlsx')

const unzip = require('unzipper')
const zip = unzip.Parse();
rs.pipe(zip)
  .on('entry', function (entry) {
  console.log('entry ---', entry);
  const fileName = entry.path;
  const { type } = entry; // 'Directory' or 'File'
  const size = entry.vars.uncompressedSize; // There is also compressedSize;
  if (fileName === "this IS the file I'm looking for") {
    entry.pipe(fs.createWriteStream('output/path'));
  } else {
    entry.autodrain();
  }
  })

const saxStream = require('sax').createStream(false);
saxStream.on('error', function (e) {
  console.error('error!', e);
});
saxStream.on('opentag', function (node) {
  console.log('node ---', node);
});
saxStream.on('text', (text) => console.log('text ---', typeof text, text));

const unzip = require('unzipper');
const saxStream = require('sax').createStream(false);
const zip = unzip.Parse();
saxStream.on('error', function (e) {
  console.error('error!', e);
});
saxStream.on('opentag', function (node) {
  console.log('node ---', node);
});
saxStream.on('text', (text) => {
  console.log('text ---', typeof text, text)
});
rs.pipe(zip)
  .on('entry', function (entry) {
  console.log('entry ---', entry);
  entry.pipe(saxStream)
  })