|
|
@@ -0,0 +1,774 @@
|
|
|
+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)
|
|
|
+}
|
