1041 lines
30 KiB
Go
1041 lines
30 KiB
Go
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 描述一个外接 App（Yumi/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
|
||
}
|
||
|
||
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(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)
|
||
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.dashboards.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"`
|
||
}
|
||
|
||
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"`
|
||
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
|
||
}
|
||
|
||
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
|
||
}
|
||
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])
|
||
rangeTotal, rangeRegion, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, query.StartMS, query.EndMS)
|
||
if err != nil {
|
||
data.err = err
|
||
} else {
|
||
data.rangeTotal, data.rangeRegion = rangeTotal, rangeRegion
|
||
monthTotal, monthRegion, err := s.kpiWindow(ctx, app, resolve, query.StatTZ, monthStart.UnixMilli(), monthEnd.UnixMilli())
|
||
if err != nil {
|
||
data.err = err
|
||
} else {
|
||
data.monthTotal, data.monthRegion = monthTotal, monthRegion
|
||
}
|
||
}
|
||
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}
|
||
}
|
||
row := KpiRow{
|
||
OperatorUserID: user.ID,
|
||
OperatorName: user.Name,
|
||
OperatorAccount: user.Username,
|
||
Team: user.Team,
|
||
TeamID: user.TeamID,
|
||
AppCode: appCode,
|
||
AppName: app.AppName,
|
||
RegionID: scope.RegionID,
|
||
}
|
||
row.RegionCode, row.RegionName = regionDisplay(catalog[appCode], scope.RegionID)
|
||
if target, ok := targetByKey[kpiTargetKey(user.ID, appCode, scope.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, scope.RegionID)
|
||
monthMetrics := scopeMetrics(data.monthTotal, data.monthRegion, scope.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
|
||
})
|
||
|
||
summary := KpiSummary{}
|
||
operatorSeen := map[uint]struct{}{}
|
||
for _, row := range rows {
|
||
if row.RangeRechargeUSDMinor != nil {
|
||
summary.RangeRechargeUSDMinor += *row.RangeRechargeUSDMinor
|
||
}
|
||
if row.MonthRechargeUSDMinor != nil {
|
||
summary.MonthRechargeUSDMinor += *row.MonthRechargeUSDMinor
|
||
}
|
||
summary.MonthTargetUSDMinor += row.MonthTargetUSDMinor
|
||
operatorSeen[row.OperatorUserID] = struct{}{}
|
||
}
|
||
summary.OperatorCount = len(operatorSeen)
|
||
if summary.MonthTargetUSDMinor > 0 {
|
||
rate := safeRatio(summary.MonthRechargeUSDMinor, summary.MonthTargetUSDMinor)
|
||
summary.MonthAttainmentRate = &rate
|
||
}
|
||
result := &KpiResult{
|
||
PeriodMonth: periodMonth,
|
||
MonthStartMS: monthStart.UnixMilli(),
|
||
MonthEndMS: monthEnd.UnixMilli(),
|
||
Items: rows,
|
||
Summary: summary,
|
||
}
|
||
if len(appErrors) > 0 {
|
||
result.AppErrors = appErrors
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
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.dashboards.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)
|
||
}
|
||
|
||
// 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 region.RegionID == regionID {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func regionDisplay(regions []RegionInfo, regionID int64) (string, string) {
|
||
if regionID == 0 {
|
||
return "", "全部区域"
|
||
}
|
||
for _, region := range regions {
|
||
if 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 = ®ionGroup{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 ""
|
||
}
|