You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
502 lines
12 KiB
Go
502 lines
12 KiB
Go
package provider
|
|
|
|
import (
|
|
"context"
|
|
"joybuywork/internal/app"
|
|
"joybuywork/internal/consts"
|
|
"joybuywork/internal/consts/ckey"
|
|
"joybuywork/internal/repository/cache"
|
|
"joybuywork/internal/repository/dao"
|
|
"joybuywork/internal/repository/model"
|
|
"joybuywork/internal/repository/mtype"
|
|
"joybuywork/internal/repository/withdefault"
|
|
"joybuywork/internal/utils"
|
|
"joybuywork/internal/utils/merr"
|
|
"joybuywork/proto/go_proto/workerr"
|
|
"fmt"
|
|
"github.com/golang-module/carbon/v2"
|
|
"gits.branchcn.com/backend/microcode/plugins/derivative/lock"
|
|
"gorm.io/gen"
|
|
"gorm.io/gen/field"
|
|
"time"
|
|
)
|
|
|
|
type FissionBehaviorTotal struct {
|
|
ctx context.Context
|
|
be *merr.BusinessError
|
|
cpnId int64
|
|
}
|
|
|
|
// NewFissionBehaviorTotal 用户行为统计数据
|
|
func NewFissionBehaviorTotal(ctx context.Context, cpnId int64) *FissionBehaviorTotal {
|
|
return &FissionBehaviorTotal{ctx: ctx, be: merr.NewBE(ctx), cpnId: cpnId}
|
|
}
|
|
|
|
// StatHour 设置小时活动归档数据
|
|
func (s *FissionBehaviorTotal) StatHour(mark string, at int64, userMark string) {
|
|
t := carbon.CreateFromTimestamp(at)
|
|
|
|
start, end := t.StartOfHour(), t.EndOfHour()
|
|
hour := t.Format(consts.YmdH)
|
|
|
|
//次数key
|
|
key := ckey.FissionBehaviorTotalCycleHour(s.cpnId, hour)
|
|
userKey := ckey.FissionBehaviorTotalCycleHourUser(s.cpnId, hour, mark)
|
|
|
|
// 如果key不存在
|
|
if app.Redis.Cache.Exists(s.ctx, key).Val() == 0 {
|
|
_ = s._existStat(mark, start, end, key, userKey)
|
|
}
|
|
|
|
// 次数
|
|
app.Redis.Cache.HIncrBy(s.ctx, key, mark, 1)
|
|
app.Redis.Cache.Expire(s.ctx, key, 2*time.Hour)
|
|
|
|
//人数
|
|
app.Redis.Cache.HSet(s.ctx, userKey, userMark, 1)
|
|
app.Redis.Cache.Expire(s.ctx, userKey, 2*time.Hour)
|
|
return
|
|
}
|
|
|
|
// StatDay 设置天纬度的统计数据
|
|
func (s *FissionBehaviorTotal) StatDay(mark string, at int64, userMark string) {
|
|
t := carbon.CreateFromTimestamp(at)
|
|
|
|
start, end := t.StartOfDay(), t.EndOfDay()
|
|
day := t.ToShortDateString()
|
|
|
|
//次数key
|
|
key := ckey.FissionBehaviorTotalCycleDay(s.cpnId, day)
|
|
userKey := ckey.FissionBehaviorTotalCycleDayUser(s.cpnId, day, mark)
|
|
|
|
// 如果key不存在
|
|
if app.Redis.Cache.Exists(s.ctx, key).Val() == 0 {
|
|
_ = s._existStat(mark, start, end, key, userKey)
|
|
}
|
|
|
|
//次数
|
|
app.Redis.Cache.HIncrBy(s.ctx, key, mark, 1)
|
|
app.Redis.Cache.Expire(s.ctx, key, 2*time.Hour)
|
|
|
|
//人数
|
|
app.Redis.Cache.HSet(s.ctx, userKey, userMark, 1)
|
|
app.Redis.Cache.Expire(s.ctx, userKey, 2*time.Hour)
|
|
}
|
|
|
|
type BehaviorNum struct {
|
|
Total int64
|
|
Mark string
|
|
}
|
|
|
|
// CycleHour 获取小时活动归档数据
|
|
func (s *FissionBehaviorTotal) CycleHour(t carbon.Carbon, marks ...string) (hourNumMap, hourUserNumMap map[string]int64, err error) {
|
|
// 小时数据统计归档
|
|
hourStr := t.Format(consts.YmdH)
|
|
key := ckey.FissionBehaviorTotalCycleHour(s.cpnId, hourStr)
|
|
if len(marks) == 0 {
|
|
var numMap = map[string]string{}
|
|
numMap, err = app.Redis.Cache.HGetAll(s.ctx, key).Result()
|
|
if err != nil {
|
|
err = s.be.RedisError(err)
|
|
return
|
|
}
|
|
|
|
hourNumMap, hourUserNumMap = make(map[string]int64, len(numMap)), make(map[string]int64, len(numMap))
|
|
|
|
for mark, num := range numMap {
|
|
hourNumMap[mark] = utils.StrToInt64(num)
|
|
userKey := ckey.FissionBehaviorTotalCycleHourUser(s.cpnId, hourStr, mark)
|
|
hourUserNumMap[mark] = app.Redis.Cache.HLen(s.ctx, userKey).Val()
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// 如果没有数据就填充好
|
|
start, end := t.StartOfHour(), t.EndOfHour()
|
|
for _, mark := range marks {
|
|
if !app.Redis.Cache.HExists(s.ctx, key, mark).Val() {
|
|
err = s._existStat(mark, start, end, key, ckey.FissionBehaviorTotalCycleHourUser(s.cpnId, hourStr, mark))
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
hourNumMap, hourUserNumMap = make(map[string]int64, len(marks)), make(map[string]int64, len(marks))
|
|
var vals []interface{}
|
|
vals, err = app.Redis.Cache.HMGet(s.ctx, key, marks...).Result()
|
|
if err != nil {
|
|
err = s.be.RedisError(err)
|
|
return
|
|
}
|
|
|
|
var numMap = map[string]string{}
|
|
for i, v := range marks {
|
|
if vals[i] == nil {
|
|
continue
|
|
}
|
|
numMap[v] = fmt.Sprint(vals[i])
|
|
}
|
|
|
|
for mark, num := range numMap {
|
|
hourNumMap[mark] = utils.StrToInt64(num)
|
|
userKey := ckey.FissionBehaviorTotalCycleHourUser(s.cpnId, hourStr, mark)
|
|
hourUserNumMap[mark] = app.Redis.Cache.HLen(s.ctx, userKey).Val()
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// CycleDay 获取天活动归档数据
|
|
func (s *FissionBehaviorTotal) CycleDay(t carbon.Carbon, marks ...string) (dayNumMap, dayUserNumMap map[string]int64, err error) {
|
|
dayStr := t.ToShortDateString()
|
|
key := ckey.FissionBehaviorTotalCycleDay(s.cpnId, dayStr)
|
|
|
|
if len(marks) == 0 {
|
|
var numMap = map[string]string{}
|
|
numMap, err = app.Redis.Cache.HGetAll(s.ctx, key).Result()
|
|
if err != nil {
|
|
err = s.be.RedisError(err)
|
|
return
|
|
}
|
|
|
|
dayNumMap, dayUserNumMap = make(map[string]int64, len(numMap)), make(map[string]int64, len(numMap))
|
|
|
|
for k, v := range numMap {
|
|
dayNumMap[k] = utils.StrToInt64(v)
|
|
userKey := ckey.FissionBehaviorTotalCycleDayUser(s.cpnId, dayStr, k)
|
|
dayUserNumMap[k] = app.Redis.Cache.HLen(s.ctx, userKey).Val()
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// 如果没有数据就填充好
|
|
start, end := t.StartOfDay(), t.EndOfDay()
|
|
for _, mark := range marks {
|
|
if !app.Redis.Cache.HExists(s.ctx, key, mark).Val() {
|
|
err = s._existStat(mark, start, end, key, ckey.FissionBehaviorTotalCycleDayUser(s.cpnId, dayStr, mark))
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
dayNumMap, dayUserNumMap = make(map[string]int64, len(marks)), make(map[string]int64, len(marks))
|
|
var vals []interface{}
|
|
vals, err = app.Redis.Cache.HMGet(s.ctx, key, marks...).Result()
|
|
if err != nil {
|
|
err = s.be.RedisError(err)
|
|
return
|
|
}
|
|
|
|
var numMap = map[string]string{}
|
|
for i, v := range marks {
|
|
if vals[i] == nil {
|
|
continue
|
|
}
|
|
numMap[v] = fmt.Sprint(vals[i])
|
|
}
|
|
|
|
for k, v := range numMap {
|
|
dayNumMap[k] = utils.StrToInt64(v)
|
|
userKey := ckey.FissionBehaviorTotalCycleDayUser(s.cpnId, dayStr, k)
|
|
dayUserNumMap[k] = app.Redis.Cache.HLen(s.ctx, userKey).Val()
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// FissionTotalAdd 玩法活动总数值统计
|
|
func (s *FissionBehaviorTotal) FissionTotalAdd(num int64, bt int32, mark string, newUser bool) (err error) {
|
|
m := dao.FissionTotal
|
|
|
|
var update []field.AssignExpr
|
|
update = append(update, m.Num.Add(num))
|
|
if newUser {
|
|
update = append(update, m.UserNum.Add(1))
|
|
}
|
|
|
|
cc := cache.NewFissionTotal(s.ctx)
|
|
id, err := cc.FindByUniqIdx(s.cpnId, mark)
|
|
if err != nil {
|
|
return s.be.CacheError(err)
|
|
}
|
|
|
|
defer func() {
|
|
if err != nil || id == 0 {
|
|
return
|
|
}
|
|
|
|
_, err = m.WithContext(s.ctx).Where(m.ID.Eq(id)).UpdateSimple(update...)
|
|
if err != nil {
|
|
err = s.be.MysqlError(err)
|
|
return
|
|
}
|
|
}()
|
|
|
|
if id != 0 {
|
|
return
|
|
}
|
|
|
|
lc := app.NewFollower(s.ctx, ckey.FissionBehaviorTotalCreateOnce(s.cpnId, mark), lock.SetExpire(10*time.Second)).Lock()
|
|
defer lc.Release()
|
|
if err != nil {
|
|
return workerr.BusyError
|
|
}
|
|
|
|
id, err = cc.FindByUniqIdx(s.cpnId, mark)
|
|
if err != nil {
|
|
return s.be.MysqlError(err)
|
|
}
|
|
|
|
if id != 0 {
|
|
return
|
|
}
|
|
|
|
total := withdefault.FissionTotal(&model.FissionTotal{
|
|
Mark: mark,
|
|
BehaviorType: bt,
|
|
CampaignID: s.cpnId,
|
|
Num: num,
|
|
UserNum: utils.Ternary[int64](newUser, 1, 0),
|
|
})
|
|
|
|
err = m.WithContext(s.ctx).Create(total)
|
|
if err != nil {
|
|
return s.be.MysqlError(err)
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
type statDayNum struct {
|
|
Num int64
|
|
UserNum int64
|
|
BehaviorType int32
|
|
}
|
|
|
|
// BehaviorTotalCycle 行为周期归档
|
|
func (s *FissionBehaviorTotal) BehaviorTotalCycle(statType int8, now carbon.Carbon, btm map[string]int32) (err error) {
|
|
var marks []string
|
|
for mark := range btm {
|
|
marks = append(marks, mark)
|
|
}
|
|
|
|
var start carbon.Carbon
|
|
var numMap, userNumMap = make(map[string]int64, len(marks)), make(map[string]int64, len(marks))
|
|
if statType == mtype.FissionTotalCycleTypeHour {
|
|
start = now.StartOfHour()
|
|
|
|
numMap, userNumMap, err = s.CycleHour(now, marks...)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
} else {
|
|
start = now.StartOfDay()
|
|
numMap, userNumMap, err = s.CycleDay(now, marks...)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
|
|
var statMap = map[string]*statDayNum{}
|
|
for mark, bt := range btm {
|
|
statMap[mark] = &statDayNum{
|
|
Num: numMap[mark],
|
|
UserNum: userNumMap[mark],
|
|
BehaviorType: bt,
|
|
}
|
|
marks = append(marks, mark)
|
|
}
|
|
|
|
m := dao.FissionTotalCycle
|
|
list, err := m.WithContext(s.ctx).Where(
|
|
m.CampaignID.Eq(s.cpnId),
|
|
m.StatType.Eq(statType),
|
|
m.StatTime.Eq(start.Timestamp()),
|
|
m.Mark.In(marks...),
|
|
).Find()
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
var recordMap = make(map[string]*model.FissionTotalCycle)
|
|
for _, v := range list {
|
|
recordMap[v.Mark] = v
|
|
}
|
|
|
|
var add []*model.FissionTotalCycle
|
|
for mark, total := range statMap {
|
|
if record, ok := recordMap[mark]; ok {
|
|
_, err = m.WithContext(s.ctx).Where(m.ID.Eq(record.ID)).UpdateSimple(m.Num.Value(total.Num), m.UserNum.Value(total.UserNum))
|
|
if err != nil {
|
|
return
|
|
}
|
|
} else if total.Num > 0 {
|
|
add = append(add, withdefault.FissionTotalCycle(&model.FissionTotalCycle{
|
|
CampaignID: s.cpnId,
|
|
StatTime: start.Timestamp(),
|
|
StatType: statType,
|
|
BehaviorType: total.BehaviorType,
|
|
Mark: mark,
|
|
Num: total.Num,
|
|
UserNum: total.UserNum,
|
|
}))
|
|
}
|
|
}
|
|
|
|
if len(add) > 0 {
|
|
err = m.WithContext(s.ctx).Create(add...)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) Num(mark string, start, end carbon.Carbon) (total int64) {
|
|
fbm := dao.FissionBehavior
|
|
|
|
total, _ = fbm.WithContext(s.ctx).Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.Mark.Eq(mark),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).Count()
|
|
|
|
return
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) UserNum(mark string, start, end carbon.Carbon) (total int64) {
|
|
fbm := dao.FissionBehavior
|
|
|
|
total, _ = fbm.WithContext(s.ctx).Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.Mark.Eq(mark),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).Group(fbm.UserMark).Count()
|
|
|
|
return
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) NumType(bt int32, start, end carbon.Carbon) (total int64) {
|
|
fbm := dao.FissionBehavior
|
|
|
|
total, _ = fbm.WithContext(s.ctx).Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.BehaviorType.Eq(bt),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).Count()
|
|
|
|
return
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) UserNumType(bt int32, start, end carbon.Carbon) (total int64) {
|
|
fbm := dao.FissionBehavior
|
|
|
|
total, _ = fbm.WithContext(s.ctx).Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.BehaviorType.Eq(bt),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).Group(fbm.UserMark).Count()
|
|
|
|
return
|
|
}
|
|
|
|
type MarkTotal struct {
|
|
Mark string
|
|
Total int64
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) MarkNumType(bt int32, start, end carbon.Carbon) (numStat []*MarkTotal) {
|
|
fbm := dao.FissionBehavior
|
|
_ = fbm.WithContext(s.ctx).
|
|
Select(
|
|
fbm.Mark,
|
|
fbm.ID.Count().As("total"),
|
|
).
|
|
Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.BehaviorType.Eq(bt),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).
|
|
Group(fbm.Mark).
|
|
Scan(&numStat)
|
|
|
|
return
|
|
}
|
|
|
|
func (s *FissionBehaviorTotal) MarkUserNumType(bt int32, start, end carbon.Carbon) (numStat []*MarkTotal) {
|
|
fbm := dao.FissionBehavior
|
|
|
|
fbmAs := fbm.As("fbm")
|
|
_ = gen.Table(
|
|
fbm.WithContext(s.ctx).
|
|
Select(
|
|
fbm.Mark,
|
|
fbm.UserMark,
|
|
).
|
|
Where(
|
|
fbm.CampaignID.Eq(s.cpnId),
|
|
fbm.BehaviorType.Eq(bt),
|
|
fbm.BehaviorTime.Gte(start.Timestamp()),
|
|
fbm.BehaviorTime.Lte(end.Timestamp()),
|
|
).
|
|
Group(fbm.Mark, fbm.UserMark).
|
|
As("fbm")).
|
|
Select(
|
|
fbmAs.Mark,
|
|
fbmAs.UserMark.Count().As("total"),
|
|
).
|
|
Group(fbmAs.Mark).
|
|
Scan(&numStat)
|
|
|
|
return
|
|
}
|
|
|
|
// 不存在统计就补充
|
|
func (s *FissionBehaviorTotal) _existStat(mark string, start, end carbon.Carbon, key string, userKey string) (err error) {
|
|
lc := app.NewRedisLock(s.ctx, ckey.ForLock(key), lock.SetExpire(10*time.Second))
|
|
if !lc.SoftLock() {
|
|
return
|
|
}
|
|
defer lc.Release()
|
|
|
|
app.Redis.Cache.HIncrBy(s.ctx, key, mark, s.Num(mark, start, end))
|
|
app.Redis.Cache.Expire(s.ctx, key, 2*time.Hour)
|
|
|
|
// 人数获取
|
|
m := dao.FissionBehavior
|
|
var userMarks []string
|
|
err = m.WithContext(s.ctx).
|
|
Select(m.UserMark.Distinct()).
|
|
Where(
|
|
m.CampaignID.Eq(s.cpnId),
|
|
m.Mark.Eq(mark),
|
|
m.BehaviorTime.Gte(start.Timestamp()),
|
|
m.BehaviorTime.Lte(end.Timestamp()),
|
|
).
|
|
Group(m.UserMark).
|
|
Pluck(m.UserMark, &userMarks)
|
|
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
for _, userMark := range userMarks {
|
|
if app.Redis.Cache.HSetNX(s.ctx, userKey, userMark, 1).Val() {
|
|
app.Redis.Cache.Expire(s.ctx, userKey, 2*time.Hour)
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|