/** * 数据包(.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)()) } /** * 读取并校验数据包。 * * 校验不做「尽力而为」:行数或 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 reader) { 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 }