datatools/doFreeClueSign/job/mamager.go

package job

import (
	"context"
	"doFreeClueSign/public"
	"fmt"
	"github.com/gogf/gf/v2/container/gmap"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gcron"
	"github.com/gogf/gf/v2/os/gctx"
	"github.com/gogf/gf/v2/os/gfile"
	"github.com/gogf/gf/v2/util/gconv"
	"time"
)

type JobManager struct {
	payUser *gmap.Map //付费账户
	lastRun *struct {
		BindPhone   string `json:"bindPhone"`
		AgainSub    string `json:"againSub"`
		NewActivity string `json:"newActivity"`
	}
}

func InitJobManager() *JobManager {
	var (
		err                         error
		ctx                         = gctx.New()
		bindPhoneAndSubAgainCronStr = g.Cfg().MustGet(ctx, "bindPhoneAndSubAgain").String()
		activityUserCronStr         = g.Cfg().MustGet(ctx, "activityUser").String()
	)

	job := &JobManager{
		payUser: gmap.New(true),
	}
	job.LoadLastRun()

	if bindPhoneAndSubAgainCronStr != "" {
		_, err = gcron.Add(ctx, bindPhoneAndSubAgainCronStr, func(ctx context.Context) {
			job.LoadPayUser()
			job.LoadAgainSubUser()
			job.LoadBindPhoneUser()
		}, "bindPhoneAndSubAgain")
		if err != nil {
			panic(err)
		}
		gcron.Start("bindPhoneAndSubAgain")
	}

	if activityUserCronStr != "" {
		_, err = gcron.Add(ctx, activityUserCronStr, func(ctx context.Context) {
			job.LoadPayUser()
			job.LoadActivityUser()
		}, "activityUser")
		if err != nil {
			panic(err)
		}
		gcron.Start("activityUser")
	}

	return job
}

func (jm *JobManager) LoadLastRun() {
	err := gconv.Struct(gfile.GetContents("runSign.json"), &jm.lastRun)
	if err != nil {
		g.Log().Errorf(context.TODO(), "加载上次运行配置日常")
	}
	now := time.Now().Format(time.DateTime)
	if jm.lastRun.BindPhone == "" {
		jm.lastRun.BindPhone = now
	}
	if jm.lastRun.AgainSub == "" {
		jm.lastRun.AgainSub = now
	}
	if jm.lastRun.NewActivity == "" {
		jm.lastRun.NewActivity = now
	}
	return
}

func (jm *JobManager) SaveLastRun() error {
	return gfile.PutContents("runSign.json", gconv.String(jm.lastRun))
}

func (jm *JobManager) LoadPayUser() {
	newMap := gmap.New(true)
	for i, v := range public.GetPayUser() {
		newMap.Set(i, v)
	}
	jm.payUser = newMap
}

func (jm *JobManager) FilterPayUserAndSaveDb(ctx context.Context, value interface{}) error {
	var (
		sql           = `INSERT INTO freeClubSign `
		val           []interface{}
		now           = time.Now().Format(time.DateTime)
		operationTime string
	)
	if msg, ok := value.(*public.AgainSubUserMsg); ok {
		if jm.payUser.Contains(msg.MgoUserID) {
			return nil
		}
		operationTime = time.Unix(msg.TimeStamp, 0).Format(time.DateTime)
		sql = `(mogUserId,sub_again_date,create_time)VALUES (?, ?, ?)ON DUPLICATE KEY UPDATE sub_again_date=?`
		val = append(val, msg.MgoUserID, operationTime, now, operationTime)
	} else if msg, ok := value.(*public.BindMsg); ok {
		if jm.payUser.Contains(msg.MgoUserID) {
			return nil
		}
		operationTime = time.Unix(msg.TimeStamp, 0).Format(time.DateTime)
		sql += `(mogUserId,phone,bind_phone_date,create_time)VALUES (?, ?, ?, ?)ON DUPLICATE KEY UPDATE phone=?, bind_phone_date=?`
		val = append(val, msg.MgoUserID, msg.Phone, operationTime, now, msg.Phone, operationTime)
	} else if msg, ok := value.(*public.NewActiveMsg); ok {
		if jm.payUser.Contains(msg.MgoUserID) {
			return nil
		}
		operationTime = time.Unix(msg.TimeStamp, 0).Format(time.DateTime)
		sql = `(mogUserId,act_again_date,create_time)VALUES (?, ?, ?)ON DUPLICATE KEY UPDATE act_again_date=?`
		val = append(val, msg.MgoUserID, operationTime, now, operationTime)
	} else {
		return fmt.Errorf("未知类型")
	}
	_, err := g.DB("bi_service").Exec(context.TODO(), sql, val...)
	if err != nil {
		return err
	}
	return nil
}