joybuywork/internal/service/crons/action/stat.go

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
}