source.go 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249
  1. // Package migrate 数据迁移引擎(P1-04/T5,ADR-003 归一 pipeline 落地)。
  2. //
  3. // 职责:从 legacy 双库(yuansiwei_i 教学 + finance_bk 教务,只读)读取三源用户数据,
  4. // 按 P0-06 归一规则 + P0-07 置信分级执行匹配,幂等 upsert 写入目标态 5 表
  5. // (t_user/t_identity/t_wechat_union/t_membership/t_legacy_mapping)。
  6. //
  7. // 设计铁律:
  8. // - **幂等**:重跑不产生重复行(按 legacy id 顺序处理 + 存在性检查)
  9. // - **先归一后匹配**:归一失败绝不写脏 canonical(phone_invalid → 仅记录 raw)
  10. // - **置信分级**:高置信(种子 + phone 唯一)自动写;中/低/冲突 → 仅出报告
  11. // - **事务边界**:整批在单个 GORM 事务中(dry-run 不写)
  12. // - **ULID 发放**:id.NewCanonicalUserIDAt(regTime) 保证时间有序
  13. // - **legacy 只读**:Source 连接仅执行 SELECT,无任何 INSERT/UPDATE/DELETE
  14. package migrate
  15. import (
  16. "context"
  17. "fmt"
  18. "strings"
  19. "time"
  20. "gorm.io/driver/mysql"
  21. "gorm.io/gorm"
  22. "gorm.io/gorm/logger"
  23. )
  24. // gormSilent 静默 logger(迁移批量读写高频 SQL,不刷日志;SLOW SQL 阈值噪音在此屏蔽)。
  25. var gormSilent = logger.Default.LogMode(logger.Silent)
  26. // gormMigrateCfg 迁移包共用的 GORM 配置(不建物理 FK + 静默日志)。
  27. func gormMigrateCfg() *gorm.Config {
  28. return &gorm.Config{
  29. DisableForeignKeyConstraintWhenMigrating: true,
  30. Logger: gormSilent,
  31. }
  32. }
  33. // isMissingTable 判断 GORM 错误是否为"表不存在"(MySQL 1146)。
  34. // dry-run / 幂等读路径据此容错:目标表未建时按"空表"处理(全量首次迁移)。
  35. func isMissingTable(err error) bool {
  36. if err == nil {
  37. return false
  38. }
  39. msg := err.Error()
  40. return strings.Contains(msg, "1146") || strings.Contains(msg, "doesn't exist")
  41. }
  42. // ---- legacy 数据源结构体(GORM 只读映射,不 AutoMigrate)----
  43. // LegacyUser 教学侧 yuansiwei_i.t_user(Canonical 主表,68107 行)。
  44. type LegacyUser struct {
  45. ID int32 `gorm:"column:id;primaryKey"`
  46. Username string `gorm:"column:username"`
  47. RealName *string `gorm:"column:real_name"`
  48. Password string `gorm:"column:password"`
  49. Salt string `gorm:"column:salt"`
  50. Role int32 `gorm:"column:role"`
  51. Email *string `gorm:"column:email"`
  52. PhoneNumber *string `gorm:"column:phone_number"`
  53. Avatar *string `gorm:"column:avatar"`
  54. RegTime *time.Time `gorm:"column:reg_time"`
  55. Status int32 `gorm:"column:status"`
  56. ParentID int32 `gorm:"column:parent_id"`
  57. GradeID int32 `gorm:"column:grade_id"`
  58. }
  59. // TableName 指定 legacy 表名。
  60. func (LegacyUser) TableName() string { return "t_user" }
  61. // LegacyWeixinUser 教学侧 yuansiwei_i.t_weixin_user(微信身份,3491 行)。
  62. type LegacyWeixinUser struct {
  63. OpenID string `gorm:"column:openid;primaryKey"`
  64. Nickname string `gorm:"column:nickname"`
  65. UserID int32 `gorm:"column:userid"`
  66. UnionID *string `gorm:"column:unionid"`
  67. AppID *string `gorm:"column:appid"`
  68. }
  69. // TableName 指定 legacy 表名。
  70. func (LegacyWeixinUser) TableName() string { return "t_weixin_user" }
  71. // LegacyAccountAssociation 教学侧 yuansiwei_i.t_user_account_association(现映射种子,16077 行)。
  72. type LegacyAccountAssociation struct {
  73. UserID int32 `gorm:"column:userid;primaryKey"`
  74. MarkingSystemUserID *int32 `gorm:"column:marking_system_userid"`
  75. }
  76. // TableName 指定 legacy 表名。
  77. func (LegacyAccountAssociation) TableName() string { return "t_user_account_association" }
  78. // LegacySchool 教学侧 yuansiwei_i.t_school(全国名录,仅取被引用的 23 所)。
  79. type LegacySchool struct {
  80. ID int32 `gorm:"column:id;primaryKey"`
  81. Name string `gorm:"column:name"`
  82. Status int8 `gorm:"column:status"` // tinyint
  83. }
  84. // TableName 指定 legacy 表名。
  85. func (LegacySchool) TableName() string { return "t_school" }
  86. // LegacySchoolCampus 教学侧 yuansiwei_i.t_school_campus(校区,63 行)。
  87. type LegacySchoolCampus struct {
  88. ID int32 `gorm:"column:id;primaryKey"`
  89. Name string `gorm:"column:name"`
  90. SchoolID int32 `gorm:"column:school_id"`
  91. Status int8 `gorm:"column:status"` // tinyint
  92. Code *string `gorm:"column:code"`
  93. }
  94. // TableName 指定 legacy 表名。
  95. func (LegacySchoolCampus) TableName() string { return "t_school_campus" }
  96. // LegacyUserSchoolCampus 教务侧 finance_bk.t_user_school_campus(用户-机构-校区绑定,26308 行)。
  97. type LegacyUserSchoolCampus struct {
  98. ID int32 `gorm:"column:id;primaryKey"`
  99. UserID int32 `gorm:"column:userid"`
  100. SchoolID int32 `gorm:"column:schoolid"`
  101. CampusID int32 `gorm:"column:campusid"`
  102. }
  103. // TableName 指定 legacy 表名(finance_bk 库)。
  104. func (LegacyUserSchoolCampus) TableName() string { return "t_user_school_campus" }
  105. // ---- 数据源接口 ----
  106. // Source 封装 legacy 双库的只读 GORM 连接(教学 + 教务)。
  107. //
  108. // 两个 *gorm.DB 分别指向 yuansiwei_i 和 finance_bk(同一 MySQL 实例,不同 database);
  109. // 若 DSN 未指定 database,用 UseDB 切换。
  110. type Source struct {
  111. teaching *gorm.DB // yuansiwei_i(t_user/t_weixin_user/t_user_account_association/t_school/t_school_campus)
  112. finance *gorm.DB // finance_bk(t_user_school_campus)
  113. }
  114. // NewSource 创建 legacy 数据源(两个 DSN 指向同一 MySQL 实例的不同库)。
  115. //
  116. // 入参:
  117. // - teachingDsn string:教学库 DSN(如 "dev:pass@tcp(127.0.0.1:3306)/yuansiwei_i?charset=utf8mb4&parseTime=True&loc=Local")。
  118. // - financeDsn string:教务库 DSN(如 "dev:pass@tcp(127.0.0.1:3306)/finance_bk?charset=utf8mb4&parseTime=True&loc=Local")。
  119. //
  120. // 出参:
  121. // - *Source:就绪的只读数据源。
  122. // - error:连接失败时返回。
  123. func NewSource(teachingDsn, financeDsn string) (*Source, error) {
  124. var missing []string
  125. if teachingDsn == "" {
  126. missing = append(missing, "teachingDsn")
  127. }
  128. if financeDsn == "" {
  129. missing = append(missing, "financeDsn")
  130. }
  131. if len(missing) > 0 {
  132. return nil, fmt.Errorf("migrate: legacy DSN 不能为空: %v", missing)
  133. }
  134. teaching, err := gorm.Open(mysql.Open(teachingDsn), gormMigrateCfg())
  135. if err != nil {
  136. return nil, fmt.Errorf("migrate: 教学库连接失败: %w", err)
  137. }
  138. if sqlDB, err := teaching.DB(); err == nil {
  139. if err := sqlDB.Ping(); err != nil {
  140. _ = sqlDB.Close()
  141. return nil, fmt.Errorf("migrate: 教学库 Ping 失败: %w", err)
  142. }
  143. }
  144. finance, err := gorm.Open(mysql.Open(financeDsn), gormMigrateCfg())
  145. if err != nil {
  146. return nil, fmt.Errorf("migrate: 教务库连接失败: %w", err)
  147. }
  148. if sqlDB, err := finance.DB(); err == nil {
  149. if err := sqlDB.Ping(); err != nil {
  150. _ = sqlDB.Close()
  151. return nil, fmt.Errorf("migrate: 教务库 Ping 失败: %w", err)
  152. }
  153. }
  154. return &Source{teaching: teaching, finance: finance}, nil
  155. }
  156. // Close 关闭两个连接池。
  157. func (s *Source) Close() error {
  158. var firstErr error
  159. if s.teaching != nil {
  160. if sqlDB, err := s.teaching.DB(); err == nil {
  161. _ = sqlDB.Close()
  162. }
  163. }
  164. if s.finance != nil {
  165. if sqlDB, err := s.finance.DB(); err == nil {
  166. _ = sqlDB.Close()
  167. } else {
  168. firstErr = err
  169. }
  170. }
  171. return firstErr
  172. }
  173. // ---- 读取方法(全部 SELECT,无写操作)----
  174. // LoadAllUsers 读取 t_user 全量(68107 行,按 id 排序保证幂等处理顺序)。
  175. func (s *Source) LoadAllUsers(ctx context.Context) ([]LegacyUser, error) {
  176. var users []LegacyUser
  177. err := s.teaching.WithContext(ctx).Model(&LegacyUser{}).Order("id").Find(&users).Error
  178. return users, err
  179. }
  180. // LoadAllWeixinUsers 读取 t_weixin_user 全量(3491 行)。
  181. func (s *Source) LoadAllWeixinUsers(ctx context.Context) ([]LegacyWeixinUser, error) {
  182. var rows []LegacyWeixinUser
  183. err := s.teaching.WithContext(ctx).Model(&LegacyWeixinUser{}).Order("openid").Find(&rows).Error
  184. return rows, err
  185. }
  186. // LoadAllAssociations 读取 t_user_account_association 全量(16077 行,现映射种子)。
  187. func (s *Source) LoadAllAssociations(ctx context.Context) ([]LegacyAccountAssociation, error) {
  188. var rows []LegacyAccountAssociation
  189. err := s.teaching.WithContext(ctx).Model(&LegacyAccountAssociation{}).Order("userid").Find(&rows).Error
  190. return rows, err
  191. }
  192. // LoadSchools 读取 t_school 中被 t_user_school_campus 引用的学校(23 所)。
  193. func (s *Source) LoadSchools(ctx context.Context, schoolIDs []int32) ([]LegacySchool, error) {
  194. if len(schoolIDs) == 0 {
  195. return nil, nil
  196. }
  197. var rows []LegacySchool
  198. err := s.teaching.WithContext(ctx).Model(&LegacySchool{}).Where("id IN ?", schoolIDs).Find(&rows).Error
  199. return rows, err
  200. }
  201. // LoadCampuses 读取 t_school_campus 全量(63 行)。
  202. func (s *Source) LoadCampuses(ctx context.Context) ([]LegacySchoolCampus, error) {
  203. var rows []LegacySchoolCampus
  204. err := s.teaching.WithContext(ctx).Model(&LegacySchoolCampus{}).Order("id").Find(&rows).Error
  205. return rows, err
  206. }
  207. // LoadUserSchoolCampus 读取 finance_bk.t_user_school_campus 全量(26308 行)。
  208. func (s *Source) LoadUserSchoolCampus(ctx context.Context) ([]LegacyUserSchoolCampus, error) {
  209. var rows []LegacyUserSchoolCampus
  210. err := s.finance.WithContext(ctx).Model(&LegacyUserSchoolCampus{}).Order("id").Find(&rows).Error
  211. return rows, err
  212. }
  213. // LoadDistinctSchoolIDs 从 t_user_school_campus 提取去重 schoolid(23 所)。
  214. func (s *Source) LoadDistinctSchoolIDs(ctx context.Context) ([]int32, error) {
  215. var ids []int32
  216. err := s.finance.WithContext(ctx).Model(&LegacyUserSchoolCampus{}).
  217. Distinct().Pluck("schoolid", &ids).Error
  218. return ids, err
  219. }