feat: 迁移商务工作台指标和仪表盘榜单
This commit is contained in:
Yvan
@@ -0,0 +1,181 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type workbenchRepo struct{ data Provider }
|
||||
|
||||
func NewWorkbenchRepo(data Provider) business.WorkbenchRepo { return &workbenchRepo{data: data} }
|
||||
func workbenchAvailable(db *gorm.DB, scope business.AnalyticsScope, sources []string, kind business.WorkbenchRankKind) (bool, bool, error) {
|
||||
empty, available, err := prepareAnalytics(db, scope, business.AnalyticsFilter{Sources: sources})
|
||||
if err != nil || empty || !available {
|
||||
return empty, available, err
|
||||
}
|
||||
tables := []string{}
|
||||
if kind == "" || kind == business.WorkbenchPayHigh || kind == business.WorkbenchPayLow {
|
||||
tables = append(tables, "tab_sw_report_day")
|
||||
}
|
||||
if kind == "" || kind == business.WorkbenchRegister {
|
||||
tables = append(tables, "tab_sw_report_day_key")
|
||||
}
|
||||
for _, table := range tables {
|
||||
available, err := businessTableAvailable(db, table)
|
||||
if err != nil || !available {
|
||||
return false, available, err
|
||||
}
|
||||
}
|
||||
return false, true, nil
|
||||
}
|
||||
func (r *workbenchRepo) Summary(ctx context.Context, scope business.AnalyticsScope, sources []string, w business.WorkbenchWindow) (*business.WorkbenchTotals, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
out := &business.WorkbenchTotals{Available: true}
|
||||
empty, available, err := workbenchAvailable(db, scope, sources, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out.Available = available
|
||||
if empty || !available {
|
||||
return out, nil
|
||||
}
|
||||
mode := 0
|
||||
if scope.IDCardDedup {
|
||||
mode = 1
|
||||
}
|
||||
for _, period := range []struct {
|
||||
day time.Time
|
||||
target *business.WorkbenchDay
|
||||
}{{w.Today, &out.Today}, {w.Yesterday, &out.Yesterday}} {
|
||||
f := business.AnalyticsFilter{Sources: sources, Begin: period.day.Format("2006-01-02"), End: period.day.Format("2006-01-02")}
|
||||
var counts []portraitCountPO
|
||||
if err := analyticsQuery(db, scope, f, "tab_sw_report_day_key", mode).Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Where("r.metric IN (1,2)").Group("r.metric").Scan(&counts).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
period.target.Counts = toPortraitCounts(counts)
|
||||
if err := analyticsQuery(db, scope, f, "tab_sw_report_day", 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(&period.target.Pay).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
// 活跃渠道按自身注册或正充值统计，不能归并一级渠道后再计数。
|
||||
active := map[business.AnalyticsKey]bool{}
|
||||
var registrations, pays []analyticsAggPO
|
||||
if err := analyticsQuery(db, scope, f, "tab_sw_report_day_key", mode).Where("r.metric=1").Select("r.source_db,r.promote_id AS entity_id").Group("r.source_db,r.promote_id").Having("COUNT(DISTINCT r.identity_key)>0").Scan(®istrations).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if err := analyticsQuery(db, scope, f, "tab_sw_report_day", 0).Select("r.source_db,r.promote_id AS entity_id").Group("r.source_db,r.promote_id").Having("SUM(r.total_pay)>0").Scan(&pays).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, po := range append(registrations, pays...) {
|
||||
if po.ID != nil {
|
||||
active[business.AnalyticsKey{SourceDB: po.SourceDB, ID: *po.ID}] = true
|
||||
}
|
||||
}
|
||||
period.target.ActiveChannels = int64(len(active))
|
||||
if err := analyticsBasePromotes(db, scope, f, "p").Where("p.create_time BETWEEN ? AND ?", period.day.Unix(), period.day.AddDate(0, 0, 1).Unix()-1).Count(&period.target.NewChannels).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
}
|
||||
for _, period := range []struct {
|
||||
begin, end time.Time
|
||||
target *string
|
||||
}{{w.MonthStart, w.Today, &out.MonthPay}, {w.PreviousMonthStart, w.PreviousMonthEnd, &out.PreviousMonthPay}} {
|
||||
f := business.AnalyticsFilter{Sources: sources, Begin: period.begin.Format("2006-01-02"), End: period.end.Format("2006-01-02")}
|
||||
if err := analyticsQuery(db, scope, f, "tab_sw_report_day", 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(period.target).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
type workbenchPromotePO struct {
|
||||
ID int64
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
Account string
|
||||
CreateTime, LastLoginTime *int64
|
||||
}
|
||||
|
||||
func (r *workbenchRepo) Ranks(ctx context.Context, scope business.AnalyticsScope, f business.AnalyticsFilter, kind business.WorkbenchRankKind) (*business.WorkbenchRankData, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
out := &business.WorkbenchRankData{Rows: []*business.WorkbenchRankRow{}, Available: true}
|
||||
empty, available, err := workbenchAvailable(db, scope, f.Sources, kind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out.Available = available
|
||||
if empty || !available {
|
||||
return out, nil
|
||||
}
|
||||
rows := map[business.AnalyticsKey]*business.WorkbenchRankRow{}
|
||||
rowFor := func(source string, id int64) *business.WorkbenchRankRow {
|
||||
key := business.AnalyticsKey{SourceDB: source, ID: id}
|
||||
if rows[key] == nil {
|
||||
rows[key] = &business.WorkbenchRankRow{AnalyticsKey: key, Value: "0"}
|
||||
}
|
||||
return rows[key]
|
||||
}
|
||||
if kind == business.WorkbenchPayLow || kind == business.WorkbenchUnlogin {
|
||||
var channels []workbenchPromotePO
|
||||
if err := analyticsBasePromotes(db, scope, f, "p").Select("p.id,p.source_db,p.account,p.create_time,p.last_login_time").Where("p.promote_level=1").Scan(&channels).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, po := range channels {
|
||||
row := rowFor(po.SourceDB, po.ID)
|
||||
row.Name, row.CreatedUnix, row.LastLoginUnix = po.Account, po.CreateTime, po.LastLoginTime
|
||||
}
|
||||
}
|
||||
if kind != business.WorkbenchUnlogin {
|
||||
mode, table := 0, "tab_sw_report_day"
|
||||
if kind == business.WorkbenchRegister {
|
||||
table = "tab_sw_report_day_key"
|
||||
if scope.IDCardDedup {
|
||||
mode = 1
|
||||
}
|
||||
}
|
||||
q := analyticsQuery(db, scope, f, table, mode)
|
||||
id := "r.promote_id"
|
||||
if kind == business.WorkbenchRegister {
|
||||
q = q.Where("r.metric=1")
|
||||
if scope.IncludeSub {
|
||||
// 旧工作台的注册榜使用 INNER JOIN + CASE，区别于渠道统计的历史 LEFT JOIN 口径。
|
||||
q = q.Joins("INNER JOIN tab_promote root_p ON root_p.source_db=r.source_db AND root_p.id=r.promote_id")
|
||||
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 root_p.id END"
|
||||
}
|
||||
} else {
|
||||
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 = "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"
|
||||
}
|
||||
value := "COALESCE(SUM(r.total_pay),0)"
|
||||
if kind == business.WorkbenchRegister {
|
||||
value = "COUNT(DISTINCT r.identity_key)"
|
||||
}
|
||||
var agg []analyticsAggPO
|
||||
if err := q.Select("r.source_db," + id + " AS entity_id," + value + " AS amount").Group("r.source_db," + id).Scan(&agg).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, po := range agg {
|
||||
if po.ID != nil {
|
||||
rowFor(po.SourceDB, *po.ID).Value = po.Amount
|
||||
}
|
||||
}
|
||||
keys := make([]business.AnalyticsKey, 0, len(rows))
|
||||
for key := range rows {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
for start := 0; start < len(keys); start += 300 {
|
||||
var names []workbenchPromotePO
|
||||
if err := db.Table("tab_promote").Select("id,source_db,account").Where("(source_db,id) IN ?", analyticsPairs(keys[start:min(start+300, len(keys))])).Scan(&names).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, po := range names {
|
||||
rowFor(po.SourceDB, po.ID).Name = po.Account
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, row := range rows {
|
||||
out.Rows = append(out.Rows, row)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"kra/internal/biz/business"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestWorkbenchStorageSummaryAndRankRules(t *testing.T) {
|
||||
db, _ := prepareChannels(t)
|
||||
ctx := context.Background()
|
||||
day := time.Date(2026, 11, 4, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
|
||||
for _, sql := range []string{
|
||||
"ALTER TABLE tab_promote ADD COLUMN last_login_time INTEGER",
|
||||
"INSERT INTO tab_promote(id,source_db,account,promote_level,parent_id,top_promote_id,busier_id,create_time) VALUES(105,'lx','zero-channel',1,0,0,14,1000)",
|
||||
"DELETE FROM tab_sw_report_day",
|
||||
"DELETE FROM tab_sw_report_day_key",
|
||||
"INSERT INTO tab_sw_report_day VALUES('lx',100,1,0,'2026-11-04',10),('lx',101,1,0,'2026-11-04',20),('hw',100,1,0,'2026-11-04',40),('lx',103,1,0,'2026-11-04',-5),('lx',0,1,0,'2026-11-04',5),('lx',999,1,0,'2026-11-04',7),('lx',100,1,1,'2026-11-04',9999),('lx',100,1,0,'2026-11-03',5),('hw',100,1,0,'2026-11-03',5),('lx',100,1,0,'2026-11-01',1),('lx',100,1,0,'2026-10-01',8),('lx',100,1,0,'2026-10-04',12),('lx',100,1,0,'2026-10-05',999)",
|
||||
"INSERT INTO tab_sw_report_day_key VALUES('lx',100,1,1,'2026-11-04',1,'A'),('lx',101,1,1,'2026-11-04',1,'A'),('hw',100,1,1,'2026-11-04',1,'A'),('lx',100,1,1,'2026-11-04',2,'A'),('lx',101,1,1,'2026-11-04',2,'B'),('hw',100,1,1,'2026-11-04',2,'A'),('lx',100,1,0,'2026-11-04',1,'1'),('lx',101,1,0,'2026-11-04',1,'2'),('hw',100,1,0,'2026-11-04',1,'1'),('lx',0,1,1,'2026-11-04',1,'official'),('lx',999,1,1,'2026-11-04',1,'orphan')",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Exec("UPDATE tab_promote SET create_time=? WHERE source_db='lx' AND id IN(100,101)", day.Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w := business.WorkbenchWindow{Today: day, Yesterday: day.AddDate(0, 0, -1), MonthStart: day.AddDate(0, 0, -3), PreviousMonthStart: day.AddDate(0, -1, -3), PreviousMonthEnd: day.AddDate(0, -1, 0)}
|
||||
r := NewWorkbenchRepo(testData{db})
|
||||
scope := business.AnalyticsScope{Viewer: channelTestScope(), IncludeSub: true, IDCardDedup: true}
|
||||
sources := []string{"lx", "hw"}
|
||||
out, err := r.Summary(ctx, scope, sources, w)
|
||||
if err != nil || out.Today.Counts.Register != 1 || out.Today.Counts.Active != 2 || portraitTestMoney(out.Today.Pay) != "65.00" || out.Today.ActiveChannels != 3 || out.Today.NewChannels != 2 || portraitTestMoney(out.MonthPay) != "76.00" || portraitTestMoney(out.PreviousMonthPay) != "20.00" {
|
||||
t.Fatalf("工作台统计口径错误: %+v %v", out, err)
|
||||
}
|
||||
scope.IDCardDedup = false
|
||||
out, err = r.Summary(ctx, scope, sources, w)
|
||||
if err != nil || out.Today.Counts.Register != 3 || portraitTestMoney(out.Today.Pay) != "65.00" {
|
||||
t.Fatal("模式0应按平台前缀去重，金额不变")
|
||||
}
|
||||
scope.IncludeSub = false
|
||||
out, err = r.Summary(ctx, scope, sources, w)
|
||||
if err != nil || out.Today.Counts.Register != 2 || out.Today.NewChannels != 1 || out.Today.ActiveChannels != 2 || portraitTestMoney(out.Today.Pay) != "50.00" {
|
||||
t.Fatal("下级偏好未作用到渠道数和金额")
|
||||
}
|
||||
scope = business.AnalyticsScope{Viewer: business.UserScope{All: true}, IncludeSub: true, IDCardDedup: true}
|
||||
f := business.AnalyticsFilter{Sources: sources, Begin: "2026-11-04", End: "2026-11-04"}
|
||||
rows, err := r.Ranks(ctx, scope, f, business.WorkbenchRegister)
|
||||
if err != nil || len(rows.Rows) != 2 {
|
||||
t.Fatalf("注册榜含下级时应排除无渠道实体的官方/孤立记录: %+v %v", rows, err)
|
||||
}
|
||||
for _, row := range rows.Rows {
|
||||
if row.ID != 100 || row.Value != "1" {
|
||||
t.Fatal("注册榜应整棵树去重")
|
||||
}
|
||||
}
|
||||
rows, err = r.Ranks(ctx, scope, f, business.WorkbenchPayLow)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
values := map[business.AnalyticsKey]string{}
|
||||
for _, row := range rows.Rows {
|
||||
values[row.AnalyticsKey] = portraitTestMoney(row.Value)
|
||||
}
|
||||
for id, want := range map[int64]string{0: "5.00", 100: "30.00", 103: "-5.00", 105: "0.00", 999: "7.00"} {
|
||||
if values[business.AnalyticsKey{SourceDB: "lx", ID: id}] != want {
|
||||
t.Fatalf("最低榜漏掉零实体或汇总错误: %+v", values)
|
||||
}
|
||||
}
|
||||
scope.Viewer = channelTestScope()
|
||||
rows, err = r.Ranks(ctx, scope, f, business.WorkbenchUnlogin)
|
||||
if err != nil || len(rows.Rows) != 3 {
|
||||
t.Fatal("未登录榜只取可见一级实体")
|
||||
}
|
||||
if err := db.Exec("DROP TABLE tab_sw_report_day").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, err = r.Ranks(ctx, scope, f, business.WorkbenchRegister)
|
||||
if err != nil || !rows.Available {
|
||||
t.Fatal("注册榜不应依赖金额汇总表")
|
||||
}
|
||||
out, err = r.Summary(ctx, scope, sources, w)
|
||||
if err != nil || out.Available {
|
||||
t.Fatal("六卡缺表应明确未接入")
|
||||
}
|
||||
}
|
||||
@@ -44,6 +44,7 @@ var ProviderSet = wire.NewSet(
|
||||
databusiness.NewLedgerRepo,
|
||||
databusiness.NewAnalyticsRepo,
|
||||
databusiness.NewServerAnalysisRepo,
|
||||
databusiness.NewWorkbenchRepo,
|
||||
NewIntegrationRuntime,
|
||||
wire.Bind(new(datasystem.Provider), new(*Data)),
|
||||
wire.Bind(new(datasystem.DatabaseProvider), new(*Data)),
|
||||
|
||||
Reference in New Issue
Block a user