workTasks/userSign/userAnalysis/analysisManager.go

package userAnalysis

import (
	"context"
	"fmt"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"strings"
)

type (
	UserAnalysis struct {
		UserMapping      map[string]BaseUserId //职位id、mongouserId对应的baseUserId
		EntUserIdMapping map[int64]BaseUserId  //ent_id对应的baseUserId
		FullBaseUserId   map[BaseUserId]bool   //全量BaseUserId

		BinPhone, BindMail   map[BaseUserId]bool
		Vip, Vip15, Vip30    map[BaseUserId]bool
		Member, MemberExpire map[BaseUserId]bool

		UnSubUser map[string]bool             //取关
		NewUser   map[int]map[BaseUserId]bool //新用户

		TestUser map[BaseUserId]bool
	}
	BaseUserId int64

	AnalysisRes struct {
		Name, Code  string              //标签名字
		Data        map[BaseUserId]bool //数据
		SaveOldData bool                //累计统计时需要设置为true 是否保留旧数据
	}
)

// UpdateTag 更新标签对应的用户Bitmap，因sql太长所以拆分成batchSize插入
func (ar *AnalysisRes) UpdateTag(ctx context.Context) {
	const batchSize = 5000
	var (
		updateBatch = [][]string{}
		tmpArr      = make([]string, 0, batchSize)
		total       = len(ar.Data)
		index       = 0
	)
	g.Log().Infof(ctx, "UpdateTag code %s 更新%d个开始", ar.Code, len(ar.Data))
	for id, ok := range ar.Data {
		if !ok {
			continue
		}
		tmpArr = append(tmpArr, fmt.Sprintf("toUInt64(%d)", id))
		if len(tmpArr) == batchSize {
			updateBatch = append(updateBatch, tmpArr)
			tmpArr = make([]string, 0, batchSize)
		}
		index++
		if index == total {
			updateBatch = append(updateBatch, tmpArr)
		}
	}

	if len(updateBatch) == 0 {
		if !ar.SaveOldData {
			execSql := fmt.Sprintf(`ALTER TABLE dwd_d_tag UPDATE bitobj = bitmapBuild([toUInt64(0)]) WHERE code = '%v';`, ar.Code)
			if _, err := g.DB().Exec(ctx, execSql); err != nil {
				g.Log().Errorf(ctx, "更新标签%s 滞空异常 %v", ar.Code, err)
			}
		}
	} else {
		for i, batch := range updateBatch {
			if i == 0 && !ar.SaveOldData {
				execSql := fmt.Sprintf(`ALTER TABLE dwd_d_tag UPDATE bitobj = bitmapBuild([%v]) WHERE code = '%v';`, strings.Join(batch, ","), ar.Code)
				if _, err := g.DB().Exec(ctx, execSql); err != nil {
					g.Log().Errorf(ctx, "更新标签%s [%d]异常 %v", ar.Code, i, err)
				}
			} else {
				execSql := fmt.Sprintf(`ALTER TABLE dwd_d_tag UPDATE bitobj = bitmapOr(bitobj,bitmapBuild([%v])) WHERE code = '%v';`, strings.Join(batch, ","), ar.Code)
				if _, err := g.DB().Exec(ctx, execSql); err != nil {
					g.Log().Errorf(ctx, "更新标签%s [%d]异常 %v", ar.Code, i, err)
				}
			}
		}
	}

	g.Log().Infof(ctx, "UpdateTag code %s 更新%d个完成", ar.Code, len(ar.Data))
}

func NewManager() *UserAnalysis {
	return &UserAnalysis{
		UserMapping:      map[string]BaseUserId{},
		EntUserIdMapping: map[int64]BaseUserId{},
		FullBaseUserId:   map[BaseUserId]bool{},
		BinPhone:         map[BaseUserId]bool{},
		BindMail:         map[BaseUserId]bool{},
		Vip:              map[BaseUserId]bool{},
		Vip15:            map[BaseUserId]bool{},
		Vip30:            map[BaseUserId]bool{},
		Member:           map[BaseUserId]bool{},
		MemberExpire:     map[BaseUserId]bool{},
		TestUser:         map[BaseUserId]bool{},
		UnSubUser:        map[string]bool{},
		NewUser:          map[int]map[BaseUserId]bool{},
	}
}

func (ua *UserAnalysis) LoadMapping() error {
	//加载baseUserId对应关系
	data, err := g.DB().Query(ctx, "SELECT mgoUserId,positionId,baseUserId,phone FROM dwd_mgo_position")
	if err != nil {
		return err
	}
	var (
		newMapping             = map[string]BaseUserId{}
		fullBaseUserId         = map[BaseUserId]bool{}
		phoneBaseUserIdMapping = map[string]BaseUserId{}
		entIdBaseUserIdMapping = map[int64]BaseUserId{}
	)
	for _, m := range data.List() {
		var (
			mgoUserId  = gconv.String(m["mgoUserId"])
			positionId = gconv.String(m["positionId"])
			baseUserId = BaseUserId(gconv.Int64(m["baseUserId"]))
			phone      = gconv.String(m["phone"])
		)
		newMapping[mgoUserId] = baseUserId
		newMapping[positionId] = baseUserId
		fullBaseUserId[baseUserId] = true
		if phone != "" {
			phoneBaseUserIdMapping[phone] = baseUserId
		}
	}
	ua.UserMapping = newMapping
	ua.FullBaseUserId = fullBaseUserId

	//加载ent_id和BaseUserId对应关系
	dataEntRes, err := g.DB("jianyu").Query(ctx, "SELECT id,phone FROM entniche_user")
	if err != nil {
		return err
	}
	for _, m := range dataEntRes {
		var (
			entId = gconv.Int64(m["id"])
			phone = gconv.String(m["phone"])
		)
		if bId, ok := phoneBaseUserIdMapping[phone]; ok {
			entIdBaseUserIdMapping[entId] = bId
		}
	}
	ua.EntUserIdMapping = entIdBaseUserIdMapping
	return nil
}