diff --git a/deploy/0.2.1/etl.sh b/deploy/0.2.1/etl.sh new file mode 100644 index 0000000..cbe47b7 --- /dev/null +++ b/deploy/0.2.1/etl.sh @@ -0,0 +1,13 @@ +#!/bin/bash +# 一次性 ETL: 清空 0.2.1 库 -> 从现网 MariaDB 只读快照导入 +# 现网连接串从 iboard 容器环境变量读取 (不回显), 源库仅 SELECT +set -e +cd /app/iboard-0.2.1/deploy/0.2.1 +SRC=$(docker inspect iboard --format '{{range .Config.Env}}{{println .}}{{end}}' | grep '^DATABASE_URL=' | cut -d= -f2-) +[ -n "$SRC" ] || { echo '未取到现网 DATABASE_URL'; exit 1; } + +echo '--- truncate 0.2.1 库 ---' +docker compose exec -T pg psql -U iboard021 -d iboard021 -c "TRUNCATE TABLE \"User\",\"Project\",\"ProjectMember\",\"BoardColumn\",\"Task\",\"TaskMember\",\"TaskTag\",\"Habit\",\"CheckIn\",\"CheckList\",\"CheckListItem\",\"CheckListTemplate\",\"Article\",\"Strategy\",\"Indicator\",\"IndicatorPoint\",\"News\",\"Alert\",\"ServerNode\",\"ServerNodeHistory\",\"ServiceCheck\",\"Econ_SowInventory\",\"ScraperCache\",\"OilPriceHistory\",\"MihomeAccount\",\"MihomeHome\",\"MihomeDevice\",\"MihomeEnvSample\",\"MihomeEnvHour\" RESTART IDENTITY CASCADE;" + +echo '--- etl run ---' +docker compose exec -T -e SRC_DATABASE_URL="$SRC" app node scripts/etl-mariadb-to-pg.mjs diff --git a/scripts/etl-mariadb-to-pg.mjs b/scripts/etl-mariadb-to-pg.mjs new file mode 100644 index 0000000..1386760 --- /dev/null +++ b/scripts/etl-mariadb-to-pg.mjs @@ -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) })