1505 lines
45 KiB
Go
1505 lines
45 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
|
||
)
|
||
|
||
var appTrackingFunnelSupportedApps = map[string]struct{}{
|
||
"fami": {},
|
||
"huwaa": {},
|
||
"lalu": {},
|
||
}
|
||
|
||
// 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
|
||
|
||
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"`
|
||
}
|
||
|
||
// 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"`
|
||
}
|
||
|
||
type FunnelQuery struct {
|
||
StatTZ string
|
||
StartMS int64
|
||
EndMS int64
|
||
AppCodes []string
|
||
RegionID int64
|
||
}
|
||
|
||
type AppFunnel 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"`
|
||
Steps []map[string]any `json:"steps"`
|
||
Cohorts []map[string]any `json:"cohorts"`
|
||
Totals map[string]any `json:"totals"`
|
||
UpdatedAtMS int64 `json:"updated_at_ms"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
type FunnelResult struct {
|
||
Apps []AppFunnel `json:"apps"`
|
||
Access AccessInfo `json:"access"`
|
||
}
|
||
|
||
type RequirementsQuery struct {
|
||
StatTZ string
|
||
StartMS int64
|
||
EndMS int64
|
||
AppCodes []string
|
||
RegionID int64
|
||
RegionIDs []int64
|
||
Section string
|
||
UserRole string
|
||
PayerType string
|
||
}
|
||
|
||
type AppRequirements 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"`
|
||
Sections []map[string]any `json:"sections"`
|
||
UpdatedAtMS int64 `json:"updated_at_ms"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
type RequirementsResult struct {
|
||
Apps []AppRequirements `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) Funnel(ctx context.Context, access repository.MoneyAccess, query FunnelQuery) (*FunnelResult, error) {
|
||
apps, err := s.listAllowedApps(ctx, query.AppCodes, access)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
apps = filterAppTrackingFunnelApps(apps)
|
||
catalog, err := s.regionCatalog(ctx, apps)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
results := make([]AppFunnel, 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.appFunnel(ctx, app, catalog[app.AppCode], access, query)
|
||
}(index, app)
|
||
}
|
||
wg.Wait()
|
||
|
||
return &FunnelResult{
|
||
Apps: results,
|
||
Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) Requirements(ctx context.Context, access repository.MoneyAccess, query RequirementsQuery) (*RequirementsResult, 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([]AppRequirements, 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.appRequirements(ctx, app, catalog[app.AppCode], access, query)
|
||
}(index, app)
|
||
}
|
||
wg.Wait()
|
||
|
||
return &RequirementsResult{
|
||
Apps: results,
|
||
Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)},
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) appRequirements(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query RequirementsQuery) AppRequirements {
|
||
out := AppRequirements{
|
||
AppCode: app.AppCode,
|
||
AppName: app.AppName,
|
||
Kind: app.Kind,
|
||
Sections: []map[string]any{},
|
||
}
|
||
if app.Kind != appKindHyapp {
|
||
out.Error = "该 App 暂未接入数据需求宽表"
|
||
return out
|
||
}
|
||
allowAll, allowedIDs := allowedRegions(access, app.AppCode)
|
||
allowedIDs = normalizeRegionIDs(regions, allowedIDs)
|
||
if !allowAll && len(allowedIDs) == 0 {
|
||
out.Error = "没有该 App 的数据权限"
|
||
return out
|
||
}
|
||
requestRegionIDs := normalizeRegionIDs(regions, query.RegionIDs)
|
||
if len(query.RegionIDs) > 0 && len(requestRegionIDs) == 0 {
|
||
out.Error = "该 App 未配置所选区域"
|
||
return out
|
||
}
|
||
if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) {
|
||
out.Error = "没有该区域的数据权限"
|
||
return out
|
||
}
|
||
for _, regionID := range requestRegionIDs {
|
||
if !access.Allows(app.AppCode, regionID) {
|
||
out.Error = "没有该区域的数据权限"
|
||
return out
|
||
}
|
||
}
|
||
if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) {
|
||
out.Error = "该 App 未配置此区域"
|
||
return out
|
||
}
|
||
restricted := !allowAll && query.RegionID == 0 && len(requestRegionIDs) == 0
|
||
regionIDs := []int64(nil)
|
||
if restricted {
|
||
// 用户没有全局范围时，admin 只把允许的大区 ID 传给 statistics-service；查询层仍只读聚合投影，不回扫业务库补权限。
|
||
out.Restricted = true
|
||
out.AllowedRegionIDs = allowedIDs
|
||
regionIDs = allowedIDs
|
||
} else if len(requestRegionIDs) > 0 {
|
||
// 顶栏多选区域时保留多区域筛选；单区域仍走 region_id 以兼容旧统计入口。
|
||
regionIDs = requestRegionIDs
|
||
}
|
||
requirements, err := s.dashboards.SocialRequirements(ctx, dashboard.SocialRequirementsQuery{
|
||
AppCode: app.AppCode,
|
||
StatTZ: query.StatTZ,
|
||
StartMS: query.StartMS,
|
||
EndMS: query.EndMS,
|
||
RegionID: query.RegionID,
|
||
RegionIDs: regionIDs,
|
||
Section: query.Section,
|
||
UserRole: query.UserRole,
|
||
PayerType: query.PayerType,
|
||
})
|
||
if err != nil {
|
||
out.Error = fmt.Sprintf("获取数据需求指标失败: %v", err)
|
||
return out
|
||
}
|
||
out.Sections = anyRowSlice(requirements["sections"])
|
||
out.UpdatedAtMS = time.Now().UnixMilli()
|
||
return out
|
||
}
|
||
|
||
func (s *Service) appFunnel(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query FunnelQuery) AppFunnel {
|
||
out := AppFunnel{
|
||
AppCode: app.AppCode,
|
||
AppName: app.AppName,
|
||
Kind: app.Kind,
|
||
Steps: []map[string]any{},
|
||
Cohorts: []map[string]any{},
|
||
Totals: map[string]any{},
|
||
}
|
||
if app.Kind != appKindHyapp {
|
||
out.Error = "该 App 暂未接入 App 埋点漏斗"
|
||
return out
|
||
}
|
||
if !supportsAppTrackingFunnel(app.AppCode) {
|
||
out.Error = "该 App 暂未接入 App 埋点漏斗"
|
||
return out
|
||
}
|
||
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
|
||
}
|
||
restricted := !allowAll && query.RegionID == 0
|
||
if restricted {
|
||
out.Restricted = true
|
||
out.AllowedRegionIDs = allowedIDs
|
||
}
|
||
regionIDs := []int64(nil)
|
||
if restricted {
|
||
regionIDs = allowedIDs
|
||
}
|
||
funnel, err := s.dashboards.AppTrackingFunnel(ctx, dashboard.AppTrackingFunnelQuery{
|
||
AppCode: app.AppCode,
|
||
StatTZ: query.StatTZ,
|
||
StartMS: query.StartMS,
|
||
EndMS: query.EndMS,
|
||
RegionID: query.RegionID,
|
||
RegionIDs: regionIDs,
|
||
})
|
||
if err != nil {
|
||
out.Error = fmt.Sprintf("获取埋点漏斗失败: %v", err)
|
||
return out
|
||
}
|
||
out.Steps = anyRowSlice(funnel["steps"])
|
||
out.Cohorts = anyRowSlice(funnel["cohorts"])
|
||
if totals, ok := funnel["totals"].(map[string]any); ok {
|
||
out.Totals = totals
|
||
}
|
||
out.UpdatedAtMS = time.Now().UnixMilli()
|
||
return out
|
||
}
|
||
|
||
func filterAppTrackingFunnelApps(apps []AppInfo) []AppInfo {
|
||
out := make([]AppInfo, 0, len(apps))
|
||
for _, app := range apps {
|
||
if app.Kind == appKindHyapp && supportsAppTrackingFunnel(app.AppCode) {
|
||
out = append(out, app)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func supportsAppTrackingFunnel(appCode string) bool {
|
||
_, ok := appTrackingFunnelSupportedApps[appctx.Normalize(appCode)]
|
||
return ok
|
||
}
|
||
|
||
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"`
|
||
DataError string `json:"data_error,omitempty"`
|
||
}
|
||
|
||
// KpiOperatorAppRow 是 人×App 级小计：该运营在此 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"`
|
||
DataError string `json:"data_error,omitempty"`
|
||
}
|
||
|
||
type KpiSummary struct {
|
||
RangeRechargeUSDMinor int64 `json:"range_recharge_usd_minor"`
|
||
MonthRechargeUSDMinor int64 `json:"month_recharge_usd_minor"`
|
||
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"`
|
||
Summary KpiSummary `json:"summary"`
|
||
AppErrors map[string]any `json:"app_errors,omitempty"`
|
||
}
|
||
|
||
// Kpi 按运营人员负责范围（App×区域）聚合充值/新增；无全量数据范围时只返回本人行。
|
||
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()
|
||
|
||
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)
|
||
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")
|
||
}
|
||
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
|
||
})
|
||
|
||
operatorAppRows := buildOperatorAppRows(rows)
|
||
summary := buildKpiSummary(rows, operatorAppRows)
|
||
|
||
result := &KpiResult{
|
||
PeriodMonth: periodMonth,
|
||
MonthStartMS: monthStart.UnixMilli(),
|
||
MonthEndMS: monthEnd.UnixMilli(),
|
||
Items: rows,
|
||
OperatorAppRows: operatorAppRows,
|
||
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) []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)
|
||
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))
|
||
}
|
||
|
||
// buildKpiSummary 只服务列表头部元信息；充值合计按人×App 小计相加，避免同一运营多区域重复。
|
||
func buildKpiSummary(rows []KpiRow, operatorAppRows []KpiOperatorAppRow) KpiSummary {
|
||
summary := KpiSummary{}
|
||
operatorSeen := map[uint]struct{}{}
|
||
for _, row := range rows {
|
||
operatorSeen[row.OperatorUserID] = struct{}{}
|
||
}
|
||
summary.OperatorCount = len(operatorSeen)
|
||
|
||
for _, row := range operatorAppRows {
|
||
if row.RangeRechargeUSDMinor != nil {
|
||
summary.RangeRechargeUSDMinor += *row.RangeRechargeUSDMinor
|
||
}
|
||
if row.MonthRechargeUSDMinor != nil {
|
||
summary.MonthRechargeUSDMinor += *row.MonthRechargeUSDMinor
|
||
}
|
||
}
|
||
return summary
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
// 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 = ®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
|
||
}
|
||
flattenRetentionMetrics(out, overview["retention"])
|
||
return out
|
||
}
|
||
|
||
func flattenRetentionMetrics(out map[string]any, retention any) {
|
||
retentionRow, ok := retention.(map[string]any)
|
||
if !ok {
|
||
return
|
||
}
|
||
for _, item := range []struct {
|
||
day string
|
||
rateKey string
|
||
usersKey string
|
||
baseKey string
|
||
}{
|
||
{day: "day1", rateKey: "d1_retention_rate", usersKey: "d1_retention_users", baseKey: "d1_retention_base_users"},
|
||
{day: "day7", rateKey: "d7_retention_rate", usersKey: "d7_retention_users", baseKey: "d7_retention_base_users"},
|
||
{day: "day30", rateKey: "d30_retention_rate", usersKey: "d30_retention_users", baseKey: "d30_retention_base_users"},
|
||
} {
|
||
// statistics-service 的历史契约把整段留存放在 retention 子对象里；
|
||
// Databi v2 的 App 行统一读取扁平字段，这里只做字段搬运，不重算口径。
|
||
out[item.rateKey] = retentionRow[item.day+"_rate"]
|
||
out[item.usersKey] = retentionRow[item.day+"_users"]
|
||
out[item.baseKey] = retentionRow[item.day+"_base_users"]
|
||
}
|
||
}
|
||
|
||
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 ""
|
||
}
|