package sys import ( "context" "fmt" "hotgo/internal/consts" "hotgo/internal/dao" "hotgo/internal/library/hgorm/handler" "hotgo/internal/model/entity" "hotgo/internal/model/input/sysin" "hotgo/internal/service" "math" "sort" "time" "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" ) type dashboardAggregateSpec struct { DimensionType string DimensionValue string MetricCode string SQL string Args []interface{} } const ( dashboardDimensionGlobal = "global" dashboardDimensionChannel = "channel" ) type dashboardMetricDefinition struct { Code string Title string ValueType string Aggregation string Description string RouteName string Sensitive bool } var dashboardMetricDefinitions = []dashboardMetricDefinition{ {Code: "member_total", Title: "用户总量", ValueType: "count", Aggregation: "latest", Description: "截至区间结束日的去重有效会员数", RouteName: ""}, {Code: "member_new", Title: "新增用户", ValueType: "count", Aggregation: "sum", Description: "区间内新增有效会员数", RouteName: ""}, {Code: "recharge_member", Title: "充值用户", ValueType: "count", Aggregation: "sum", Description: "区间内至少成功充值一次的去重会员数", RouteName: ""}, {Code: "product_total", Title: "商品总量", ValueType: "count", Aggregation: "latest", Description: "截至区间结束日的有效商品数", RouteName: ""}, {Code: "trade_order", Title: "订单量", ValueType: "count", Aggregation: "sum", Description: "区间内创建的业务订单数", RouteName: ""}, {Code: "trade_order", Title: "订单金额", ValueType: "amount", Aggregation: "sum", Description: "区间内业务订单总金额", RouteName: "", Sensitive: true}, {Code: "recharge_success", Title: "成功充值", ValueType: "amount", Aggregation: "sum", Description: "区间内已成功入账的充值金额", RouteName: "", Sensitive: true}, {Code: "recharge_pending", Title: "待处理充值", ValueType: "amount", Aggregation: "latest", Description: "截至区间结束日的待处理充值金额", RouteName: "", Sensitive: true}, {Code: "withdrawal_success", Title: "成功提现", ValueType: "amount", Aggregation: "sum", Description: "区间内已成功支付的提现金额", RouteName: "", Sensitive: true}, {Code: "withdrawal_pending", Title: "待处理提现", ValueType: "amount", Aggregation: "latest", Description: "截至区间结束日的待审核及处理中提现金额", RouteName: "", Sensitive: true}, {Code: "commission", Title: "任务佣金", ValueType: "amount", Aggregation: "sum", Description: "区间内结算的任务佣金", RouteName: "", Sensitive: true}, {Code: "referral_commission", Title: "下级返佣", ValueType: "amount", Aggregation: "sum", Description: "区间内结算的层级返佣", RouteName: "", Sensitive: true}, {Code: "interest_transfer", Title: "利息宝转入", ValueType: "amount", Aggregation: "sum", Description: "区间内转入利息宝的金额", RouteName: "", Sensitive: true}, {Code: "wallet_available", Title: "可用余额", ValueType: "amount", Aggregation: "latest", Description: "截至区间结束日的钱包可用余额", RouteName: "", Sensitive: true}, {Code: "wallet_frozen", Title: "冻结余额", ValueType: "amount", Aggregation: "latest", Description: "截至区间结束日的钱包冻结余额", RouteName: "", Sensitive: true}, } type sSysDashboard struct{} func NewSysDashboard() *sSysDashboard { return &sSysDashboard{} } func init() { service.RegisterSysDashboard(NewSysDashboard()) } // RefreshDaily rebuilds recent dashboard snapshots from business source tables. func (s *sSysDashboard) RefreshDaily(ctx context.Context, days int) error { if days < 1 { days = 1 } if days > 90 { days = 90 } return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { for offset := days - 1; offset >= 0; offset-- { day := time.Now().AddDate(0, 0, -offset) start := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, day.Location()) end := start.AddDate(0, 0, 1) if err := s.refreshDashboardDay(ctx, start, end); err != nil { return err } } return nil }) } func (s *sSysDashboard) refreshDashboardDay(ctx context.Context, start, end time.Time) error { dc := dao.DashboardDailyStat.Columns() if _, err := dao.DashboardDailyStat.Ctx(ctx).Where(dc.StatDate, start.Format("2006-01-02")).Delete(); err != nil { return gerror.Wrap(err, "清理看板日汇总失败") } mc, pc, oc := dao.AppMember.Columns(), dao.Product.Columns(), dao.TradeOrder.Columns() rc, wc, lc := dao.RechargeOrder.Columns(), dao.WithdrawalOrder.Columns(), dao.WalletLedger.Columns() wac, cc := dao.WalletAccount.Columns(), dao.PaymentChannel.Columns() specs := []dashboardAggregateSpec{ {dashboardDimensionGlobal, "all", "member_total", fmt.Sprintf("SELECT %s tenant_id,COUNT(*) metric_count,0 metric_amount FROM %s WHERE %s IS NULL AND %s=? AND %s=? AND %s=? AND %s=? AND %s=? AND %s=? AND %s=? AND %s=? AND %s=? AND r.%s=? AND w.%s 3600 { res.Warning = "看板汇总数据延迟超过1小时,请检查汇总任务。" } } return } func normalizeDashboardRange(in *sysin.DashboardSummaryInp) (time.Time, time.Time) { now := time.Now() start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) end := start if in.StartDate != nil && in.EndDate != nil { start = time.Date(in.StartDate.Year(), time.Month(in.StartDate.Month()), in.StartDate.Day(), 0, 0, 0, 0, now.Location()) end = time.Date(in.EndDate.Year(), time.Month(in.EndDate.Month()), in.EndDate.Day(), 0, 0, 0, 0, now.Location()) } return start, end } func (s *sSysDashboard) queryRows(ctx context.Context, start, end time.Time) (rows []*entity.DashboardDailyStat, err error) { columns := dao.DashboardDailyStat.Columns() mod := handler.Model(dao.DashboardDailyStat.Ctx(ctx), &handler.Option{FilterTenant: true}). WhereBetween(columns.StatDate, start.Format("2006-01-02"), end.Format("2006-01-02")). WhereIn(columns.DimensionType, []string{dashboardDimensionGlobal, dashboardDimensionChannel}). OrderAsc(columns.StatDate).OrderAsc(columns.Id) if err = mod.Scan(&rows); err != nil { return nil, gerror.Wrap(err, "获取看板汇总数据失败") } return } func aggregateDashboardMetric(rows []*entity.DashboardDailyStat, definition dashboardMetricDefinition) int64 { var value int64 var latestDate string for _, row := range rows { if row.DimensionType != dashboardDimensionGlobal || row.MetricCode != definition.Code { continue } rowValue := row.MetricCount if definition.ValueType == "amount" { rowValue = row.MetricAmount } date := "" if row.StatDate != nil { date = row.StatDate.Format("Y-m-d") } if definition.Aggregation == "latest" { if date >= latestDate { latestDate = date value = rowValue } } else { value += rowValue } } return value } func dashboardChangeRate(value, previous int64) float64 { if previous == 0 { if value == 0 { return 0 } return 100 } return math.Round((float64(value-previous)/float64(previous))*10000) / 100 } func (s *sSysDashboard) buildTrend(start, end time.Time, rows []*entity.DashboardDailyStat) []*sysin.DashboardTrendPointModel { points := make(map[string]*sysin.DashboardTrendPointModel) for day := start; !day.After(end); day = day.AddDate(0, 0, 1) { date := day.Format("2006-01-02") points[date] = &sysin.DashboardTrendPointModel{Date: date} } for _, row := range rows { if row.DimensionType != dashboardDimensionGlobal || row.StatDate == nil { continue } point := points[row.StatDate.Format("Y-m-d")] if point == nil { continue } switch row.MetricCode { case "trade_order": point.OrderAmount += row.MetricAmount case "recharge_success": point.RechargeAmount += row.MetricAmount case "withdrawal_success": point.WithdrawalAmount += row.MetricAmount case "commission": point.CommissionAmount += row.MetricAmount } } dates := make([]string, 0, len(points)) for date := range points { dates = append(dates, date) } sort.Strings(dates) result := make([]*sysin.DashboardTrendPointModel, 0, len(dates)) for _, date := range dates { result = append(result, points[date]) } return result } func (s *sSysDashboard) buildChannels(currentRows, previousRows []*entity.DashboardDailyStat, canViewAmounts bool) []*sysin.DashboardChannelModel { labels := map[string]string{"wechat": "微信", "alipay": "支付宝", "bank": "银行卡"} channels := make(map[string]*sysin.DashboardChannelModel, len(labels)) for code, label := range labels { channels[code] = &sysin.DashboardChannelModel{Channel: code, Label: label, Masked: !canViewAmounts} } fillChannelSummary(channels, currentRows, labels, false, !canViewAmounts) fillChannelSummary(channels, previousRows, labels, true, !canViewAmounts) codes := make([]string, 0, len(channels)) for code := range channels { codes = append(codes, code) } sort.Strings(codes) result := make([]*sysin.DashboardChannelModel, 0, len(codes)) for _, code := range codes { if channels[code].Masked { channels[code].RechargeAmount = 0 channels[code].WithdrawalAmount = 0 channels[code].PreviousRechargeAmount = 0 channels[code].PreviousWithdrawalAmount = 0 } result = append(result, channels[code]) } return result } func fillChannelSummary(channels map[string]*sysin.DashboardChannelModel, rows []*entity.DashboardDailyStat, labels map[string]string, previous, masked bool) { for _, row := range rows { if row.DimensionType != dashboardDimensionChannel { continue } channel := channels[row.DimensionValue] if channel == nil { label := labels[row.DimensionValue] if label == "" { label = row.DimensionValue } channel = &sysin.DashboardChannelModel{Channel: row.DimensionValue, Label: label, Masked: masked} channels[row.DimensionValue] = channel } switch row.MetricCode { case "recharge_success": if previous { channel.PreviousRechargeCount += row.MetricCount channel.PreviousRechargeAmount += row.MetricAmount } else { channel.RechargeCount += row.MetricCount channel.RechargeAmount += row.MetricAmount } case "withdrawal_success": if previous { channel.PreviousWithdrawalCount += row.MetricCount channel.PreviousWithdrawalAmount += row.MetricAmount } else { channel.WithdrawalCount += row.MetricCount channel.WithdrawalAmount += row.MetricAmount } } } } func dashboardDataStatus(rows []*entity.DashboardDailyStat) (bool, *gtime.Time) { var latest *gtime.Time for _, row := range rows { if row.CalculatedAt != nil && (latest == nil || row.CalculatedAt.After(latest)) { latest = row.CalculatedAt } } return len(rows) > 0, latest }