fix: 补齐渠道全量合计筛选导出及有效端目标遗漏

This commit is contained in:
Yvan
2026-11-05 10:57:06 +08:00
parent 297a892518
commit 1aa15d3dc8
22 changed files with 545 additions and 88 deletions
+68 -5
View File
@@ -2,6 +2,7 @@ package business
import (
"context"
"math/big"
"sort"
"strings"
"time"
@@ -59,7 +60,13 @@ func prepareAnalytics(db *gorm.DB, scope business.AnalyticsScope, f business.Ana
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)
q := db.Table(table+" r").Where("r.dedup_mode = ? AND r.game_id > 0", mode)
if f.Begin != "" {
q = q.Where("r.stat_date >= ?", f.Begin)
}
if f.End != "" {
q = q.Where("r.stat_date <= ?", f.End)
}
if len(f.Sources) > 0 {
q = q.Where("r.source_db IN ?", f.Sources)
}
@@ -165,6 +172,7 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
var global []portraitCountPO
globalPay := "0"
withGlobal := kind == business.ChannelAnalytics && scope.Viewer.All
metricFiltered := kind == business.ChannelAnalytics && (f.HasRegister != nil || f.HasPay != nil)
group, queryContext := errgroup.WithContext(ctx)
// 每个请求最多两个并行查询,复用同一存储客户端;结果读取完后再组装 DO。
group.SetLimit(2)
@@ -183,7 +191,7 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
}
return nil
})
if withGlobal {
if withGlobal && !metricFiltered {
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 {
@@ -208,6 +216,23 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
putAnalyticsCount(&rowFor(po).Counts, po.Metric, po.Count)
}
}
// Keep visible channels with no events in the selected period so that
// "no registration / no recharge" can find genuinely inactive channels.
if kind == business.ChannelAnalytics && len(f.GameNames) == 0 {
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"
}
var channels []analyticsAggPO
if err := analyticsPromotes(db, scope, f, "p").Select("DISTINCT p.source_db," + root + " AS entity_id").Scan(&channels).Error; err != nil {
return nil, business.ErrStorage
}
for _, po := range channels {
if po.ID != nil {
rowFor(po)
}
}
}
if kind == business.ChannelAnalytics && !scope.Viewer.All {
if err := filterAnalyticsRows(db, scope, f, rowMap); err != nil {
return nil, err
@@ -218,6 +243,33 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
return nil, err
}
}
if metricFiltered {
for key, row := range rowMap {
pay, ok := new(big.Rat).SetString(row.Pay)
if !ok {
return nil, business.ErrStorage
}
if f.HasRegister != nil && (row.Counts.Register > 0) != *f.HasRegister || f.HasPay != nil && (pay.Sign() > 0) != *f.HasPay {
delete(rowMap, key)
}
}
// Recompute the administrator's distinct-player totals over matched
// channel trees, before pagination. Summing per-row ratios is incorrect.
if withGlobal && len(rowMap) > 0 {
keys := make([]business.AnalyticsKey, 0, len(rowMap))
for key := range rowMap {
keys = append(keys, key)
}
q, id := analyticsRoot(db, kind, scope, f, "tab_sw_report_day_key", mode, false)
if err := q.Where("(r.source_db,"+id+") IN ?", analyticsPairs(keys)).Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&global).Error; err != nil {
return nil, business.ErrStorage
}
q, id = analyticsRoot(db, kind, scope, f, "tab_sw_report_day", 0, true)
if err := q.Where("(r.source_db,"+id+") IN ?", analyticsPairs(keys)).Select("COALESCE(SUM(r.total_pay),0)").Scan(&globalPay).Error; err != nil {
return nil, business.ErrStorage
}
}
}
if withGlobal {
totals := toPortraitCounts(global)
result.GlobalCounts, result.GlobalPay = &totals, globalPay
@@ -227,15 +279,26 @@ func (r *analyticsRepo) Aggregate(ctx context.Context, kind business.AnalyticsKi
}
if kind == business.ChannelAnalytics {
location := time.FixedZone("Asia/Shanghai", 8*3600)
begin, _ := time.ParseInLocation("2006-01-02", f.Begin, location)
end, _ := time.ParseInLocation("2006-01-02", f.End, location)
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"
}
keys := []business.AnalyticsKey{}
for key := range rowMap {
keys = append(keys, key)
}
for start := 0; start < len(keys); start += 300 {
var count int64
if err := db.Table("sys_business_channel_effectiveness e").Joins("JOIN sys_business_effectiveness_state s ON s.id=1 AND s.rule_key=e.rule_key").Where("(e.source_db,e.promote_id) IN ? AND e.qualified_at>=? AND e.qualified_at<?", analyticsPairs(keys[start:min(start+300, len(keys))]), begin.Unix(), end.AddDate(0, 0, 1).Unix()).Count(&count).Error; err != nil {
q := analyticsPromotes(db, scope, f, "p").Joins("JOIN sys_business_channel_effectiveness e ON e.source_db=p.source_db AND e.promote_id=p.id").Joins("JOIN sys_business_effectiveness_state s ON s.id=1 AND s.rule_key=e.rule_key").Where("(p.source_db,"+root+") IN ? AND e.qualified_at IS NOT NULL", analyticsPairs(keys[start:min(start+300, len(keys))]))
if f.Begin != "" {
begin, _ := time.ParseInLocation("2006-01-02", f.Begin, location)
q = q.Where("e.qualified_at >= ?", begin.Unix())
}
if f.End != "" {
end, _ := time.ParseInLocation("2006-01-02", f.End, location)
q = q.Where("e.qualified_at < ?", end.AddDate(0, 0, 1).Unix())
}
if err := q.Count(&count).Error; err != nil {
return nil, business.ErrStorage
}
result.EffectiveCount += count
+64
View File
@@ -3,6 +3,7 @@ package business
import (
"context"
"testing"
"time"
"gorm.io/gorm"
"kra/internal/biz/business"
@@ -59,6 +60,69 @@ func TestAnalyticsChannelTreeAndAdminOfficialRows(t *testing.T) {
t.Fatalf("超管全局去重或官方/历史渠道遗漏: %+v %v", data, err)
}
}
func TestAnalyticsFiltersIncludeIdleChannelsAndRecalculateTotals(t *testing.T) {
db, repo, scope, filter := prepareAnalyticsData(t)
if err := db.Exec("INSERT INTO tab_promote(id,source_db,account,promote_level,parent_id,top_promote_id,busier_id) VALUES(200,'lx','idle',1,0,0,14)").Error; err != nil {
t.Fatal(err)
}
yes, no := true, false
filter.HasRegister, filter.HasPay = &no, &no
for _, viewer := range []business.UserScope{scope.Viewer, {All: true}} {
scope.Viewer = viewer
data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil {
t.Fatal(err)
}
found := false
for _, row := range data.Rows {
if row.Counts.Register != 0 || portraitTestMoney(row.Pay) != "0.00" {
t.Fatalf("筛选后的每行必须同时满足两个条件: %+v", row)
}
found = found || row.SourceDB == "lx" && row.ID == 200
}
if !found {
t.Fatal("整个期间没有事件的可见渠道不能遗漏")
}
if viewer.All && (data.GlobalCounts.Register != 0 || portraitTestMoney(data.GlobalPay) != "0.00") {
t.Fatalf("管理员合计未跟随过滤: %+v", data)
}
}
filter.HasRegister, filter.HasPay = &yes, &yes
filter.Sources, filter.PromoteName = []string{"lx"}, "root-channel"
data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || len(data.Rows) != 1 || portraitTestMoney(data.GlobalPay) != "60.60" || data.GlobalCounts.Register != data.Rows[0].Counts.Register {
t.Fatalf("平台、名称及指标筛选须共同约束列表与全量合计: %+v %v", data, err)
}
}
func TestAnalyticsEffectiveCountIncludesQualifiedDescendants(t *testing.T) {
db, repo, scope, filter := prepareAnalyticsData(t)
qualified := time.Date(2026, 11, 2, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).Unix()
if err := db.Exec("UPDATE sys_business_effectiveness_state SET rule_key='audit' WHERE id=1").Error; err != nil {
t.Fatal(err)
}
for _, key := range []business.ChannelKey{{SourceDB: "lx", ID: 100}, {SourceDB: "lx", ID: 101}, {SourceDB: "yq", ID: 100}} {
if err := db.Create(&effectivenessPO{RuleKey: "audit", SourceDB: key.SourceDB, PromoteID: key.ID, QualifiedAt: &qualified, WindowPay: "3000"}).Error; err != nil {
t.Fatal(err)
}
}
filter.PromoteName = "root-channel"
data, err := repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || len(data.Rows) != 1 || data.EffectiveCount != 2 {
t.Fatalf("合并渠道树展示仍须计入下级达标,且不能混入权限外同号渠道: %+v %v", data, err)
}
scope.IncludeSub = false
data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || data.EffectiveCount != 1 {
t.Fatalf("关闭包含下级后只统计自身: %+v %v", data, err)
}
filter.Begin = "2026-11-03"
data, err = repo.Aggregate(context.Background(), business.ChannelAnalytics, scope, filter)
if err != nil || data.EffectiveCount != 0 {
t.Fatalf("有效端必须按达标时间归属统计区间: %+v %v", data, err)
}
}
func TestAnalyticsFiltersKeepHierarchyAndSnapshotNames(t *testing.T) {
db, repo, scope, filter := prepareAnalyticsData(t)
filter.PromoteName = "child"
+35 -5
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"strings"
"time"
"gorm.io/gorm"
"kra/internal/biz/business"
@@ -18,11 +19,13 @@ type channelPO struct {
ParentID, TopPromoteID, BusierID uint64
CreateTime *int64
LastLoginTime *int64 `gorm:"column:last_login_time"`
MobilePhone string
SettlementDayPeriod *int
}
func (channelPO) TableName() string { return "tab_promote" }
func toBizChannel(p channelPO) *business.Channel {
return &business.Channel{ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.ID}, Account: p.Account, Nickname: p.Nickname, ParentName: p.ParentName, RealName: p.RealName, Level: p.PromoteLevel, RegisterType: p.RegisterType, Status: p.Status, LockStatus: p.LockStatus, ParentID: p.ParentID, TopPromoteID: p.TopPromoteID, BusierID: p.BusierID, CreatedUnix: p.CreateTime, LastLoginUnix: p.LastLoginTime, QualifiedAt: p.QualifiedAt}
return &business.Channel{ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.ID}, Account: p.Account, Nickname: p.Nickname, ParentName: p.ParentName, RealName: p.RealName, Level: p.PromoteLevel, RegisterType: p.RegisterType, Status: p.Status, LockStatus: p.LockStatus, ParentID: p.ParentID, TopPromoteID: p.TopPromoteID, BusierID: p.BusierID, CreatedUnix: p.CreateTime, LastLoginUnix: p.LastLoginTime, QualifiedAt: p.QualifiedAt, Phone: p.MobilePhone, SettlementDays: p.SettlementDayPeriod}
}
type channelRepo struct{ data Provider }
@@ -58,7 +61,7 @@ func (r *channelRepo) Bindings(ctx context.Context, scope business.UserScope, so
}
func (r *channelRepo) List(ctx context.Context, scope business.UserScope, filter business.ChannelFilter, options ...business.ListOption) (*business.ChannelPage, error) {
db := r.data.DB().WithContext(ctx)
page := &business.ChannelPage{Rows: []*business.Channel{}, Available: true}
page := &business.ChannelPage{Rows: []*business.Channel{}, Available: true, StatusCounts: &business.ChannelStatusCounts{}}
bindings := channelBindingQuery(db, scope, filter)
constrain := !scope.All || filter.BusierUserID != nil || filter.BusierAccount != ""
if constrain {
@@ -89,6 +92,26 @@ func (r *channelRepo) List(ctx context.Context, scope business.UserScope, filter
keyword := strings.ToLower(strings.TrimSpace(filter.Keyword))
q = q.Where("INSTR(LOWER(p.account),?)>0 OR INSTR(LOWER(p.nickname),?)>0 OR INSTR(LOWER(p.real_name),?)>0", keyword, keyword, keyword)
}
if filter.Phone != "" {
q = q.Where("INSTR(p.mobile_phone,?) > 0", filter.Phone)
}
if filter.SettlementDays != nil {
q = q.Where("p.settlement_day_period = ?", *filter.SettlementDays)
}
for _, period := range []struct {
column string
begin, end *time.Time
}{
{"p.create_time", filter.CreatedBegin, filter.CreatedEnd},
{"p.last_login_time", filter.LoginBegin, filter.LoginEnd},
} {
if period.begin != nil {
q = q.Where(period.column+" >= ?", period.begin.Unix())
}
if period.end != nil {
q = q.Where(period.column+" < ?", period.end.AddDate(0, 0, 1).Unix())
}
}
for _, field := range []struct {
column string
value *int
@@ -105,14 +128,21 @@ func (r *channelRepo) List(ctx context.Context, scope business.UserScope, filter
q = q.Where("e.qualified_at IS NULL")
}
}
if err := q.Count(&page.Total).Error; err != nil {
var counts struct{ Total, Normal, Locked, Terminated int64 }
if err := q.Session(&gorm.Session{}).Select("COUNT(*) AS total,COALESCE(SUM(CASE WHEN COALESCE(p.lock_status,0)=0 THEN 1 ELSE 0 END),0) AS normal,COALESCE(SUM(CASE WHEN COALESCE(p.lock_status,0) NOT IN (0,-2) THEN 1 ELSE 0 END),0) AS locked,COALESCE(SUM(CASE WHEN p.lock_status=-2 THEN 1 ELSE 0 END),0) AS terminated").Scan(&counts).Error; err != nil {
return nil, business.ErrStorage
}
page.Total = counts.Total
page.StatusCounts = &business.ChannelStatusCounts{Normal: counts.Normal, Locked: counts.Locked, Terminated: counts.Terminated}
var pos []channelPO
o := business.NewListOptions(options...)
keysQuery := q.Select("p.id,p.source_db").Order("p.create_time DESC,p.id DESC,p.source_db ASC").Offset(o.Offset).Limit(o.Limit)
order := "p.create_time DESC,p.id DESC,p.source_db ASC"
if filter.SortField == "createdAt" && filter.Ascending {
order = "p.create_time ASC,p.id ASC,p.source_db ASC"
}
keysQuery := q.Select("p.id,p.source_db").Order(order).Offset(o.Offset).Limit(o.Limit)
pageQuery := joinEffectiveness(db.Table("tab_promote p").Joins("JOIN (?) page_ids ON page_ids.id=p.id AND page_ids.source_db=p.source_db", keysQuery))
if err := pageQuery.Select("p.id,p.source_db,p.account,p.nickname,p.parent_name,p.real_name,p.promote_level,p.register_type,p.status,p.lock_status,p.parent_id,p.top_promote_id,p.busier_id,p.create_time,e.qualified_at").Order("p.create_time DESC,p.id DESC,p.source_db ASC").Scan(&pos).Error; err != nil {
if err := pageQuery.Select("p.*,e.qualified_at").Order(order).Scan(&pos).Error; err != nil {
return nil, business.ErrStorage
}
if len(pos) == 0 {
+38
View File
@@ -3,7 +3,9 @@ package business
import (
"context"
"math/big"
"strconv"
"testing"
"time"
"gorm.io/gorm"
"kra/internal/biz/business"
@@ -40,6 +42,42 @@ func prepareChannels(t *testing.T) (*gorm.DB, business.ChannelRepo) {
return db, NewChannelRepo(testData{db})
}
func TestChannelMetadataFiltersSortingAndFullStatusCounts(t *testing.T) {
db, repo := prepareChannels(t)
day := time.Date(2026, 11, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
for _, sql := range []string{"ALTER TABLE tab_promote ADD COLUMN mobile_phone TEXT", "ALTER TABLE tab_promote ADD COLUMN settlement_day_period INTEGER", "ALTER TABLE tab_promote ADD COLUMN last_login_time INTEGER"} {
if err := db.Exec(sql).Error; err != nil {
t.Fatal(err)
}
}
for i, status := range []int{0, -1, -2} {
if err := db.Exec("UPDATE tab_promote SET create_time=?,last_login_time=?,mobile_phone=?,settlement_day_period=7,lock_status=? WHERE source_db='lx' AND id=?", day.Unix()+int64(i+1)*100, day.Unix()+1000, "1381234001"+strconv.Itoa(i), status, 100+i).Error; err != nil {
t.Fatal(err)
}
}
scope := business.UserScope{SelfID: 9, OnlySelf: true}
seven := 7
filter := business.ChannelFilter{Phone: "4001", SettlementDays: &seven, CreatedBegin: &day, CreatedEnd: &day, LoginBegin: &day, LoginEnd: &day, SortField: "createdAt", Ascending: true}
page, err := repo.List(context.Background(), scope, filter, business.ListLimit(1))
if err != nil || page.Total != 3 || len(page.Rows) != 1 || page.Rows[0].ID != 100 || page.Rows[0].Phone != "13812340010" || page.Rows[0].SettlementDays == nil || *page.Rows[0].SettlementDays != 7 || page.Rows[0].LastLoginUnix == nil {
t.Fatalf("筛选、排序或元数据遗漏: %+v %v", page, err)
}
if page.StatusCounts.Normal != 1 || page.StatusCounts.Locked != 1 || page.StatusCounts.Terminated != 1 {
t.Fatalf("状态汇总必须覆盖全量三条而非当前页: %+v", page.StatusCounts)
}
filter.Ascending = false
page, err = repo.List(context.Background(), scope, filter, business.ListLimit(1))
if err != nil || page.Rows[0].ID != 102 {
t.Fatalf("降序必须作用于所有页: %+v %v", page, err)
}
locked := -1
filter.LockStatus = &locked
page, err = repo.List(context.Background(), scope, filter)
if err != nil || page.Total != 1 || page.StatusCounts.Normal != 0 || page.StatusCounts.Locked != 1 || page.StatusCounts.Terminated != 0 {
t.Fatalf("筛选变化应同步刷新汇总: %+v %v", page, err)
}
}
func channelTestScope() business.UserScope {
return business.UserScope{SelfID: 8, IncludeSelf: true, DescendantPath: "0,7,8"}
}