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

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

	AnalysisRes struct {
		Name, Code string              //标签名字
		Data       map[BaseUserId]bool //数据
	}
)

// 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
	)
	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)
		}
	}

	for i, batch := range updateBatch {
		if i == 0 {
			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, "code %s 更新%d个完成", ar.Code, len(ar.Data))
}

func NewManager() *UserAnalysis {
	return &UserAnalysis{
		UserMapping:  map[string]BaseUserId{},
		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{},
	}
}

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{}
		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
		if phone != "" {
			phoneBaseUserIdMapping[phone] = baseUserId
		}
	}
	ua.UserMapping = newMapping

	//加载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
}