Drsakura
1e7c3fdcc1
- scripts/data-transfer.js：tables / export / import --preview|--apply，与网页版共用同一套内核。 用于 nginx/CDN 挡住 1.22MB 上传时的备用通道，也可挂 cron 做异地备份。 - readBundle：校验失败中途抛错时关闭 readStream 与 readline，避免反复上传坏包耗尽文件句柄。 - docs：补命令行用法与体积限制说明。
374 lines
13 KiB
JavaScript
374 lines
13 KiB
JavaScript
/**
|
||
* 数据包（.hdata）的写出与读取。
|
||
*
|
||
* 格式：gzip 压缩的 JSONL，三部分
|
||
* 1. header —— 格式版本、导出时间、导出人、包含的类别与表
|
||
* 2. 每行一条记录 { "t": "products", "r": { ... } }，BLOB 列以 base64 字符串承载
|
||
* 3. footer —— 每张表的行数与 sha256
|
||
*
|
||
* 为什么不是 zip、为什么不像数据库备份那样直接拷 .db：
|
||
* - .db 备份是整库，连 users / settings / sessions 一起搬，正是这次要分开的东西；
|
||
* - zip 需要额外依赖，而 gzip + JSONL 只用 Node 内置能力，还能逐行流式读写
|
||
* （合同原件是 BLOB，未来单个文件几十 MB，不能整包读进内存）；
|
||
* - 每张表单独一个 sha256，导入前就能判断包是否被截断 / 传坏，
|
||
* 而不是解析到一半才炸在某个 JSON 语法错误上。
|
||
*/
|
||
const fs = require('fs')
|
||
const zlib = require('zlib')
|
||
const crypto = require('crypto')
|
||
const readline = require('readline')
|
||
const { Readable } = require('stream')
|
||
const { pipeline } = require('stream/promises')
|
||
|
||
const { TABLES, OWNER_COLUMNS, ORG_COLUMNS, resolveCategories, CATEGORY_LIST } = require('./catalog')
|
||
|
||
const FORMAT = 'happy-data'
|
||
const FORMAT_VERSION = 1
|
||
const APP_NAME = 'happy出口通'
|
||
|
||
/** 单包上限：超过就提示按类别分开导，避免一次把几个 GB 读进内存 */
|
||
const MAX_BUNDLE_BYTES = 100 * 1024 * 1024
|
||
const MAX_ROWS = 2_000_000
|
||
|
||
/* ------------------------------------------------------------------ 元数据 */
|
||
|
||
const quote = name => `"${String(name).replace(/"/g, '""')}"`
|
||
|
||
/** 目标库 / 源库的物理结构：列、BLOB 列、主键列 */
|
||
function tableInfo(db, table) {
|
||
let columns
|
||
try {
|
||
columns = db.pragma(`table_info(${quote(table)})`)
|
||
} catch {
|
||
return null
|
||
}
|
||
if (!columns || !columns.length) return null
|
||
const pk = columns.find(column => Number(column.pk) === 1) || null
|
||
return {
|
||
columns: columns.map(column => ({ name: column.name, type: String(column.type || '').toUpperCase() })),
|
||
blobs: new Set(columns.filter(column => String(column.type || '').toUpperCase().includes('BLOB')).map(column => column.name)),
|
||
pk: pk ? pk.name : null,
|
||
autoId: !!pk && pk.name === 'id'
|
||
}
|
||
}
|
||
|
||
/** 外键：{ from: 本表列, table: 父表, to: 父表列 } */
|
||
function foreignKeys(db, table) {
|
||
let list
|
||
try {
|
||
list = db.pragma(`foreign_key_list(${quote(table)})`)
|
||
} catch {
|
||
return []
|
||
}
|
||
return (list || []).map(item => ({ from: item.from, table: item.table, to: item.to }))
|
||
}
|
||
|
||
/** 包内是否出现这个表（老库可能缺新表，按实际存在处理） */
|
||
function tableExists(db, table) {
|
||
return !!db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(table)
|
||
}
|
||
|
||
/* ------------------------------------------------------------------ 导出 */
|
||
|
||
function encodeValue(value, isBlob) {
|
||
if (value === null || value === undefined) return null
|
||
if (isBlob) return Buffer.isBuffer(value) ? value.toString('base64') : Buffer.from(String(value)).toString('base64')
|
||
if (Buffer.isBuffer(value)) return value.toString('base64')
|
||
return value
|
||
}
|
||
|
||
function decodeValue(value, isBlob) {
|
||
if (value === null || value === undefined) return null
|
||
if (isBlob) return Buffer.from(String(value), 'base64')
|
||
return value
|
||
}
|
||
|
||
/**
|
||
* 把一张表读成「行对象」，并按需附上父记录的自然键快照。
|
||
*
|
||
* 为什么要附 `__refs`：只导出订单时，订单里的 customer_id 在目标库里根本不存在。
|
||
* 有了客户公司名这份快照，导入方就能按公司名找到客户、找不到才补一个空壳，
|
||
* 而不是整行订单都导不进去。
|
||
*/
|
||
function* rowsOfTable(db, table, { info, refs, refCache }) {
|
||
const columns = info.columns.map(column => column.name)
|
||
const select = columns.map(quote).join(', ')
|
||
const orderBy = info.autoId ? 'id' : quote(info.pk || columns[0])
|
||
const statement = db.prepare(`SELECT ${select} FROM ${quote(table)} ORDER BY ${orderBy}`)
|
||
|
||
for (const row of statement.iterate()) {
|
||
const payload = {}
|
||
for (const name of columns) {
|
||
const value = encodeValue(row[name], info.blobs.has(name))
|
||
/* 空值不写进包里，体积能小一截（products 有 1576 行、不少列是空的） */
|
||
if (value !== null) payload[name] = value
|
||
}
|
||
|
||
const refValues = {}
|
||
for (const ref of refs) {
|
||
const raw = row[ref.from]
|
||
if (raw === null || raw === undefined) continue
|
||
const snapshot = refSnapshot(db, ref, raw, refCache)
|
||
if (snapshot) refValues[ref.from] = snapshot
|
||
}
|
||
if (Object.keys(refValues).length) payload.__refs = refValues
|
||
|
||
yield payload
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 父记录的自然键快照。
|
||
* 父表主键本身就是自然键（products.sku）时，值已经在行里，不用再查；
|
||
* 主键是自增 id（customers / suppliers / product_groups）时按 id 取一次并缓存。
|
||
*/
|
||
function refSnapshot(db, ref, value, cache) {
|
||
if (ref.to && ref.to !== 'id') return value
|
||
const spec = TABLES.find(item => item.table === ref.table)
|
||
if (!spec || !spec.key.length) return null
|
||
const cacheKey = `${ref.table}:${value}`
|
||
if (cache.has(cacheKey)) return cache.get(cacheKey)
|
||
let snapshot = null
|
||
try {
|
||
const row = db.prepare(`SELECT ${spec.key.map(quote).join(', ')} FROM ${quote(ref.table)} WHERE id = ?`).get(value)
|
||
if (row) snapshot = spec.key.length === 1 ? row[spec.key[0]] : Object.fromEntries(spec.key.map(key => [key, row[key]]))
|
||
} catch { /* 父记录已被删：留空，导入时按「找不到父记录」处理 */ }
|
||
cache.set(cacheKey, snapshot)
|
||
return snapshot
|
||
}
|
||
|
||
/** 选中类别 → 实际要导出的表（跳过源库里不存在的表） */
|
||
function planTables(db, categories) {
|
||
const selected = new Set(resolveCategories(categories))
|
||
return TABLES.filter(spec => selected.has(spec.category) && tableExists(db, spec.table))
|
||
}
|
||
|
||
/**
|
||
* 生成数据包的行流（不含 gzip）。
|
||
* 交给 Readable.from 之后由 gzip 压到 HTTP 响应里 —— 逐行 yield，不整包驻留内存。
|
||
*/
|
||
async function* bundleLines(db, { categories, exporter }) {
|
||
const plans = planTables(db, categories)
|
||
const header = {
|
||
type: 'header',
|
||
format: FORMAT,
|
||
version: FORMAT_VERSION,
|
||
app: APP_NAME,
|
||
exportedAt: new Date().toISOString(),
|
||
categories: resolveCategories(categories),
|
||
tableNames: plans.map(spec => spec.table),
|
||
exporter: exporter ? { id: exporter.id, username: exporter.username, display_name: exporter.display_name } : null
|
||
}
|
||
yield JSON.stringify(header) + '\n'
|
||
|
||
const tableMeta = {}
|
||
let totalRows = 0
|
||
|
||
for (const spec of plans) {
|
||
const info = tableInfo(db, spec.table)
|
||
if (!info) continue
|
||
const refs = foreignKeys(db, spec.table).filter(ref =>
|
||
!OWNER_COLUMNS.has(ref.from) && !ORG_COLUMNS.has(ref.from)
|
||
&& TABLES.some(item => item.table === ref.table))
|
||
const refCache = new Map()
|
||
const hash = crypto.createHash('sha256')
|
||
let count = 0
|
||
|
||
for (const payload of rowsOfTable(db, spec.table, { info, refs, refCache })) {
|
||
const line = JSON.stringify({ t: spec.table, r: payload })
|
||
hash.update(line)
|
||
count++
|
||
yield line + '\n'
|
||
}
|
||
|
||
tableMeta[spec.table] = { rows: count, sha256: hash.digest('hex') }
|
||
totalRows += count
|
||
}
|
||
|
||
yield JSON.stringify({
|
||
type: 'footer',
|
||
version: FORMAT_VERSION,
|
||
tables: tableMeta,
|
||
rows: totalRows,
|
||
categories: header.categories
|
||
}) + '\n'
|
||
}
|
||
|
||
/** 把数据包写到任意可写流（HTTP 响应 / 文件） */
|
||
async function writeBundle(db, output, options) {
|
||
const gzip = zlib.createGzip({ level: 6 })
|
||
await pipeline(Readable.from(bundleLines(db, options)), gzip, output)
|
||
}
|
||
|
||
/** 导出到文件（命令行工具与测试用） */
|
||
async function writeBundleFile(db, filePath, options) {
|
||
await writeBundle(db, fs.createWriteStream(filePath), options)
|
||
return fs.statSync(filePath).size
|
||
}
|
||
|
||
/* ------------------------------------------------------------------ 读取 */
|
||
|
||
function openMaybeGzip(filePath) {
|
||
const fd = fs.openSync(filePath, 'r')
|
||
const magic = Buffer.alloc(2)
|
||
const read = fs.readSync(fd, magic, 0, 2, 0)
|
||
fs.closeSync(fd)
|
||
const isGzip = read === 2 && magic[0] === 0x1f && magic[1] === 0x8b
|
||
return fs.createReadStream(filePath).pipe(isGzip ? zlib.createGunzip() : new (require('stream').PassThrough)())
|
||
}
|
||
|
||
/**
|
||
* 逐行吐出，并在**任何**退出路径上关掉底层流。
|
||
*
|
||
* 校验失败时 readBundle 会中途抛错，如果不在这里兜住，文件句柄就跟着异常一起漏掉 ——
|
||
* 服务器上反复上传坏包会把 fd 耗光（排查起来极难，因为错误信息只有「数据包损坏」）。
|
||
* 包在外层 for-await 里的生成器，在 break / throw / 正常结束三种情况下都会走到 finally。
|
||
*/
|
||
async function* linesOf(reader, input) {
|
||
try {
|
||
for await (const line of reader) yield line
|
||
} finally {
|
||
try { reader.close() } catch { /* 已经关了 */ }
|
||
try { input.destroy() } catch { /* 已经释放 */ }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 读取并校验数据包。
|
||
*
|
||
* 校验不做「尽力而为」：行数或 sha256 对不上就整体拒绝，并指出是哪张表 ——
|
||
* 半个包导进去比导不进去更难收拾。
|
||
*/
|
||
async function readBundle(filePath) {
|
||
const stat = fs.statSync(filePath)
|
||
if (stat.size > MAX_BUNDLE_BYTES) {
|
||
const error = new Error(`数据包 ${(stat.size / 1024 / 1024).toFixed(1)} MB，超过 ${MAX_BUNDLE_BYTES / 1024 / 1024} MB 上限。请按类别分开导出。`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
|
||
const input = openMaybeGzip(filePath)
|
||
const reader = readline.createInterface({ input, crlfDelay: Infinity })
|
||
|
||
let header = null
|
||
let footer = null
|
||
const rows = new Map()
|
||
const hashes = new Map()
|
||
const counts = new Map()
|
||
let lineNo = 0
|
||
|
||
for await (const raw of linesOf(reader, input)) {
|
||
const line = raw.trim()
|
||
if (!line) continue
|
||
lineNo++
|
||
let parsed
|
||
try {
|
||
parsed = JSON.parse(line)
|
||
} catch {
|
||
const error = new Error(`数据包第 ${lineNo} 行不是合法 JSON，文件可能已损坏`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
|
||
if (parsed.type === 'header') {
|
||
if (header) {
|
||
const error = new Error('数据包出现重复的头部')
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
if (parsed.format !== FORMAT) {
|
||
const error = new Error('这不是本系统的数据包（缺少 happy-data 标识）')
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
if (Number(parsed.version) > FORMAT_VERSION) {
|
||
const error = new Error(`数据包版本 ${parsed.version} 高于本系统支持的 ${FORMAT_VERSION}，请先升级系统`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
header = parsed
|
||
continue
|
||
}
|
||
|
||
if (parsed.type === 'footer') {
|
||
footer = parsed
|
||
continue
|
||
}
|
||
|
||
if (!header) {
|
||
const error = new Error('数据包缺少头部，无法识别格式')
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
if (!parsed.t || !parsed.r || typeof parsed.r !== 'object') {
|
||
const error = new Error(`数据包第 ${lineNo} 行缺少表名或记录内容`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
|
||
if (!rows.has(parsed.t)) {
|
||
rows.set(parsed.t, [])
|
||
hashes.set(parsed.t, crypto.createHash('sha256'))
|
||
counts.set(parsed.t, 0)
|
||
}
|
||
const list = rows.get(parsed.t)
|
||
if (list.length >= MAX_ROWS) {
|
||
const error = new Error(`数据包内 ${parsed.t} 超过 ${MAX_ROWS} 行，请按类别分开导入`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
list.push(parsed.r)
|
||
counts.set(parsed.t, counts.get(parsed.t) + 1)
|
||
hashes.get(parsed.t).update(line)
|
||
}
|
||
|
||
if (!header) {
|
||
const error = new Error('数据包为空或缺少头部')
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
if (!footer) {
|
||
const error = new Error('数据包缺少结尾清单，可能传输中断。请重新导出后上传。')
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
|
||
const mismatches = []
|
||
for (const [table, meta] of Object.entries(footer.tables || {})) {
|
||
const actualRows = counts.get(table) || 0
|
||
if (actualRows !== meta.rows) mismatches.push(`${table} 行数应为 ${meta.rows}，实际 ${actualRows}`)
|
||
else {
|
||
const actualHash = (hashes.get(table) || crypto.createHash('sha256')).digest('hex')
|
||
if (actualHash !== meta.sha256) mismatches.push(`${table} 校验值不符`)
|
||
}
|
||
}
|
||
if (mismatches.length) {
|
||
const error = new Error(`数据包校验失败：${mismatches.slice(0, 4).join('；')}${mismatches.length > 4 ? ` 等 ${mismatches.length} 项` : ''}。请重新导出后上传。`)
|
||
error.statusCode = 400
|
||
throw error
|
||
}
|
||
|
||
return {
|
||
header,
|
||
footer,
|
||
rows,
|
||
size: stat.size,
|
||
categories: (header.categories || []).filter(key => CATEGORY_LIST.some(item => item.key === key))
|
||
}
|
||
}
|
||
|
||
module.exports = {
|
||
FORMAT,
|
||
FORMAT_VERSION,
|
||
MAX_BUNDLE_BYTES,
|
||
tableInfo,
|
||
foreignKeys,
|
||
tableExists,
|
||
decodeValue,
|
||
bundleLines,
|
||
writeBundle,
|
||
writeBundleFile,
|
||
planTables,
|
||
readBundle
|
||
}
|