2026-07-03 22:42:48 +08:00

1392 lines
43 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package databi
import (
"context"
"fmt"
"log/slog"
"sort"
"strconv"
"strings"
"sync"
"time"
"hyapp-admin-server/internal/appctx"
"hyapp-admin-server/internal/integration/userclient"
"hyapp-admin-server/internal/model"
"hyapp-admin-server/internal/modules/appregistry"
"hyapp-admin-server/internal/modules/dashboard"
"hyapp-admin-server/internal/modules/payment"
"hyapp-admin-server/internal/repository"
)
const (
appKindHyapp = "hyapp"
appKindLegacy = "legacy"
overviewConcurrency = 4
)
// LegacyRegionCatalogSource 由 payment.MongoMoneyRegionSource 实现;
// 社交 BI 与财务范围共用同一套 legacy 区域目录与合成 region_id 口径。
type LegacyRegionCatalogSource interface {
RegionCatalog(ctx context.Context) ([]payment.RegionCatalogEntry, error)
}
// LegacyAppDescriptor 描述一个外接 AppYumi/Aslan来源于 dashboard 外部源配置。
type LegacyAppDescriptor struct {
AppCode string
AppName string
}
type Service struct {
store *repository.Store
dashboards *dashboard.DashboardService
apps *appregistry.Service
user userclient.Client
legacyApps []LegacyAppDescriptor
legacyRegionCatalogs []LegacyRegionCatalogSource
overviewMu sync.Mutex
overviewCalls map[string]*overviewCall
}
// overviewCall 合并同参数的统计查询BI 页面的 overview 与 kpi 会并发请求同一窗口,
// legacy Mongo 聚合昂贵同窗口只算一次并短暂缓存BI 数据分钟级新鲜度足够)。
type overviewCall struct {
done chan struct{}
data map[string]any
err error
expiresAt time.Time
}
const overviewCacheTTL = 45 * time.Second
func (s *Service) statisticsOverview(ctx context.Context, query dashboard.StatisticsQuery) (map[string]any, error) {
key := fmt.Sprintf("%s|%s|%d|%d|%d", appctx.Normalize(query.AppCode), strings.TrimSpace(query.StatTZ), query.StartMS, query.EndMS, query.RegionID)
s.overviewMu.Lock()
if s.overviewCalls == nil {
s.overviewCalls = map[string]*overviewCall{}
}
if call, ok := s.overviewCalls[key]; ok {
expired := !call.expiresAt.IsZero() && time.Now().After(call.expiresAt)
if !expired {
s.overviewMu.Unlock()
<-call.done
return call.data, call.err
}
delete(s.overviewCalls, key)
}
call := &overviewCall{done: make(chan struct{})}
s.overviewCalls[key] = call
s.overviewMu.Unlock()
// 共享结果不随首个请求方取消:其余等待者可能仍需要它。
call.data, call.err = s.dashboards.StatisticsOverview(context.WithoutCancel(ctx), query)
s.overviewMu.Lock()
if call.err != nil {
// 失败不缓存,后续请求立即重试。
delete(s.overviewCalls, key)
} else {
call.expiresAt = time.Now().Add(overviewCacheTTL)
}
s.overviewMu.Unlock()
close(call.done)
return call.data, call.err
}
type ServiceOption func(*Service)
func WithLegacyApps(apps ...LegacyAppDescriptor) ServiceOption {
return func(s *Service) {
for _, app := range apps {
appCode := appctx.Normalize(app.AppCode)
if appCode == "" {
continue
}
app.AppCode = appCode
s.legacyApps = append(s.legacyApps, app)
}
}
}
func WithLegacyRegionCatalogs(catalogs ...LegacyRegionCatalogSource) ServiceOption {
return func(s *Service) {
for _, catalog := range catalogs {
if catalog != nil {
s.legacyRegionCatalogs = append(s.legacyRegionCatalogs, catalog)
}
}
}
}
func NewService(store *repository.Store, dashboards *dashboard.DashboardService, apps *appregistry.Service, user userclient.Client, options ...ServiceOption) *Service {
service := &Service{
store: store,
dashboards: dashboards,
apps: apps,
user: user,
}
for _, option := range options {
if option != nil {
option(service)
}
}
return service
}
type AppInfo struct {
AppCode string `json:"app_code"`
AppName string `json:"app_name"`
Kind string `json:"kind"`
}
type RegionInfo struct {
AppCode string `json:"app_code"`
RegionID int64 `json:"region_id"`
RegionCode string `json:"region_code"`
RegionName string `json:"region_name"`
Countries []string `json:"countries"`
}
type ScopeInfo struct {
AppCode string `json:"app_code"`
RegionID int64 `json:"region_id"`
}
type OperatorInfo struct {
UserID uint `json:"user_id"`
Name string `json:"name"`
Account string `json:"account"`
Team string `json:"team"`
TeamID *uint `json:"team_id"`
Status string `json:"status"`
Scopes []ScopeInfo `json:"scopes"`
}
type AccessInfo struct {
All bool `json:"all"`
Scopes []ScopeInfo `json:"scopes"`
}
type MasterData struct {
Apps []AppInfo `json:"apps"`
Regions []RegionInfo `json:"regions"`
Operators []OperatorInfo `json:"operators"`
Access AccessInfo `json:"access"`
Permissions map[string]bool `json:"permissions,omitempty"`
}
// Master 返回社交 BI 的筛选主数据:当前用户可见的 App、区域目录、运营人员及各自负责范围。
func (s *Service) Master(ctx context.Context, access repository.MoneyAccess, actorUserID uint, viewAll bool) (*MasterData, error) {
apps, err := s.listAllowedApps(ctx, nil, access)
if err != nil {
return nil, err
}
catalog, err := s.regionCatalog(ctx, apps)
if err != nil {
return nil, err
}
regions := []RegionInfo{}
for _, app := range apps {
allowAll, allowedIDs := allowedRegions(access, app.AppCode)
allowedSet := int64Set(normalizeRegionIDs(catalog[app.AppCode], allowedIDs))
for _, region := range catalog[app.AppCode] {
if !allowAll {
if _, ok := allowedSet[region.RegionID]; !ok {
continue
}
}
regions = append(regions, region)
}
}
operators, err := s.listOperators(actorUserID, viewAll)
if err != nil {
return nil, err
}
return &MasterData{
Apps: apps,
Regions: regions,
Operators: operators,
Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
}, nil
}
func (s *Service) listOperators(actorUserID uint, viewAll bool) ([]OperatorInfo, error) {
users, scopesByUser, err := s.store.ListMoneyScopeOperators()
if err != nil {
return nil, err
}
operators := make([]OperatorInfo, 0, len(users))
for _, user := range users {
if !viewAll && user.ID != actorUserID {
continue
}
scopes := scopesByUser[user.ID]
operators = append(operators, OperatorInfo{
UserID: user.ID,
Name: user.Name,
Account: user.Username,
Team: user.Team,
TeamID: user.TeamID,
Status: user.Status,
Scopes: scopeInfos(scopes),
})
}
return operators, nil
}
type OverviewQuery struct {
StatTZ string
StartMS int64
EndMS int64
AppCodes []string
RegionID int64
}
type AppOverview struct {
AppCode string `json:"app_code"`
AppName string `json:"app_name"`
Kind string `json:"kind"`
Restricted bool `json:"restricted"`
AllowedRegionIDs []int64 `json:"allowed_region_ids,omitempty"`
Total map[string]any `json:"total"`
DailySeries []map[string]any `json:"daily_series"`
RegionBreakdown []map[string]any `json:"region_breakdown"`
DailyRegionBreakdown []map[string]any `json:"daily_region_breakdown"`
CountryBreakdown []map[string]any `json:"country_breakdown"`
DailyCountryBreakdown []map[string]any `json:"daily_country_breakdown"`
ReportMetricSources any `json:"report_metric_sources,omitempty"`
UpdatedAtMS any `json:"updated_at_ms,omitempty"`
Error string `json:"error,omitempty"`
}
type OverviewResult struct {
Apps []AppOverview `json:"apps"`
Access AccessInfo `json:"access"`
}
// Overview 服务端并发聚合各 App 的统计总览hyapp App 走 statistics-service
// legacy App 走 dashboard 外部源;国家行按区域目录卷积出大区行,并按用户数据范围裁剪。
func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, query OverviewQuery) (*OverviewResult, error) {
apps, err := s.listAllowedApps(ctx, query.AppCodes, access)
if err != nil {
return nil, err
}
catalog, err := s.regionCatalog(ctx, apps)
if err != nil {
return nil, err
}
results := make([]AppOverview, len(apps))
var wg sync.WaitGroup
semaphore := make(chan struct{}, overviewConcurrency)
for index, app := range apps {
wg.Add(1)
go func(index int, app AppInfo) {
defer wg.Done()
semaphore <- struct{}{}
defer func() { <-semaphore }()
results[index] = s.appOverview(ctx, app, catalog[app.AppCode], access, query)
}(index, app)
}
wg.Wait()
return &OverviewResult{
Apps: results,
Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
}, nil
}
func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query OverviewQuery) AppOverview {
out := AppOverview{
AppCode: app.AppCode,
AppName: app.AppName,
Kind: app.Kind,
Total: map[string]any{},
DailySeries: []map[string]any{},
RegionBreakdown: []map[string]any{},
DailyRegionBreakdown: []map[string]any{},
CountryBreakdown: []map[string]any{},
DailyCountryBreakdown: []map[string]any{},
}
allowAll, allowedIDs := allowedRegions(access, app.AppCode)
allowedIDs = normalizeRegionIDs(regions, allowedIDs)
if !allowAll && len(allowedIDs) == 0 {
out.Error = "没有该 App 的数据权限"
return out
}
if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) {
out.Error = "没有该区域的数据权限"
return out
}
if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) {
out.Error = "该 App 未配置此区域"
return out
}
overview, err := s.statisticsOverview(ctx, dashboard.StatisticsQuery{
AppCode: app.AppCode,
StatTZ: query.StatTZ,
StartMS: query.StartMS,
EndMS: query.EndMS,
SeriesStartMS: query.StartMS,
SeriesEndMS: query.EndMS,
RegionID: query.RegionID,
})
if err != nil {
out.Error = fmt.Sprintf("获取统计数据失败: %v", err)
return out
}
countryRows := anyRowSlice(overview["country_breakdown"])
dailyCountryRows := anyRowSlice(overview["daily_country_breakdown"])
resolve := regionResolver(regions)
restricted := !allowAll && query.RegionID == 0
if restricted {
allowedSet := int64Set(allowedIDs)
countryRows = filterRowsByRegion(countryRows, resolve, allowedSet)
dailyCountryRows = filterRowsByRegion(dailyCountryRows, resolve, allowedSet)
out.Restricted = true
out.AllowedRegionIDs = allowedIDs
totalAcc := newMetricAccumulator()
for _, row := range countryRows {
totalAcc.addRow(row)
}
out.Total = totalAcc.toMap()
out.DailySeries = rollupDaily(dailyCountryRows)
} else {
out.Total = topLevelMetrics(overview)
out.DailySeries = anyRowSlice(overview["daily_series"])
}
out.RegionBreakdown = rollupByRegion(countryRows, resolve)
out.DailyRegionBreakdown = rollupDailyByRegion(dailyCountryRows, resolve)
out.CountryBreakdown = countryRows
out.DailyCountryBreakdown = dailyCountryRows
out.ReportMetricSources = overview["report_metric_sources"]
out.UpdatedAtMS = overview["updated_at_ms"]
return out
}
type KpiQuery struct {
StatTZ string
StartMS int64
EndMS int64
PeriodMonth string
AppCodes []string
OperatorUserID uint
ActorUserID uint
ViewAll bool
}
type KpiRow struct {
OperatorUserID uint `json:"operator_user_id"`
OperatorName string `json:"operator_name"`
OperatorAccount string `json:"operator_account"`
Team string `json:"team"`
TeamID *uint `json:"team_id"`
AppCode string `json:"app_code"`
AppName string `json:"app_name"`
RegionID int64 `json:"region_id"`
RegionCode string `json:"region_code"`
RegionName string `json:"region_name"`
RangeRechargeUSDMinor *int64 `json:"range_recharge_usd_minor"`
RangeNewUsers *int64 `json:"range_new_users"`
RangeActiveUsers *int64 `json:"range_active_users"`
MonthRechargeUSDMinor *int64 `json:"month_recharge_usd_minor"`
MonthTargetUSDMinor int64 `json:"month_target_usd_minor"`
DailyTargetUSDMinor int64 `json:"daily_target_usd_minor"`
MonthAttainmentRate *float64 `json:"month_attainment_rate"`
DataError string `json:"data_error,omitempty"`
}
// KpiOperatorAppRow 是 人×App 级小计:该运营在此 App 下全部负责区域的合计,
// 目标取自 admin_databi_kpi_targets 的 region_id=0 行(该运营的整 App 目标)。
type KpiOperatorAppRow struct {
OperatorUserID uint `json:"operator_user_id"`
OperatorName string `json:"operator_name"`
OperatorAccount string `json:"operator_account"`
Team string `json:"team"`
TeamID *uint `json:"team_id"`
AppCode string `json:"app_code"`
AppName string `json:"app_name"`
RegionCount int `json:"region_count"`
RegionNames string `json:"region_names"`
WholeAppScope bool `json:"whole_app_scope"`
RangeRechargeUSDMinor *int64 `json:"range_recharge_usd_minor"`
RangeNewUsers *int64 `json:"range_new_users"`
MonthRechargeUSDMinor *int64 `json:"month_recharge_usd_minor"`
MonthTargetUSDMinor int64 `json:"month_target_usd_minor"`
DailyTargetUSDMinor int64 `json:"daily_target_usd_minor"`
MonthAttainmentRate *float64 `json:"month_attainment_rate"`
DataError string `json:"data_error,omitempty"`
}
// KpiAppRow 是 App 级总目标行:实际值为整 App 充值(不分运营的全员共同目标)。
type KpiAppRow struct {
AppCode string `json:"app_code"`
AppName string `json:"app_name"`
OperatorCount int `json:"operator_count"`
RangeRechargeUSDMinor *int64 `json:"range_recharge_usd_minor"`
MonthRechargeUSDMinor *int64 `json:"month_recharge_usd_minor"`
MonthTargetUSDMinor int64 `json:"month_target_usd_minor"`
DailyTargetUSDMinor int64 `json:"daily_target_usd_minor"`
MonthAttainmentRate *float64 `json:"month_attainment_rate"`
DataError string `json:"data_error,omitempty"`
}
type KpiSummary struct {
RangeRechargeUSDMinor int64 `json:"range_recharge_usd_minor"`
MonthRechargeUSDMinor int64 `json:"month_recharge_usd_minor"`
MonthTargetUSDMinor int64 `json:"month_target_usd_minor"`
MonthAttainmentRate *float64 `json:"month_attainment_rate"`
OperatorCount int `json:"operator_count"`
}
type KpiResult struct {
PeriodMonth string `json:"period_month"`
MonthStartMS int64 `json:"month_start_ms"`
MonthEndMS int64 `json:"month_end_ms"`
Items []KpiRow `json:"items"`
OperatorAppRows []KpiOperatorAppRow `json:"operator_app_rows"`
AppRows []KpiAppRow `json:"app_rows"`
Summary KpiSummary `json:"summary"`
AppErrors map[string]any `json:"app_errors,omitempty"`
}
// Kpi 按运营人员负责范围App×区域聚合充值/新增,并对齐月度 KPI 目标计算达成率。
// 无 view-all 权限时只返回本人行,即"我的 KPI"。
func (s *Service) Kpi(ctx context.Context, access repository.MoneyAccess, query KpiQuery) (*KpiResult, error) {
users, scopesByUser, err := s.store.ListMoneyScopeOperators()
if err != nil {
return nil, err
}
operators := make([]model.User, 0, len(users))
for _, user := range users {
if !query.ViewAll && user.ID != query.ActorUserID {
continue
}
if query.OperatorUserID > 0 && user.ID != query.OperatorUserID {
continue
}
operators = append(operators, user)
}
requestedApps := map[string]struct{}{}
for _, appCode := range query.AppCodes {
appCode = appctx.Normalize(appCode)
if appCode != "" {
requestedApps[appCode] = struct{}{}
}
}
appCodeSet := map[string]struct{}{}
for _, user := range operators {
for _, scope := range scopesByUser[user.ID] {
appCode := appctx.Normalize(scope.AppCode)
if appCode == "" {
continue
}
if len(requestedApps) > 0 {
if _, ok := requestedApps[appCode]; !ok {
continue
}
}
appCodeSet[appCode] = struct{}{}
}
}
apps, err := s.listApps(ctx)
if err != nil {
return nil, err
}
appsByCode := map[string]AppInfo{}
relevantApps := []AppInfo{}
for _, app := range apps {
appsByCode[app.AppCode] = app
if _, ok := appCodeSet[app.AppCode]; ok {
relevantApps = append(relevantApps, app)
}
}
catalog, err := s.regionCatalog(ctx, relevantApps)
if err != nil {
return nil, err
}
location, err := time.LoadLocation(firstNonEmptyString(query.StatTZ, "Asia/Shanghai"))
if err != nil {
return nil, fmt.Errorf("load kpi timezone %s: %w", query.StatTZ, err)
}
periodMonth, monthStart, monthEnd, err := resolveKpiMonth(query.PeriodMonth, query.EndMS, location)
if err != nil {
return nil, err
}
dataByApp := make(map[string]*appKpiData, len(relevantApps))
var mu sync.Mutex
var wg sync.WaitGroup
semaphore := make(chan struct{}, overviewConcurrency)
for _, app := range relevantApps {
wg.Add(1)
go func(app AppInfo) {
defer wg.Done()
semaphore <- struct{}{}
defer func() { <-semaphore }()
data := &appKpiData{}
resolve := regionResolver(catalog[app.AppCode])
// 展示区间与当月两个窗口互不依赖,并行取;同参数窗口会被 statisticsOverview 合并/命中缓存。
var windowWG sync.WaitGroup
windowWG.Add(2)
var rangeErr, monthErr error
go func() {
defer windowWG.Done()
total, region, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, query.StartMS, query.EndMS)
if err != nil {
rangeErr = err
return
}
data.rangeTotal, data.rangeRegion = total, region
}()
go func() {
defer windowWG.Done()
total, region, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, monthStart.UnixMilli(), monthEnd.UnixMilli())
if err != nil {
monthErr = err
return
}
data.monthTotal, data.monthRegion = total, region
}()
windowWG.Wait()
if rangeErr != nil {
data.err = rangeErr
} else if monthErr != nil {
data.err = monthErr
}
mu.Lock()
dataByApp[app.AppCode] = data
mu.Unlock()
}(app)
}
wg.Wait()
operatorIDs := make([]uint, 0, len(operators))
for _, user := range operators {
operatorIDs = append(operatorIDs, user.ID)
}
targets, err := s.store.ListDatabiKpiTargets([]string{periodMonth}, operatorIDs)
if err != nil {
return nil, err
}
targetByKey := map[string]model.DatabiKpiTarget{}
for _, target := range targets {
targetByKey[kpiTargetKey(target.UserID, target.AppCode, target.RegionID)] = target
}
rows := []KpiRow{}
appErrors := map[string]any{}
for _, user := range operators {
for _, scope := range scopesByUser[user.ID] {
appCode := appctx.Normalize(scope.AppCode)
if len(requestedApps) > 0 {
if _, ok := requestedApps[appCode]; !ok {
continue
}
}
app, ok := appsByCode[appCode]
if !ok {
app = AppInfo{AppCode: appCode, AppName: appCode, Kind: appKindLegacy}
}
// 历史 scope 里的 legacy 区域 ID 可能被前端 float64 圆整过,先还原成目录精确 ID 再做展示与取数。
regionID := normalizeRegionID(catalog[appCode], scope.RegionID)
row := KpiRow{
OperatorUserID: user.ID,
OperatorName: user.Name,
OperatorAccount: user.Username,
Team: user.Team,
TeamID: user.TeamID,
AppCode: appCode,
AppName: app.AppName,
RegionID: regionID,
}
row.RegionCode, row.RegionName = regionDisplay(catalog[appCode], regionID)
if target, ok := lookupKpiTarget(targetByKey, user.ID, appCode, scope.RegionID, regionID); ok {
row.MonthTargetUSDMinor = target.TargetUSDMinor
row.DailyTargetUSDMinor = target.DailyTargetUSDMinor
}
data := dataByApp[appCode]
if data == nil {
row.DataError = "该 App 未接入统计数据"
} else if data.err != nil {
row.DataError = fmt.Sprintf("获取统计数据失败: %v", data.err)
appErrors[appCode] = row.DataError
} else {
rangeMetrics := scopeMetrics(data.rangeTotal, data.rangeRegion, regionID)
monthMetrics := scopeMetrics(data.monthTotal, data.monthRegion, regionID)
row.RangeRechargeUSDMinor = metricPointer(rangeMetrics, "recharge_usd_minor")
row.RangeNewUsers = metricPointer(rangeMetrics, "new_users")
row.RangeActiveUsers = metricPointer(rangeMetrics, "active_users")
row.MonthRechargeUSDMinor = metricPointer(monthMetrics, "recharge_usd_minor")
if row.MonthTargetUSDMinor > 0 && row.MonthRechargeUSDMinor != nil {
rate := safeRatio(*row.MonthRechargeUSDMinor, row.MonthTargetUSDMinor)
row.MonthAttainmentRate = &rate
}
}
rows = append(rows, row)
}
}
sort.Slice(rows, func(i, j int) bool {
left, right := rows[i], rows[j]
leftMonth := int64(0)
if left.MonthRechargeUSDMinor != nil {
leftMonth = *left.MonthRechargeUSDMinor
}
rightMonth := int64(0)
if right.MonthRechargeUSDMinor != nil {
rightMonth = *right.MonthRechargeUSDMinor
}
if leftMonth != rightMonth {
return leftMonth > rightMonth
}
if left.OperatorUserID != right.OperatorUserID {
return left.OperatorUserID < right.OperatorUserID
}
if left.AppCode != right.AppCode {
return left.AppCode < right.AppCode
}
return left.RegionID < right.RegionID
})
appTargets, err := s.store.ListDatabiAppKpiTargets([]string{periodMonth})
if err != nil {
return nil, err
}
appTargetByCode := map[string]model.DatabiAppKpiTarget{}
for _, target := range appTargets {
appTargetByCode[appctx.Normalize(target.AppCode)] = target
}
operatorAppRows := buildOperatorAppRows(rows, targetByKey)
appRows := buildAppRows(relevantApps, dataByApp, appTargetByCode, rows)
summary := buildKpiSummary(rows, operatorAppRows, appRows)
result := &KpiResult{
PeriodMonth: periodMonth,
MonthStartMS: monthStart.UnixMilli(),
MonthEndMS: monthEnd.UnixMilli(),
Items: rows,
OperatorAppRows: operatorAppRows,
AppRows: appRows,
Summary: summary,
}
if len(appErrors) > 0 {
result.AppErrors = appErrors
}
return result, nil
}
type appKpiData struct {
rangeTotal map[string]any
rangeRegion map[int64]map[string]any
monthTotal map[string]any
monthRegion map[int64]map[string]any
err error
}
func nullableAdd(sum *int64, value *int64) *int64 {
if value == nil {
return sum
}
if sum == nil {
out := *value
return &out
}
out := *sum + *value
return &out
}
// buildOperatorAppRows 把 人×App×区域 明细行归并成 人×App 小计:
// 该运营有整 App 范围region 0时直接取那一行其值即整 App 口径),否则各区域可空求和。
func buildOperatorAppRows(rows []KpiRow, targetByKey map[string]model.DatabiKpiTarget) []KpiOperatorAppRow {
type group struct {
row KpiOperatorAppRow
regionNames []string
wholeRow *KpiRow
}
groups := map[string]*group{}
order := []string{}
for index := range rows {
row := rows[index]
key := strconv.FormatUint(uint64(row.OperatorUserID), 10) + "|" + row.AppCode
item := groups[key]
if item == nil {
item = &group{row: KpiOperatorAppRow{
AppCode: row.AppCode,
AppName: row.AppName,
OperatorAccount: row.OperatorAccount,
OperatorName: row.OperatorName,
OperatorUserID: row.OperatorUserID,
Team: row.Team,
TeamID: row.TeamID,
}}
groups[key] = item
order = append(order, key)
}
item.row.RegionCount++
item.regionNames = append(item.regionNames, row.RegionName)
if row.DataError != "" && item.row.DataError == "" {
item.row.DataError = row.DataError
}
if row.RegionID == 0 {
item.row.WholeAppScope = true
item.wholeRow = &rows[index]
continue
}
item.row.RangeRechargeUSDMinor = nullableAdd(item.row.RangeRechargeUSDMinor, row.RangeRechargeUSDMinor)
item.row.RangeNewUsers = nullableAdd(item.row.RangeNewUsers, row.RangeNewUsers)
item.row.MonthRechargeUSDMinor = nullableAdd(item.row.MonthRechargeUSDMinor, row.MonthRechargeUSDMinor)
}
out := make([]KpiOperatorAppRow, 0, len(order))
for _, key := range order {
item := groups[key]
if item.wholeRow != nil {
// 整 App 范围行的取数已是整 App 口径,区域行是它的子集,避免相加重复计数。
item.row.RangeRechargeUSDMinor = item.wholeRow.RangeRechargeUSDMinor
item.row.RangeNewUsers = item.wholeRow.RangeNewUsers
item.row.MonthRechargeUSDMinor = item.wholeRow.MonthRechargeUSDMinor
}
item.row.RegionNames = joinRegionNames(item.regionNames)
if target, ok := targetByKey[kpiTargetKey(item.row.OperatorUserID, item.row.AppCode, 0)]; ok {
item.row.MonthTargetUSDMinor = target.TargetUSDMinor
item.row.DailyTargetUSDMinor = target.DailyTargetUSDMinor
} else if item.wholeRow != nil {
// 整 App 范围的运营其明细行region 0目标与 人×App 目标同键,直接沿用。
item.row.MonthTargetUSDMinor = item.wholeRow.MonthTargetUSDMinor
item.row.DailyTargetUSDMinor = item.wholeRow.DailyTargetUSDMinor
}
if item.row.MonthTargetUSDMinor > 0 && item.row.MonthRechargeUSDMinor != nil {
rate := safeRatio(*item.row.MonthRechargeUSDMinor, item.row.MonthTargetUSDMinor)
item.row.MonthAttainmentRate = &rate
}
out = append(out, item.row)
}
sort.Slice(out, func(i, j int) bool {
left := int64(0)
if out[i].MonthRechargeUSDMinor != nil {
left = *out[i].MonthRechargeUSDMinor
}
right := int64(0)
if out[j].MonthRechargeUSDMinor != nil {
right = *out[j].MonthRechargeUSDMinor
}
if left != right {
return left > right
}
if out[i].OperatorUserID != out[j].OperatorUserID {
return out[i].OperatorUserID < out[j].OperatorUserID
}
return out[i].AppCode < out[j].AppCode
})
return out
}
func joinRegionNames(names []string) string {
if len(names) <= 3 {
return strings.Join(names, " / ")
}
return strings.Join(names[:3], " / ") + fmt.Sprintf(" 等%d个", len(names))
}
// buildAppRows 生成 App 级总目标行:实际值 = 整 App 充值(与运营/区域无关的共同目标口径)。
func buildAppRows(apps []AppInfo, dataByApp map[string]*appKpiData, appTargetByCode map[string]model.DatabiAppKpiTarget, rows []KpiRow) []KpiAppRow {
operatorsByApp := map[string]map[uint]struct{}{}
for _, row := range rows {
set := operatorsByApp[row.AppCode]
if set == nil {
set = map[uint]struct{}{}
operatorsByApp[row.AppCode] = set
}
set[row.OperatorUserID] = struct{}{}
}
out := make([]KpiAppRow, 0, len(apps))
for _, app := range apps {
row := KpiAppRow{
AppCode: app.AppCode,
AppName: app.AppName,
OperatorCount: len(operatorsByApp[app.AppCode]),
}
if target, ok := appTargetByCode[app.AppCode]; ok {
row.MonthTargetUSDMinor = target.TargetUSDMinor
row.DailyTargetUSDMinor = target.DailyTargetUSDMinor
}
data := dataByApp[app.AppCode]
if data == nil {
row.DataError = "该 App 未接入统计数据"
} else if data.err != nil {
row.DataError = fmt.Sprintf("获取统计数据失败: %v", data.err)
} else {
row.RangeRechargeUSDMinor = metricPointer(data.rangeTotal, "recharge_usd_minor")
row.MonthRechargeUSDMinor = metricPointer(data.monthTotal, "recharge_usd_minor")
}
if row.MonthTargetUSDMinor > 0 && row.MonthRechargeUSDMinor != nil {
rate := safeRatio(*row.MonthRechargeUSDMinor, row.MonthTargetUSDMinor)
row.MonthAttainmentRate = &rate
}
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool {
left := int64(0)
if out[i].MonthRechargeUSDMinor != nil {
left = *out[i].MonthRechargeUSDMinor
}
right := int64(0)
if out[j].MonthRechargeUSDMinor != nil {
right = *out[j].MonthRechargeUSDMinor
}
if left != right {
return left > right
}
return out[i].AppCode < out[j].AppCode
})
return out
}
// buildKpiSummary 的口径:实际充值按 App 行求和(跨运营/区域去重);
// 月目标按 App 逐个取"最权威"的一层——App 总目标 > 人×App 目标合计 > 人×区域明细目标合计。
func buildKpiSummary(rows []KpiRow, operatorAppRows []KpiOperatorAppRow, appRows []KpiAppRow) KpiSummary {
summary := KpiSummary{}
operatorSeen := map[uint]struct{}{}
regionTargetByApp := map[string]int64{}
for _, row := range rows {
operatorSeen[row.OperatorUserID] = struct{}{}
if row.RegionID != 0 {
regionTargetByApp[row.AppCode] += row.MonthTargetUSDMinor
}
}
summary.OperatorCount = len(operatorSeen)
operatorAppTargetByApp := map[string]int64{}
for _, row := range operatorAppRows {
operatorAppTargetByApp[row.AppCode] += row.MonthTargetUSDMinor
}
for _, row := range appRows {
if row.RangeRechargeUSDMinor != nil {
summary.RangeRechargeUSDMinor += *row.RangeRechargeUSDMinor
}
if row.MonthRechargeUSDMinor != nil {
summary.MonthRechargeUSDMinor += *row.MonthRechargeUSDMinor
}
summary.MonthTargetUSDMinor += pickEffectiveAppTarget(
row.MonthTargetUSDMinor,
operatorAppTargetByApp[row.AppCode],
regionTargetByApp[row.AppCode],
)
}
if summary.MonthTargetUSDMinor > 0 {
rate := safeRatio(summary.MonthRechargeUSDMinor, summary.MonthTargetUSDMinor)
summary.MonthAttainmentRate = &rate
}
return summary
}
// pickEffectiveAppTarget一个 App 的"生效月目标"取最高层已配置的那级,避免多级目标重复相加。
func pickEffectiveAppTarget(appTarget int64, operatorAppTargets int64, regionTargets int64) int64 {
if appTarget > 0 {
return appTarget
}
if operatorAppTargets > 0 {
return operatorAppTargets
}
return regionTargets
}
func (s *Service) kpiWindow(ctx context.Context, app AppInfo, resolve func(map[string]any) (RegionInfo, bool), statTZ string, startMS int64, endMS int64) (map[string]any, map[int64]map[string]any, error) {
overview, err := s.statisticsOverview(ctx, dashboard.StatisticsQuery{
AppCode: app.AppCode,
StatTZ: statTZ,
StartMS: startMS,
EndMS: endMS,
SeriesStartMS: startMS,
SeriesEndMS: endMS,
})
if err != nil {
return nil, nil, err
}
regionRows := rollupByRegion(anyRowSlice(overview["country_breakdown"]), resolve)
regionMetrics := map[int64]map[string]any{}
for _, row := range regionRows {
if regionID, ok := rowInt64(row, "region_id"); ok {
regionMetrics[regionID] = row
}
}
return topLevelMetrics(overview), regionMetrics, nil
}
func scopeMetrics(total map[string]any, regions map[int64]map[string]any, regionID int64) map[string]any {
if regionID == 0 {
return total
}
return regions[regionID]
}
func metricPointer(row map[string]any, key string) *int64 {
if row == nil {
return nil
}
value, ok := rowInt64(row, key)
if !ok {
return nil
}
out := value
return &out
}
func kpiTargetKey(userID uint, appCode string, regionID int64) string {
return strconv.FormatUint(uint64(userID), 10) + ":" + appctx.Normalize(appCode) + ":" + strconv.FormatInt(regionID, 10)
}
// lookupKpiTarget 依次用 scope 原始 ID 与归一后的精确 ID 查目标:
// 目标行是前端写入的legacy 大区可能存的是圆整值,也可能是修复后的精确值。
func lookupKpiTarget(targetByKey map[string]model.DatabiKpiTarget, userID uint, appCode string, rawRegionID int64, regionID int64) (model.DatabiKpiTarget, bool) {
if target, ok := targetByKey[kpiTargetKey(userID, appCode, rawRegionID)]; ok {
return target, true
}
if regionID != rawRegionID {
if target, ok := targetByKey[kpiTargetKey(userID, appCode, regionID)]; ok {
return target, true
}
}
if rounded := int64(float64(regionID)); rounded != rawRegionID && rounded != regionID {
if target, ok := targetByKey[kpiTargetKey(userID, appCode, rounded)]; ok {
return target, true
}
}
return model.DatabiKpiTarget{}, false
}
// resolveKpiMonth 解析目标月份并给出该月在统计时区下的起止时间;结束时间截断到明天零点,
// 避免 legacy Mongo 源为未来日期空跑逐日聚合。
func resolveKpiMonth(periodMonth string, endMS int64, location *time.Location) (string, time.Time, time.Time, error) {
now := time.Now().In(location)
reference := now
if endMS > 0 {
reference = time.UnixMilli(endMS - 1).In(location)
}
month := strings.TrimSpace(periodMonth)
if month == "" {
month = reference.Format("2006-01")
}
parsed, err := time.ParseInLocation("2006-01", month, location)
if err != nil {
return "", time.Time{}, time.Time{}, fmt.Errorf("period_month must be formatted as YYYY-MM")
}
monthStart := time.Date(parsed.Year(), parsed.Month(), 1, 0, 0, 0, 0, location)
monthEnd := monthStart.AddDate(0, 1, 0)
tomorrow := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location).AddDate(0, 0, 1)
if monthEnd.After(tomorrow) {
monthEnd = tomorrow
}
if !monthEnd.After(monthStart) {
monthEnd = monthStart.AddDate(0, 0, 1)
}
return month, monthStart, monthEnd, nil
}
func (s *Service) listApps(ctx context.Context) ([]AppInfo, error) {
out := []AppInfo{}
seen := map[string]struct{}{}
if s.apps != nil {
hyappApps, err := s.apps.ListApps(ctx)
if err != nil {
return nil, fmt.Errorf("list hyapp apps: %w", err)
}
for _, app := range hyappApps {
appCode := appctx.Normalize(app.AppCode)
if appCode == "" {
continue
}
if _, ok := seen[appCode]; ok {
continue
}
seen[appCode] = struct{}{}
out = append(out, AppInfo{
AppCode: appCode,
AppName: firstNonEmptyString(app.AppName, appCode),
Kind: appKindHyapp,
})
}
}
for _, app := range s.legacyApps {
if _, ok := seen[app.AppCode]; ok {
continue
}
seen[app.AppCode] = struct{}{}
out = append(out, AppInfo{
AppCode: app.AppCode,
AppName: firstNonEmptyString(app.AppName, app.AppCode),
Kind: appKindLegacy,
})
}
return out, nil
}
func (s *Service) listAllowedApps(ctx context.Context, requested []string, access repository.MoneyAccess) ([]AppInfo, error) {
apps, err := s.listApps(ctx)
if err != nil {
return nil, err
}
requestedSet := map[string]struct{}{}
for _, appCode := range requested {
appCode = appctx.Normalize(appCode)
if appCode != "" {
requestedSet[appCode] = struct{}{}
}
}
out := []AppInfo{}
for _, app := range apps {
if len(requestedSet) > 0 {
if _, ok := requestedSet[app.AppCode]; !ok {
continue
}
}
if !access.All {
allowAll, allowedIDs := allowedRegions(access, app.AppCode)
if !allowAll && len(allowedIDs) == 0 {
continue
}
}
out = append(out, app)
}
return out, nil
}
// regionCatalog 汇总每个 App 的区域目录hyapp App 共用 user-service 的区域主数据,
// legacy App 读各自 Mongo sys_region_config与财务范围同一套合成 region_id
// 单个目录源故障时降级返回其余目录(行仍可按行内 region_id 分组),不让整个 BI 查询失败。
func (s *Service) regionCatalog(ctx context.Context, apps []AppInfo) (map[string][]RegionInfo, error) {
out := map[string][]RegionInfo{}
hasHyapp := false
legacyNeeded := map[string]struct{}{}
for _, app := range apps {
if app.Kind == appKindHyapp {
hasHyapp = true
} else {
legacyNeeded[app.AppCode] = struct{}{}
}
}
hyappRegions := []RegionInfo{}
if hasHyapp && s.user != nil {
regions, err := s.user.ListRegions(ctx, userclient.ListRegionsRequest{Caller: "admin-server", Status: "active"})
if err != nil {
slog.Warn("databi_list_hyapp_regions_failed", "error", err)
}
for _, region := range regions {
if region == nil {
continue
}
hyappRegions = append(hyappRegions, RegionInfo{
RegionID: region.RegionID,
RegionCode: region.RegionCode,
RegionName: firstNonEmptyString(region.Name, region.RegionCode),
Countries: normalizeCountryCodes(region.Countries),
})
}
}
legacyRegions := map[string][]RegionInfo{}
if len(legacyNeeded) > 0 {
for _, catalog := range s.legacyRegionCatalogs {
entries, err := catalog.RegionCatalog(ctx)
if err != nil {
slog.Warn("databi_list_legacy_regions_failed", "error", err)
continue
}
for _, entry := range entries {
appCode := appctx.Normalize(entry.AppCode)
if _, ok := legacyNeeded[appCode]; !ok {
continue
}
legacyRegions[appCode] = append(legacyRegions[appCode], RegionInfo{
AppCode: appCode,
RegionID: entry.RegionID,
RegionCode: entry.RegionCode,
RegionName: firstNonEmptyString(entry.RegionName, entry.RegionCode),
Countries: normalizeCountryCodes(entry.Countries),
})
}
}
}
for _, app := range apps {
if app.Kind == appKindHyapp {
regions := make([]RegionInfo, 0, len(hyappRegions))
for _, region := range hyappRegions {
region.AppCode = app.AppCode
regions = append(regions, region)
}
out[app.AppCode] = regions
continue
}
out[app.AppCode] = legacyRegions[app.AppCode]
}
return out, nil
}
func allowedRegions(access repository.MoneyAccess, appCode string) (bool, []int64) {
if access.All {
return true, nil
}
appCode = appctx.Normalize(appCode)
ids := []int64{}
for _, scope := range access.Scopes {
if appctx.Normalize(scope.AppCode) != appCode {
continue
}
if scope.RegionID == 0 {
return true, nil
}
ids = append(ids, scope.RegionID)
}
return false, ids
}
func regionResolver(regions []RegionInfo) func(map[string]any) (RegionInfo, bool) {
byID := map[int64]RegionInfo{}
byCountry := map[string]RegionInfo{}
for _, region := range regions {
byID[region.RegionID] = region
for _, code := range region.Countries {
if _, ok := byCountry[code]; !ok {
byCountry[code] = region
}
}
}
return func(row map[string]any) (RegionInfo, bool) {
if regionID, ok := rowInt64(row, "region_id"); ok && regionID > 0 {
if region, ok := byID[regionID]; ok {
return region, true
}
return RegionInfo{RegionID: regionID, RegionName: "区域 " + strconv.FormatInt(regionID, 10)}, true
}
countryCode := strings.ToUpper(rowString(row, "country_code", "countryCode"))
if countryCode != "" {
if region, ok := byCountry[countryCode]; ok {
return region, true
}
}
return RegionInfo{}, false
}
}
func regionInCatalog(regions []RegionInfo, regionID int64) bool {
for _, region := range regions {
if repository.RegionIDLooseEqual(region.RegionID, regionID) {
return true
}
}
return false
}
func normalizeRegionIDs(regions []RegionInfo, regionIDs []int64) []int64 {
out := make([]int64, 0, len(regionIDs))
for _, regionID := range regionIDs {
out = append(out, normalizeRegionID(regions, regionID))
}
return out
}
// normalizeRegionID 把前端/历史数据里被 float64 圆整过的 legacy 区域 ID 还原成目录里的精确 ID
// 目录没有对应项时原样返回(显示与取数各自兜底)。
func normalizeRegionID(regions []RegionInfo, regionID int64) int64 {
if regionID == 0 {
return 0
}
for _, region := range regions {
if repository.RegionIDLooseEqual(region.RegionID, regionID) {
return region.RegionID
}
}
return regionID
}
func regionDisplay(regions []RegionInfo, regionID int64) (string, string) {
if regionID == 0 {
return "", "全部区域"
}
for _, region := range regions {
if repository.RegionIDLooseEqual(region.RegionID, regionID) {
return region.RegionCode, region.RegionName
}
}
return "", "区域 " + strconv.FormatInt(regionID, 10)
}
func filterRowsByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool), allowed map[int64]struct{}) []map[string]any {
out := make([]map[string]any, 0, len(rows))
for _, row := range rows {
region, ok := resolve(row)
if !ok {
continue
}
if _, ok := allowed[region.RegionID]; !ok {
continue
}
out = append(out, row)
}
return out
}
// rollupByRegion 把国家行按区域目录归并成大区行;无法归属任何区域的国家进入
// region_id=0 的"未划分区域"行,保证与整体合计能对账。
func rollupByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool)) []map[string]any {
type regionGroup struct {
region RegionInfo
acc *metricAccumulator
}
groups := map[int64]*regionGroup{}
order := []int64{}
for _, row := range rows {
region, ok := resolve(row)
if !ok {
region = RegionInfo{RegionID: 0, RegionCode: "UNASSIGNED", RegionName: "未划分区域"}
}
group := groups[region.RegionID]
if group == nil {
group = &regionGroup{region: region, acc: newMetricAccumulator()}
groups[region.RegionID] = group
order = append(order, region.RegionID)
}
group.acc.addRow(row)
}
out := make([]map[string]any, 0, len(order))
for _, regionID := range order {
group := groups[regionID]
row := group.acc.toMap()
row["region_id"] = group.region.RegionID
row["region_code"] = group.region.RegionCode
row["region_name"] = group.region.RegionName
out = append(out, row)
}
sort.Slice(out, func(i, j int) bool {
left, _ := rowInt64(out[i], "recharge_usd_minor")
right, _ := rowInt64(out[j], "recharge_usd_minor")
if left != right {
return left > right
}
leftID, _ := rowInt64(out[i], "region_id")
rightID, _ := rowInt64(out[j], "region_id")
return leftID < rightID
})
return out
}
func rollupDailyByRegion(rows []map[string]any, resolve func(map[string]any) (RegionInfo, bool)) []map[string]any {
byDay := map[string][]map[string]any{}
days := []string{}
for _, row := range rows {
day := rowString(row, "stat_day", "label")
if day == "" {
continue
}
if _, ok := byDay[day]; !ok {
days = append(days, day)
}
byDay[day] = append(byDay[day], row)
}
sort.Strings(days)
out := []map[string]any{}
for _, day := range days {
for _, row := range rollupByRegion(byDay[day], resolve) {
row["stat_day"] = day
row["label"] = day
out = append(out, row)
}
}
return out
}
func rollupDaily(rows []map[string]any) []map[string]any {
byDay := map[string]*metricAccumulator{}
days := []string{}
for _, row := range rows {
day := rowString(row, "stat_day", "label")
if day == "" {
continue
}
acc := byDay[day]
if acc == nil {
acc = newMetricAccumulator()
byDay[day] = acc
days = append(days, day)
}
acc.addRow(row)
}
sort.Strings(days)
out := make([]map[string]any, 0, len(days))
for _, day := range days {
row := byDay[day].toMap()
row["stat_day"] = day
row["label"] = day
out = append(out, row)
}
return out
}
// topLevelMetrics 复制统计源顶层指标并剥离结构性字段,避免响应里重复携带明细数组。
func topLevelMetrics(overview map[string]any) map[string]any {
out := make(map[string]any, len(overview))
for key, value := range overview {
switch key {
case "daily_series", "country_breakdown", "daily_country_breakdown", "report_metric_sources", "game_ranking", "lucky_gift_pools", "user_risk", "daily_region_breakdown", "region_breakdown":
continue
}
out[key] = value
}
return out
}
func scopeInfos(scopes []model.UserMoneyScope) []ScopeInfo {
out := make([]ScopeInfo, 0, len(scopes))
for _, scope := range scopes {
out = append(out, ScopeInfo{AppCode: appctx.Normalize(scope.AppCode), RegionID: scope.RegionID})
}
return out
}
func int64Set(values []int64) map[int64]struct{} {
out := make(map[int64]struct{}, len(values))
for _, value := range values {
out[value] = struct{}{}
}
return out
}
func normalizeCountryCodes(codes []string) []string {
out := make([]string, 0, len(codes))
seen := map[string]struct{}{}
for _, code := range codes {
normalized := strings.ToUpper(strings.TrimSpace(code))
if normalized == "" {
continue
}
if _, ok := seen[normalized]; ok {
continue
}
seen[normalized] = struct{}{}
out = append(out, normalized)
}
return out
}
func firstNonEmptyString(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}