feat: 迁移渠道与游戏统计及全量汇总

2026-11-04 01:08:33 +08:00
parent 6703a66f0d
commit fc27120950
31 changed files with 1492 additions and 6 deletions
@@ -0,0 +1,307 @@
+package business
+
+import (
+	"context"
+	"sort"
+
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+type analyticsRepo struct{ data Provider }
+
+func NewAnalyticsRepo(data Provider) business.AnalyticsRepo { return &analyticsRepo{data: data} }
+func analyticsBasePromotes(db *gorm.DB, scope business.AnalyticsScope, f business.AnalyticsFilter, alias string) *gorm.DB {
+	q := ledgerBasePromotes(db, business.LedgerScope{Viewer: scope.Viewer, IncludeSub: scope.IncludeSub, Unrestricted: scope.Viewer.All}, business.LedgerFilter{Sources: f.Sources}, alias)
+	if f.PromoteType != nil {
+		q = q.Where(alias+".register_type = ?", *f.PromoteType)
+	}
+	return q
+}
+func analyticsPromotes(db *gorm.DB, scope business.AnalyticsScope, f business.AnalyticsFilter, alias string) *gorm.DB {
+	q := analyticsBasePromotes(db, scope, f, alias)
+	if f.PromoteName == "" {
+		return q
+	}
+	hits := analyticsBasePromotes(db, scope, f, "hit").Select("1").Where("hit.account LIKE ?", "%"+f.PromoteName+"%").Where("hit.source_db=" + alias + ".source_db")
+	if scope.IncludeSub {
+		root := "CASE WHEN hit.promote_level=1 THEN hit.id WHEN hit.top_promote_id>0 THEN hit.top_promote_id ELSE hit.id END"
+		hits = hits.Where("(" + alias + ".id=(" + root + ") OR " + alias + ".top_promote_id=(" + root + "))")
+	} else {
+		hits = hits.Where("hit.id=" + alias + ".id")
+	}
+	return q.Where("EXISTS (?)", hits)
+}
+func prepareAnalytics(db *gorm.DB, scope business.AnalyticsScope, f business.AnalyticsFilter) (bool, bool, error) {
+	if !scope.Viewer.All {
+		var binds []struct{ UserID uint }
+		if err := channelBindingQuery(db, scope.Viewer, business.ChannelFilter{Sources: f.Sources}).Select("b.user_id").Limit(1).Scan(&binds).Error; err != nil {
+			return false, false, business.ErrStorage
+		}
+		if len(binds) == 0 {
+			return true, true, nil
+		}
+	}
+	available, err := businessTableAvailable(db, "tab_promote")
+	if err != nil || !available {
+		return false, available, err
+	}
+	var rows []struct{ ID int64 }
+	if err := analyticsPromotes(db, scope, f, "p").Select("p.id").Limit(1).Scan(&rows).Error; err != nil {
+		return false, false, business.ErrStorage
+	}
+	return len(rows) == 0, true, nil
+}
+func analyticsQuery(db *gorm.DB, scope business.AnalyticsScope, f business.AnalyticsFilter, table string, mode int) *gorm.DB {
+	q := db.Table(table+" r").Where("r.stat_date >= ? AND r.stat_date <= ? AND r.dedup_mode = ? AND r.game_id > 0", f.Begin, f.End, mode)
+	if len(f.Sources) > 0 {
+		q = q.Where("r.source_db IN ?", f.Sources)
+	}
+	if len(f.GameNames) > 0 {
+		q = q.Where("r.game_name IN ?", f.GameNames)
+	}
+	if scope.Viewer.All && scope.IncludeSub && f.PromoteType == nil && f.PromoteName == "" {
+		return q
+	}
+	return q.Where("EXISTS (?)", analyticsPromotes(db, scope, f, "p").Select("1").Where("p.source_db=r.source_db AND p.id=r.promote_id"))
+}
+
+type analyticsAggPO struct {
+	SourceDB string `gorm:"column:source_db"`
+	ID       *int64 `gorm:"column:entity_id"`
+	Metric   int
+	Count    int64
+	Amount   string
+}
+
+func analyticsRoot(db *gorm.DB, kind business.AnalyticsKind, scope business.AnalyticsScope, f business.AnalyticsFilter, table string, mode int, pay bool) (*gorm.DB, string) {
+	q := analyticsQuery(db, scope, f, table, mode)
+	id := "r.game_id"
+	if kind == business.ChannelAnalytics {
+		id = "r.promote_id"
+		if scope.IncludeSub {
+			q = q.Joins("LEFT JOIN tab_promote root_p ON root_p.source_db=r.source_db AND root_p.id=r.promote_id")
+			id = "COALESCE(NULLIF(root_p.top_promote_id,0),r.promote_id)"
+			if pay {
+				id = "CASE WHEN root_p.promote_level=1 THEN root_p.id WHEN root_p.top_promote_id>0 THEN root_p.top_promote_id ELSE COALESCE(root_p.id,r.promote_id) END"
+			}
+		}
+	}
+	return q, id
+}
+func putAnalyticsCount(counts *business.PortraitCounts, metric int, count int64) {
+	switch metric {
+	case 1:
+		counts.Register = count
+	case 2:
+		counts.Active = count
+	case 3:
+		counts.Pay = count
+	case 4:
+		counts.NewPay = count
+	}
+}
+func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKind, scope business.AnalyticsScope, f business.AnalyticsFilter) (*business.AnalyticsData, error) {
+	db := r.data.DB().WithContext(ctx)
+	result := &business.AnalyticsData{Rows: []*business.AnalyticsRow{}, GlobalPay: "0", Available: true}
+	empty, available, err := prepareAnalytics(db, scope, f)
+	if err != nil {
+		return nil, err
+	}
+	result.Available = available
+	if empty || !available {
+		return result, nil
+	}
+	for _, table := range []string{"tab_sw_report_day", "tab_sw_report_day_key"} {
+		available, err := businessTableAvailable(db, table)
+		if err != nil {
+			return nil, err
+		}
+		if !available {
+			result.Available = false
+			return result, nil
+		}
+	}
+	mode := 0
+	if scope.IDCardDedup {
+		mode = 1
+	}
+	rowMap := map[business.AnalyticsKey]*business.AnalyticsRow{}
+	rowFor := func(po analyticsAggPO) *business.AnalyticsRow {
+		key := business.AnalyticsKey{SourceDB: po.SourceDB, ID: *po.ID}
+		if rowMap[key] == nil {
+			rowMap[key] = &business.AnalyticsRow{AnalyticsKey: key, Pay: "0"}
+		}
+		return rowMap[key]
+	}
+	var pays []analyticsAggPO
+	q, id := analyticsRoot(db, kind, scope, f, "tab_sw_report_day", 0, true)
+	if err := q.Select("r.source_db," + id + " AS entity_id,COALESCE(SUM(r.total_pay),0) AS amount").Group("r.source_db," + id).Scan(&pays).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	for _, po := range pays {
+		if po.ID == nil {
+			continue
+		}
+		rowFor(po).Pay = po.Amount
+	}
+	var counts []analyticsAggPO
+	q, id = analyticsRoot(db, kind, scope, f, "tab_sw_report_day_key", mode, false)
+	if err := q.Select("r.source_db," + id + " AS entity_id,r.metric,COUNT(DISTINCT r.identity_key) AS count").Group("r.source_db," + id + ",r.metric").Scan(&counts).Error; err != nil {
+		return nil, business.ErrStorage
+	}
+	for _, po := range counts {
+		if po.ID == nil {
+			continue
+		}
+		putAnalyticsCount(&rowFor(po).Counts, po.Metric, po.Count)
+	}
+	if kind == business.ChannelAnalytics && !scope.Viewer.All {
+		if err := filterAnalyticsRows(db, scope, f, rowMap); err != nil {
+			return nil, err
+		}
+	}
+	if kind == business.ChannelAnalytics && scope.Viewer.All {
+		var global []portraitCountPO
+		q := analyticsQuery(db, scope, f, "tab_sw_report_day_key", mode)
+		if err := q.Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&global).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+		totals := toPortraitCounts(global)
+		result.GlobalCounts = &totals
+		if err := analyticsQuery(db, scope, f, "tab_sw_report_day", 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(&result.GlobalPay).Error; err != nil {
+			return nil, business.ErrStorage
+		}
+	}
+	if len(rowMap) == 0 {
+		return result, nil
+	}
+	if kind == business.GameAnalytics {
+		available, err := businessTableAvailable(db, "tab_game")
+		if err != nil {
+			return nil, err
+		}
+		if !available {
+			result.Available = false
+			return result, nil
+		}
+	}
+	if err := fillAnalyticsNames(db, kind, scope, f, rowMap); err != nil {
+		return nil, err
+	}
+	for _, row := range rowMap {
+		result.Rows = append(result.Rows, row)
+	}
+	sort.Slice(result.Rows, func(i, j int) bool {
+		a, b := result.Rows[i], result.Rows[j]
+		if a.SourceDB != b.SourceDB {
+			return a.SourceDB < b.SourceDB
+		}
+		return a.ID < b.ID
+	})
+	return result, nil
+}
+func analyticsPairs(keys []business.AnalyticsKey) [][]any {
+	result := make([][]any, 0, len(keys))
+	for _, key := range keys {
+		result = append(result, []any{key.SourceDB, key.ID})
+	}
+	return result
+}
+func filterAnalyticsRows(db *gorm.DB, scope business.AnalyticsScope, f business.AnalyticsFilter, rows map[business.AnalyticsKey]*business.AnalyticsRow) error {
+	keys := []business.AnalyticsKey{}
+	for key := range rows {
+		keys = append(keys, key)
+	}
+	allowed := map[business.AnalyticsKey]bool{}
+	root := "p.id"
+	if scope.IncludeSub {
+		root = "CASE WHEN p.promote_level=1 THEN p.id WHEN p.top_promote_id>0 THEN p.top_promote_id ELSE p.id END"
+	}
+	for start := 0; start < len(keys); start += 300 {
+		var selected []analyticsAggPO
+		q := analyticsPromotes(db, scope, f, "p").Select("DISTINCT p.source_db,"+root+" AS entity_id").Where("(p.source_db,"+root+") IN ?", analyticsPairs(keys[start:min(start+300, len(keys))]))
+		if err := q.Scan(&selected).Error; err != nil {
+			return business.ErrStorage
+		}
+		for _, row := range selected {
+			if row.ID != nil {
+				allowed[business.AnalyticsKey{SourceDB: row.SourceDB, ID: *row.ID}] = true
+			}
+		}
+	}
+	for key := range rows {
+		if !allowed[key] {
+			delete(rows, key)
+		}
+	}
+	return nil
+}
+func fillAnalyticsNames(db *gorm.DB, kind business.AnalyticsKind, scope business.AnalyticsScope, f business.AnalyticsFilter, rows map[business.AnalyticsKey]*business.AnalyticsRow) error {
+	keys := []business.AnalyticsKey{}
+	for key := range rows {
+		keys = append(keys, key)
+	}
+	type entityPO struct {
+		ID           int64
+		SourceDB     string `gorm:"column:source_db"`
+		Name         string
+		RegisterType *int
+		BusierID     *int64
+	}
+	promotes := map[business.AnalyticsKey]entityPO{}
+	bindings := map[business.PlatformKey]struct{ account, user string }{}
+	pairs := []business.PlatformKey{}
+	pairSeen := map[business.PlatformKey]bool{}
+	for start := 0; start < len(keys); start += 300 {
+		var entities []entityPO
+		q := db.Table("tab_game").Select("id,source_db,game_name AS name")
+		if kind == business.ChannelAnalytics {
+			q = db.Table("tab_promote").Select("id,source_db,account AS name,register_type,busier_id")
+		}
+		if err := q.Where("(source_db,id) IN ?", analyticsPairs(keys[start:min(start+300, len(keys))])).Scan(&entities).Error; err != nil {
+			return business.ErrStorage
+		}
+		for _, entity := range entities {
+			key := business.AnalyticsKey{SourceDB: entity.SourceDB, ID: entity.ID}
+			if row := rows[key]; row != nil {
+				row.Name = entity.Name
+				row.PromoteType = entity.RegisterType
+				promotes[key] = entity
+			}
+			if kind == business.ChannelAnalytics && entity.BusierID != nil && *entity.BusierID > 0 {
+				pair := business.PlatformKey{SourceDB: entity.SourceDB, BusierID: uint64(*entity.BusierID)}
+				if !pairSeen[pair] {
+					pairs = append(pairs, pair)
+					pairSeen[pair] = true
+				}
+			}
+		}
+	}
+	for start := 0; start < len(pairs); start += 300 {
+		var names []struct {
+			SourceDB                        string `gorm:"column:source_db"`
+			PlatformBusierID                uint64
+			PlatformBusierAccount, Username string
+		}
+		if err := channelBindingQuery(db, scope.Viewer, business.ChannelFilter{Sources: f.Sources}).Select("b.source_db,b.platform_busier_id,b.platform_busier_account,u.username").Where("(b.source_db,b.platform_busier_id) IN ?", portraitBindingPairs(pairs[start:min(start+300, len(pairs))])).Order("b.user_id").Scan(&names).Error; err != nil {
+			return business.ErrStorage
+		}
+		for _, name := range names {
+			key := business.PlatformKey{SourceDB: name.SourceDB, BusierID: name.PlatformBusierID}
+			if _, ok := bindings[key]; !ok {
+				bindings[key] = struct{ account, user string }{name.PlatformBusierAccount, name.Username}
+			}
+		}
+	}
+	for key, entity := range promotes {
+		if entity.BusierID == nil || *entity.BusierID <= 0 {
+			continue
+		}
+		if name, ok := bindings[business.PlatformKey{SourceDB: entity.SourceDB, BusierID: uint64(*entity.BusierID)}]; ok {
+			rows[key].BindingAccount = name.account
+			rows[key].BindingUserName = name.user
+		}
+	}
+	return nil
+}
@@ -0,0 +1,111 @@
+package business
+
+import (
+	"context"
+	"testing"
+
+	"gorm.io/gorm"
+	"kra/internal/biz/business"
+)
+
+func prepareAnalyticsData(t *testing.T) (*gorm.DB, business.AnalyticsRepo, business.AnalyticsScope, business.AnalyticsFilter) {
+	t.Helper()
+	db, _ := prepareChannels(t)
+	for _, sql := range []string{
+		"ALTER TABLE tab_sw_report_day ADD COLUMN game_name TEXT",
+		"ALTER TABLE tab_sw_report_day_key ADD COLUMN game_name TEXT",
+		"UPDATE tab_sw_report_day SET game_name=CASE WHEN game_id=1 THEN '历史游戏甲' ELSE '游戏乙' END",
+		"UPDATE tab_sw_report_day_key SET game_name=CASE WHEN game_id=1 THEN '历史游戏甲' ELSE '游戏乙' END",
+		"UPDATE tab_sw_report_day_key SET identity_key='A' WHERE (source_db='hw' OR promote_id=103) AND metric=1",
+		"INSERT INTO tab_sw_report_day(source_db,promote_id,game_id,dedup_mode,stat_date,total_pay,game_name) VALUES('lx',0,1,0,'2026-11-01',5,'历史游戏甲'),('lx',999,1,0,'2026-11-01',7,'历史游戏甲')",
+		"INSERT INTO tab_sw_report_day_key(source_db,promote_id,game_id,dedup_mode,stat_date,metric,identity_key,game_name) VALUES('lx',0,1,1,'2026-11-01',1,'official','历史游戏甲'),('lx',999,1,1,'2026-11-01',1,'orphan','历史游戏甲'),('lx',100,1,1,'2026-11-01',2,'A','历史游戏甲'),('lx',103,1,1,'2026-11-02',2,'A','历史游戏甲'),('hw',100,1,1,'2026-11-01',2,'A','历史游戏甲'),('lx',100,1,1,'2026-11-01',3,'A','历史游戏甲'),('lx',103,1,1,'2026-11-02',3,'A','历史游戏甲'),('hw',100,1,1,'2026-11-01',3,'A','历史游戏甲')",
+	} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	scope := business.AnalyticsScope{Viewer: channelTestScope(), IncludeSub: true, IDCardDedup: true}
+	filter := business.AnalyticsFilter{Sources: []string{"lx", "hw"}, Begin: "2026-11-01", End: "2026-11-03"}
+	return db, NewAnalyticsRepo(testData{db}), scope, filter
+}
+func TestAnalyticsChannelTreeAndAdminOfficialRows(t *testing.T) {
+	_, repo, scope, filter := prepareAnalyticsData(t)
+	data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(data.Rows) != 3 || data.GlobalCounts != nil {
+		t.Fatalf("普通商务统计范围错误: %+v", data)
+	}
+	var lx *business.AnalyticsRow
+	for _, row := range data.Rows {
+		if row.SourceDB == "lx" && row.ID == 100 {
+			lx = row
+		}
+		if row.ID == 0 || row.ID == 999 {
+			t.Fatal("受约束商务看到官方/未知渠道")
+		}
+	}
+	if lx == nil || lx.Counts.Register != 2 || portraitTestMoney(lx.Pay) != "60.60" || lx.Name != "root-channel" || lx.BusinessName() != "lx-business" {
+		t.Fatalf("渠道树去重、充值或商务展示错误: %+v", lx)
+	}
+	scope.Viewer = business.UserScope{All: true}
+	data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 5 || data.GlobalCounts.Register != 4 || data.GlobalCounts.Active != 2 || data.GlobalCounts.Pay != 1 || portraitTestMoney(data.GlobalPay) != "2070.60" {
+		t.Fatalf("超管全局去重或官方/历史渠道遗漏: %+v %v", data, err)
+	}
+}
+func TestAnalyticsFiltersKeepHierarchyAndSnapshotNames(t *testing.T) {
+	db, repo, scope, filter := prepareAnalyticsData(t)
+	filter.PromoteName = "child"
+	data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 1 || data.Rows[0].ID != 100 || data.Rows[0].Counts.Register != 2 || portraitTestMoney(data.Rows[0].Pay) != "60.60" {
+		t.Fatalf("模糊渠道筛选应扩到同树可见渠道: %+v %v", data, err)
+	}
+	if err := db.Exec("UPDATE tab_promote SET register_type=1 WHERE source_db='lx' AND id=101").Error; err != nil {
+		t.Fatal(err)
+	}
+	filter.PromoteName = ""
+	kind := 1
+	filter.PromoteType = &kind
+	data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 1 || data.Rows[0].ID != 100 || portraitTestMoney(data.Rows[0].Pay) != "20.20" || data.Rows[0].PromoteType == nil || *data.Rows[0].PromoteType != 0 {
+		t.Fatalf("类型筛选应限制参与渠道，同时显示归根后的当前资料: %+v %v", data, err)
+	}
+	filter.PromoteType = nil
+	filter.GameNames = []string{"历史游戏甲"}
+	data, err = repo.Aggregate(context.Background(), business.GameAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 2 {
+		t.Fatalf("按游戏快照名筛选错误: %+v %v", data, err)
+	}
+	for _, row := range data.Rows {
+		if row.SourceDB == "lx" && (row.ID != 1 || row.Name != "游戏甲" || row.Counts.Register != 1) {
+			t.Fatalf("应按报表快照筛选、用当前游戏名展示: %+v", row)
+		}
+	}
+	filter.GameNames = []string{"游戏甲"}
+	data, err = repo.Aggregate(context.Background(), business.GameAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 0 {
+		t.Fatal("不能用当前游戏表名替代报表快照过滤")
+	}
+}
+func TestAnalyticsMissingScopeAndTablesDoNotBroaden(t *testing.T) {
+	db, repo, scope, filter := prepareAnalyticsData(t)
+	scope.IncludeSub = false
+	data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || len(data.Rows) != 2 {
+		t.Fatal("关闭下级仅保留一级渠道统计")
+	}
+	if err := db.Exec("DROP TABLE tab_sw_report_day_key").Error; err != nil {
+		t.Fatal(err)
+	}
+	data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || data.Available {
+		t.Fatal("缺身份键表不能返回零汇总")
+	}
+	scope.Viewer = business.UserScope{SelfID: 500, OnlySelf: true}
+	data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
+	if err != nil || !data.Available || len(data.Rows) != 0 {
+		t.Fatal("无绑定应为确定空范围")
+	}
+}
@@ -42,6 +42,7 @@ var ProviderSet = wire.NewSet(
 	databusiness.NewBusinessPortraitRepo,
 	databusiness.NewChannelPortraitRepo,
 	databusiness.NewLedgerRepo,
+	databusiness.NewAnalyticsRepo,
 	NewIntegrationRuntime,
 	wire.Bind(new(datasystem.Provider), new(*Data)),
 	wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),