package migrate import ( "context" "fmt" "time" "gorm.io/driver/mysql" "gorm.io/gorm" identitymodel "yuansiwei/unified-backend/internal/identity/model" "yuansiwei/unified-backend/internal/identity/id" "yuansiwei/unified-backend/internal/migrate/norm" orgmodel "yuansiwei/unified-backend/internal/org/model" ) // Options 迁移运行配置。 type Options struct { // TeachingDSN 教学库 yuansiwei_i(只读源)。 TeachingDSN string // FinanceDSN 教务库 finance_bk(只读源)。 FinanceDSN string // TargetDSN 目标态 unified 库(读写,写入 5 表)。 TargetDSN string // DryRun 仅预演:读库做存在性检查 + 产出报告,不执行任何写入。 DryRun bool // Now 时间锚点(regTime 为空时用于 ULID 发放 / created_at)。可注入便于测试。 Now time.Time } // Runner 迁移编排器(P1-04/T5 主体)。 // // 职责: // 1. 读 legacy 双库(只读); // 2. 按 P0-06 归一 + P0-07 置信分级规划; // 3. 幂等 upsert 写目标态 5 表(t_institution/t_campus 维度 + t_user/t_identity/ // t_wechat_union/t_membership/t_legacy_mapping); // 4. 产出 MigrateReport(含 P0-08 冲突/异常清单)。 // // 事务边界:所有写入在单个 GORM 事务中(dry-run 不进入事务,仅读)。 type Runner struct { opts Options src *Source target *gorm.DB // wxCache:legacy userid → 微信身份(Run 时填充,writeIdentities/writeWechatUnion 消费)。 // 放在 Runner 上避免大参数透传;非并发(单事务串行处理)。 wxCache map[int32]*LegacyWeixinUser } // NewRunner 构建迁移编排器(连接 legacy 双库 + 目标库,均 fail-fast)。 func NewRunner(opts Options) (*Runner, error) { if opts.Now.IsZero() { opts.Now = time.Now() } src, err := NewSource(opts.TeachingDSN, opts.FinanceDSN) if err != nil { return nil, err } target, err := gorm.Open(gormMySQLTarget(opts.TargetDSN), gormMigrateCfg()) if err != nil { _ = src.Close() return nil, fmt.Errorf("migrate: 目标库连接失败: %w", err) } if sqlDB, err := target.DB(); err == nil { if err := sqlDB.Ping(); err != nil { _ = src.Close() return nil, fmt.Errorf("migrate: 目标库 Ping 失败: %w", err) } } return &Runner{opts: opts, src: src, target: target}, nil } // Close 关闭所有连接。 func (r *Runner) Close() error { var err error if r.src != nil { err = r.src.Close() } if r.target != nil { if sqlDB, derr := r.target.DB(); derr == nil { _ = sqlDB.Close() } } return err } // EnsureTables 确保目标态 12 表存在(dev 期 AutoMigrate;生产以 DDL 为准,幂等对齐)。 // 迁移前调用,避免"写入时表不存在"。 func (r *Runner) EnsureTables(ctx context.Context) error { tables := append(identitymodel.AllTables(), orgmodel.AllTables()...) return r.target.WithContext(ctx).AutoMigrate(tables...) } // Run 执行迁移(单事务)。 // // 出参: // - *MigrateReport:完整报告(含冲突/异常清单,P0-08 输入)。 // - error:仅当读取/事务提交失败时返回;单行写入失败记录进 report.Warnings 不中断。 func (r *Runner) Run(ctx context.Context) (*MigrateReport, error) { rep := &MigrateReport{DryRun: r.opts.DryRun} // ---- 读取 legacy 全量(只读)---- users, err := r.src.LoadAllUsers(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_user 失败: %w", err) } rep.UsersTotal = len(users) wx, err := r.src.LoadAllWeixinUsers(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_weixin_user 失败: %w", err) } // 填充微信缓存(userid → 身份,1:1) r.wxCache = make(map[int32]*LegacyWeixinUser, len(wx)) for i := range wx { if wx[i].UserID > 0 { if _, exists := r.wxCache[wx[i].UserID]; !exists { r.wxCache[wx[i].UserID] = &wx[i] } } } assoc, err := r.src.LoadAllAssociations(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_user_account_association 失败: %w", err) } schoolIDs, err := r.src.LoadDistinctSchoolIDs(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_user_school_campus distinct schoolid 失败: %w", err) } schools, err := r.src.LoadSchools(ctx, schoolIDs) if err != nil { return rep, fmt.Errorf("migrate: 读 t_school 失败: %w", err) } rep.InstitutionsTotal = len(schools) campuses, err := r.src.LoadCampuses(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_school_campus 失败: %w", err) } rep.CampusesTotal = len(campuses) binds, err := r.src.LoadUserSchoolCampus(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_user_school_campus 失败: %w", err) } // ---- 规划(纯逻辑:ULID 发放 + 置信分级 + 冲突检测)---- // 已有映射(重跑幂等:legacy id → canonical_user_id 复用) existingCanon, err := r.loadExistingMappings(ctx) if err != nil { return rep, fmt.Errorf("migrate: 读 t_legacy_mapping 失败: %w", err) } userSet := make(map[int32]struct{}, len(users)) for i := range users { userSet[users[i].ID] = struct{}{} } // 微信孤儿(openid 的 userid 不在 t_user) for i := range wx { if _, ok := userSet[wx[i].UserID]; !ok { rep.WechatOrphans = append(rep.WechatOrphans, ConflictItem{ LegacyID: wx[i].UserID, Detail: fmt.Sprintf("openid=%s 的 userid 不在 t_user,丢弃", wx[i].OpenID), }) } } // 种子映射(userid → 阅卷/标记系统 id),供 t_legacy_mapping.legacy_teaching_user_id(P1-05 §2 拍定)。 teachingByUserID := make(map[int32]int32, len(assoc)) for i := range assoc { if assoc[i].UserID > 0 && assoc[i].MarkingSystemUserID != nil && *assoc[i].MarkingSystemUserID > 0 { if _, exists := teachingByUserID[assoc[i].UserID]; !exists { teachingByUserID[assoc[i].UserID] = *assoc[i].MarkingSystemUserID } } } planner := NewPlanner(users, assoc, wx) // ULID 发放:已映射复用,新 id 发放(时间有序:以 regTime 为锚) canonByID := make(map[int32]string, len(users)) for i := range users { u := &users[i] if c, ok := existingCanon[u.ID]; ok && c != "" { canonByID[u.ID] = c continue } var t time.Time if u.RegTime != nil { t = *u.RegTime } else { t = r.opts.Now } c, err := id.NewCanonicalUserIDAt(t) if err != nil { return rep, fmt.Errorf("migrate: ULID 发放失败(legacy id=%d): %w", u.ID, err) } canonByID[u.ID] = c } // 决策表(每个 user 的匹配决策) decisionByID := make(map[int32]MatchDecision, len(users)) for i := range users { u := &users[i] d := planner.Decide(u.ID, u.PhoneNumber) decisionByID[u.ID] = d // phone 撞号冲突清单(每组只记一次,避免重复) if d.PhoneConflict { if u.PhoneNumber != nil { rep.PhoneConflicts = appendUniqueConflict(rep.PhoneConflicts, ConflictItem{ LegacyID: u.ID, Detail: fmt.Sprintf("phone=%s 撞号(同归一 phone 多账号,不自动合并)", normPhoneDisplay(*u.PhoneNumber)), }) } } if d.PhoneInvalid && u.PhoneNumber != nil { rep.PhoneInvalid = append(rep.PhoneInvalid, ConflictItem{ LegacyID: u.ID, Detail: fmt.Sprintf("phone 归一失败,原始值=%q", *u.PhoneNumber), }) } } // ---- 写入(单事务;dry-run 仅读不写)---- if r.opts.DryRun { // dry-run:只做存在性预检 + 计数(不实际写入)。 // 机构/校区/用户/身份/成员 的新建量按"目标库已有则复用"估算。 r.simulateCounts(ctx, rep, schools, campuses, users, canonByID, binds) return rep, nil } err = r.target.Transaction(func(tx *gorm.DB) error { // ① 机构(维度表,幂等:已存在则复用) instByID, err := r.writeInstitutions(ctx, tx, schools, rep) if err != nil { return err } // ② 校区(依赖机构) campusByID, err := r.writeCampuses(ctx, tx, campuses, instByID, rep) if err != nil { return err } // ③ 用户 + 身份 + 微信锚点 + 遗留映射(按 legacy id 序,幂等) usedUsername := map[string]int32{} // 归一 username → 首个占用者 usedPhone := map[string]int32{} // 归一 phone → 首个占用者 existUsername := r.existingUsernames(ctx, tx) existPhone := r.existingIdentities(ctx, tx, identitymodel.IdentityTypePhone, "") for k, v := range existUsername { usedUsername[k] = v } for k, v := range existPhone { usedPhone[k] = v } for i := range users { u := &users[i] canon := canonByID[u.ID] d := decisionByID[u.ID] if err := r.writeUser(ctx, tx, u, canon, d, rep, usedUsername, existUsername); err != nil { rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_user legacy_id=%d 失败: %v", u.ID, err)) continue } r.writeIdentities(ctx, tx, u, canon, d, rep, usedPhone, existPhone) r.writeWechatUnion(ctx, tx, u.ID, canon, planner, rep) if err := r.writeLegacyMapping(ctx, tx, u, canon, d, teachingByUserID, rep); err != nil { rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_legacy_mapping legacy_id=%d 失败: %v", u.ID, err)) } } // ④ 成员关系(依赖 canonical + 机构/校区) r.writeMemberships(ctx, tx, binds, canonByID, instByID, campusByID, users, rep) return nil }) if err != nil { return rep, fmt.Errorf("migrate: 事务提交失败(已回滚): %w", err) } return rep, nil } // normPhoneDisplay 展示用(截断,防超长)。 func normPhoneDisplay(raw string) string { c, ok := norm.Phone(raw) if ok { return c } return raw } // appendUniqueConflict 去重追加冲突项(同 LegacyID 只记一次)。 func appendUniqueConflict(items []ConflictItem, item ConflictItem) []ConflictItem { for _, e := range items { if e.LegacyID == item.LegacyID { return items } } return append(items, item) } // ---- 幂等辅助(读目标库现有状态)---- // loadExistingMappings 读 t_legacy_mapping,返回 legacy_i_user_id → canonical_user_id。 // // dry-run 容错:目标表未建(MySQL 1146 表不存在)时按"空表"处理(全量首次迁移,ULID 全发放)。 // 这保证 dry-run 纯只读、不依赖目标库已建表。 func (r *Runner) loadExistingMappings(ctx context.Context) (map[int32]string, error) { m := make(map[int32]string) var rows []identitymodel.LegacyMapping if err := r.target.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Find(&rows).Error; err != nil { if isMissingTable(err) { return m, nil // 表不存在(dry-run 首次)→ 空映射 } return nil, err } for i := range rows { if rows[i].LegacyIUserID != nil && *rows[i].LegacyIUserID > 0 { m[*rows[i].LegacyIUserID] = rows[i].CanonicalUserID } } return m, nil } // existingUsernames 读 t_user 现有 username(归一 username → canonical_user_id 占位,用于冲突检测)。 func (r *Runner) existingUsernames(ctx context.Context, tx *gorm.DB) map[string]int32 { m := make(map[string]int32) var users []identitymodel.User if err := tx.WithContext(ctx).Model(&identitymodel.User{}).Find(&users).Error; err != nil { return m } for i := range users { if users[i].Username != nil && *users[i].Username != "" { canon := norm.Username(*users[i].Username) if _, ok := m[canon]; !ok { m[canon] = 0 // 占位(无 legacy id,仅标记"已占用") } } } return m } // existingIdentities 读 t_identity 现有指定 type 的 identifier(归一值 → 占位)。 func (r *Runner) existingIdentities(ctx context.Context, tx *gorm.DB, typ, appid string) map[string]int32 { m := make(map[string]int32) var idents []identitymodel.Identity q := tx.WithContext(ctx).Model(&identitymodel.Identity{}).Where("type = ?", typ) if err := q.Find(&idents).Error; err != nil { return m } for i := range idents { m[idents[i].Identifier] = 0 } return m } // ---- 写入方法(每个方法在 dry-run 外执行;幂等:存在则跳过)---- // writeInstitutions 写 t_institution(幂等:按 name 查现有复用)。 func (r *Runner) writeInstitutions(ctx context.Context, tx *gorm.DB, schools []LegacySchool, rep *MigrateReport) (map[int32]int64, error) { instByID := make(map[int32]int64, len(schools)) for i := range schools { s := &schools[i] var existing []orgmodel.Institution tx.WithContext(ctx).Model(&orgmodel.Institution{}). Where("institution_name = ?", s.Name).Limit(1).Find(&existing) if len(existing) > 0 { instByID[s.ID] = existing[0].InstitutionID continue } inst := &orgmodel.Institution{ InstitutionName: s.Name, Status: mapOrgStatus(s.Status), CreatedAt: r.opts.Now, UpdatedAt: r.opts.Now, } if err := tx.WithContext(ctx).Create(inst).Error; err != nil { return instByID, err } instByID[s.ID] = inst.InstitutionID rep.InstitutionsCreated++ } return instByID, nil } // writeCampuses 写 t_campus(幂等:按 campus_name + institution 复用)。 func (r *Runner) writeCampuses(ctx context.Context, tx *gorm.DB, campuses []LegacySchoolCampus, instByID map[int32]int64, rep *MigrateReport) (map[int32]int64, error) { campusByID := make(map[int32]int64, len(campuses)) for i := range campuses { c := &campuses[i] // 孤儿校区(school_id 不在被引用机构集):institution_id 置 0(机构级/平台级) instID, ok := instByID[c.SchoolID] if !ok { instID = 0 } var existing []orgmodel.Campus tx.WithContext(ctx).Model(&orgmodel.Campus{}). Where("campus_name = ? AND institution_id = ?", c.Name, instID).Limit(1).Find(&existing) if len(existing) > 0 { campusByID[c.ID] = existing[0].CampusID continue } campus := &orgmodel.Campus{ InstitutionID: instID, CampusName: c.Name, CampusCode: c.Code, Status: mapOrgStatus(c.Status), CreatedAt: r.opts.Now, UpdatedAt: r.opts.Now, } if err := tx.WithContext(ctx).Create(campus).Error; err != nil { return campusByID, err } campusByID[c.ID] = campus.CampusID rep.CampusesCreated++ } return campusByID, nil } // writeUser 写 t_user(幂等:canonical_user_id 存在则跳过;username 冲突置 NULL)。 func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string, d MatchDecision, rep *MigrateReport, usedUsername map[string]int32, existUsername map[string]int32) error { var n int64 tx.WithContext(ctx).Model(&identitymodel.User{}).Where("canonical_user_id = ?", canon).Count(&n) if n > 0 { rep.UsersSkipped++ return nil } // username:归一后若已占用(存量或本批首个外)→ 置 NULL + 记冲突 var username *string if u.Username != "" { canonUsername := norm.Username(u.Username) if _, taken := usedUsername[canonUsername]; !taken { s := u.Username username = &s usedUsername[canonUsername] = u.ID } else { rep.UsernameConflicts = append(rep.UsernameConflicts, ConflictItem{ LegacyID: u.ID, Detail: fmt.Sprintf("username=%q 归一后与已有用户重复,置 NULL", u.Username), }) } } _ = existUsername // 存量 username 冲突已在 usedUsername 合并 displayName := u.RealName user := &identitymodel.User{ CanonicalUserID: canon, Username: username, DisplayName: displayName, AvatarURL: u.Avatar, PersonType: mapPersonType(u.Role), PasswordHash: &u.Password, PasswordSalt: &u.Salt, HashVersion: strPtr(identitymodel.HashVersionShaSalt), // I 库=教学侧,sha_salt(ADR-016 惰性升级) Status: mapUserStatus(u.Status), CreatedAt: r.opts.Now, UpdatedAt: r.opts.Now, } if u.RegTime != nil { user.CreatedAt = *u.RegTime } if err := tx.WithContext(ctx).Create(user).Error; err != nil { return err } rep.UsersCreated++ if u.Status == 0 || u.Status == 3 { rep.UsersDisabled++ } return nil } // writeIdentities 写 t_identity(phone/email/username/wechat;幂等 + 全局唯一键冲突处理)。 func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string, d MatchDecision, rep *MigrateReport, usedPhone map[string]int32, existPhone map[string]int32) { // phone(强归一;撞号→本批首个占用,其余跳过+冲突已记) if u.PhoneNumber != nil && *u.PhoneNumber != "" { canonPhone, ok := norm.Phone(*u.PhoneNumber) if ok && !d.PhoneConflict { if _, taken := usedPhone[canonPhone]; !taken { if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypePhone, canonPhone, "", rep); err == nil { usedPhone[canonPhone] = u.ID rep.IdentitiesPhone++ } } } } // email(轻归一) if u.Email != nil && *u.Email != "" { canonEmail, ok := norm.Email(*u.Email) if ok { if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeEmail, canonEmail, "", rep); err == nil { rep.IdentitiesEmail++ } } } // username(轻归一;与 t_user.username 一致) if u.Username != "" { canonUsername := norm.Username(u.Username) if canonUsername != "" { if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeUsername, canonUsername, "", rep); err == nil { rep.IdentitiesUsername++ } } } // wechat(openid + appid,经 t_weixin_user) if w := r.wechatOf(u.ID); w != nil { openid, ok := norm.WechatOpenID(w.OpenID) if ok { appid := "" if w.AppID != nil { appid = *w.AppID } if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeWechat, openid, appid, rep); err == nil { rep.IdentitiesWechat++ } } } _ = existPhone } // wechatOf 取某 legacy user 的微信身份(runner 持有 planner 不方便,直接查 wx 缓存)。 // 注:实现上在 Run 里通过闭包注入更合适;此处用字段缓存。 func (r *Runner) wechatOf(userID int32) *LegacyWeixinUser { return r.wxCache[userID] } // upsertIdentity 幂等写入单条 identity(唯一键 type+identifier+appid 已存在则跳过)。 func (r *Runner) upsertIdentity(ctx context.Context, tx *gorm.DB, canon, typ, identifier, appid string, rep *MigrateReport) error { var n int64 tx.WithContext(ctx).Model(&identitymodel.Identity{}). Where("type = ? AND identifier = ? AND appid = ?", typ, identifier, appid).Count(&n) if n > 0 { rep.IdentityDuplicates++ return nil } it := &identitymodel.Identity{ CanonicalUserID: canon, Type: typ, Identifier: identifier, AppID: appid, Verified: false, CreatedAt: r.opts.Now, } return tx.WithContext(ctx).Create(it).Error } // writeWechatUnion 写 t_wechat_union(unionid 锚点;幂等:canonical/unionid 已存在则跳过)。 func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32, canon string, planner *Planner, rep *MigrateReport) { w := r.wxCache[userID] if w == nil || w.UnionID == nil || *w.UnionID == "" { return } // 同 unionid 可能挂多个 canonical(跨 appid 同人):已存在任一 canonical 的该 unionid → 跳过 var n int64 tx.WithContext(ctx).Model(&identitymodel.WechatUnion{}).Where("unionid = ?", *w.UnionID).Count(&n) if n > 0 { return } wu := &identitymodel.WechatUnion{ CanonicalUserID: canon, UnionID: *w.UnionID, CreatedAt: r.opts.Now, UpdatedAt: r.opts.Now, } if err := tx.WithContext(ctx).Create(wu).Error; err == nil { rep.WechatUnionCreated++ } else { rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_wechat_union unionid=%s 失败: %v", *w.UnionID, err)) } } // writeLegacyMapping 写 t_legacy_mapping(canonical_user_id UNIQUE 幂等;重跑复用 ULID 时天然存在则跳过)。 // // 三源 user_id(P1-05 §2 拍定): // - LegacyIUserID = 旧A yuansiwei_i.t_user.id(本地主源,恒有) // - LegacyTeachingUserID = 阅卷/标记系统 id(t_user_account_association.marking_system_userid,仅种子用户有) // - LegacyAcademicUserID = 教务 user id(本地恒 NULL,finance_bk 无 user 主表) func (r *Runner) writeLegacyMapping(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string, d MatchDecision, teachingByUserID map[int32]int32, rep *MigrateReport) error { var n int64 tx.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Where("canonical_user_id = ?", canon).Count(&n) if n > 0 { return nil // 已存在(重跑幂等) } legacyID := u.ID role := mapRoleCode(u.Role) lm := &identitymodel.LegacyMapping{ CanonicalUserID: canon, LegacyIUserID: &legacyID, // 旧A yuansiwei_i.t_user.id InstitutionID: 0, // 平台级(canonical 跨机构;代表机构非身份属性,P1-05 §6 拍定) CampusID: 0, Role: &role, MatchSource: strPtr(d.MatchSource), MatchConfidence: d.Confidence, AuditRef: strPtr(fmt.Sprintf("migrate/v1.2/%d", u.ID)), MergedAt: &r.opts.Now, CreatedAt: r.opts.Now, UpdatedAt: r.opts.Now, } // 阅卷/标记系统 id(种子用户才有;P1-05 §2:marking_system_userid → legacy_teaching_user_id) if markingID, ok := teachingByUserID[u.ID]; ok { lm.LegacyTeachingUserID = &markingID } if err := tx.WithContext(ctx).Create(lm).Error; err != nil { return err } rep.LegacyMappingsCreated++ return nil } // writeMemberships 写 t_membership(幂等:唯一键 canonical+campus+role)。 func (r *Runner) writeMemberships(ctx context.Context, tx *gorm.DB, binds []LegacyUserSchoolCampus, canonByID map[int32]string, instByID map[int32]int64, campusByID map[int32]int64, users []LegacyUser, rep *MigrateReport) { userByID := make(map[int32]*LegacyUser, len(users)) for i := range users { userByID[users[i].ID] = &users[i] } for i := range binds { b := &binds[i] u, ok := userByID[b.UserID] if !ok { // 绑定用户不在 t_user(孤儿)→ 跳过 + 警告 rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过:legacy 绑定 userid=%d 不在 t_user", b.UserID)) continue } canon, ok := canonByID[b.UserID] if !ok { continue } instID, ok1 := instByID[b.SchoolID] campusID, ok2 := campusByID[b.CampusID] if !ok1 || !ok2 { rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过:institution/campus 映射缺失 schoolid=%d campusid=%d", b.SchoolID, b.CampusID)) continue } role := mapRoleCode(u.Role) // 幂等:唯一键 (canonical, campus, role) var n int64 tx.WithContext(ctx).Model(&orgmodel.Membership{}). Where("canonical_user_id = ? AND campus_id = ? AND role = ?", canon, campusID, role).Count(&n) if n > 0 { continue } m := &orgmodel.Membership{ CanonicalUserID: canon, InstitutionID: instID, CampusID: campusID, Role: role, Status: orgmodel.MembershipStatusActive, JoinedAt: &r.opts.Now, CreatedAt: r.opts.Now, } if err := tx.WithContext(ctx).Create(m).Error; err == nil { rep.MembershipsCreated++ } else { rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_membership legacy_userid=%d 失败: %v", b.UserID, err)) } } } // simulateCounts dry-run 计数预检(只读,不写):估算各表新建量。 func (r *Runner) simulateCounts(ctx context.Context, rep *MigrateReport, schools []LegacySchool, campuses []LegacySchoolCampus, users []LegacyUser, canonByID map[int32]string, binds []LegacyUserSchoolCampus) { existingUsers := r.countTable(ctx, "t_user") existingInst := r.countTable(ctx, "t_institution") existingCampus := r.countTable(ctx, "t_campus") existingMapping := r.countTable(ctx, "t_legacy_mapping") // 幂等估算:新建 = 目标量 - 已存在量(取下限 0) if len(users) > int(existingUsers) { rep.UsersCreated = len(users) - int(existingUsers) } if len(schools) > int(existingInst) { rep.InstitutionsCreated = len(schools) - int(existingInst) } if len(campuses) > int(existingCampus) { rep.CampusesCreated = len(campuses) - int(existingCampus) } if len(users) > int(existingMapping) { rep.LegacyMappingsCreated = len(users) - int(existingMapping) } // 身份/成员/锚点按"全量首次写入"估算(重跑时这些由唯一键去重,量趋 0) for i := range users { u := &users[i] _ = canonByID[u.ID] if u.PhoneNumber != nil && *u.PhoneNumber != "" { if _, ok := norm.Phone(*u.PhoneNumber); ok { rep.IdentitiesPhone++ } } if u.Email != nil && *u.Email != "" { if _, ok := norm.Email(*u.Email); ok { rep.IdentitiesEmail++ } } if u.Username != "" { rep.IdentitiesUsername++ } } for i := range users { if w := r.wxCache[users[i].ID]; w != nil { rep.IdentitiesWechat++ if w.UnionID != nil && *w.UnionID != "" { rep.WechatUnionCreated++ } } } rep.MembershipsCreated = len(binds) } // countTable 统计表行数(dry-run 幂等估算用)。 func (r *Runner) countTable(ctx context.Context, table string) int64 { var n int64 _ = r.target.WithContext(ctx).Table(table).Count(&n).Error return n } // ---- 映射函数 ---- // mapUserStatus legacy status int → 目标态 string。 // 1=active,0/2/3=disabled(P0-02 §4 拍板:0/3 都迁 disabled 保留可追溯)。 func mapUserStatus(legacy int32) string { if legacy == 1 { return identitymodel.UserStatusActive } return identitymodel.UserStatusDisabled } // mapOrgStatus 机构/校区 legacy status(tinyint)→ 目标态 string。 func mapOrgStatus(legacy int8) string { if legacy == 1 { return orgmodel.StatusActive } return orgmodel.StatusDisabled } // mapPersonType legacy role 数字码 → 目标态 person_type(P0-02 §3.1)。 func mapPersonType(legacy int32) string { switch legacy { case 2: return identitymodel.PersonTypeStudent case 4: return identitymodel.PersonTypeTeacher case 3: return identitymodel.PersonTypeParent default: return identitymodel.PersonTypeStaff } } // mapRoleCode legacy role 数字码 → 目标态 7 角色字符串码(P0-02 §2.5 拍板映射)。 func mapRoleCode(legacy int32) string { switch legacy { case 1: return orgmodel.RoleSuperAdmin case 2: return orgmodel.RoleStudent case 3: return orgmodel.RoleParent case 4: return orgmodel.RoleTeacher case 5, 6: return orgmodel.RoleAcademic // admin/redactor → 教务 case 8: return orgmodel.RoleFinance case 7, 9: return orgmodel.RoleMarket // prototype/marketer → 市场 default: return orgmodel.RoleAcademic // 未知角色兜底为教务(运营角色) } } // strPtr 字符串指针辅助。 func strPtr(s string) *string { return &s } // gormMySQLTarget 目标库 GORM 驱动(mysql.Open 返回 Dialector,与 infra/db 同款)。 func gormMySQLTarget(dsn string) gorm.Dialector { return mysql.Open(dsn) }