batch3_run.go 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326
  1. package fixture
  2. import (
  3. "context"
  4. "database/sql"
  5. "fmt"
  6. "io"
  7. "sort"
  8. )
  9. // 本文件是第三批的流程编排与报告排版。
  10. // OrgReport 是一次「搬 org 域」的结果。
  11. type OrgReport struct {
  12. Options OrgOptions
  13. Schools int
  14. Campuses int
  15. Warnings []string
  16. InstIDs []int64
  17. CampIDs []int64
  18. }
  19. // PrepareOrg 把选中的机构与其校区搬进目标库(D1 原值插入)。
  20. func PrepareOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgReport, error) {
  21. src, err := cfg.OpenOrgSource()
  22. if err != nil {
  23. return nil, err
  24. }
  25. defer src.Close()
  26. target, err := cfg.OpenTarget()
  27. if err != nil {
  28. return nil, err
  29. }
  30. defer target.Close()
  31. rep := &OrgReport{Options: opts}
  32. schools, err := extractSchools(ctx, src, opts.InstitutionIDs)
  33. if err != nil {
  34. return nil, err
  35. }
  36. if len(schools) == 0 {
  37. return nil, fmt.Errorf("t_school 里找不到任何一个指定机构 id(%v)", opts.InstitutionIDs)
  38. }
  39. // 请求了但库里没有的 id 必须显式报出来(不允许"少迁一个却当成功")
  40. found := map[int64]bool{}
  41. for _, s := range schools {
  42. found[s.ID] = true
  43. }
  44. var missing []int64
  45. for _, id := range opts.InstitutionIDs {
  46. if !found[id] {
  47. missing = append(missing, id)
  48. }
  49. }
  50. if len(missing) > 0 {
  51. return nil, fmt.Errorf("指定的机构 id 在 t_school 里不存在:%v(拒绝静默跳过)", missing)
  52. }
  53. campuses, err := extractCampuses(ctx, src, opts.InstitutionIDs)
  54. if err != nil {
  55. return nil, err
  56. }
  57. if !opts.WithAllCampuses {
  58. legacy, err := cfg.OpenLegacy()
  59. if err != nil {
  60. return nil, err
  61. }
  62. refd, err := extractReferencedCampuses(ctx, legacy)
  63. legacy.Close()
  64. if err != nil {
  65. return nil, err
  66. }
  67. var kept []LegacyCampus
  68. for _, c := range campuses {
  69. if refd[c.ID] {
  70. kept = append(kept, c)
  71. }
  72. }
  73. campuses = kept
  74. }
  75. // 机构
  76. for _, s := range schools {
  77. ti, w, err := MapInstitution(s)
  78. if err != nil {
  79. return nil, err
  80. }
  81. rep.Warnings = append(rep.Warnings, w...)
  82. if err := upsertInstitution(ctx, target, ti); err != nil {
  83. return nil, err
  84. }
  85. rep.InstIDs = append(rep.InstIDs, ti.InstitutionID)
  86. }
  87. rep.Schools = len(schools)
  88. // 校区
  89. for _, c := range campuses {
  90. tc, w, err := MapCampus(c)
  91. if err != nil {
  92. return nil, err
  93. }
  94. rep.Warnings = append(rep.Warnings, w...)
  95. if err := upsertCampus(ctx, target, tc); err != nil {
  96. return nil, err
  97. }
  98. rep.CampIDs = append(rep.CampIDs, tc.CampusID)
  99. }
  100. rep.Campuses = len(campuses)
  101. sort.Slice(rep.CampIDs, func(i, j int) bool { return rep.CampIDs[i] < rep.CampIDs[j] })
  102. return rep, nil
  103. }
  104. // OrgIdemResult 是 org 域「跑两次」的比对结论。
  105. type OrgIdemResult struct {
  106. InstBefore, InstAfter map[int64]string
  107. CampBefore, CampAfter map[int64]string
  108. }
  109. // AllOK 表示两张表都完全幂等(行数与 id 集合不变)。
  110. func (r *OrgIdemResult) AllOK() bool {
  111. return sameIDSet(r.InstBefore, r.InstAfter) && sameIDSet(r.CampBefore, r.CampAfter)
  112. }
  113. func sameIDSet(a, b map[int64]string) bool {
  114. if len(a) != len(b) {
  115. return false
  116. }
  117. for k := range a {
  118. if _, ok := b[k]; !ok {
  119. return false
  120. }
  121. }
  122. return true
  123. }
  124. // IdempotencyCheckOrg 先拍快照 → 再跑一次完全相同的映射 → 比对。
  125. func IdempotencyCheckOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgIdemResult, error) {
  126. ib, cb, err := SnapshotOrg(ctx, cfg)
  127. if err != nil {
  128. return nil, err
  129. }
  130. if _, err := PrepareOrg(ctx, cfg, opts); err != nil {
  131. return nil, fmt.Errorf("第二次映射失败(幂等被破坏):%w", err)
  132. }
  133. ia, ca, err := SnapshotOrg(ctx, cfg)
  134. if err != nil {
  135. return nil, err
  136. }
  137. return &OrgIdemResult{InstBefore: ib, InstAfter: ia, CampBefore: cb, CampAfter: ca}, nil
  138. }
  139. // OrgCleanReport 是清理结果。
  140. type OrgCleanReport struct {
  141. DeletedInst int64
  142. DeletedCamp int64
  143. Remaining map[string]int64
  144. }
  145. // CleanupOrg 删除第三批造的数据(按 D1 id 精确删)。
  146. func CleanupOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgCleanReport, error) {
  147. src, err := cfg.OpenOrgSource()
  148. if err != nil {
  149. return nil, err
  150. }
  151. defer src.Close()
  152. target, err := cfg.OpenTarget()
  153. if err != nil {
  154. return nil, err
  155. }
  156. defer target.Close()
  157. schools, err := extractSchools(ctx, src, opts.InstitutionIDs)
  158. if err != nil {
  159. return nil, err
  160. }
  161. campuses, err := extractCampuses(ctx, src, opts.InstitutionIDs)
  162. if err != nil {
  163. return nil, err
  164. }
  165. rep := &OrgCleanReport{Remaining: map[string]int64{}}
  166. // 先删校区(引用方),再删机构
  167. for _, c := range campuses {
  168. res, err := target.ExecContext(ctx, `DELETE FROM t_campus WHERE campus_id = ?`, c.ID)
  169. if err != nil {
  170. return nil, err
  171. }
  172. n, _ := res.RowsAffected()
  173. rep.DeletedCamp += n
  174. }
  175. for _, s := range schools {
  176. res, err := target.ExecContext(ctx, `DELETE FROM t_institution WHERE institution_id = ?`, s.ID)
  177. if err != nil {
  178. return nil, err
  179. }
  180. n, _ := res.RowsAffected()
  181. rep.DeletedInst += n
  182. }
  183. for _, t := range []string{"t_institution", "t_campus"} {
  184. var n int64
  185. if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil {
  186. return nil, err
  187. }
  188. rep.Remaining[t] = n
  189. }
  190. return rep, nil
  191. }
  192. // CountOrgTables 统计 org 两张表行数。
  193. func CountOrgTables(ctx context.Context, cfg Config) (map[string]int64, error) {
  194. db, err := cfg.OpenTarget()
  195. if err != nil {
  196. return nil, err
  197. }
  198. defer db.Close()
  199. inst, camp, err := countTargetOrg(ctx, db)
  200. if err != nil {
  201. return nil, err
  202. }
  203. return map[string]int64{"t_institution": inst, "t_campus": camp}, nil
  204. }
  205. // PrintOrgReport 打印搬数据结果。
  206. func PrintOrgReport(w io.Writer, r *OrgReport) {
  207. fmt.Fprintf(w, "== 第三批 org 域映射 ==\n")
  208. fmt.Fprintf(w, "机构 id 集合:%v(%d 个)\n", r.Options.InstitutionIDs, len(r.Options.InstitutionIDs))
  209. fmt.Fprintf(w, "迁移校区策略:WithAllCampuses=%v\n", r.Options.WithAllCampuses)
  210. fmt.Fprintf(w, "写入:t_institution=%d 行,t_campus=%d 行\n", r.Schools, r.Campuses)
  211. fmt.Fprintf(w, "institution_id:%v\n", r.InstIDs)
  212. fmt.Fprintf(w, "campus_id:%v\n", r.CampIDs)
  213. if len(r.Warnings) > 0 {
  214. fmt.Fprintf(w, "映射告警(%d):\n", len(r.Warnings))
  215. for _, x := range r.Warnings {
  216. fmt.Fprintf(w, " - %s\n", x)
  217. }
  218. } else {
  219. fmt.Fprintf(w, "映射告警:无\n")
  220. }
  221. }
  222. // PrintOrgVerify 打印核对结果。
  223. func PrintOrgVerify(w io.Writer, r *OrgVerifyReport) {
  224. fmt.Fprintf(w, "== 第三批逐列核对(列 × 行 聚合)==\n")
  225. last := ""
  226. for _, c := range r.Cols {
  227. if c.Table != last {
  228. cnt := r.TargetInst
  229. if c.Table == "t_campus" {
  230. cnt = r.TargetCamp
  231. }
  232. fmt.Fprintf(w, "\n-- %s(target %d 行)--\n", c.Table, cnt)
  233. last = c.Table
  234. }
  235. st := markOK
  236. if !c.OK() {
  237. st = markFail
  238. }
  239. fmt.Fprintf(w, " %s %-18s %d/%d 相等 %s\n", st, c.Column, c.Equal, c.Total, c.Note)
  240. for _, s := range c.Samples {
  241. fmt.Fprintf(w, " ✘ %s\n", s)
  242. }
  243. }
  244. fmt.Fprintf(w, "\n== 引用完整性(逐个来源)==\n")
  245. fmt.Fprintf(w, " %-28s %-8s %-10s %-10s %-10s %s\n", "来源", "行数", "机构数", "校区数", "孤儿机构", "孤儿校区")
  246. for _, s := range r.RefStats {
  247. st := markOK
  248. if s.OrphanInst > 0 || s.OrphanCampus > 0 {
  249. st = markFail
  250. }
  251. fmt.Fprintf(w, " %s %-26s %-8d %-10d %-10d %-10d %d\n",
  252. st, s.Source, s.Rows, s.DistinctInst, s.DistinctCampus, s.OrphanInst, s.OrphanCampus)
  253. if len(s.OrphanInstList) > 0 || len(s.OrphanCampusList) > 0 {
  254. fmt.Fprintf(w, " 孤儿机构 id:%v 孤儿校区 id:%v\n", s.OrphanInstList, s.OrphanCampusList)
  255. }
  256. }
  257. fmt.Fprintf(w, " (campus_id=0 表示机构级、不建行,已按来源单独统计,不计入孤儿)\n")
  258. for _, s := range r.RefStats {
  259. if s.CampusZeroRows > 0 {
  260. fmt.Fprintf(w, " %-26s campus_id=0 的行数:%d\n", s.Source, s.CampusZeroRows)
  261. }
  262. }
  263. if len(r.CrossOrg) > 0 {
  264. fmt.Fprintf(w, "\n== R17 标注:cross-org-campus(只标注,不修正)==\n")
  265. for _, c := range r.CrossOrg {
  266. fmt.Fprintf(w, " · %s\n", c)
  267. }
  268. }
  269. fmt.Fprintf(w, "\n== 语义断言 ==\n")
  270. for _, a := range r.Assertions {
  271. st := markOK
  272. if !a.Pass {
  273. st = markFail
  274. }
  275. fmt.Fprintf(w, " %s %s\n %s\n", st, a.Name, a.Detail)
  276. }
  277. fmt.Fprintf(w, "\n核对结论:%s(不一致列 + 失败断言 = %d)\n", verdict(r.OK()), r.FailCount())
  278. }
  279. // PrintOrgIdem 打印幂等结论。
  280. func PrintOrgIdem(w io.Writer, r *OrgIdemResult) {
  281. fmt.Fprintf(w, "== 幂等实测(同一批跑两次)==\n")
  282. st := markOK
  283. if !r.AllOK() {
  284. st = markFail
  285. }
  286. fmt.Fprintf(w, " %s t_institution:%d 行 → %d 行\n", st, len(r.InstBefore), len(r.InstAfter))
  287. fmt.Fprintf(w, " %s t_campus:%d 行 → %d 行\n", st, len(r.CampBefore), len(r.CampAfter))
  288. fmt.Fprintf(w, " id 集合不变:institution=%v campus=%v\n",
  289. sameIDSet(r.InstBefore, r.InstAfter), sameIDSet(r.CampBefore, r.CampAfter))
  290. fmt.Fprintf(w, "\n幂等结论:%s\n", verdict(r.AllOK()))
  291. }
  292. // PrintOrgClean 打印清理结果。
  293. func PrintOrgClean(w io.Writer, r *OrgCleanReport) {
  294. fmt.Fprintf(w, "== 第三批清理 ==\n")
  295. fmt.Fprintf(w, " t_campus 删除 %d 行,剩余全表 %d 行\n", r.DeletedCamp, r.Remaining["t_campus"])
  296. fmt.Fprintf(w, " t_institution 删除 %d 行,剩余全表 %d 行\n", r.DeletedInst, r.Remaining["t_institution"])
  297. ok := r.Remaining["t_campus"] == 0 && r.Remaining["t_institution"] == 0
  298. fmt.Fprintf(w, "两表回 0 行:%s\n", verdict(ok))
  299. }
  300. var _ = sql.ErrNoRows