feat: 迁移商务与游戏商务画像及交互看板

This commit is contained in:
Yvan
2026-11-03 23:11:46 +08:00
parent 7f8105f6fb
commit 0de822f121
30 changed files with 1959 additions and 26 deletions
+299
View File
@@ -0,0 +1,299 @@
package business
import (
"context"
"errors"
"strconv"
"gorm.io/gorm"
"kra/internal/biz/business"
)
type businessPortraitRepo struct {
data Provider
platforms PlatformProvider
}
func NewBusinessPortraitRepo(data Provider, platforms PlatformProvider) business.BusinessPortraitRepo {
return &businessPortraitRepo{data: data, platforms: platforms}
}
func (r *businessPortraitRepo) Bindings(ctx context.Context, scope business.UserScope, sources []string, key *business.PlatformKey) ([]*business.Binding, error) {
q := channelBindingQuery(r.data.DB().WithContext(ctx), scope, business.ChannelFilter{Sources: sources})
if key != nil {
q = q.Where("b.source_db = ? AND b.platform_busier_id = ?", key.SourceDB, key.BusierID)
}
var pos []bindingPO
if err := q.Select("b.*").Order("b.user_id,b.source_db,b.platform_busier_id").Scan(&pos).Error; err != nil {
return nil, business.ErrStorage
}
rows := make([]*business.Binding, 0, len(pos))
for _, po := range pos {
rows = append(rows, toBizBinding(po))
}
return rows, nil
}
func portraitBindingPairs(pairs []business.PlatformKey) [][]any {
result := make([][]any, 0, len(pairs))
for _, pair := range pairs {
result = append(result, []any{pair.SourceDB, pair.BusierID})
}
return result
}
func (r *businessPortraitRepo) Channels(ctx context.Context, pairs []business.PlatformKey, includeSub bool) ([]*business.PortraitChannel, bool, error) {
rows := []*business.PortraitChannel{}
if len(pairs) == 0 {
return rows, true, nil
}
db := r.data.DB().WithContext(ctx)
available, err := businessTableAvailable(db, "tab_promote")
if err != nil {
return nil, false, err
}
if !available {
return rows, false, nil
}
seen := map[business.ChannelKey]bool{}
for start := 0; start < len(pairs); start += 300 {
q := db.Table("tab_promote").Select("id,source_db,account,promote_level,top_promote_id,create_time").Where("(source_db,busier_id) IN ?", portraitBindingPairs(pairs[start:min(start+300, len(pairs))]))
if !includeSub {
q = q.Where("promote_level = 1")
}
var pos []channelPO
if err := q.Order("source_db,id").Scan(&pos).Error; err != nil {
return nil, false, business.ErrStorage
}
for _, po := range pos {
key := business.ChannelKey{SourceDB: po.SourceDB, ID: po.ID}
if !seen[key] {
rows = append(rows, &business.PortraitChannel{ChannelKey: key, Account: po.Account, Level: po.PromoteLevel, TopPromoteID: po.TopPromoteID, CreatedUnix: po.CreateTime})
seen[key] = true
}
}
}
return rows, true, nil
}
type portraitPlatformPO struct {
ID uint64
Account, RealName string
Level, Status *int
Commerce1, Commerce2 *int64
CreatedUnix *int64 `gorm:"column:create_time"`
}
func (r *businessPortraitRepo) PlatformBusiness(ctx context.Context, key business.PlatformKey) (*business.PortraitPlatformBusiness, error) {
db, err := r.platforms.Database(key.SourceDB)
if err != nil || db == nil {
return nil, business.ErrPlatformUnavailable
}
var po portraitPlatformPO
if err := db.WithContext(ctx).Table("tab_promote_business").Select("id,account,real_name,level,status,commerce1,commerce2,create_time").Where("id = ?", key.BusierID).Take(&po).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, business.ErrPlatformUnavailable
}
return &business.PortraitPlatformBusiness{ID: po.ID, Account: po.Account, RealName: po.RealName, Level: po.Level, Status: po.Status, Commerce1: po.Commerce1, Commerce2: po.Commerce2, CreatedUnix: po.CreatedUnix}, nil
}
func portraitPromoteScope(db *gorm.DB, scope business.PortraitReportScope, sourceColumn, promoteColumn string) *gorm.DB {
q := db.Table("tab_promote scope_promote").Select("1").Where("scope_promote.source_db = " + sourceColumn + " AND scope_promote.id = " + promoteColumn)
if len(scope.Pairs) == 0 {
return q.Where("1=0")
}
q = q.Where("(scope_promote.source_db,scope_promote.busier_id) IN ?", portraitBindingPairs(scope.Pairs))
if !scope.IncludeSub {
q = q.Where("scope_promote.promote_level = 1")
}
return q
}
func portraitReportQuery(db *gorm.DB, scope business.PortraitReportScope, table, begin, end 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", begin, end, mode)
if len(scope.Sources) > 0 {
q = q.Where("r.source_db IN ?", scope.Sources)
}
return q.Where("EXISTS (?)", portraitPromoteScope(db, scope, "r.source_db", "r.promote_id"))
}
func portraitIdentity(db *gorm.DB, dedup bool) string {
if dedup {
return "r.identity_key"
}
if db.Dialector.Name() == "mysql" {
return "CONCAT(r.source_db,'|',r.identity_key)"
}
return "(r.source_db || '|' || r.identity_key)"
}
type portraitCountPO struct {
Metric int
Count int64
}
func toPortraitCounts(rows []portraitCountPO) business.PortraitCounts {
result := business.PortraitCounts{}
for _, row := range rows {
switch row.Metric {
case 1:
result.Register = row.Count
case 2:
result.Active = row.Count
case 3:
result.Pay = row.Count
case 4:
result.NewPay = row.Count
}
}
return result
}
func (r *businessPortraitRepo) Statistics(ctx context.Context, scope business.PortraitReportScope) (*business.PortraitStatistics, error) {
result := &business.PortraitStatistics{CurrentPay: "0", PreviousPay: "0", RegisterSeries: map[string]string{}, PaySeries: map[string]string{}, RegisterRanks: []*business.PortraitStatRank{}, PayRanks: []*business.PortraitStatRank{}, Available: true}
if len(scope.Pairs) == 0 {
return result, nil
}
db := r.data.DB().WithContext(ctx)
tables := []string{"tab_promote", "tab_sw_report_day", "tab_sw_report_day_key"}
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
}
}
mode := 0
if scope.IDCardDedup {
mode = 1
}
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")
for _, period := range []struct {
begin, end string
counts *business.PortraitCounts
pay *string
}{{begin, end, &result.Current, &result.CurrentPay}, {previousBegin, previousEnd, &result.Previous, &result.PreviousPay}} {
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
}
*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
}
}
if scope.Range.Hourly {
if err := portraitHourly(db, scope, result); err != nil {
return nil, err
}
} else {
if err := portraitDaily(db, scope, mode, begin, end, result); err != nil {
return nil, err
}
}
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
return result, nil
}
func portraitDaily(db *gorm.DB, scope business.PortraitReportScope, mode int, begin, end string, result *business.PortraitStatistics) error {
dateExpression := "SUBSTR(CAST(r.stat_date AS CHAR),1,10)"
var registers []struct {
Bucket string
Count int64
}
q := portraitReportQuery(db, scope, "tab_sw_report_day_key", begin, end, mode).Where("r.metric = 1")
if err := q.Select(dateExpression + " AS bucket,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group(dateExpression).Scan(&registers).Error; err != nil {
return business.ErrStorage
}
for _, row := range registers {
result.RegisterSeries[row.Bucket] = strconv.FormatInt(row.Count, 10)
}
var payments []struct{ Bucket, Amount string }
if err := portraitReportQuery(db, scope, "tab_sw_report_day", begin, end, 0).Select(dateExpression + " AS bucket,COALESCE(SUM(r.total_pay),0) AS amount").Group(dateExpression).Scan(&payments).Error; err != nil {
return business.ErrStorage
}
for _, row := range payments {
result.PaySeries[row.Bucket] = row.Amount
}
return nil
}
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()
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("EXISTS (?)", portraitPromoteScope(db, scope, "r.source_db", "r.promote_id"))
if len(scope.Sources) > 0 {
q = q.Where("r.source_db IN ?", scope.Sources)
}
var rows []struct {
Hour int
Value string
}
if db.Dialector.Name() == "mysql" {
// 保留原 Mapper 的数据库会话时区口径,最终实库验收同时核对时区配置。
hour := "HOUR(FROM_UNIXTIME(r." + 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 {
return business.ErrStorage
}
}
for _, row := range rows {
if row.Hour >= 0 && row.Hour <= 23 {
metric.series[strconv.Itoa(row.Hour)] = row.Value
}
}
}
return nil
}
func portraitRankQuery(db *gorm.DB, scope business.PortraitReportScope, mode int, begin, end string, pay bool) ([]*business.PortraitStatRank, error) {
table, aggregate := "tab_sw_report_day_key", "COUNT(DISTINCT r.identity_key)"
if pay {
table, aggregate = "tab_sw_report_day", "COALESCE(SUM(r.total_pay),0)"
}
var selectArgs []any
if !pay && scope.CountLimit > 0 {
// 人数上限须先于 TOP N 排序,避免截取后才限制导致并列结果不同。
aggregate = "CASE WHEN COUNT(DISTINCT r.identity_key)>? THEN ? ELSE COUNT(DISTINCT r.identity_key) END"
selectArgs = []any{scope.CountLimit, scope.CountLimit}
}
q := portraitReportQuery(db, scope, table, begin, end, mode)
if !pay {
q = q.Where("r.metric = 1")
}
root := "r.promote_id"
if scope.IncludeSub || pay {
q = q.Joins("JOIN tab_promote rank_promote ON rank_promote.source_db=r.source_db AND rank_promote.id=r.promote_id")
root = "CASE WHEN rank_promote.promote_level=1 THEN rank_promote.id WHEN rank_promote.top_promote_id>0 THEN rank_promote.top_promote_id ELSE rank_promote.id END"
}
var rows []struct {
SourceDB string `gorm:"column:source_db"`
PromoteID uint64
Value string
}
if err := q.Select("r.source_db,"+root+" AS promote_id,"+aggregate+" AS value", selectArgs...).Group("r.source_db," + root).Order("value DESC," + root + ",r.source_db").Limit(scope.RankLimit).Scan(&rows).Error; err != nil {
return nil, business.ErrStorage
}
result := make([]*business.PortraitStatRank, 0, len(rows))
for _, row := range rows {
result = append(result, &business.PortraitStatRank{ChannelKey: business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}, Value: row.Value})
}
return result, nil
}
@@ -0,0 +1,153 @@
package business
import (
"context"
"math/big"
"testing"
"time"
"gorm.io/gorm"
"kra/internal/biz/business"
)
func preparePortraitData(t *testing.T) (*gorm.DB, business.BusinessPortraitRepo, business.PortraitReportScope) {
t.Helper()
db, _ := prepareChannels(t)
zone := time.FixedZone("CST", 8*3600)
start := time.Date(2026, 11, 3, 0, 0, 0, 0, zone)
for _, sql := range []string{
"UPDATE tab_sw_report_day_key SET identity_key='A' WHERE source_db='hw' AND metric=1",
"INSERT INTO tab_sw_report_day_key VALUES('hw',100,1,1,'2026-11-01',2,'C'),('lx',100,1,1,'2026-11-01',3,'A'),('hw',100,1,1,'2026-11-01',3,'A')",
"INSERT INTO tab_sw_report_day_key VALUES('hw',100,1,0,'2026-11-01',1,'u1')",
"ALTER TABLE tab_user ADD COLUMN register_time INTEGER",
"ALTER TABLE tab_spend ADD COLUMN pay_time INTEGER",
} {
if err := db.Exec(sql).Error; err != nil {
t.Fatal(err)
}
}
if err := db.Exec("UPDATE tab_user SET register_time = ?", start.Unix()+3600).Error; err != nil {
t.Fatal(err)
}
if err := db.Exec("UPDATE tab_spend SET pay_time = ?", start.Unix()+3600).Error; err != nil {
t.Fatal(err)
}
scope := business.PortraitReportScope{Pairs: []business.PlatformKey{{SourceDB: "lx", BusierID: 14}, {SourceDB: "hw", BusierID: 14}}, Sources: []string{"lx", "hw"}, IncludeSub: true, IDCardDedup: true, RankLimit: 5, Range: business.PortraitRange{Begin: time.Date(2026, 11, 1, 0, 0, 0, 0, zone), End: start, PreviousBegin: time.Date(2026, 10, 1, 0, 0, 0, 0, zone), PreviousEnd: time.Date(2026, 10, 3, 0, 0, 0, 0, zone)}}
return db, NewBusinessPortraitRepo(testData{db}, channelDatabases{}), scope
}
func portraitTestMoney(s string) string {
n, ok := new(big.Rat).SetString(s)
if !ok {
return "invalid"
}
return n.FloatString(2)
}
func TestPortraitBindingsAndChannelsFollowBusinessTree(t *testing.T) {
_, repo, _ := preparePortraitData(t)
ctx := context.Background()
bindings, err := repo.Bindings(ctx, channelTestScope(), []string{"lx", "hw"}, nil)
if err != nil || len(bindings) != 2 {
t.Fatalf("本人及下级绑定范围错误: %+v %v", bindings, err)
}
for _, b := range bindings {
if b.UserID == 10 || b.Status != "0" {
t.Fatal("混入平级或未通过绑定")
}
}
pairs := []business.PlatformKey{}
for _, b := range bindings {
pairs = append(pairs, b.PlatformKey)
}
channels, available, err := repo.Channels(ctx, pairs, true)
if err != nil || !available || len(channels) != 5 {
t.Fatalf("渠道范围不正确: %+v %v", channels, err)
}
channels, _, err = repo.Channels(ctx, pairs, false)
if err != nil || len(channels) != 2 {
t.Fatal("关闭下级时只能取一级渠道")
}
outside := business.PlatformKey{SourceDB: "lx", BusierID: 16}
bindings, err = repo.Bindings(ctx, channelTestScope(), []string{"lx"}, &outside)
if err != nil || len(bindings) != 0 {
t.Fatal("绑定过滤绕过商务树")
}
}
func TestPortraitSummaryDedupsAcrossPlatformsAndRanksWithinTrees(t *testing.T) {
_, repo, scope := preparePortraitData(t)
stats, err := repo.Statistics(context.Background(), scope)
if err != nil {
t.Fatal(err)
}
if stats.Current.Register != 3 || stats.Current.Active != 1 || stats.Current.Pay != 1 || portraitTestMoney(stats.CurrentPay) != "2058.60" || portraitTestMoney(stats.PreviousPay) != "1.10" {
t.Fatalf("全局去重或金额口径错误: %+v", stats)
}
if stats.RegisterSeries["2026-11-01"] != "1" || stats.RegisterSeries["2026-11-02"] != "3" {
t.Fatalf("日趋势错误: %+v", stats.RegisterSeries)
}
var lxRoot, hwRoot string
for _, rank := range stats.RegisterRanks {
if rank.ID == 100 && rank.SourceDB == "lx" {
lxRoot = rank.Value
}
if rank.ID == 100 && rank.SourceDB == "hw" {
hwRoot = rank.Value
}
}
if lxRoot != "2" || hwRoot != "1" {
t.Fatalf("渠道树注册排行不能直接累加子渠道或混淆同号平台: %+v", stats.RegisterRanks)
}
scope.IDCardDedup = false
stats, err = repo.Statistics(context.Background(), scope)
if err != nil || stats.Current.Register != 3 || portraitTestMoney(stats.CurrentPay) != "2058.60" {
t.Fatalf("模式0没有区分跨平台同号身份或改变金额: %+v %v", stats, err)
}
scope.IDCardDedup = true
scope.IncludeSub = false
stats, err = repo.Statistics(context.Background(), scope)
if err != nil || stats.Current.Register != 1 || portraitTestMoney(stats.CurrentPay) != "1009.10" {
t.Fatalf("关闭下级没有收窄: %+v %v", stats, err)
}
}
func TestPortraitHourlyUsesRawMainAccountsAndHalfOpenDay(t *testing.T) {
db, repo, scope := preparePortraitData(t)
scope.Range.Begin = scope.Range.End
scope.Range.Hourly = true
stats, err := repo.Statistics(context.Background(), scope)
if err != nil || stats.RegisterSeries["1"] != "5" || portraitTestMoney(stats.PaySeries["1"]) != "200.90" {
t.Fatalf("小时趋势错误: %+v %v", stats, err)
}
end := scope.Range.End.AddDate(0, 0, 1).Unix()
if err := db.Exec("INSERT INTO tab_user(source_db,id,promote_id,puid,idcard,register_time) VALUES('lx',900,100,0,'boundary',?)", end).Error; err != nil {
t.Fatal(err)
}
stats, err = repo.Statistics(context.Background(), scope)
if err != nil || stats.RegisterSeries["0"] != "" || stats.RegisterSeries["1"] != "5" {
t.Fatal("次日零点被计入当日")
}
if err := db.Exec("DROP TABLE tab_sw_report_day_key").Error; err != nil {
t.Fatal(err)
}
stats, err = repo.Statistics(context.Background(), scope)
if err != nil || stats.Available {
t.Fatal("缺汇总键表不应返回假零指标")
}
channels, available, err := repo.Channels(context.Background(), nil, true)
if err != nil || !available || len(channels) != 0 {
t.Fatal("无绑定应是已知空范围")
}
}
func TestPortraitCountLimitAppliesBeforeTopN(t *testing.T) {
db, repo, scope := preparePortraitData(t)
if err := db.Exec("INSERT INTO tab_sw_report_day_key VALUES('lx',103,1,1,'2026-11-02',1,'N2'),('lx',103,1,1,'2026-11-02',1,'N3')").Error; err != nil {
t.Fatal(err)
}
scope.CountLimit = 1
scope.RankLimit = 1
scope.Sources = []string{"lx"}
stats, err := repo.Statistics(context.Background(), scope)
if err != nil || len(stats.RegisterRanks) != 1 || stats.RegisterRanks[0].ID != 100 || stats.RegisterRanks[0].Value != "1" {
t.Fatalf("人数归一上限必须先于排行截取: %+v %v", stats, err)
}
}
+1
View File
@@ -39,6 +39,7 @@ var ProviderSet = wire.NewSet(
databusiness.NewActivationRepo,
databusiness.NewChannelRepo,
databusiness.NewChannelWriteRepo,
databusiness.NewBusinessPortraitRepo,
NewIntegrationRuntime,
wire.Bind(new(datasystem.Provider), new(*Data)),
wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),