feat(0.2.1): M4-b ETL —— 现网 MariaDB 只读快照导入 PG (uuid/code/number 回填, 时序表限量, 序列校准)
This commit is contained in:
jif2.zhang
@@ -0,0 +1,189 @@
|
||||
// 一次性数据迁移: 现网 MariaDB 只读快照 -> 0.2.1 PostgreSQL
|
||||
// 运行 (在 app 容器内, 具备 mariadb 驱动与 PG 客户端):
|
||||
// SRC_DATABASE_URL=mysql://... node scripts/etl-mariadb-to-pg.mjs
|
||||
// 约束: 源库仅 SELECT; 目标库须为空 (Project 计数为 0), 除非 ETL_FORCE=1
|
||||
// 迁移内容见 docs/V0.2/离线编辑模块/0.2.1实施方案.md §6:
|
||||
// - 任务域 6 表回填 uuid; Project 回填短编号 code; Task 按项目内 (createdAt,id) 顺序回填 number
|
||||
// - 时序大表 (ServerNodeHistory/MihomeEnv*) 仅取最近 N 条
|
||||
// - 显式 id 插入后逐表 setval 校准序列
|
||||
import 'dotenv/config'
|
||||
import crypto from 'node:crypto'
|
||||
import mariadb from 'mariadb'
|
||||
import { PrismaPg } from '@prisma/adapter-pg'
|
||||
import { PrismaClient } from '../generated/prisma/client.ts'
|
||||
|
||||
// 顺序: 先父后子
|
||||
const TABLES = [
|
||||
['User', 'user'],
|
||||
['Project', 'project'],
|
||||
['ProjectMember', 'projectMember'],
|
||||
['BoardColumn', 'boardColumn'],
|
||||
['Habit', 'habit'],
|
||||
['CheckList', 'checkList'],
|
||||
['CheckListTemplate', 'checkListTemplate'],
|
||||
['CheckListItem', 'checkListItem'],
|
||||
['CheckIn', 'checkIn'],
|
||||
['Task', 'task'],
|
||||
['TaskMember', 'taskMember'],
|
||||
['TaskTag', 'taskTag'],
|
||||
['Article', 'article'],
|
||||
['Strategy', 'strategy'],
|
||||
['Indicator', 'indicator'],
|
||||
['IndicatorPoint', 'indicatorPoint'],
|
||||
['News', 'news'],
|
||||
['Alert', 'alert'],
|
||||
['ServerNode', 'serverNode'],
|
||||
['ServerNodeHistory', 'serverNodeHistory'],
|
||||
['ServiceCheck', 'serviceCheck'],
|
||||
['Econ_SowInventory', 'econ_SowInventory'],
|
||||
['ScraperCache', 'scraperCache'],
|
||||
['OilPriceHistory', 'oilPriceHistory'],
|
||||
['MihomeAccount', 'mihomeAccount'],
|
||||
['MihomeHome', 'mihomeHome'],
|
||||
['MihomeDevice', 'mihomeDevice'],
|
||||
['MihomeEnvSample', 'mihomeEnvSample'],
|
||||
['MihomeEnvHour', 'mihomeEnvHour'],
|
||||
]
|
||||
|
||||
const ROW_CAP = { ServerNodeHistory: 20000, MihomeEnvSample: 20000, MihomeEnvHour: 20000 }
|
||||
|
||||
const JSON_COLUMNS = {
|
||||
Strategy: ['data'],
|
||||
ScraperCache: ['metadata'],
|
||||
MihomeEnvSample: ['raw'],
|
||||
MihomeEnvHour: ['raw'],
|
||||
Project: ['notes', 'taskKinds'],
|
||||
Task: ['extras'],
|
||||
CheckListTemplate: ['items'],
|
||||
}
|
||||
|
||||
const BIGINT_COLUMNS = {
|
||||
ServerNode: ['memTotal', 'diskTotal', 'ramUsed', 'swapUsed', 'diskUsed', 'netIn', 'netOut', 'uptime'],
|
||||
ServerNodeHistory: ['ram', 'disk', 'netIn', 'netOut'],
|
||||
}
|
||||
|
||||
const UUID_TABLES = new Set(['Project', 'ProjectMember', 'BoardColumn', 'Task', 'TaskMember', 'TaskTag'])
|
||||
|
||||
function parseUrl(url) {
|
||||
const u = new URL(url)
|
||||
return {
|
||||
host: u.hostname,
|
||||
port: Number(u.port) || 3306,
|
||||
user: decodeURIComponent(u.username),
|
||||
password: decodeURIComponent(u.password),
|
||||
database: u.pathname.replace(/^\//, ''),
|
||||
}
|
||||
}
|
||||
|
||||
function parseJson(v) {
|
||||
if (v === null || v === undefined || typeof v !== 'string') return v
|
||||
try { return JSON.parse(v) } catch { return v }
|
||||
}
|
||||
|
||||
function toBigInt(v) {
|
||||
if (v === null || v === undefined || typeof v === 'bigint') return v
|
||||
return BigInt(v)
|
||||
}
|
||||
|
||||
function randCode(len) {
|
||||
return crypto.randomBytes(len * 2).toString('hex').toUpperCase().replace(/[^A-Z0-9]/g, '').padEnd(len, 'X').slice(0, len)
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const src = process.env.SRC_DATABASE_URL
|
||||
if (!src) throw new Error('SRC_DATABASE_URL 未配置')
|
||||
const cfg = parseUrl(src)
|
||||
|
||||
const conn = await mariadb.createConnection({
|
||||
host: cfg.host, port: cfg.port, user: cfg.user, password: cfg.password, database: cfg.database, connectTimeout: 8000,
|
||||
})
|
||||
console.log('[etl] 源库已连接 (只读):', `${cfg.host}:${cfg.port}/${cfg.database}`)
|
||||
|
||||
const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL })
|
||||
const prisma = new PrismaClient({ adapter })
|
||||
|
||||
const existing = await prisma.project.count()
|
||||
if (existing > 0 && process.env.ETL_FORCE !== '1') {
|
||||
console.log(`[etl] 目标库已有 ${existing} 个项目, 跳过 (ETL_FORCE=1 可强制)`)
|
||||
await conn.end()
|
||||
return
|
||||
}
|
||||
|
||||
// Project.code 回填: 优先按名称推导字母, 冲突加随机尾缀
|
||||
const projectRows = await conn.query('SELECT * FROM `Project`')
|
||||
const usedCodes = new Set()
|
||||
for (const p of projectRows) {
|
||||
const letters = String(p.name || '').toUpperCase().replace(/[^A-Z0-9]/g, '')
|
||||
let cand = ''
|
||||
for (let i = 0; i < 30 && !cand; i++) {
|
||||
const base = i === 0
|
||||
? (letters.length >= 2 ? letters.slice(0, 8) : ('P' + String(p.id) + randCode(4)).slice(0, 8))
|
||||
: (letters.slice(0, 5) + randCode(3)).slice(0, 8)
|
||||
if (base.length >= 2 && !usedCodes.has(base)) cand = base
|
||||
}
|
||||
if (!cand) cand = randCode(6)
|
||||
usedCodes.add(cand)
|
||||
p.code = cand
|
||||
}
|
||||
|
||||
// Task.number 回填: 项目内按 (createdAt, id) 顺序 1..n
|
||||
const taskRows = (await conn.query('SELECT * FROM `Task`')).sort((a, b) => {
|
||||
if (a.projectId !== b.projectId) return a.projectId - b.projectId
|
||||
const ta = a.createdAt ? new Date(a.createdAt).getTime() : 0
|
||||
const tb = b.createdAt ? new Date(b.createdAt).getTime() : 0
|
||||
if (ta !== tb) return ta - tb
|
||||
return a.id - b.id
|
||||
})
|
||||
const taskNumber = new Map()
|
||||
for (const r of taskRows) {
|
||||
const key = String(r.projectId)
|
||||
const next = (taskNumber.get(key) || 0) + 1
|
||||
taskNumber.set(key, next)
|
||||
r.number = next
|
||||
}
|
||||
|
||||
const counts = {}
|
||||
for (const [table, model] of TABLES) {
|
||||
let rows
|
||||
if (table === 'Project') rows = projectRows
|
||||
else if (table === 'Task') rows = taskRows
|
||||
else {
|
||||
const cap = ROW_CAP[table]
|
||||
rows = await conn.query(cap ? `SELECT * FROM \`${table}\` ORDER BY id DESC LIMIT ${cap}` : `SELECT * FROM \`${table}\``)
|
||||
}
|
||||
if (rows.length) {
|
||||
const jsonCols = JSON_COLUMNS[table] || []
|
||||
const bigCols = BIGINT_COLUMNS[table] || []
|
||||
for (const r of rows) {
|
||||
for (const c of jsonCols) r[c] = parseJson(r[c])
|
||||
for (const c of bigCols) r[c] = toBigInt(r[c])
|
||||
if (UUID_TABLES.has(table)) r.uuid = crypto.randomUUID()
|
||||
}
|
||||
const chunk = 500
|
||||
for (let i = 0; i < rows.length; i += chunk) {
|
||||
await prisma[model].createMany({ data: rows.slice(i, i + chunk) })
|
||||
}
|
||||
}
|
||||
counts[table] = rows.length
|
||||
console.log(`[etl] ${table}: ${rows.length}`)
|
||||
}
|
||||
|
||||
// 序列校准: 显式插入 id 后必须 setval, 否则新插入主键冲突
|
||||
for (const [table] of TABLES) {
|
||||
await prisma.$executeRawUnsafe(
|
||||
`SELECT setval(pg_get_serial_sequence('"${table}"', 'id'), COALESCE((SELECT MAX(id) FROM "${table}"), 0) + 1, false)`,
|
||||
)
|
||||
}
|
||||
|
||||
await conn.end()
|
||||
const summary = {
|
||||
users: await prisma.user.count(),
|
||||
projects: await prisma.project.count(),
|
||||
tasks: await prisma.task.count(),
|
||||
checkIns: await prisma.checkIn.count(),
|
||||
news: await prisma.news.count(),
|
||||
}
|
||||
console.log('[etl] 完成:', JSON.stringify(summary))
|
||||
}
|
||||
|
||||
main().catch((e) => { console.error('[etl] 失败:', e); process.exit(1) })
|
||||
Reference in New Issue
Block a user