stat_center/skd/job/blocstatdistrict/method.go

package blocstatdistrict

import (
	"context"
	"stat/common/log"
	"stat/common/utils"
	"stat/models/accessmod"
	"stat/models/channelCodeBlocStatMod"
	"stat/models/rechargemod"
	"stat/models/registermod"
	"time"

	"go.mongodb.org/mongo-driver/bson"
)

const (
	DirectVisitCount string = "directVisitCount" // 访问数
	RechargeAmount   string = "rechargeAmount"   // 充值金额
	NewDirectCount   string = "newDirectCount"   // 新增
)

func defaultTimeMap(defaultStatDate time.Time) map[string]time.Time {
	return map[string]time.Time{
		DirectVisitCount: defaultStatDate, // 访问数
		RechargeAmount:   defaultStatDate, // 充值金额
		NewDirectCount:   defaultStatDate, // 新增
	}
}

// 获取统计方法
type StatMethod = func(ctx context.Context, start time.Time, end time.Time, appIds []int32) error

// 统计方法汇总
var statMethodMap = map[string]StatMethod{
	DirectVisitCount: DirectVisitCountMethod, // 访问数
	RechargeAmount:   RechargeAmountMethod,   // 充值金额
	NewDirectCount:   NewDirectCountMethod,   // 新增
}

// DirectVisitCountMethod 直推访问数
func DirectVisitCountMethod(ctx context.Context, start time.Time, end time.Time, appIds []int32) error {
	type districtVisitInfo struct {
		DirectVisitCount  int64 // 直推访问数
		FissionVisitCount int64 // 裂变访问数
	}

	for _, appid := range appIds {

		districtVisitMap := make(map[string]*districtVisitInfo)

		err := accessmod.VisitInfoByTimeShard(ctx, appid, start, end, utils.GlobalShardHour, func(item *accessmod.AccessMod) {
			if len(item.DistrictCode) > 0 && !item.BlocDeduction {
				if item.IsDirect {
					if value, ok := districtVisitMap[item.DistrictCode]; !ok {
						districtVisitMap[item.DistrictCode] = &districtVisitInfo{
							DirectVisitCount: 1,
						}
					} else {
						value.DirectVisitCount += 1
					}
				} else if !item.IsDirect { // 裂变
					if value, ok := districtVisitMap[item.DistrictCode]; !ok {
						districtVisitMap[item.DistrictCode] = &districtVisitInfo{
							FissionVisitCount: 1,
						}
					} else {
						value.FissionVisitCount += 1
					}
				}
			}
		})
		if err != nil {
			log.Error("VisitCountMethod VisitInfoByTimeShard ", log.E(err))
			return err
		}
		// 插入数据库
		for k, v := range districtVisitMap {
			// 更新渠道存
			err = channelCodeBlocStatMod.UpsertOneDistrictVisitStat(ctx, k, appid, start, v.DirectVisitCount, v.FissionVisitCount)
			if err != nil {
				log.Error("VisitCountMethod UpsertOneDistrictVisitStat ", log.E(err))
				return err
			}
		}
	}
	return nil
}

// RechargeAmountMethod  用户充值
func RechargeAmountMethod(ctx context.Context, start time.Time, end time.Time, appIds []int32) error {
	type districtRechargeUserIds struct {
		RechangeAmount  int64
		FissionAmount   int64
		DeductionAmount int64
	}
	for _, appid := range appIds {
		rechangeUserMap := make(map[string]*districtRechargeUserIds)
		// 获取充值金额

		filter := bson.M{"appId": appid, "successAt": bson.M{"$gte": start, "$lt": end}, "districtCode": bson.M{"$gt": ""}}
		rechargeList, _ := rechargemod.FindManyRechargeInfo(ctx, filter)

		for _, rechargeInfo := range rechargeList {
			if len(rechargeInfo.DistrictCode) > 0 { // 充值不存在扣量
				if rechargeInfo.IsDirect {
					if value, ok := rechangeUserMap[rechargeInfo.DistrictCode]; !ok {
						rechangeUserMap[rechargeInfo.DistrictCode] = &districtRechargeUserIds{
							RechangeAmount: rechargeInfo.Money,
						}
					} else {
						value.RechangeAmount += rechargeInfo.Money
					}
				} else {
					if value, ok := rechangeUserMap[rechargeInfo.DistrictCode]; !ok {
						rechangeUserMap[rechargeInfo.DistrictCode] = &districtRechargeUserIds{
							FissionAmount: rechargeInfo.Money,
						}
					} else {

						value.FissionAmount += rechargeInfo.Money
					}
				}
			}
		}
		// 插入数据库
		for k, v := range rechangeUserMap {
			// 更新渠道存
			err := channelCodeBlocStatMod.UpsertOneDistrictRechargeStat(ctx, k, appid, start, v.RechangeAmount, v.FissionAmount, v.DeductionAmount)
			if err != nil {
				log.Error("VisitCountMethod UpsertOneDistrictVisitStat ", log.E(err))
				return err
			}
		}
	}
	return nil
}

// NewDirectCountMethod 新增
func NewDirectCountMethod(ctx context.Context, start time.Time, end time.Time, appIds []int32) error {
	type districtRegister struct {
		NewDirectCount    int64
		NewFissionCount   int64
		NewDeductionCount int64
	}
	for _, appid := range appIds {
		rechangeUserMap := make(map[string]*districtRegister)

		// 获取充值金额
		err := registermod.RegisterUsersInfoByTimeShard(ctx, appid, start, end, utils.GlobalShardHour, func(registerInfo *registermod.RegisterMod) {
			if len(registerInfo.DistrictCode) > 0 {
				if registerInfo.IsDeduction {
					if value, ok := rechangeUserMap[registerInfo.DistrictCode]; !ok {
						rechangeUserMap[registerInfo.DistrictCode] = &districtRegister{
							NewDeductionCount: 1,
						}
					} else {
						value.NewDeductionCount += 1
					}
				} else {
					if registerInfo.IsDirect {
						if value, ok := rechangeUserMap[registerInfo.DistrictCode]; !ok {
							rechangeUserMap[registerInfo.DistrictCode] = &districtRegister{
								NewDirectCount: 1,
							}
						} else {

							value.NewDirectCount += 1
						}
					} else {
						if value, ok := rechangeUserMap[registerInfo.DistrictCode]; !ok {
							rechangeUserMap[registerInfo.DistrictCode] = &districtRegister{
								NewFissionCount: 1,
							}
						} else {

							value.NewFissionCount += 1
						}
					}
				}
			}
		})
		if err != nil {
			log.Error("NewRechargeCountMethod TotalDistricRechargeAmount ", log.E(err))
			return err
		}
		// 插入数据库
		for k, v := range rechangeUserMap {
			// 更新渠道存
			err := channelCodeBlocStatMod.UpsertOneUserRegisterStat(ctx, k, appid, start, v.NewDirectCount, v.NewFissionCount, v.NewDeductionCount)
			if err != nil {
				log.Error("VisitCountMethod UpsertOneDistrictVisitStat ", log.E(err))
				return err
			}
		}
	}
	return nil
}