feat:荟聚迁移二期-判断用户是否在群组中

entity/db.go

@@ -3,6 +3,7 @@ package entity
 import (
 	"app.yhyue.com/moapp/jybase/mongodb"
 	"app.yhyue.com/moapp/jybase/mysql"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/go-xorm/xorm"
 )
 
@@ -13,3 +14,4 @@ var SaveEngine *xorm.Engine
 var Mongo mongodb.MongodbSim
 var MainMysql *mysql.Mysql
 var BaseMysql *mysql.Mysql
+var ClickhouseConn driver.Conn

go.mod

@@ -5,6 +5,7 @@ go 1.19
 require (
 	app.yhyue.com/moapp/jybase v0.0.0-20240424025716-c77615e9004e
 	app.yhyue.com/moapp/jypkg v1.15.7
+	github.com/ClickHouse/clickhouse-go/v2 v2.2.0
 	github.com/garyburd/redigo v1.6.2
 	github.com/gin-contrib/sessions v0.0.5
 	github.com/go-sql-driver/mysql v1.7.1
@@ -13,8 +14,11 @@ require (
 	github.com/zeromicro/go-zero v1.5.3
 	google.golang.org/grpc v1.56.1
 	google.golang.org/protobuf v1.31.0
+
 )
 
+replace github.com/go-xorm/xorm v0.7.9 => gitea.com/xorm/xorm v0.7.9
+
 require (
 	bp.jydev.jianyu360.cn/BaseService/entManageApplication v0.0.0-20230214091519-89a98c01ab0e // indirect
 	bp.jydev.jianyu360.cn/BaseService/powerCheckCenter v0.0.0-20230225125145-431a4f70093a // indirect
@@ -68,12 +72,15 @@ require (
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/openzipkin/zipkin-go v0.4.1 // indirect
+	github.com/paulmach/orb v0.7.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
+	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.15.1 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/ugorji/go/codec v1.1.7 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
@@ -110,6 +117,7 @@ require (
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gorm.io/driver/mysql v1.0.5 // indirect

rpc/etc/activity.yaml

@@ -1,8 +1,8 @@
 Name: activity.rpc
 ListenOn: 127.0.0.1:8086
 RedisC:
-  Host: 192.168.3.11
-  addr: 192.168.3.11:1712
+  Host: 192.168.3.149
+  addr: 192.168.3.149:1712
   modules: other
   poolMaxSize: 10
   poolMaxIdle: 20
@@ -12,11 +12,11 @@ RedisC:
   idleTimeOut: 240
 Etcd:
   Hosts:
-    - 127.0.0.1:2379
+    - 192.168.3.206:2379
   Key: activity.rpc
   Timeout: 20000
-DataSource: root:Topnet123@tcp(192.168.3.11:3366)/jyactivities?charset=utf8mb4&parseTime=true&loc=Local
-EntDataSource: root:Topnet123@tcp(192.168.3.11:3366)/jianyu?charset=utf8mb4&parseTime=true&loc=Local
+DataSource: "root:=PDT49#80Z!RVv52_z@tcp(192.168.3.217:4000)/jyactivities?charset=utf8mb4&parseTime=true&loc=Local"
+EntDataSource: "root:=PDT49#80Z!RVv52_z@tcp(192.168.3.217:4000)/jianyu?charset=utf8mb4&parseTime=true&loc=Local"
 MaxConn: 10
 MaxIdle: 20
 MaxLife: 100
@@ -65,4 +65,11 @@ mysql:
     userName: root
     password: '=PDT49#80Z!RVv52_z'
     maxOpenConns: 5
-    maxIdleConns: 5
+    maxIdleConns: 5
+clickhouse:
+  Addr: 192.168.3.207:19000
+  UserName: jytop
+  Password: pwdTopJy123
+  DbName: information
+  MaxIdleConns: 5
+  MaxOpenConns: 5

rpc/init/db.go

@@ -6,6 +6,9 @@ import (
 	"app.yhyue.com/moapp/jyMarketing/util"
 	"app.yhyue.com/moapp/jybase/mongodb"
 	"app.yhyue.com/moapp/jybase/mysql"
+	"context"
+	"fmt"
+	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/go-xorm/xorm"
 	"github.com/zeromicro/go-zero/core/logx"
 	"time"
@@ -102,3 +105,33 @@ func MysqlInit(C config.Config) {
 	}
 	//mongo初始化
 }
+func ConnectClickhouse(C config.Config) error {
+	var (
+		ctx = context.Background()
+		err error
+	)
+	entity.ClickhouseConn, err = clickhouse.Open(&clickhouse.Options{
+		Addr:         []string{C.Clickhouse.Address},
+		DialTimeout:  10 * time.Second,
+		MaxIdleConns: C.Clickhouse.MaxIdleConns,
+		MaxOpenConns: C.Clickhouse.MaxOpenConns,
+		Auth: clickhouse.Auth{
+			Database: C.Clickhouse.DbName,
+			Username: C.Clickhouse.Username,
+			Password: C.Clickhouse.Password,
+		},
+		Debugf: func(format string, v ...interface{}) {
+			fmt.Printf(format, v)
+		},
+	})
+	if err != nil {
+		return err
+	}
+	if err := entity.ClickhouseConn.Ping(ctx); err != nil {
+		if exception, ok := err.(*clickhouse.Exception); ok {
+			fmt.Printf("Exception [%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
+		}
+		return err
+	}
+	return nil
+}

rpc/init/init.go

@@ -3,6 +3,7 @@ package init
 import (
 	"app.yhyue.com/moapp/jyMarketing/entity"
 	"app.yhyue.com/moapp/jyMarketing/rpc/internal/config"
+	"app.yhyue.com/moapp/jybase/log"
 	"app.yhyue.com/moapp/jypkg/compatible"
 	"app.yhyue.com/moapp/jypkg/middleground"
 	"flag"
@@ -35,6 +36,11 @@ func init() {
 	conf.MustLoad(*configFile, &C)
 	MysqlInit(C)
 	MongoDBInit(C.Mongo)
+	err := ConnectClickhouse(C)
+	if err != nil {
+		log.Fatal("clickhouse 连接失败")
+		return
+	}
 	Middleground = middleground.NewMiddleground(C.Etcd.Hosts).
 		RegUserCenter(C.Middleground.UserCenterKey).
 		RegPowerCheckCenter(C.Middleground.PowerCheckCenterKey).

rpc/internal/config/config.go

@@ -32,7 +32,8 @@ type Config struct {
 		ResourceCenterKey    string
 		EntManageApplication string
 	}
-	Mysql Mysql
+	Mysql      Mysql
+	Clickhouse Clickhouse
 }
 
 type Redis struct {
@@ -67,3 +68,12 @@ type MysqlStruct struct {
 	MaxOpenConns int    `json:"maxOpenConns"`
 	MaxIdleConns int    `json:"maxIdleConns"`
 }
+
+type Clickhouse struct {
+	Username     string `json:"username"`
+	Password     string `json:"password"`
+	Address      string `json:"address"`
+	DbName       string `json:"dbName"`
+	MaxIdleConns int    `json:"maxIdleConns"`
+	MaxOpenConns int    `json:"maxOpenConns"`
+}

service/activityService.go

@@ -6,6 +6,7 @@ import (
 	C "app.yhyue.com/moapp/jyMarketing/rpc/init"
 	"app.yhyue.com/moapp/jyMarketing/util"
 	"app.yhyue.com/moapp/jybase/common"
+	"app.yhyue.com/moapp/jybase/mongodb"
 	"fmt"
 	_ "github.com/garyburd/redigo/redis"
 	"github.com/zeromicro/go-zero/core/logx"
@@ -839,19 +840,26 @@ func ActivityProduct(activityId int64) (int64, string, []entity.ProductJson) {
 	return entity.SuccessCode, "查询对应的产品信息成功", productNameList
 }
 
-func IsInGroup(userId string, groupId []string) (bool, string) {
-	tmp := []int64{}
-	var exist bool
-	orm := entity.TiEngine
-	err := orm.Select("count(*)").Table("customers_user").Alias("u").Where("u.userid=?", userId).Join("left", "groupcustomers as g", "g.customerid = u.customerid").In("g.groupid", groupId).Find(&tmp)
-	if err != nil {
-		logx.Info("查询用户是否在分组中失败", err)
-		return false, "查询用户是否在分组中失败"
+func IsInGroup(userId string, groupId []string) (exist bool, err string) {
+	// 用户id 不是mongoId    直接返回
+	// 选择标签分组时 只处理个人身份领取优惠券
+	if !mongodb.IsObjectIdHex(userId) {
+		return false, ""
+	}
+	selectGroupId := []int64{}
+	for i := 0; i < len(groupId); i++ {
+		tmpId, err := strconv.ParseInt(groupId[i], 10, 64)
+		if err != nil {
+			continue
+		}
+		selectGroupId = append(selectGroupId, tmpId)
 	}
-	if len(tmp) > 0 && tmp[0] > 0 {
-		exist = true
+	if len(selectGroupId) == 0 {
+		return false, ""
 	}
-	return exist, ""
+	uic := NewUserIdConstructor(selectGroupId, 0)
+	count := uic.CountUser(userId)
+	return common.If(count > 0, true, false).(bool), ""
 }
 
 // 活动参与数量是否达上限

service/userGroupService.go

@@ -0,0 +1,283 @@
+package service
+
+import (
+	"app.yhyue.com/moapp/jyMarketing/entity"
+	"app.yhyue.com/moapp/jybase/common"
+	"context"
+	"fmt"
+	"log"
+	"strings"
+)
+
+const (
+	logical_operator_normal = 0                           // 正常运算标签
+	logical_operator_not    = 1                           // 非运算标签
+	tag_operator_and        = 1                           // 且
+	tag_operator_or         = 2                           // 或
+	Tabledwd_d_tag          = "pub_tags.dwd_d_tag"        // 标签用户表  todo 后边调整
+	Tabledwd_mgo_position   = "pub_tags.dwd_mgo_position" // base_user_id 对应的mgoid
+	FullUserTagSql          = `SELECT groupBitmapAndState(bitobj) as userIds from pub_tags.dwd_d_tag ddt WHERE  ddt.id=2017`
+	andSql                  = `SELECT groupBitmapAndState(bitobj) as userIds from pub_tags.dwd_d_tag ddt WHERE  ddt.id in (%s) `
+	orSql                   = `SELECT groupBitmapOrState(bitobj) as userIds from pub_tags.dwd_d_tag ddt WHERE  ddt.id in (%s) `
+	hasAllSql               = ` bitmapHasAll( ddut.bitobj,bitmapBuild(%s)) `
+	hasAnySql               = ` bitmapHasAny( ddut.bitobj,bitmapBuild(%s)) `
+	notHasAllSql            = ` not bitmapHasAll( ddut.bitobj,bitmapBuild(%s)) `
+	notHasAnySql            = ` not bitmapHasAny( ddut.bitobj,bitmapBuild(%s)) `
+	countUserSql            = `SELECT
+	COUNT(1) as count
+FROM
+	pub_tags.dwd_d_user_tag ddut
+LEFT JOIN pub_tags.dwd_mgo_position dmp ON
+	(ddut.baseUserId = toUInt64(dmp.baseUserId))
+WHERE
+	dmp.mgoUserId = '%s'
+	AND dmp.type= 0  AND (%s)`
+)
+
+// UserIdConstructor 用户群组标签转换
+type UserIdConstructor struct {
+	groupFilter      []int64         // 群组过滤条件
+	userGtFilter     int64           // 用户过滤条件  暂停发消息时 用的
+	userGroupTagList []*UserGroupTag // 用户群组标签列表 （整理后的）
+	baseQuerySQL     string          // 查询群组下base_user_id 的sql
+	countUserSQL     string          // 查询用户标签是否符合群组标签的sql
+}
+
+type UserGroupTag struct {
+	GroupId     int64   // 群组id
+	TagOperator int64   // 群组内关系
+	NormalTag   []int64 // 正常标签
+	NotTag      []int64 // 非标签
+}
+
+func NewUserIdConstructor(groupFilter []int64, userGtFilter int64) (u *UserIdConstructor) {
+	u = &UserIdConstructor{
+		groupFilter:      groupFilter,
+		userGtFilter:     userGtFilter,
+		userGroupTagList: []*UserGroupTag{},
+	}
+	return
+}
+
+// GetGroupTags 获取用户群组标签信息
+func (u *UserIdConstructor) getGroupTags() *[]map[string]interface{} {
+	groupIdFilter := []string{}
+	groupIdValue := []interface{}{}
+	where := ""
+	for i := 0; i < len(u.groupFilter); i++ {
+		groupIdFilter = append(groupIdFilter, "?")
+		groupIdValue = append(groupIdValue, u.groupFilter[i])
+	}
+	where = fmt.Sprintf("where ugt.group_id in (%s)", strings.Join(groupIdFilter, ","))
+	query := fmt.Sprintf(`SELECT ugt.group_id,ug.tag_operator,ugt.tag_id,ugt.logical_operator FROM convertlabsync.user_group_tag ugt left join user_group ug  on (ugt.group_id=ug.id) %s`, where)
+	rs := entity.BaseMysql.SelectBySql(query, groupIdValue...)
+	return rs
+}
+
+// InitTagList 处理成方便用的数组
+func (u *UserIdConstructor) InitTagList() bool {
+	rs := u.getGroupTags()
+	if rs == nil || len(*rs) == 0 {
+		return false
+	}
+	groupMap := map[int64]*UserGroupTag{}
+	for i := 0; i < len(*rs); i++ {
+		groupId := common.Int64All((*rs)[i]["group_id"])
+		tagOperator := common.Int64All((*rs)[i]["tag_operator"])
+		tagId := common.Int64All((*rs)[i]["tag_id"])
+		logicalOperator := common.IntAll((*rs)[i]["logical_operator"])
+		if _, ok := groupMap[groupId]; !ok {
+			groupMap[groupId] = &UserGroupTag{
+				GroupId:     groupId,
+				TagOperator: tagOperator,
+				NormalTag:   []int64{},
+				NotTag:      []int64{},
+			}
+		}
+		// 追加
+		switch logicalOperator {
+		case logical_operator_normal:
+			groupMap[groupId].NormalTag = append(groupMap[groupId].NormalTag, tagId)
+		case logical_operator_not:
+			groupMap[groupId].NotTag = append(groupMap[groupId].NotTag, tagId)
+		}
+	}
+	for _, v := range groupMap {
+		u.userGroupTagList = append(u.userGroupTagList, v)
+	}
+	return true
+}
+
+// 转换成sql   这需要判断
+// '正常标签'这里指不是非运算
+// toBaseQuerySQL 转换成查询baseUserId 的sql
+func (u *UserIdConstructor) toBaseQuerySQL() string {
+	sqlList := []string{} // 包含多个群组的sql
+	for i := 0; i < len(u.userGroupTagList); i++ {
+		// 拼接群组内sql
+		groupTag := u.userGroupTagList[i]
+		normalTagSQL, notTagSQL := "", ""
+		tagSql := ""
+		switch groupTag.TagOperator {
+		case tag_operator_and:
+			if len(groupTag.NormalTag) > 0 { // 正常标签
+				normalTagList := []string{}
+				for j := 0; j < len(groupTag.NormalTag); j++ {
+					normalTagList = append(normalTagList, fmt.Sprint(groupTag.NormalTag[j]))
+				}
+				normalTagSQL = fmt.Sprintf(andSql, strings.Join(normalTagList, ","))
+			}
+			if len(groupTag.NotTag) > 0 { // 非标签
+				notTagList := []string{}
+				for j := 0; j < len(groupTag.NotTag); j++ {
+					notTagList = append(notTagList, fmt.Sprint(groupTag.NotTag[j]))
+				}
+				notTagSQL = fmt.Sprintf(orSql, strings.Join(notTagList, ","))
+			}
+			// 同时有：  正常标签 - 非标签
+			if normalTagSQL != "" && notTagSQL != "" {
+				tagSql = fmt.Sprintf("SELECT  bitmapAndnot((%s),(%s)) as userIds", normalTagSQL, notTagSQL)
+			} else if normalTagSQL != "" {
+				// 只有正常标签 ： 正常标签
+				tagSql = normalTagSQL
+			} else if notTagSQL != "" {
+				// 只有非标签 ：   全量标签-非标签
+				tagSql = fmt.Sprintf("SELECT  bitmapAndnot((%s),(%s)) as userIds", FullUserTagSql, notTagSQL)
+			}
+
+		case tag_operator_or:
+			if len(groupTag.NormalTag) > 0 { // 正常标签
+				normalTagList := []string{}
+				for j := 0; j < len(groupTag.NormalTag); j++ {
+					normalTagList = append(normalTagList, fmt.Sprint(groupTag.NormalTag[j]))
+				}
+				normalTagSQL = fmt.Sprintf(orSql, strings.Join(normalTagList, ","))
+			}
+			if len(groupTag.NotTag) > 0 { // 非标签
+				notTagList := []string{}
+				for j := 0; j < len(groupTag.NotTag); j++ {
+					notTagList = append(notTagList, fmt.Sprint(groupTag.NotTag[j]))
+				}
+				notTagSQL = fmt.Sprintf(andSql, strings.Join(notTagList, ","))
+			}
+			// 同时有：  正常标签 ∪ (U-(B∩C∩D....))  U:全量标签 B、C、D... 非标签
+			if normalTagSQL != "" && notTagSQL != "" {
+				tmpNotTagSql := fmt.Sprintf("SELECT  bitmapAndnot((%s),(%s)) as userIds", FullUserTagSql, notTagSQL)
+				tagSql = fmt.Sprintf("SELECT  bitmapOr((%s),(%s)) as userIds", FullUserTagSql, tmpNotTagSql)
+			} else if normalTagSQL != "" {
+				// 只有正常标签
+				tagSql = normalTagSQL
+			} else if notTagSQL != "" {
+				// 只有非标签：    U-(B∩C∩D....)  U:全量标签 B、C、D... 非标签
+				tagSql = fmt.Sprintf("SELECT  bitmapAndnot((%s),(%s)) as userIds", FullUserTagSql, notTagSQL)
+			}
+		}
+		sqlList = append(sqlList, tagSql)
+	}
+	// 如果用户有过滤
+	if u.userGtFilter > 0 {
+		u.baseQuerySQL = fmt.Sprintf("SELECT  arrayFilter(x -> x >%v,bitmapToArray( groupBitmapOrState(userIds))) as userIds from (%s)", u.userGtFilter, strings.Join(sqlList, " UNION    DISTINCT  "))
+	} else {
+		u.baseQuerySQL = fmt.Sprintf("SELECT  bitmapToArray( groupBitmapOrState(  userIds)) as userIds  from (%s)", strings.Join(sqlList, " UNION    DISTINCT  "))
+	}
+	fmt.Println("baseQuerySQL:", u.baseQuerySQL)
+	return u.baseQuerySQL
+}
+
+// 从数据库查询
+func (u *UserIdConstructor) QueryBaseUserIdList() (userList []int64) {
+	if !u.InitTagList() {
+		return []int64{}
+	}
+	rows := entity.ClickhouseConn.QueryRow(context.Background(), u.toBaseQuerySQL())
+	if err := rows.Scan(&userList); err != nil {
+		log.Println("QueryBaseUserIdList err:", err)
+		return
+	}
+	return userList
+}
+
+// 判断活动群组id 和 用户身上的的标签是否匹配
+// 分组之间用 or 连接
+// 分组内
+// 且： 正常标签： bitmapHasAll  and  (not bitmapHasAny )
+// 或：bitmapHasAny  or (not bitmapHasAny())
+func (u *UserIdConstructor) toCountUserSQL(userId string) string {
+	sqlList := []string{} // 包含多个群组的sql
+	for i := 0; i < len(u.userGroupTagList); i++ {
+		// 拼接群组内sql
+		groupTag := u.userGroupTagList[i]
+		normalTagSQL, notTagSQL := "", ""
+		tagSql := ""
+		switch groupTag.TagOperator {
+		case tag_operator_and:
+			if len(groupTag.NormalTag) > 0 { // 正常标签
+				normalTagList := []string{}
+				for j := 0; j < len(groupTag.NormalTag); j++ {
+					normalTagList = append(normalTagList, fmt.Sprint(groupTag.NormalTag[j]))
+				}
+				normalTagSQL = fmt.Sprintf(hasAllSql, strings.Join(normalTagList, ","))
+			}
+			if len(groupTag.NotTag) > 0 { // 非标签
+				notTagList := []string{}
+				for j := 0; j < len(groupTag.NotTag); j++ {
+					notTagList = append(notTagList, fmt.Sprint(groupTag.NotTag[j]))
+				}
+				notTagSQL = fmt.Sprintf(notHasAnySql, strings.Join(notTagList, ","))
+			}
+			// 同时有：  正常标签 and 非标签
+			if normalTagSQL != "" && notTagSQL != "" {
+				tagSql = fmt.Sprintf("(%s and %s)", normalTagSQL, notTagSQL)
+			} else if normalTagSQL != "" {
+				// 只有正常标签 ： 正常标签
+				tagSql = fmt.Sprintf("(%s)", normalTagSQL)
+			} else if notTagSQL != "" {
+				// 只有非标签 ：
+				tagSql = fmt.Sprintf("(%s)", notTagSQL)
+			}
+
+		case tag_operator_or:
+			if len(groupTag.NormalTag) > 0 { // 正常标签
+				normalTagList := []string{}
+				for j := 0; j < len(groupTag.NormalTag); j++ {
+					normalTagList = append(normalTagList, fmt.Sprint(groupTag.NormalTag[j]))
+				}
+				normalTagSQL = fmt.Sprintf(hasAnySql, strings.Join(normalTagList, ","))
+			}
+			if len(groupTag.NotTag) > 0 { // 非标签
+				notTagList := []string{}
+				for j := 0; j < len(groupTag.NotTag); j++ {
+					notTagList = append(notTagList, fmt.Sprint(groupTag.NotTag[j]))
+				}
+				notTagSQL = fmt.Sprintf(notHasAllSql, strings.Join(notTagList, ","))
+			}
+			// 同时有：  正常标签  or 非标签
+			if normalTagSQL != "" && notTagSQL != "" {
+				tagSql = fmt.Sprintf(" (%s or %s) ", normalTagSQL, notTagSQL)
+			} else if normalTagSQL != "" {
+				// 只有正常标签 ： 正常标签
+				tagSql = fmt.Sprintf(" (%s) ", normalTagSQL)
+			} else if notTagSQL != "" {
+				// 只有非标签 ：
+				tagSql = fmt.Sprintf(" (%s) ", notTagSQL)
+			}
+		}
+		sqlList = append(sqlList, tagSql)
+	}
+	u.countUserSQL = fmt.Sprintf(countUserSql, userId, strings.Join(sqlList, " or "))
+	fmt.Println("baseQuerySQL:", u.baseQuerySQL)
+	return u.countUserSQL
+}
+
+// 从数据库查询
+func (u *UserIdConstructor) CountUser(userId string) (count int64) {
+	if !u.InitTagList() {
+		return 0
+	}
+	rows := entity.ClickhouseConn.QueryRow(context.Background(), u.toCountUserSQL(userId))
+	if err := rows.Scan(&count); err != nil {
+		log.Println("QueryBaseUserIdList err:", err)
+		return
+	}
+	return count
+}