Drsakura
1a759ef37d
数据迁移不再整体替换 trade.db —— 那条路会把账号、权限组、组织、设置（含大模型密钥）、 登录会话与审计日志一起搬到目标部署。改为按类别搬运业务数据，系统数据永远是目标部署自己那份。 - backend/lib/dataTransfer：表目录（30 张业务表，依赖顺序）、gzip JSONL 数据包（每表 sha256）、导入内核 - 归属统一给导入人；冲突跳过不覆盖；父记录按 __refs 自然键快照补壳 - 流水表（price_history）按整行内容判重：用 (sku, changed_at) 当键实测 643 行丢 186 行 - 预览在库副本上真跑一遍导入，报告与真导同源；导入前 VACUUM INTO 备份，单事务写入 - 新权限点 data.transfer（仅超管组），导出/导入均写审计 - 前端「数据迁移」页：按类别勾选导出、上传后预览逐表明细、确认导入；移除设置里的静态占位「数据交换」 - 测试：单测 7 项（含目录外键顺序、BLOB、重复导入、包被改坏）、API 端到端 1 项（两个真实实例互传） - 用真实库副本彩排：3854 行 0 失败，29 张业务表行数全部一致，包 1.22 MB
326 lines
12 KiB
JavaScript
326 lines
12 KiB
JavaScript
/**
|
||
* 业务数据导入内核。
|
||
*
|
||
* 三条硬规则（对应 Wayne 2026-10-23 定的口径）：
|
||
* ① **归属一律给导入人** —— owner_id / created_by / assigned_to 全部改写成本次导入的账号；
|
||
* ② **冲突跳过，绝不覆盖** —— 线上已有同一条记录就跳过，并在报告里点名；
|
||
* ③ **父记录可补壳** —— 只导订单时，客户按公司名找不到就建一条「只有公司名」的客户，
|
||
* 报告里单列「自动补建」，导完提示补全信息。
|
||
*
|
||
* 这个函数是「纯函数式」的：给它一个已打开的库实例，它只在这个实例里写。
|
||
* 因此同一个实现既能跑在**真实库**（apply），也能跑在**库的副本**（preview）——
|
||
* 预览报告与实际导入完全同源，不存在「预览说能导、真导却失败」的两套逻辑。
|
||
*/
|
||
const { TABLES, OWNER_COLUMNS, ORG_COLUMNS, resolveCategories, CATEGORY_LIST } = require('./catalog')
|
||
const { tableInfo, foreignKeys, tableExists, decodeValue } = require('./bundle')
|
||
|
||
const quote = name => `"${String(name).replace(/"/g, '""')}"`
|
||
const MAX_WARNINGS_PER_TABLE = 20
|
||
|
||
/** 导入目标组织：优先取导入人所属组织，否则退回第一个组织 */
|
||
function targetOrgId(db, user) {
|
||
if (user && user.org_id) return user.org_id
|
||
const row = db.prepare('SELECT id FROM organizations ORDER BY id LIMIT 1').get()
|
||
return row ? row.id : null
|
||
}
|
||
|
||
function makeContext(db, user) {
|
||
return {
|
||
db,
|
||
userId: user ? user.id : null,
|
||
orgId: targetOrgId(db, user),
|
||
now: new Date().toISOString(),
|
||
ids: new Map(), // table -> Map(包内旧 id -> 本库新 id)
|
||
stubTotals: new Map(),
|
||
statements: new Map()
|
||
}
|
||
}
|
||
|
||
function idMap(ctx, table) {
|
||
if (!ctx.ids.has(table)) ctx.ids.set(table, new Map())
|
||
return ctx.ids.get(table)
|
||
}
|
||
|
||
/**
|
||
* 按自然键找已存在的记录（供去重与父记录解析共用）。
|
||
*
|
||
* 键里允许出现空值：product_groups 的键是「名称 + 品牌」，而品牌经常为空 ——
|
||
* 若要求每列都非空才算可比较，这类表就会永远判定「不存在」，重复导入时成倍增长。
|
||
* 所以空值按 `IS NULL` 比较，只要键里至少有一列有值就参与去重。
|
||
*/
|
||
function findByKey(db, table, keySpec, values) {
|
||
const clauses = []
|
||
const params = []
|
||
for (const key of keySpec) {
|
||
const value = values[key]
|
||
if (value === undefined) return null
|
||
if (value === null || value === '') {
|
||
clauses.push(`(${quote(key)} IS NULL OR ${quote(key)} = '')`)
|
||
} else {
|
||
clauses.push(`${quote(key)} = ?`)
|
||
params.push(value)
|
||
}
|
||
}
|
||
if (!clauses.length) return null
|
||
try {
|
||
return db.prepare(`SELECT * FROM ${quote(table)} WHERE ${clauses.join(' AND ')} LIMIT 1`).get(...params) || null
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
/** 父记录的自然键快照 → 列值对象 */
|
||
function keyValuesFromSnapshot(parentSpec, snapshot) {
|
||
if (snapshot === null || snapshot === undefined) return null
|
||
if (typeof snapshot === 'object') return snapshot
|
||
return { [parentSpec.key[0]]: snapshot }
|
||
}
|
||
/** 补一条最小父记录（客户只有公司名、供应商只有名字、产品组只有名字） */
|
||
function createStub(ctx, parentSpec, keyValues) {
|
||
const value = keyValues ? keyValues[parentSpec.key[0]] : null
|
||
if (!value) return null
|
||
const { db, now } = ctx
|
||
if (parentSpec.table === 'customers') {
|
||
/* 补出来的客户直接归导入人、状态置为名下（不给 pooled 状态，避免导完冒出一堆公海客户） */
|
||
const result = db.prepare(`
|
||
INSERT INTO customers (company, owner_id, created_by, org_id, status, pool_status, created_at)
|
||
VALUES (?, ?, ?, ?, 'active', 'owned', ?)
|
||
`).run(String(value), ctx.userId, ctx.userId, ctx.orgId, now)
|
||
ctx.stubTotals.set('customers', (ctx.stubTotals.get('customers') || 0) + 1)
|
||
return result.lastInsertRowid
|
||
}
|
||
if (parentSpec.table === 'suppliers') {
|
||
const result = db.prepare('INSERT INTO suppliers (name, created_at) VALUES (?, ?)').run(String(value), now)
|
||
ctx.stubTotals.set('suppliers', (ctx.stubTotals.get('suppliers') || 0) + 1)
|
||
return result.lastInsertRowid
|
||
}
|
||
if (parentSpec.table === 'product_groups') {
|
||
const result = db.prepare('INSERT INTO product_groups (name, created_at) VALUES (?, ?)').run(String(value), now)
|
||
ctx.stubTotals.set('product_groups', (ctx.stubTotals.get('product_groups') || 0) + 1)
|
||
return result.lastInsertRowid
|
||
}
|
||
return null
|
||
}
|
||
|
||
/**
|
||
* 解析一个外键值：
|
||
* ① 本包里刚导进去的同一条（id 映射表命中）
|
||
* ② 本库已存在（按父记录自然键查）
|
||
* ③ 都没有 → 允许补壳的补一个，否则置空
|
||
*/
|
||
function resolveParent(ctx, ref, rawValue, snapshot) {
|
||
const parentSpec = TABLES.find(item => item.table === ref.table)
|
||
const mapped = ctx.ids.get(ref.table)
|
||
if (mapped && mapped.has(rawValue)) return mapped.get(rawValue)
|
||
|
||
if (parentSpec && parentSpec.key && parentSpec.key.length) {
|
||
/* 父表主键就是自然键（products.sku）：父记录在就沿用原值，不在就置空由约束决定成败 */
|
||
if (ref.to && ref.to !== 'id') {
|
||
const exists = ctx.db.prepare(`SELECT 1 FROM ${quote(ref.table)} WHERE ${quote(ref.to)} = ?`).get(rawValue)
|
||
return exists ? rawValue : null
|
||
}
|
||
const keyValues = keyValuesFromSnapshot(parentSpec, snapshot)
|
||
if (keyValues) {
|
||
const row = findByKey(ctx.db, ref.table, parentSpec.key, keyValues)
|
||
if (row) {
|
||
if (mapped) mapped.set(rawValue, row.id)
|
||
return row.id
|
||
}
|
||
if (parentSpec.stub) {
|
||
const created = createStub(ctx, parentSpec, keyValues)
|
||
if (created) {
|
||
if (mapped) mapped.set(rawValue, created)
|
||
return created
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return null
|
||
}
|
||
|
||
function prepareStatement(ctx, sql) {
|
||
if (!ctx.statements.has(sql)) ctx.statements.set(sql, ctx.db.prepare(sql))
|
||
return ctx.statements.get(sql)
|
||
}
|
||
|
||
/** 单表导入 */
|
||
function importTable(ctx, spec, rows, info) {
|
||
const stat = {
|
||
table: spec.table,
|
||
label: spec.label,
|
||
category: spec.category,
|
||
rows: rows.length,
|
||
insert: 0,
|
||
skip: 0,
|
||
failed: 0,
|
||
warnings: []
|
||
}
|
||
const warn = message => {
|
||
if (stat.warnings.length < MAX_WARNINGS_PER_TABLE) stat.warnings.push(message)
|
||
}
|
||
|
||
const columnsByName = new Map(info.columns.map(column => [column.name, column]))
|
||
const fkByColumn = new Map(foreignKeys(ctx.db, spec.table).map(ref => [ref.from, ref]))
|
||
const unknownColumns = new Set()
|
||
|
||
/* 去重用的列：普通表用目录里定义的业务键；流水表（price_history）用整行内容。
|
||
整行比对时排除自增主键与「归属人 / 组织」—— 这两类本来就会被改写成本次导入的人，
|
||
不该参与「是不是同一条记录」的判断。 */
|
||
const keyColumns = spec.contentKey
|
||
? info.columns
|
||
.map(column => column.name)
|
||
.filter(name => !(info.autoId && name === info.pk) && !OWNER_COLUMNS.has(name) && !ORG_COLUMNS.has(name))
|
||
: spec.key
|
||
|
||
/* 流水表允许同一个包里出现逐列完全相同的两行（源库确实会这样：一次批量改价写两条）。
|
||
它们对线上来说是两条独立的历史记录，必须都带过去；只跟**导入前**existing 的比对。
|
||
普通表不允许：两个同货号的行塞进去会撞主键，那是数据本身有问题，跳过并报告更合适。 */
|
||
const duplicatesInsidePackage = new Set()
|
||
|
||
for (let index = 0; index < rows.length; index++) {
|
||
const raw = rows[index]
|
||
try {
|
||
const values = {}
|
||
const refs = raw.__refs && typeof raw.__refs === 'object' ? raw.__refs : {}
|
||
|
||
for (const name of Object.keys(raw)) {
|
||
if (name === '__refs') continue
|
||
const column = columnsByName.get(name)
|
||
if (!column) { unknownColumns.add(name); continue }
|
||
const value = decodeValue(raw[name], info.blobs.has(name))
|
||
|
||
/* 归属人 / 归属组织：一律换成「本次导入的人 / 当前组织」 */
|
||
if (OWNER_COLUMNS.has(name)) {
|
||
values[name] = value === null ? null : ctx.userId
|
||
continue
|
||
}
|
||
if (ORG_COLUMNS.has(name)) {
|
||
values[name] = ctx.orgId
|
||
continue
|
||
}
|
||
|
||
const ref = fkByColumn.get(name)
|
||
if (ref) {
|
||
if (value === null) { values[name] = null; continue }
|
||
const mapped = resolveParent(ctx, ref, value, refs[name])
|
||
if (mapped === null || mapped === undefined) {
|
||
/* 父记录既不在包里也补不出来：外键置空，由 NOT NULL 约束决定该行成败 */
|
||
values[name] = null
|
||
} else {
|
||
values[name] = mapped
|
||
}
|
||
continue
|
||
}
|
||
values[name] = value
|
||
}
|
||
|
||
/* 客户：归属强制给导入人；本来就是公海的客户改成「名下有主」，
|
||
否则会出现「归我所有却标记在公海」这种自相矛盾的状态。 */
|
||
if (spec.table === 'customers') {
|
||
values.owner_id = ctx.userId
|
||
values.created_by = ctx.userId
|
||
values.org_id = ctx.orgId
|
||
if (values.pool_status === 'public') {
|
||
values.pool_status = 'owned'
|
||
values.pool_at = null
|
||
values.pool_note = null
|
||
}
|
||
}
|
||
|
||
/* 去重：键里至少一列有值才判定；某列为空按 NULL 比较，不因此放弃去重
|
||
（否则「品牌为空的产品组」这种表每次重导都会翻倍） */
|
||
const keyValues = {}
|
||
let keyReady = false
|
||
for (const key of keyColumns) {
|
||
const value = values[key]
|
||
if (value === undefined || value === null || value === '') {
|
||
keyValues[key] = null
|
||
continue
|
||
}
|
||
keyValues[key] = value
|
||
keyReady = true
|
||
}
|
||
if (keyReady) {
|
||
const fingerprint = keyColumns.map(key => `${key}=${keyValues[key] === null ? '\u0000' : keyValues[key]}`).join('\u0001')
|
||
const repeatInsidePackage = duplicatesInsidePackage.has(fingerprint)
|
||
const existing = (!repeatInsidePackage || !spec.contentKey)
|
||
? findByKey(ctx.db, spec.table, keyColumns, keyValues)
|
||
: null
|
||
if (existing) {
|
||
stat.skip++
|
||
if (info.autoId && raw.id !== undefined && raw.id !== null && existing[info.pk]) {
|
||
idMap(ctx, spec.table).set(raw.id, existing[info.pk])
|
||
}
|
||
continue
|
||
}
|
||
duplicatesInsidePackage.add(fingerprint)
|
||
}
|
||
|
||
const names = Object.keys(values).filter(name => !(info.autoId && name === info.pk))
|
||
if (!names.length) throw new Error('没有可写入的字段')
|
||
|
||
const sql = `INSERT INTO ${quote(spec.table)} (${names.map(quote).join(', ')}) VALUES (${names.map(() => '?').join(', ')})`
|
||
const result = prepareStatement(ctx, sql).run(...names.map(name => values[name]))
|
||
|
||
stat.insert++
|
||
if (info.autoId) {
|
||
const oldId = raw.id
|
||
if (oldId !== undefined && oldId !== null) idMap(ctx, spec.table).set(oldId, result.lastInsertRowid)
|
||
}
|
||
} catch (error) {
|
||
stat.failed++
|
||
warn(`第 ${index + 1} 行：${String(error.message || error).slice(0, 160)}`)
|
||
}
|
||
}
|
||
|
||
if (unknownColumns.size) {
|
||
warn(`目标库缺少列 ${[...unknownColumns].join('、')}（版本较旧？），这些字段已忽略`)
|
||
}
|
||
return stat
|
||
}
|
||
|
||
/**
|
||
* 把数据包写进给定的库实例。
|
||
* 调用方负责包在事务里（见 index.js 的 applyImport）——这里出现异常就整体回滚。
|
||
*/
|
||
function importInto(db, bundle, { categories, user }) {
|
||
const selected = new Set(resolveCategories(categories))
|
||
const ctx = makeContext(db, user)
|
||
const tables = []
|
||
const missing = []
|
||
|
||
for (const spec of TABLES) {
|
||
if (!selected.has(spec.category)) continue
|
||
const rows = bundle.rows.get(spec.table)
|
||
if (!rows || !rows.length) continue
|
||
if (!tableExists(db, spec.table)) { missing.push(spec.table); continue }
|
||
const info = tableInfo(db, spec.table)
|
||
if (!info) { missing.push(spec.table); continue }
|
||
tables.push(importTable(ctx, spec, rows, info))
|
||
}
|
||
|
||
const totals = tables.reduce((acc, item) => {
|
||
acc.rows += item.rows
|
||
acc.insert += item.insert
|
||
acc.skip += item.skip
|
||
acc.failed += item.failed
|
||
return acc
|
||
}, { rows: 0, insert: 0, skip: 0, failed: 0 })
|
||
|
||
const stubbed = [...ctx.stubTotals.entries()].map(([table, count]) => ({
|
||
table,
|
||
label: (TABLES.find(item => item.table === table) || {}).label || table,
|
||
count
|
||
}))
|
||
|
||
return {
|
||
categories: CATEGORY_LIST.filter(item => selected.has(item.key)).map(item => ({ key: item.key, label: item.label })),
|
||
tables,
|
||
totals,
|
||
stubbed,
|
||
missing
|
||
}
|
||
}
|
||
|
||
module.exports = { importInto, targetOrgId }
|