diff --git a/internal/data/business/business_portrait.go b/internal/data/business/business_portrait.go index 82d9062..b3f8f8d 100644 --- a/internal/data/business/business_portrait.go +++ b/internal/data/business/business_portrait.go @@ -5,6 +5,7 @@ import ( "errors" "strconv" + "golang.org/x/sync/errgroup" "gorm.io/gorm" "kra/internal/biz/business" ) @@ -161,15 +162,13 @@ func (r *businessPortraitRepo) Statistics(ctx context.Context, scope business.Po if scope.Range.Hourly { tables = append(tables, "tab_user", "tab_spend") } - for _, table := range tables { - available, err := businessTableAvailable(db, table) - if err != nil { - return nil, err - } - if !available { - result.Available = false - return result, nil - } + available, err := businessTablesAvailable(db, tables...) + if err != nil { + return nil, err + } + if !available { + result.Available = false + return result, nil } mode := 0 if scope.IDCardDedup { @@ -177,49 +176,77 @@ func (r *businessPortraitRepo) Statistics(ctx context.Context, scope business.Po } begin, end := scope.Range.Begin.Format("2006-01-02"), scope.Range.End.Format("2006-01-02") previousBegin, previousEnd := scope.Range.PreviousBegin.Format("2006-01-02"), scope.Range.PreviousEnd.Format("2006-01-02") - periods := []struct { - begin, end string - counts *business.PortraitCounts - pay *string - }{{begin, end, &result.Current, &result.CurrentPay}} - if !scope.NoCompare { - periods = append(periods, struct { - begin, end string - counts *business.PortraitCounts - pay *string - }{previousBegin, previousEnd, &result.Previous, &result.PreviousPay}) + // 合并本期和对比期的扫描;本月同期之间的日期空隙不能混入统计。 + group, queryCtx := errgroup.WithContext(ctx) + group.SetLimit(2) + queryDB := db.WithContext(queryCtx) + summaryQuery := func(table string, dedup int) *gorm.DB { + if scope.NoCompare { + return portraitReportQuery(queryDB, scope, table, begin, end, dedup) + } + return portraitReportQuery(queryDB, scope, table, min(begin, previousBegin), max(end, previousEnd), dedup). + Where("(r.stat_date >= ? AND r.stat_date <= ?) OR (r.stat_date >= ? AND r.stat_date <= ?)", begin, end, previousBegin, previousEnd) } - for _, period := range periods { - var rows []portraitCountPO - q := portraitReportQuery(db, scope, "tab_sw_report_day_key", period.begin, period.end, mode) - if err := q.Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&rows).Error; err != nil { - return nil, business.ErrStorage + group.Go(func() error { + var rows []struct { + Metric int + Current, Previous int64 } - *period.counts = toPortraitCounts(rows) - if err := portraitReportQuery(db, scope, "tab_sw_report_day", period.begin, period.end, 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(period.pay).Error; err != nil { - return nil, business.ErrStorage + identity := portraitIdentity(queryDB, scope.IDCardDedup) + selectSQL := "r.metric,COUNT(DISTINCT CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN " + identity + " END) AS current" + args := []any{begin, end} + if !scope.NoCompare { + selectSQL += ",COUNT(DISTINCT CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN " + identity + " END) AS previous" + args = append(args, previousBegin, previousEnd) } - } - if scope.Range.Hourly { - if err := portraitHourly(db, scope, result); err != nil { - return nil, err + if err := summaryQuery("tab_sw_report_day_key", mode).Select(selectSQL, args...).Group("r.metric").Scan(&rows).Error; err != nil { + return business.ErrStorage } - } else { - if err := portraitDaily(db, scope, mode, begin, end, result); err != nil { - return nil, err + current, previous := make([]portraitCountPO, 0, len(rows)), make([]portraitCountPO, 0, len(rows)) + for _, row := range rows { + current = append(current, portraitCountPO{Metric: row.Metric, Count: row.Current}) + previous = append(previous, portraitCountPO{Metric: row.Metric, Count: row.Previous}) } - } + result.Current, result.Previous = toPortraitCounts(current), toPortraitCounts(previous) + return nil + }) + group.Go(func() error { + var amounts struct{ Current, Previous string } + selectSQL := "COALESCE(SUM(CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN r.total_pay ELSE 0 END),0) AS current" + args := []any{begin, end} + if !scope.NoCompare { + selectSQL += ",COALESCE(SUM(CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN r.total_pay ELSE 0 END),0) AS previous" + args = append(args, previousBegin, previousEnd) + } + if err := summaryQuery("tab_sw_report_day", 0).Select(selectSQL, args...).Scan(&amounts).Error; err != nil { + return business.ErrStorage + } + result.CurrentPay = amounts.Current + if !scope.NoCompare { + result.PreviousPay = amounts.Previous + } + return nil + }) + group.Go(func() error { + if scope.Range.Hourly { + return portraitHourly(queryDB, scope, result) + } + return portraitDaily(queryDB, scope, mode, begin, end, result) + }) if scope.RankLimit > 0 { - register, err := portraitRankQuery(db, scope, mode, begin, end, false) - if err != nil { - return nil, err - } - result.RegisterRanks = register - pay, err := portraitRankQuery(db, scope, 0, begin, end, true) - if err != nil { - return nil, err - } - result.PayRanks = pay + group.Go(func() error { + register, err := portraitRankQuery(queryDB, scope, mode, begin, end, false) + result.RegisterRanks = register + return err + }) + group.Go(func() error { + pay, err := portraitRankQuery(queryDB, scope, 0, begin, end, true) + result.PayRanks = pay + return err + }) + } + if err := group.Wait(); err != nil { + return nil, err } return result, nil } @@ -247,11 +274,17 @@ func portraitDaily(db *gorm.DB, scope business.PortraitReportScope, mode int, be } func portraitHourly(db *gorm.DB, scope business.PortraitReportScope, result *business.PortraitStatistics) error { start, end := scope.Range.Begin.Unix(), scope.Range.End.AddDate(0, 0, 1).Unix() + // register_time 是历史字符串列;数值比较不能利用其字符串索引。 + // 复用已有数值生成列和时间范围索引,未迁移的库保留 CAST 回退。 + registerTime, err := registrationTimeExpression(db) + if err != nil { + return err + } for _, metric := range []struct { table, column, aggregate, condition string series map[string]string - }{{"tab_user", "register_time", "COUNT(*)", "(r.puid = 0 OR r.puid IS NULL)", result.RegisterSeries}, {"tab_spend", "pay_time", "COALESCE(SUM(r.pay_amount),0)", "r.pay_status = 1", result.PaySeries}} { - q := db.Table(metric.table+" r").Where("r."+metric.column+" >= ? AND r."+metric.column+" < ?", start, end).Where(metric.condition).Where("(r.source_db,r.promote_id) IN (?)", portraitPromoteScope(db, scope)) + }{{"tab_user", registerTime, "COUNT(*)", "(r.puid = 0 OR r.puid IS NULL)", result.RegisterSeries}, {"tab_spend", "r.pay_time", "COALESCE(SUM(r.pay_amount),0)", "r.pay_status = 1", result.PaySeries}} { + q := db.Table(metric.table+" r").Where(metric.column+" >= ? AND "+metric.column+" < ?", start, end).Where(metric.condition).Where("(r.source_db,r.promote_id) IN (?)", portraitPromoteScope(db, scope)) if len(scope.Sources) > 0 { q = q.Where("r.source_db IN ?", scope.Sources) } @@ -261,12 +294,12 @@ func portraitHourly(db *gorm.DB, scope business.PortraitReportScope, result *bus } if db.Dialector.Name() == "mysql" { // 保留原 Mapper 的数据库会话时区口径,最终实库验收同时核对时区配置。 - hour := "HOUR(FROM_UNIXTIME(r." + metric.column + "))" + hour := "HOUR(FROM_UNIXTIME(" + metric.column + "))" if err := q.Select(hour + " AS hour," + metric.aggregate + " AS value").Group(hour).Scan(&rows).Error; err != nil { return business.ErrStorage } } else { - if err := q.Select("CAST((r."+metric.column+" - ?) / 3600 AS INTEGER) AS hour,"+metric.aggregate+" AS value", start).Group("hour").Scan(&rows).Error; err != nil { + if err := q.Select("CAST(("+metric.column+" - ?) / 3600 AS INTEGER) AS hour,"+metric.aggregate+" AS value", start).Group("hour").Scan(&rows).Error; err != nil { return business.ErrStorage } } diff --git a/internal/data/business/business_portrait_test.go b/internal/data/business/business_portrait_test.go index 34d8bbd..86ce8f9 100644 --- a/internal/data/business/business_portrait_test.go +++ b/internal/data/business/business_portrait_test.go @@ -3,10 +3,13 @@ package business import ( "context" "math/big" + "reflect" + "strings" "testing" "time" "gorm.io/gorm" + "gorm.io/gorm/logger" "kra/internal/biz/business" ) @@ -151,3 +154,117 @@ func TestPortraitCountLimitAppliesBeforeTopN(t *testing.T) { t.Fatalf("人数归一上限必须先于排行截取: %+v %v", stats, err) } } + +func TestPortraitComparisonKeepsPeriodDedupAndExcludesGaps(t *testing.T) { + db, repo, scope := preparePortraitData(t) + for _, sql := range []string{ + "INSERT INTO tab_sw_report_day_key VALUES('lx',100,1,1,'2026-10-01',1,'A'),('lx',101,1,1,'2026-10-02',1,'A'),('hw',100,1,1,'2026-10-03',1,'P'),('lx',100,1,1,'2026-10-15',1,'gap'),('lx',100,1,1,'2026-10-03',2,'active'),('lx',100,1,1,'2026-10-03',3,'payer'),('lx',100,1,1,'2026-10-03',4,'new-payer')", + "INSERT INTO tab_sw_report_day VALUES('lx',100,1,0,'2026-10-03',2.20),('lx',100,1,0,'2026-10-15',99999.99)", + } { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + stats, err := repo.Statistics(context.Background(), scope) + if err != nil { + t.Fatal(err) + } + wantPrevious := business.PortraitCounts{Register: 2, Active: 1, Pay: 1, NewPay: 1} + if stats.Current.Register != 3 || stats.Previous != wantPrevious || portraitTestMoney(stats.PreviousPay) != "3.30" || portraitTestMoney(stats.CurrentPay) != "2058.60" { + t.Fatalf("合并扫描改变了周期内去重或混入同期之间的日期: %+v", stats) + } + scope.NoCompare = true + withoutCompare, err := repo.Statistics(context.Background(), scope) + if err != nil { + t.Fatal(err) + } + if withoutCompare.Current != stats.Current || withoutCompare.CurrentPay != stats.CurrentPay || withoutCompare.Previous != (business.PortraitCounts{}) || portraitTestMoney(withoutCompare.PreviousPay) != "0.00" { + t.Fatalf("关闭对比仍然返回了上期统计: %+v", withoutCompare) + } + scope.NoCompare = false + scope.Range.PreviousBegin, scope.Range.PreviousEnd = scope.Range.Begin, scope.Range.End + overlap, err := repo.Statistics(context.Background(), scope) + if err != nil || overlap.Current != overlap.Previous || overlap.CurrentPay != overlap.PreviousPay { + t.Fatalf("重叠周期必须独立统计: %+v %v", overlap, err) + } +} + +type portraitSQLCapture struct { + logger.Interface + queries []string +} + +func (l *portraitSQLCapture) Trace(_ context.Context, _ time.Time, fc func() (string, int64), _ error) { + sql, _ := fc() + if strings.Contains(sql, "HOUR(FROM_UNIXTIME(") { + l.queries = append(l.queries, sql) + } +} + +func TestMySQLPortraitHourlyUsesRegistrationRangeAndPreservesLegacyValues(t *testing.T) { + db := openRehearsalMySQL(t) + pool, _ := db.DB() + pool.SetMaxOpenConns(1) + for _, sql := range []string{ + "SET SESSION time_zone='+08:00'", + "CREATE TABLE tab_promote(id INT,source_db VARCHAR(10),busier_id INT,promote_level INT,PRIMARY KEY(id,source_db),KEY(source_db,busier_id))", + "INSERT INTO tab_promote VALUES(100,'lx',14,1),(100,'hw',15,1)", + "CREATE TABLE tab_user(id INT,source_db VARCHAR(10),promote_id INT,puid INT,register_time VARCHAR(11),PRIMARY KEY(id,source_db),KEY(promote_id,puid),KEY(register_time))", + "INSERT INTO tab_user VALUES(1,'lx',100,0,'1793638800'),(2,'lx',100,NULL,'01793638800'),(3,'lx',100,0,'1793638800 '),(4,'lx',100,1,'1793638800'),(5,'hw',100,0,'1793638800'),(6,'lx',100,0,'1793721600'),(7,'lx',100,0,'1793635199'),(8,'lx',100,0,'1793635200')", + "CREATE TABLE tab_spend(source_db VARCHAR(10),promote_id INT,pay_status INT,pay_time INT,pay_amount DECIMAL(20,2))", + "INSERT INTO tab_spend VALUES('lx',100,1,1793638800,1.10),('lx',100,0,1793638800,999.00),('hw',100,1,1793638800,999.00),('lx',100,1,1793721600,999.00)", + } { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + oldUsers := make([]map[string]any, 2000) + for i := range oldUsers { + oldUsers[i] = map[string]any{"id": i + 100, "source_db": "lx", "promote_id": 100 + i%200, "puid": 0, "register_time": "1793000000"} + } + if err := db.Table("tab_user").CreateInBatches(oldUsers, 200).Error; err != nil { + t.Fatal(err) + } + channels := make([]map[string]any, 199) + for i := range channels { + channels[i] = map[string]any{"id": i + 101, "source_db": "lx", "busier_id": 14, "promote_level": 1} + } + if err := db.Table("tab_promote").CreateInBatches(channels, 200).Error; err != nil { + t.Fatal(err) + } + day := time.Date(2026, 11, 3, 0, 0, 0, 0, time.FixedZone("CST", 8*3600)) + scope := business.PortraitReportScope{Pairs: []business.PlatformKey{{SourceDB: "lx", BusierID: 14}}, Sources: []string{"lx", "hw"}, IncludeSub: true, Range: business.PortraitRange{Begin: day, End: day, Hourly: true}} + read := func(queryDB *gorm.DB) *business.PortraitStatistics { + t.Helper() + out := &business.PortraitStatistics{RegisterSeries: map[string]string{}, PaySeries: map[string]string{}} + if err := portraitHourly(queryDB, scope, out); err != nil { + t.Fatal(err) + } + return out + } + before := read(db) + if !reflect.DeepEqual(before.RegisterSeries, map[string]string{"0": "1", "1": "3"}) || portraitTestMoney(before.PaySeries["1"]) != "1.10" || len(before.PaySeries) != 1 { + t.Fatalf("旧时间字符串、主账号、跨平台或零点边界口径错误: %+v", before) + } + if err := migrateRegistrationIndex(db); err != nil { + t.Fatal(err) + } + if err := db.Exec("ANALYZE TABLE tab_user,tab_promote").Error; err != nil { + t.Fatal(err) + } + capture := &portraitSQLCapture{Interface: logger.Default} + after := read(db.Session(&gorm.Session{Logger: capture})) + if !reflect.DeepEqual(before, after) || len(capture.queries) != 2 { + t.Fatalf("已有索引和未迁移时的结果不同: before=%+v after=%+v", before, after) + } + var plan []struct{ Table, Key, Type string } + if err := db.Raw("EXPLAIN " + capture.queries[0]).Scan(&plan).Error; err != nil { + t.Fatal(err) + } + for _, step := range plan { + if step.Table == "r" && step.Key == "idx_sw_user_registration" && step.Type == "range" { + return + } + } + t.Fatalf("小时注册趋势没有按数值时间索引做范围查找: %+v", plan) +}