350 lines
12 KiB
Go
350 lines
12 KiB
Go
package business
|
|
|
|
import (
|
|
"context"
|
|
"sort"
|
|
|
|
"golang.org/x/sync/errgroup"
|
|
"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 {
|
|
if pay {
|
|
q = analyticsPayLeaves(db, q)
|
|
}
|
|
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 analyticsPayLeaves(db, q *gorm.DB) *gorm.DB {
|
|
// 金额可先按原渠道加总再归并树，减少按每天/游戏重复关联渠道资料。
|
|
leaves := q.Select("r.source_db,r.promote_id,COALESCE(SUM(r.total_pay),0) AS total_pay").Group("r.source_db,r.promote_id")
|
|
return db.Table("(?) r", leaves)
|
|
}
|
|
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, counts []analyticsAggPO
|
|
var global []portraitCountPO
|
|
globalPay := "0"
|
|
withGlobal := kind == business.ChannelAnalytics && scope.Viewer.All
|
|
group, queryContext := errgroup.WithContext(ctx)
|
|
// 每个请求最多两个并行查询，复用同一存储客户端；结果读取完后再组装 DO。
|
|
group.SetLimit(2)
|
|
queryDB := db.WithContext(queryContext)
|
|
group.Go(func() error {
|
|
q, id := analyticsRoot(queryDB, 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 business.ErrStorage
|
|
}
|
|
return nil
|
|
})
|
|
group.Go(func() error {
|
|
q, id := analyticsRoot(queryDB, 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 business.ErrStorage
|
|
}
|
|
return nil
|
|
})
|
|
if withGlobal {
|
|
group.Go(func() error {
|
|
q := analyticsQuery(queryDB, scope, f, "tab_sw_report_day_key", mode)
|
|
if err := q.Select("r.metric,COUNT(DISTINCT " + portraitIdentity(queryDB, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&global).Error; err != nil {
|
|
return business.ErrStorage
|
|
}
|
|
if err := analyticsQuery(queryDB, scope, f, "tab_sw_report_day", 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(&globalPay).Error; err != nil {
|
|
return business.ErrStorage
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
if err := group.Wait(); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, po := range pays {
|
|
if po.ID != nil {
|
|
rowFor(po).Pay = po.Amount
|
|
}
|
|
}
|
|
for _, po := range counts {
|
|
if po.ID != nil {
|
|
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 withGlobal {
|
|
totals := toPortraitCounts(global)
|
|
result.GlobalCounts, result.GlobalPay = &totals, globalPay
|
|
}
|
|
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
|
|
}
|
|
}
|
|
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 (r *analyticsRepo) FillNames(ctx context.Context, kind business.AnalyticsKind, scope business.AnalyticsScope, f business.AnalyticsFilter, rows []*business.AnalyticsRow) error {
|
|
if len(rows) == 0 {
|
|
return nil
|
|
}
|
|
byKey := make(map[business.AnalyticsKey]*business.AnalyticsRow, len(rows))
|
|
for _, row := range rows {
|
|
byKey[row.AnalyticsKey] = row
|
|
}
|
|
return fillAnalyticsNames(r.data.DB().WithContext(ctx), kind, scope, f, byKey)
|
|
}
|
|
|
|
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)
|
|
}
|
|
sort.Slice(keys, func(i, j int) bool {
|
|
if keys[i].ID != keys[j].ID {
|
|
return keys[i].ID < keys[j].ID
|
|
}
|
|
return keys[i].SourceDB < keys[j].SourceDB
|
|
})
|
|
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
|
|
}
|