workTasks/userSign/userAnalysis/analysisManager.go

package userAnalysis

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

type (
	UserAnalysis struct {
		UserMapping map[string]BaseUserId //职位id、mongouserId对应的baseUserId
		//BaseUserClubMap          map[string]BaseUserId //电销线索
		TelemarketingBaseUserMap map[string]BaseUserId //电销线索

		EntUserIdMapping map[int64]BaseUserId //ent_id对应的baseUserId
		FullBaseUserId   map[BaseUserId]bool  //全量BaseUserId

		BinPhone, BindMail   map[BaseUserId]bool
		Vip, Vip15, Vip30    map[BaseUserId]bool
		VipExpired30         map[BaseUserId]bool //已到期30天
		VipGift7DayExpired   map[BaseUserId]bool //赠送7天到期
		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{},
		VipExpired30:       map[BaseUserId]bool{}, //已到期30天
		VipGift7DayExpired: map[BaseUserId]bool{}, //赠送7天到期
		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 {
	ctx := gctx.New()
	//加载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{}
		telemarketingBaseUserMap = map[string]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

	//加载线索
	dwdBaseinfoRes, err := g.DB("subjectdb").Query(ctx, "SELECT uid,base_user_id FROM dwd_f_userbase_baseinfo WHERE userid!=''")
	if err != nil {
		return err
	}
	for _, m := range dwdBaseinfoRes.List() {
		var (
			uid          = gconv.String(m["uid"])
			base_user_id = BaseUserId(gconv.Int64(m["base_user_id"]))
		)
		if uid != "" && base_user_id != 0 {
			telemarketingBaseUserMap[uid] = base_user_id
		}
	}

	clueRes, err := g.DB("subjectdb").Query(ctx, "SELECT id,uid FROM dwd_f_crm_clue_info WHERE userid!=''")
	if err != nil {
		return err
	}
	for _, m := range clueRes.List() {
		if value, ok := telemarketingBaseUserMap[gconv.String(m["uid"])]; ok {
			telemarketingBaseUserMap[gconv.String(m["id"])] = value
		}
	}
	ua.TelemarketingBaseUserMap = telemarketingBaseUserMap
	return nil
}