batch2_extract.go 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240
  1. package fixture
  2. import (
  3. "context"
  4. "database/sql"
  5. "fmt"
  6. )
  7. // 本文件是第二批的 legacy **只读**查询。金额列一律 `CAST(... AS DECIMAL(20,3))`,
  8. // 浮点误差在读取源头就被掐掉(与上一批同一口径)。
  9. //
  10. // fixture 取样范围(可由 CLI 覆盖):
  11. // 分账树:root_userid = 395 的整棵子树(8 个节点:1 根 + 7 子)
  12. // 分账明细:userid = 395 的 4 条
  13. // 充值工单:userid = 31348 的 8 条(含 response_userid=0、3 位小数金额、NULL 时间)
  14. // 收支记录:create_userid = 395 的 10 条(含 3 位小数 cost、字典引用 15/23)
  15. // 收支字典:全表 45 条(解析 expenditure_id 用)
  16. // Batch2Options 是第二批 fixture 的取样参数。
  17. type Batch2Options struct {
  18. DivideRootUserID int64 // 分账树根(取整棵子树)
  19. BalanceRequestUserID int64 // 充值工单所属用户
  20. ExpenditureLogUserID int64 // 收支记录创建人
  21. WithExpenditureDict bool // 是否同时 D1 迁移 t_expenditure 字典(解析 expenditure_id 必需)
  22. }
  23. // DefaultBatch2Options 是默认取样(选择依据见 README §7)。
  24. func DefaultBatch2Options() Batch2Options {
  25. return Batch2Options{
  26. DivideRootUserID: 395,
  27. BalanceRequestUserID: 31348,
  28. ExpenditureLogUserID: 395,
  29. WithExpenditureDict: true,
  30. }
  31. }
  32. func extractDivides(ctx context.Context, db *sql.DB, rootUserID int64) ([]LegacyDivideInto, error) {
  33. const q = `
  34. SELECT userid, cast(ifnull(type,1) as signed),
  35. CAST(ifnull(quota,0) AS DECIMAL(20,3)),
  36. CAST(ifnull(total_income,0) AS DECIMAL(20,3)),
  37. CAST(ifnull(total_income_payed,0) AS DECIMAL(20,3)),
  38. root_userid, parent_userid,
  39. cast(ifnull(left_val,1) as signed), cast(ifnull(right_val,2) as signed),
  40. cast(ifnull(layer,1) as signed)
  41. FROM t_user_divide_into
  42. WHERE root_userid = ?
  43. ORDER BY left_val, userid`
  44. rows, err := db.QueryContext(ctx, q, rootUserID)
  45. if err != nil {
  46. return nil, err
  47. }
  48. defer rows.Close()
  49. var out []LegacyDivideInto
  50. for rows.Next() {
  51. var r LegacyDivideInto
  52. if err := rows.Scan(&r.UserID, &r.Type, &r.Quota, &r.TotalIncome, &r.TotalIncomePayed,
  53. &r.RootUserID, &r.ParentUserID, &r.LeftVal, &r.RightVal, &r.Layer); err != nil {
  54. return nil, err
  55. }
  56. out = append(out, r)
  57. }
  58. return out, rows.Err()
  59. }
  60. func extractDivideLogs(ctx context.Context, db *sql.DB, userID int64) ([]LegacyDivideIntoLog, error) {
  61. const q = `
  62. SELECT id, cast(type as signed), userid, studentid, orderid,
  63. CAST(ifnull(money,0) AS DECIMAL(20,3)),
  64. create_time, remark
  65. FROM t_user_divide_into_log
  66. WHERE userid = ?
  67. ORDER BY id`
  68. rows, err := db.QueryContext(ctx, q, userID)
  69. if err != nil {
  70. return nil, err
  71. }
  72. defer rows.Close()
  73. var out []LegacyDivideIntoLog
  74. for rows.Next() {
  75. var r LegacyDivideIntoLog
  76. if err := rows.Scan(&r.ID, &r.Type, &r.UserID, &r.StudentID, &r.OrderNo,
  77. &r.Money, &r.CreateTime, &r.Remark); err != nil {
  78. return nil, err
  79. }
  80. out = append(out, r)
  81. }
  82. return out, rows.Err()
  83. }
  84. func extractBalanceRequests(ctx context.Context, db *sql.DB, userID int64) ([]LegacyBalanceRequest, error) {
  85. const q = `
  86. SELECT id, type, userid, schoolid, campusid,
  87. CAST(money AS DECIMAL(20,3)),
  88. transfer_accounts_type, transfer_accounts_time, orderid,
  89. request_userid, request_time, response_userid, response_time,
  90. response_msg, aproved, item, imgs, remark
  91. FROM t_user_balance_request
  92. WHERE userid = ?
  93. ORDER BY request_time, id`
  94. rows, err := db.QueryContext(ctx, q, userID)
  95. if err != nil {
  96. return nil, err
  97. }
  98. defer rows.Close()
  99. var out []LegacyBalanceRequest
  100. for rows.Next() {
  101. var r LegacyBalanceRequest
  102. if err := rows.Scan(&r.BizNo, &r.Type, &r.UserID, &r.SchoolID, &r.CampusID, &r.Money,
  103. &r.TransferAccountsType, &r.TransferAccountsTime, &r.OrderNo,
  104. &r.RequestUserID, &r.RequestTime, &r.ResponseUserID, &r.ResponseTime,
  105. &r.ResponseMsg, &r.Aproved, &r.Item, &r.Imgs, &r.Remark); err != nil {
  106. return nil, err
  107. }
  108. out = append(out, r)
  109. }
  110. return out, rows.Err()
  111. }
  112. func extractExpenditureLogs(ctx context.Context, db *sql.DB, createUserID int64) ([]LegacyExpenditureLog, error) {
  113. const q = `
  114. SELECT id, schoolid, campusid, expenditureid,
  115. CAST(cost AS DECIMAL(20,3)),
  116. create_userid, create_time, year, month, remark, imgs
  117. FROM t_expenditure_log
  118. WHERE create_userid = ?
  119. ORDER BY id`
  120. rows, err := db.QueryContext(ctx, q, createUserID)
  121. if err != nil {
  122. return nil, err
  123. }
  124. defer rows.Close()
  125. var out []LegacyExpenditureLog
  126. for rows.Next() {
  127. var r LegacyExpenditureLog
  128. if err := rows.Scan(&r.BizNo, &r.SchoolID, &r.CampusID, &r.ExpenditureID, &r.Cost,
  129. &r.CreateUserID, &r.CreateTime, &r.Year, &r.Month, &r.Remark, &r.Imgs); err != nil {
  130. return nil, err
  131. }
  132. out = append(out, r)
  133. }
  134. return out, rows.Err()
  135. }
  136. func extractExpenditures(ctx context.Context, db *sql.DB) ([]LegacyExpenditure, error) {
  137. const q = `SELECT id, type, parentid, name, remark FROM t_expenditure ORDER BY id`
  138. rows, err := db.QueryContext(ctx, q)
  139. if err != nil {
  140. return nil, err
  141. }
  142. defer rows.Close()
  143. var out []LegacyExpenditure
  144. for rows.Next() {
  145. var r LegacyExpenditure
  146. if err := rows.Scan(&r.ID, &r.Type, &r.ParentID, &r.Name, &r.Remark); err != nil {
  147. return nil, err
  148. }
  149. out = append(out, r)
  150. }
  151. return out, rows.Err()
  152. }
  153. // collectUserRefs 汇总这批数据里出现的**全部 legacy 用户引用**,
  154. // 用于一次性补 t_legacy_mapping / t_user(否则引用列会全退化成 NULL)。
  155. func collectUserRefs(divides []LegacyDivideInto, dlog []LegacyDivideIntoLog,
  156. brs []LegacyBalanceRequest, elogs []LegacyExpenditureLog) []int64 {
  157. set := map[int64]bool{}
  158. add := func(v int64) {
  159. if v > 0 {
  160. set[v] = true
  161. }
  162. }
  163. for _, d := range divides {
  164. add(d.UserID)
  165. if d.RootUserID.Valid {
  166. add(d.RootUserID.Int64)
  167. }
  168. if d.ParentUserID.Valid {
  169. add(d.ParentUserID.Int64)
  170. }
  171. }
  172. for _, l := range dlog {
  173. add(l.UserID)
  174. add(l.StudentID)
  175. }
  176. for _, r := range brs {
  177. add(r.UserID)
  178. add(r.RequestUserID)
  179. if r.ResponseUserID.Valid {
  180. add(r.ResponseUserID.Int64)
  181. }
  182. }
  183. for _, e := range elogs {
  184. if e.CreateUserID.Valid {
  185. add(e.CreateUserID.Int64)
  186. }
  187. }
  188. out := make([]int64, 0, len(set))
  189. for v := range set {
  190. out = append(out, v)
  191. }
  192. sortInt64s(out)
  193. return out
  194. }
  195. func sortInt64s(v []int64) {
  196. for i := 1; i < len(v); i++ {
  197. for j := i; j > 0 && v[j] < v[j-1]; j-- {
  198. v[j], v[j-1] = v[j-1], v[j]
  199. }
  200. }
  201. }
  202. // countBatch2Legacy 统计这批在 legacy 侧的行数。
  203. func countBatch2Legacy(ctx context.Context, db *sql.DB, opts Batch2Options) (map[string]int, error) {
  204. out := map[string]int{}
  205. type q1 struct {
  206. name string
  207. sql string
  208. arg int64
  209. }
  210. for _, x := range []q1{
  211. {"t_user_divide_into", `SELECT COUNT(*) FROM t_user_divide_into WHERE root_userid = ?`, opts.DivideRootUserID},
  212. {"t_user_divide_into_log", `SELECT COUNT(*) FROM t_user_divide_into_log WHERE userid = ?`, opts.DivideRootUserID},
  213. {"t_user_balance_request", `SELECT COUNT(*) FROM t_user_balance_request WHERE userid = ?`, opts.BalanceRequestUserID},
  214. {"t_expenditure_log", `SELECT COUNT(*) FROM t_expenditure_log WHERE create_userid = ?`, opts.ExpenditureLogUserID},
  215. } {
  216. var n int
  217. if err := db.QueryRowContext(ctx, x.sql, x.arg).Scan(&n); err != nil {
  218. return nil, fmt.Errorf("%s:%w", x.name, err)
  219. }
  220. out[x.name] = n
  221. }
  222. var n int
  223. if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM t_expenditure`).Scan(&n); err != nil {
  224. return nil, err
  225. }
  226. out["t_expenditure"] = n
  227. return out, nil
  228. }