newuserTask/newuserGet/internal/logic/task.go

package logic

import (
	"context"
	"errors"
	"fmt"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"newuserGet/internal/dao"
	"newuserGet/internal/model/do"
	"newuserGet/internal/model/entity"
	"strings"
	"time"
)

func Task(ctx context.Context) {
	start := time.Now()
	defer func() {
		g.Log().Info(ctx, "本轮完成", time.Since(start))

	}()
	// get rule
	var rules []entity.NewUserSendRule
	err := dao.NewUserSendRule.Ctx(ctx).Scan(&rules)
	if err != nil {
		g.Log().Error(ctx, "获取规则失败", err)
		return
	}
	// 遍历规则
	for i := 0; i < len(rules); i++ {
		g.Log().Infof(ctx, "开始处理第%v个规则:id-%v name:%v \n", i, rules[i].Id, rules[i].Name)
		// 获取注册时间规则用户和过滤行为
		userList, err := getFromRegisterBehavior(ctx, rules[i])
		if err != nil {
			g.Log().Error(ctx, err)
			continue
		}
		if len(userList) == 0 {
			g.Log().Info(ctx, "当前规则getFromRegisterBehavior未获取到有效用户", rules[i].Id, rules[i].Name)
			continue
		}
		if rules[i].CallState != "" {
			// 过滤外呼状态
			userList, err = filterCallState(ctx, rules[i], userList)
			if err != nil {
				g.Log().Error(ctx, err)
				continue
			}
			if len(userList) == 0 {
				g.Log().Info(ctx, "当前规则filterCallState未获取到有效用户", rules[i].Id, rules[i].Name)
				continue
			}
		}
		if rules[i].Trailstatus != "" {
			// 过滤销售进程
			userList = filterTrailStatus(ctx, rules[i], userList)
		}
		// 保存用户
		saveList := []do.NewUserSendLog{}
		for j := 0; j <= len(userList); j++ {
			saveList = append(saveList, do.NewUserSendLog{
				Phone:      userList[j].Phone,
				UserId:     userList[j].MgoId,
				CreateTime: gtime.Now(),
			})
		}
		_, err = dao.NewUserSendLog.Ctx(ctx).Data(saveList).Insert()
		if err != nil {
			g.Log().Errorf(ctx, "第%v个规则 保存失败 :id-%v name:%v err:%v  %v\n", i, rules[i].Id, rules[i].Name, err, saveList)
			continue
		}
		g.Log().Infof(ctx, "完成--第%v个规则:id-%v name:%v \n", i, rules[i].Id, rules[i].Name)

	}

}

type User struct {
	Phone string `json:"phone"`
	MgoId string `json:"mgoId"`
}

const (
	OrSQL          = "SELECT  bitobj from pub_tags.dwd_d_tag ddt WHERE  ddt.id in (%s) "
	BitMapSQL      = "SELECT bitmapAnd((%s),(%s)) as userIds"
	QueryUserIdSQL = `SELECT
	DISTINCT dmp.mgoUserId as mgoId,dmp.phone as phone
from
	pub_tags.dwd_mgo_position dmp
WHERE
	bitmapHasAny( (%s),
	bitmapBuild([toUInt64(dmp.baseUserId)]))  and dmp.mgoUserId!='' and dmp.phone!='';`
	CallStateNotDialed = "0"       // 未拨打
	CallStateNotDeal   = "notDeal" // 振铃未接通
	CallStateIvr       = "ivr"     // ivr
	CallStatDealing    = "dealing" //已接听
)

// 通过时间和行为计算用户
func getFromRegisterBehavior(ctx context.Context, rule entity.NewUserSendRule) (userList []*User, err error) {
	start := time.Now()
	defer func() {
		g.Log().Info(ctx, "getFromRegisterBehavior 耗时：", time.Since(start))

	}()
	//   时间标签并 之后与行为标签交
	// 处理注册时间标签
	if rule.RegisterTagIds == "" {
		return nil, errors.New("注册时间标签id为空")
	}
	regQuery := fmt.Sprintf(OrSQL, rule.RegisterTagIds)
	filterQuery := regQuery
	if rule.BehaviorTagIds != "" {
		//拼接行为标签sql
		behQuery := fmt.Sprintf(OrSQL, rule.BehaviorTagIds)
		filterQuery = fmt.Sprintf(BitMapSQL, regQuery, behQuery)
	}
	// 处理mgoid和手机号 拼接sql
	finalQuery := fmt.Sprintf(QueryUserIdSQL, filterQuery)
	g.Log().Info(ctx, "getFromRegisterBehavior 查询sql", finalQuery)
	err = dao.DwdDUserTag.DB().Raw(finalQuery).Scan(&userList)
	if err != nil {
		return nil, err
	}
	g.Log().Info(ctx, userList)
	return
}

// 外呼状态
func filterCallState(ctx context.Context, rule entity.NewUserSendRule, userList []*User) (newUserList []*User, err error) {
	start := time.Now()
	defer func() {
		g.Log().Info(ctx, "filterCallState 耗时：", time.Since(start))

	}()
	callstateList := strings.Split(rule.CallState, ",")
	for i := 0; i < len(userList); i++ {
		for j := 0; j < len(callstateList); j++ {
			// 未拨打  判断不存在
			if callstateList[j] == CallStateNotDialed {
				exist, err := dao.VoiceRecord.Ctx(ctx).Where("CalledNo=?", userList[i].Phone).Exist()
				if err != nil {
					g.Log().Error(ctx, "filterCallState CallStateNotDialed err:", err)
					continue
				}
				if !exist {
					newUserList = append(newUserList, userList[i])
					continue
				}
			}
			// 已接听 至少一个记录
			if callstateList[j] == CallStatDealing {
				exist, err := dao.VoiceRecord.Ctx(ctx).Where("CalledNo=? and State=?", userList[i].Phone, CallStatDealing).Exist()
				if err != nil {
					g.Log().Error(ctx, "filterCallState CallStatDealing err:", err, userList[i])
					continue
				}
				if exist {
					newUserList = append(newUserList, userList[i])
					continue
				}
			}
			// 其他的 判断是否存在记录
			findState := strings.Split(callstateList[j], "#")
			exist, err := dao.VoiceRecord.Ctx(ctx).Where("CalledNo=?", userList[i].Phone).WhereNotIn(dao.VoiceRecord.Columns().CallState, findState).Exist()
			if err != nil {
				g.Log().Error(ctx, "filterCallState otherstate err:", err, findState, userList[i])
				continue
			}
			if !exist {
				continue
			}
			// 存在则查询是否包含其他记录
			exist, err = dao.VoiceRecord.Ctx(ctx).Where("CalledNo=?", userList[i].Phone).WhereNotIn(dao.VoiceRecord.Columns().CallState, findState).Exist()
			if err != nil {
				g.Log().Error(ctx, "filterCallState otherstate notin err:", err, findState, userList[i])
				continue
			}
			if !exist {
				newUserList = append(newUserList, userList[i])
				continue
			}
		}
	}
	return
}

// 跟进状态
func filterTrailStatus(ctx context.Context, rule entity.NewUserSendRule, userList []*User) (newUserList []*User) {
	start := time.Now()
	defer func() {
		g.Log().Info(ctx, "filterTrailStatus 耗时：", time.Since(start))

	}()
	for i := 0; i < len(userList); i++ {

	}
	return
}