feat(data): 命令行导入导出通道 + 修上传解析的句柄泄漏

- scripts/data-transfer.js：tables / export / import --preview|--apply，与网页版共用同一套内核。用于 nginx/CDN 挡住 1.22MB 上传时的备用通道，也可挂 cron 做异地备份。 - readBundle：校验失败中途抛错时关闭 readStream 与 readline，避免反复上传坏包耗尽文件句柄。 - docs：补命令行用法与体积限制说明。
2026-10-23 13:48:53 +08:00
parent 1a759ef37d
commit 1e7c3fdcc1
3 changed files with 203 additions and 1 deletions
@@ -217,6 +217,22 @@ function openMaybeGzip(filePath) {
  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 { /* 已经释放 */ }
+  }
+}
+
 /**
 * 读取并校验数据包。
 *
@@ -241,7 +257,7 @@ async function readBundle(filePath) {
  const counts = new Map()
  let lineNo = 0

-  for await (const raw of reader) {
+  for await (const raw of linesOf(reader, input)) {
    const line = raw.trim()
    if (!line) continue
    lineNo++
@@ -0,0 +1,162 @@
+#!/usr/bin/env node
+/**
+ * 命令行版数据迁移 —— 网页「数据迁移」的备用通道。
+ *
+ * 什么时候用它：
+ *   · 服务器的 nginx / CDN 限制了上传体积，1 MB 以上的包传不上去；
+ *   · 想用 scp 直接把包丢到服务器再导，不经过浏览器；
+ *   · 定期把业务数据导出来做异地备份（配合 cron）。
+ *
+ * 与网页版共用 backend/lib/dataTransfer 的同一套代码：同样的判重、归属改写、补壳与备份。
+ *
+ * 用法（在 backend/ 目录下执行，Node ≥ 22）：
+ *   node scripts/data-transfer.js export --out ./data.hdata.gz [--categories catalog,orders]
+ *   node scripts/data-transfer.js import ./data.hdata.gz --preview [--categories catalog]
+ *   node scripts/data-transfer.js import ./data.hdata.gz --apply [--user admin] [--categories catalog]
+ *   node scripts/data-transfer.js tables
+ *
+ * 环境变量 DATA_DIR 与网页版一致（默认 backend/data）。
+ */
+const fs = require('fs')
+const os = require('os')
+const path = require('path')
+
+const args = process.argv.slice(2)
+const command = args[0]
+
+function flag(name) {
+  const index = args.indexOf(`--${name}`)
+  if (index < 0) return undefined
+  const value = args[index + 1]
+  return !value || value.startsWith('--') ? true : value
+}
+
+function positional() {
+  return args.slice(1).filter((item, index, list) => !item.startsWith('--') && !(list[index - 1] || '').startsWith('--'))
+}
+
+function usage(message) {
+  if (message) console.error(`✗ ${message}\n`)
+  console.error(`用法：
+  node scripts/data-transfer.js tables
+  node scripts/data-transfer.js export --out <文件.hdata.gz> [--categories catalog,customers,orders,todos,prospects]
+  node scripts/data-transfer.js import <文件.hdata.gz> --preview [--categories …]
+  node scripts/data-transfer.js import <文件.hdata.gz> --apply [--user admin] [--categories …]`)
+  process.exit(message ? 1 : 0)
+}
+
+if (!command || ['-h', '--help', 'help'].includes(command)) usage()
+
+const db = require('../db/db')
+const transfer = require('../lib/dataTransfer')
+const { readBundle } = require('../lib/dataTransfer/bundle')
+const { CATEGORY_LIST } = require('../lib/dataTransfer/catalog')
+
+const DATA_DIR = process.env.DATA_DIR || path.join(__dirname, '..', 'data')
+const TRANSFER_DIR = path.join(DATA_DIR, 'transfers')
+const BACKUP_DIR = path.join(DATA_DIR, 'backups')
+
+function defaultUser() {
+  const username = flag('user')
+  const row = username
+    ? db.prepare('SELECT * FROM users WHERE username = ?').get(String(username))
+    : db.prepare("SELECT * FROM users WHERE role = 'admin' ORDER BY id LIMIT 1").get()
+  if (!row) {
+    console.error(username ? `✗ 找不到用户 ${username}` : '✗ 数据库里没有 admin 账号')
+    process.exit(1)
+  }
+  return row
+}
+
+function categories() {
+  const value = flag('categories')
+  return value && value !== true ? String(value).split(',').map(item => item.trim()).filter(Boolean) : []
+}
+
+function printPlan(plan) {
+  console.log('')
+  console.log('表名                          包内   新增   跳过   失败')
+  for (const item of plan.tables) {
+    console.log(`${item.label.padEnd(26)}${String(item.rows).padStart(6)}${String(item.insert).padStart(7)}${String(item.skip).padStart(7)}${String(item.failed).padStart(7)}`)
+  }
+  console.log('')
+  console.log(`合计：包内 ${plan.totals.rows} → 新增 ${plan.totals.insert} / 跳过 ${plan.totals.skip} / 失败 ${plan.totals.failed}`)
+  if (plan.stubbed.length) {
+    console.log(`自动补建：${plan.stubbed.map(item => `${item.label} ${item.count} 条（只有名称，请补全）`).join('、')}`)
+  }
+  const failures = plan.tables.filter(item => item.failed)
+  if (failures.length) {
+    console.log('失败明细：')
+    for (const item of failures) for (const warning of item.warnings) console.log(`  ${item.label}: ${warning}`)
+  }
+  if (plan.missing.length) console.log(`目标库缺少表：${plan.missing.join('、')}`)
+}
+
+;(async () => {
+  if (command === 'tables') {
+    for (const category of transfer.listCatalog(db)) {
+      const rows = category.tables.reduce((sum, item) => sum + item.rows, 0)
+      console.log(`${category.key.padEnd(10)}${category.label}（${rows} 行）`)
+      for (const item of category.tables) console.log(`  ${item.table.padEnd(24)}${item.label.padEnd(18)}${item.rows}`)
+    }
+    return
+  }
+
+  if (command === 'export') {
+    const out = flag('out')
+    if (!out || out === true) usage('导出需要 --out <文件路径>')
+    const target = path.resolve(String(out))
+    fs.mkdirSync(path.dirname(target), { recursive: true })
+    const started = Date.now()
+    await transfer.exportToStream(db, fs.createWriteStream(target), {
+      categories: categories(),
+      exporter: defaultUser()
+    })
+    const size = fs.statSync(target).size
+    console.log(`✓ 已导出 ${target}（${(size / 1024 / 1024).toFixed(2)} MB，${Date.now() - started}ms）`)
+    console.log('  包内不含用户、权限组、组织、设置（含 AI 密钥）、登录会话与审计日志。')
+    return
+  }
+
+  if (command === 'import') {
+    const file = positional()[0] || flag('file')
+    if (!file || file === true) usage('导入需要数据包路径')
+    const source = path.resolve(String(file))
+    if (!fs.existsSync(source)) usage(`找不到数据包：${source}`)
+    const user = defaultUser()
+
+    if (flag('preview')) {
+      /* 预览在库副本上真跑一遍，与网页版同源；线上库不受影响 */
+      const preview = await transfer.previewImport(db, source, {
+        categories: categories(),
+        user,
+        workDir: TRANSFER_DIR,
+        fileName: path.basename(source)
+      })
+      console.log(`数据包：${preview.file.name} · ${(preview.file.size / 1024 / 1024).toFixed(2)} MB · 导出于 ${preview.file.exportedAt}`)
+      printPlan(preview.plan)
+      console.log('\n这只是预览，线上库没有变化。确认无误后加 --apply 执行导入。')
+      return
+    }
+
+    if (flag('apply')) {
+      const result = await transfer.applyImport(db, source, {
+        categories: categories(),
+        user,
+        backupDir: BACKUP_DIR,
+        fileName: path.basename(source)
+      })
+      console.log(`数据包：${result.file.name} · 归属给 ${user.username}`)
+      printPlan(result.plan)
+      console.log(`\n✓ 导入完成。导入前备份：${result.backup}`)
+      return
+    }
+
+    usage('导入需要 --preview（只看报告）或 --apply（真正写入）')
+  }
+
+  usage(`未知命令：${command}`)
+})().catch(error => {
+  console.error(`✗ ${error.message}`)
+  process.exitCode = 1
+})
@@ -94,6 +94,30 @@ sleep 3 && curl -s http://127.0.0.1:4300/api/health

 反向也成立：云端 → 本地同样走这一页，包是同一个格式。

+## 命令行（备用通道）
+
+浏览器上传被 nginx / CDN 的体积限制挡住（默认 `client_max_body_size 1m`，而真实包 1.22 MB 就会 413），
+或者想把包 `scp` 上去直接在服务器导，用同仓库的 CLI：
+
+```bash
+cd /opt/1panel/www/sites/erp.foohere.de/index/backend
+
+# 各类别、各表当前行数
+/usr/bin/node scripts/data-transfer.js tables
+
+# 先预览：在库副本上真跑一遍，线上库不动
+/usr/bin/node scripts/data-transfer.js import /root/bundle.hdata.gz --preview
+
+# 确认后导入：自动备份到 data/backups/，归属给 --user 指定的账号
+/usr/bin/node scripts/data-transfer.js import /root/bundle.hdata.gz --apply --user admin
+
+# 定期导出做异地备份（配合 cron）
+/usr/bin/node scripts/data-transfer.js export --out /root/backup-$(date +%F).hdata.gz
+```
+
+与网页版**共用同一套代码**（判重、归属改写、补壳、备份完全一致），
+`--categories catalog,orders` 可以只搬某几类。
+
 ## 接口

 | 方法 | 路径 | 说明 |