package action import ( "joybuywork/internal/app" "joybuywork/internal/consts/ckey" "joybuywork/internal/repository/dao" "joybuywork/internal/repository/dbcheck" "joybuywork/internal/repository/model" "joybuywork/internal/repository/mtype" "joybuywork/internal/service/provider" "joybuywork/internal/utils" "fmt" "github.com/golang-module/carbon/v2" "github.com/panjf2000/ants/v2" "sync" "time" ) type StatHour struct { Action } func (s *StatHour) Desc() string { return "活动小时统计" } func (s *StatHour) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronStatHour(), release: true, } } func (s *StatHour) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkCampaignHour), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } bt := map[string]int32{ mtype.FissionBehaviorMarkView: mtype.FissionBehaviorTypeView, mtype.FissionBehaviorMarkClickPlay: mtype.FissionBehaviorTypeClickPlay, mtype.FissionBehaviorMarkClickShare: mtype.FissionBehaviorTypeClickShare, mtype.FissionBehaviorMarkViewDownload: mtype.FissionBehaviorTypeViewDownload, mtype.FissionBehaviorMarkClickDownload: mtype.FissionBehaviorTypeClickDownload, mtype.FissionBehaviorMarkViewWechat: mtype.FissionBehaviorTypeViewWechat, mtype.FissionBehaviorMarkAppInstall: mtype.FissionBehaviorTypeAppInstall, mtype.FissionBehaviorMarkAppOpen: mtype.FissionBehaviorTypeAppOpen, mtype.FissionBehaviorMarkViewDownloadByWx: mtype.FissionBehaviorTypeViewDownloadByWx, mtype.FissionBehaviorMarkAppRegister: mtype.FissionBehaviorTypeAppRegister, mtype.FissionBehaviorMarkSign: mtype.FissionBehaviorTypeSign, mtype.FissionBehaviorMarkSignTarget: mtype.FissionBehaviorTypeSignTarget, mtype.FissionBehaviorMarkGrow: mtype.FissionBehaviorTypeGrow, } gp, _ := ants.NewPool(len(list)) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("统计活动小时任务失败:任务id:%d", task.ID)) err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeHour, carbon.Now().SubHour(), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } } }) } return } type StatDay struct { Action } func (s *StatDay) Desc() string { return "活动天统计" } func (s *StatDay) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronStatDay(), release: true, } } func (s *StatDay) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkCampaignDay), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } bt := map[string]int32{ mtype.FissionBehaviorMarkView: mtype.FissionBehaviorTypeView, mtype.FissionBehaviorMarkClickPlay: mtype.FissionBehaviorTypeClickPlay, mtype.FissionBehaviorMarkClickShare: mtype.FissionBehaviorTypeClickShare, mtype.FissionBehaviorMarkViewDownload: mtype.FissionBehaviorTypeViewDownload, mtype.FissionBehaviorMarkClickDownload: mtype.FissionBehaviorTypeClickDownload, mtype.FissionBehaviorMarkViewWechat: mtype.FissionBehaviorTypeViewWechat, mtype.FissionBehaviorMarkAppInstall: mtype.FissionBehaviorTypeAppInstall, mtype.FissionBehaviorMarkAppOpen: mtype.FissionBehaviorTypeAppOpen, mtype.FissionBehaviorMarkViewDownloadByWx: mtype.FissionBehaviorTypeViewDownloadByWx, mtype.FissionBehaviorMarkAppRegister: mtype.FissionBehaviorTypeAppRegister, mtype.FissionBehaviorMarkSign: mtype.FissionBehaviorTypeSign, mtype.FissionBehaviorMarkSignTarget: mtype.FissionBehaviorTypeSignTarget, mtype.FissionBehaviorMarkGrow: mtype.FissionBehaviorTypeGrow, } gp, _ := ants.NewPool(10) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("统计活动天任务失败:任务id:%d", task.ID)) err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeDay, carbon.Now().SubHours(12), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } } }) } return } type CondHour struct { Action } func (s *CondHour) Desc() string { return "活动完成条件小时统计" } func (s *CondHour) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronCondHour(), release: true, } } func (s *CondHour) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkCondHour), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } gp, _ := ants.NewPool(10) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("活动完成条件小时统计失败:任务id:%d", task.ID)) var conds []*model.FissionCampaignCond condM := dao.FissionCampaignCond conds, err = condM.WithContext(s.ctx).Where(condM.CampaignID.Eq(task.CampaignID), condM.IsDel.Eq(mtype.No)).Find() if err != nil { return } var bt = map[string]int32{ mtype.FissionBehaviorMarkFinishTask: mtype.FissionBehaviorTypeFinishTask, } for _, cond := range conds { bt[utils.BehaviorFormatMark(mtype.FissionBehaviorTypeFinishCond, cond.ID)] = mtype.FissionBehaviorTypeFinishCond } err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeHour, carbon.Now().SubHour(), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } // 停止更新,把天的也停掉 var dayStop *model.FissionArchiveTask dayStop, err = m.WithContext(s.ctx).Where( m.CampaignID.Eq(task.CampaignID), m.Mark.Eq(mtype.FissionArchiveTaskMarkCondDay), m.IsStop.Eq(mtype.Yes), ).Take() if err = dbcheck.DbError(err); err != nil { return } if dayStop != nil { // 停止接收自定义事件 _, _ = condM.WithContext(s.ctx).Where(condM.CampaignID.Eq(task.CampaignID)).UpdateSimple(condM.IsDel.Value(mtype.Yes)) } } }) } return } type CondDay struct { Action } func (s *CondDay) Desc() string { return "活动完成条件天统计" } func (s *CondDay) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronCondDay(), release: true, } } func (s *CondDay) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkCondDay), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } gp, _ := ants.NewPool(10) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("活动完成条件天统计:任务id:%d", task.ID)) var conds []*model.FissionCampaignCond condM := dao.FissionCampaignCond conds, err = condM.WithContext(s.ctx).Where(condM.CampaignID.Eq(task.CampaignID), condM.IsDel.Eq(mtype.No)).Find() if err != nil { return } var bt = map[string]int32{ mtype.FissionBehaviorMarkFinishTask: mtype.FissionBehaviorTypeFinishTask, } for _, cond := range conds { bt[utils.BehaviorFormatMark(mtype.FissionBehaviorTypeFinishCond, cond.ID)] = mtype.FissionBehaviorTypeFinishCond } err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeDay, carbon.Now().SubHours(12), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } // 停止更新,把小时的也停掉 var hourStop *model.FissionArchiveTask hourStop, err = m.WithContext(s.ctx).Where( m.CampaignID.Eq(task.CampaignID), m.Mark.Eq(mtype.FissionArchiveTaskMarkCondHour), m.IsStop.Eq(mtype.Yes), ).Take() if err = dbcheck.DbError(err); err != nil { return } if hourStop != nil { // 停止接收自定义事件 _, _ = condM.WithContext(s.ctx).Where(condM.CampaignID.Eq(task.CampaignID)).UpdateSimple(condM.IsDel.Value(mtype.Yes)) } } }) } return } type AdsDay struct { Action } func (s *AdsDay) Desc() string { return "广告活动天统计" } func (s *AdsDay) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronAdsDay(), release: true, } } func (s *AdsDay) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkAdsDay), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } gp, _ := ants.NewPool(10) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("广告活动天统计:任务id:%d", task.ID)) var bt = map[string]int32{ mtype.FissionBehaviorMarkView: mtype.FissionBehaviorTypeView, mtype.FissionBehaviorMarkAppInstall: mtype.FissionBehaviorTypeAppInstall, } err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeDay, carbon.Now().SubHours(12), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID), m.EndTime.Eq(task.EndTime)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } } }) } return } type AdsHour struct { Action } func (s *AdsHour) Desc() string { return "广告活动小时统计" } func (s *AdsHour) OnceKey() *OnceKey { return &OnceKey{ exp: 5 * time.Second, key: ckey.CronAdsHour(), release: true, } } func (s *AdsHour) Exec() (err error) { m := dao.FissionArchiveTask list, err := m.WithContext(s.ctx).Where( m.EndTime.Gte(time.Now().Unix()), m.Mark.Eq(mtype.FissionArchiveTaskMarkAdsHour), m.IsStop.Eq(mtype.No), ).Find() if err != nil { err = s.BE().MysqlError(err) return } if len(list) == 0 { return } gp, _ := ants.NewPool(10) defer gp.Release() wg := &sync.WaitGroup{} defer wg.Wait() for _, v := range list { wg.Add(1) task := v _ = gp.Submit(func() { defer wg.Done() defer app.TryCatch(s.ctx, fmt.Sprintf("广告活动小时统计:任务id:%d", task.ID)) var bt = map[string]int32{ mtype.FissionBehaviorMarkView: mtype.FissionBehaviorTypeView, mtype.FissionBehaviorMarkAppInstall: mtype.FissionBehaviorTypeAppInstall, } err = provider.NewFissionBehaviorTotal(s.ctx, task.CampaignID).BehaviorTotalCycle(mtype.FissionTotalCycleTypeHour, carbon.Now().SubHour(), bt) if err != nil { return } if task.EndTime < time.Now().Unix() { _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ID), m.EndTime.Eq(task.EndTime)).UpdateSimple(m.IsStop.Value(mtype.Yes)) if err != nil { return } } }) } return }