批量更新
This commit is contained in:
@@ -0,0 +1,355 @@
|
||||
// 全量迁移：把业务表中的老 uid（yz_system_tenant_user.uid）统一替换为
|
||||
// 认证中心的 identity_id（yz_auth_identity.id）。
|
||||
//
|
||||
// 背景：统一认证上线后，令牌里的 user_id 是 identity_id；而业务表的
|
||||
// uid / user_id / create_user_id / owner_user_id / uploader_id 等字段存的仍是
|
||||
// 老表 uid。不迁移的话用户会「找不到自己的数据」，因此做全量替换而非兼容层。
|
||||
//
|
||||
// 用法（需在 go/ 目录运行，读取 conf/app.conf）：
|
||||
//
|
||||
// go run scripts/uidmigrate/migrate_uid.go -check 预检：列出受影响的表/列/行数
|
||||
// go run scripts/uidmigrate/migrate_uid.go -apply 执行迁移（记录变更日志，可回滚）
|
||||
// go run scripts/uidmigrate/migrate_uid.go -rollback 按变更日志回滚
|
||||
//
|
||||
// 说明：本脚本直接使用 database/sql，避免 beego ORM 的 Raw 取值差异。
|
||||
// 执行前请先自行备份数据库（mysqldump）。
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
beego "github.com/beego/beego/v2/server/web"
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
// tempOffset 中间值偏移：先把旧值搬到远离目标值的区间，避免新旧值重叠导致错改
|
||||
const tempOffset = 1000000000
|
||||
|
||||
// 需要处理的列名
|
||||
var uidColumns = []string{
|
||||
"uid", "user_id", "create_user_id", "update_user_id",
|
||||
"owner_user_id", "uploader_id", "operator_id", "tuid",
|
||||
}
|
||||
|
||||
// 不参与迁移的表
|
||||
var skipTables = map[string]bool{
|
||||
"yz_system_tenant_user": true,
|
||||
"yz_users": true,
|
||||
"yz_uid_migration_log": true,
|
||||
}
|
||||
|
||||
type colRef struct {
|
||||
Table string
|
||||
Col string
|
||||
}
|
||||
|
||||
var db *sql.DB
|
||||
|
||||
func main() {
|
||||
check := flag.Bool("check", false, "预检，不写数据")
|
||||
apply := flag.Bool("apply", false, "执行迁移")
|
||||
rollback := flag.Bool("rollback", false, "按变更日志回滚")
|
||||
flag.Parse()
|
||||
|
||||
if !*check && !*apply && !*rollback {
|
||||
fmt.Println("请指定 -check / -apply / -rollback")
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := beego.LoadAppConfig("ini", "conf/app.conf"); err != nil {
|
||||
log.Printf("加载 conf/app.conf 失败（若已自动加载可忽略）: %v", err)
|
||||
}
|
||||
|
||||
var err error
|
||||
db, err = openDB()
|
||||
if err != nil {
|
||||
log.Fatalf("连接数据库失败: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
switch {
|
||||
case *check:
|
||||
runCheck()
|
||||
case *apply:
|
||||
runApply()
|
||||
case *rollback:
|
||||
runRollback()
|
||||
}
|
||||
}
|
||||
|
||||
func openDB() (*sql.DB, error) {
|
||||
user, _ := beego.AppConfig.String("mysqluser")
|
||||
pass, _ := beego.AppConfig.String("mysqlpass")
|
||||
urls, _ := beego.AppConfig.String("mysqlurls")
|
||||
name, _ := beego.AppConfig.String("mysqldb")
|
||||
if user == "" || urls == "" || name == "" {
|
||||
return nil, fmt.Errorf("数据库配置(mysqluser/mysqlurls/mysqldb) 未正确设置")
|
||||
}
|
||||
dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", user, pass, urls, name)
|
||||
return sql.Open("mysql", dsn)
|
||||
}
|
||||
|
||||
// buildMapping 建立「老 uid → identity_id」映射：同企业(tid) 下账号/手机/邮箱 一致。
|
||||
//
|
||||
// 注意：老表是 utf8mb4_0900_ai_ci、新表是 utf8mb4_unicode_ci，直接 JOIN 比较字符串
|
||||
// 会触发 Illegal mix of collations，因此改为在 Go 侧匹配。
|
||||
func buildMapping() (map[uint64]uint64, error) {
|
||||
legacy := make([]struct {
|
||||
uid uint64
|
||||
tid uint64
|
||||
account, phone string
|
||||
email string
|
||||
}, 0)
|
||||
rows, err := db.Query(
|
||||
"SELECT uid, tid, IFNULL(account,''), IFNULL(phone,''), IFNULL(email,'') FROM yz_system_tenant_user WHERE delete_time IS NULL")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var it struct {
|
||||
uid uint64
|
||||
tid uint64
|
||||
account, phone string
|
||||
email string
|
||||
}
|
||||
if err := rows.Scan(&it.uid, &it.tid, &it.account, &it.phone, &it.email); err != nil {
|
||||
continue
|
||||
}
|
||||
legacy = append(legacy, it)
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
index := map[string]uint64{}
|
||||
rows2, err := db.Query(
|
||||
"SELECT identity_id, tid, IFNULL(account,''), IFNULL(phone,''), IFNULL(email,'') FROM yz_auth_tenant_user WHERE delete_time IS NULL")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows2.Next() {
|
||||
var identity, tid uint64
|
||||
var account, phone, email string
|
||||
if err := rows2.Scan(&identity, &tid, &account, &phone, &email); err != nil {
|
||||
continue
|
||||
}
|
||||
if key := mergeKey(tid, account, phone, email); key != "" {
|
||||
index[key] = identity
|
||||
}
|
||||
}
|
||||
rows2.Close()
|
||||
|
||||
mapping := map[uint64]uint64{}
|
||||
for _, it := range legacy {
|
||||
key := mergeKey(it.tid, it.account, it.phone, it.email)
|
||||
if key == "" {
|
||||
continue
|
||||
}
|
||||
if id, ok := index[key]; ok && id > 0 {
|
||||
mapping[it.uid] = id
|
||||
}
|
||||
}
|
||||
return mapping, nil
|
||||
}
|
||||
|
||||
// mergeKey 归并键：企业ID + (手机号 > 邮箱 > 账号)
|
||||
func mergeKey(tid uint64, account, phone, email string) string {
|
||||
base := fmt.Sprintf("%d:", tid)
|
||||
if v := strings.TrimSpace(phone); v != "" {
|
||||
return base + "p:" + v
|
||||
}
|
||||
if v := strings.TrimSpace(email); v != "" {
|
||||
return base + "e:" + v
|
||||
}
|
||||
if v := strings.TrimSpace(account); v != "" {
|
||||
return base + "a:" + v
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func findColumns() ([]colRef, error) {
|
||||
name, _ := beego.AppConfig.String("mysqldb")
|
||||
quoted := make([]string, 0, len(uidColumns))
|
||||
for _, c := range uidColumns {
|
||||
quoted = append(quoted, "'"+c+"'")
|
||||
}
|
||||
rows, err := db.Query(fmt.Sprintf(`
|
||||
SELECT TABLE_NAME, COLUMN_NAME
|
||||
FROM information_schema.COLUMNS
|
||||
WHERE TABLE_SCHEMA = '%s' AND COLUMN_NAME IN (%s)
|
||||
ORDER BY TABLE_NAME, COLUMN_NAME`, name, strings.Join(quoted, ",")))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]colRef, 0)
|
||||
total := 0
|
||||
for rows.Next() {
|
||||
var t, c string
|
||||
if err := rows.Scan(&t, &c); err != nil {
|
||||
continue
|
||||
}
|
||||
total++
|
||||
if skipTables[t] || strings.Contains(t, "_bak") || strings.HasPrefix(t, "yz_auth_") {
|
||||
continue
|
||||
}
|
||||
out = append(out, colRef{Table: t, Col: c})
|
||||
}
|
||||
log.Printf("扫描到 %d 个承载用户ID的列（跳过 %d 个）", len(out), total-len(out))
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func runCheck() {
|
||||
mapping, err := buildMapping()
|
||||
if err != nil {
|
||||
log.Fatalf("建立映射失败: %v", err)
|
||||
}
|
||||
if len(mapping) == 0 {
|
||||
log.Fatal("没有建立任何 uid → identity_id 映射，请确认认证中心数据是否已迁移")
|
||||
}
|
||||
log.Printf("映射关系: %d 个老 uid", len(mapping))
|
||||
for old, newUID := range mapping {
|
||||
log.Printf(" uid %d → identity %d", old, newUID)
|
||||
}
|
||||
|
||||
cols, err := findColumns()
|
||||
if err != nil {
|
||||
log.Fatalf("扫描列失败: %v", err)
|
||||
}
|
||||
|
||||
in := joinKeys(mapping)
|
||||
total := int64(0)
|
||||
affected := 0
|
||||
for _, c := range cols {
|
||||
var cnt int64
|
||||
err := db.QueryRow(fmt.Sprintf(
|
||||
"SELECT COUNT(*) FROM `%s` WHERE `%s` IN (%s)", c.Table, c.Col, in)).Scan(&cnt)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if cnt > 0 {
|
||||
log.Printf(" %-42s %-18s %d 行", c.Table, c.Col, cnt)
|
||||
total += cnt
|
||||
affected++
|
||||
}
|
||||
}
|
||||
log.Printf("合计：%d 张表的列需要更新，约 %d 行（预检完成，未写入任何数据）", affected, total)
|
||||
}
|
||||
|
||||
func runApply() {
|
||||
mapping, err := buildMapping()
|
||||
if err != nil {
|
||||
log.Fatalf("建立映射失败: %v", err)
|
||||
}
|
||||
cols, err := findColumns()
|
||||
if err != nil {
|
||||
log.Fatalf("扫描列失败: %v", err)
|
||||
}
|
||||
ensureLogTable()
|
||||
|
||||
batch := time.Now().Format("20060102150405")
|
||||
changed := 0
|
||||
for _, c := range cols {
|
||||
for oldUID, newUID := range mapping {
|
||||
if oldUID == newUID {
|
||||
continue
|
||||
}
|
||||
// 阶段一：旧值 → 临时值；阶段二：临时值 → 新值
|
||||
cnt := execUpdate(c.Table, c.Col, oldUID, oldUID+tempOffset)
|
||||
if cnt == 0 {
|
||||
continue
|
||||
}
|
||||
execUpdate(c.Table, c.Col, oldUID+tempOffset, newUID)
|
||||
saveLog(batch, c.Table, c.Col, oldUID, newUID, cnt)
|
||||
changed++
|
||||
log.Printf(" %s.%s: %d → %d（%d 行）", c.Table, c.Col, oldUID, newUID, cnt)
|
||||
}
|
||||
}
|
||||
log.Printf("迁移完成：%d 处替换，批次号 %s", changed, batch)
|
||||
log.Println("如需回滚：go run scripts/uidmigrate/migrate_uid.go -rollback")
|
||||
}
|
||||
|
||||
func runRollback() {
|
||||
rows, err := db.Query(
|
||||
"SELECT table_name, column_name, old_value, new_value FROM yz_uid_migration_log ORDER BY id DESC")
|
||||
if err != nil {
|
||||
log.Fatalf("读取变更日志失败: %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
type item struct {
|
||||
table, col string
|
||||
oldV, newV uint64
|
||||
}
|
||||
list := make([]item, 0)
|
||||
for rows.Next() {
|
||||
var it item
|
||||
if err := rows.Scan(&it.table, &it.col, &it.oldV, &it.newV); err != nil {
|
||||
continue
|
||||
}
|
||||
list = append(list, it)
|
||||
}
|
||||
if len(list) == 0 {
|
||||
log.Println("没有可回滚的记录")
|
||||
return
|
||||
}
|
||||
for _, it := range list {
|
||||
// 反向：新值 → 临时值 → 旧值
|
||||
execUpdate(it.table, it.col, it.newV, it.newV+tempOffset)
|
||||
cnt := execUpdate(it.table, it.col, it.newV+tempOffset, it.oldV)
|
||||
log.Printf(" 回滚 %s.%s: %d → %d（%d 行）", it.table, it.col, it.newV, it.oldV, cnt)
|
||||
}
|
||||
log.Println("回滚完成")
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- 工具
|
||||
|
||||
func execUpdate(table, col string, from, to uint64) int64 {
|
||||
res, err := db.Exec(fmt.Sprintf(
|
||||
"UPDATE `%s` SET `%s` = ? WHERE `%s` = ?", table, col, col), to, from)
|
||||
if err != nil {
|
||||
log.Printf("更新失败 %s.%s (%d→%d): %v", table, col, from, to, err)
|
||||
return 0
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return n
|
||||
}
|
||||
|
||||
func saveLog(batch, table, col string, oldV, newV uint64, rows int64) {
|
||||
_, err := db.Exec(
|
||||
"INSERT INTO yz_uid_migration_log (batch_no, table_name, column_name, old_value, new_value, row_count, create_time) VALUES (?,?,?,?,?,?,NOW())",
|
||||
batch, table, col, oldV, newV, rows)
|
||||
if err != nil {
|
||||
log.Printf("记录变更日志失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func ensureLogTable() {
|
||||
_, _ = db.Exec(`CREATE TABLE IF NOT EXISTS yz_uid_migration_log (
|
||||
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
batch_no VARCHAR(32) NOT NULL,
|
||||
table_name VARCHAR(128) NOT NULL,
|
||||
column_name VARCHAR(64) NOT NULL,
|
||||
old_value BIGINT UNSIGNED NOT NULL,
|
||||
new_value BIGINT UNSIGNED NOT NULL,
|
||||
row_count BIGINT NOT NULL DEFAULT 0,
|
||||
create_time DATETIME NOT NULL,
|
||||
PRIMARY KEY (id),
|
||||
KEY idx_batch (batch_no)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`)
|
||||
}
|
||||
|
||||
func joinKeys(m map[uint64]uint64) string {
|
||||
parts := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
parts = append(parts, fmt.Sprintf("%d", k))
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return "0"
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
Reference in New Issue
Block a user