datatools/clueSync/tag.go

package main

import (
	"app.yhyue.com/moapp/jybase/common"
	"context"
	"fmt"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/zeromicro/go-zero/core/logx"
	"log"
	"regexp"
	"strings"
	"time"
)

func tagAllSync() {
	//dwd_f_userbase_baseinfo -->l_registedate 注册日期 -->userid
	//dwd_f_userbase_visit_info -->date 访问时间 -->userid
	//dwd_f_userbase_search_info -->search_area去重 jianyu_subjectdb_test.d_area_code -->userid
	//dwd_f_userbase_search_info -->search_word去重 -->userid
	//dwd_f_userbase_subscribe_info -->subscribe_areas去重 jianyu_subjectdb_test.d_area_code -->userid
	//dwd_f_userbase_subscribe_info -->subscribe_keywords去重 -->userid
	//dwd_f_userbase_order_info -->product_type去重 -->userid
	//dwd_f_userbase_order_info -->product_type=VIP订阅 vip_endtime 一周内到期、一月内到期 -->userid  暂时不要了
	//dwd_f_userbase_event_info -->createtime查昨天有没有数据， 有数据-昨日浏览过 没数据-昨日未浏览 -->userid
	log.Println("用户标签定时任务开始")
	// TiDb.ExecBySql(`SET session max_execution_time=86400`)
	count := 0
	now := time.Now()
	var allUser []map[string]interface{}
	startOfDay := time.Date(now.Year(), now.Month(), now.Day()-1, 0, 0, 0, 0, now.Location())
	TiDb.SelectByBath(100, func(l *[]map[string]interface{}) bool {
		for _, v := range *l {
			allUser = append(allUser, v)
		}
		return true
	}, fmt.Sprintf(`select userid from dwd_f_userbase_visit_info where  createtime>"%s"   order by createtime asc`, startOfDay.Format(time.DateTime)))
	TiDb.Update("dwd_f_crm_attribute_label", map[string]interface{}{}, map[string]interface{}{
		"members_info": "昨日未浏览",
		"updatetime":   time.Now().Format("2006-01-02 15:04:05"),
	})
	for _, v := range allUser {
		count++
		//查询用户信息
		userId := common.ObjToString(v["userid"])
		userData := TiDb.SelectBySql("select l_registedate,userid,uid,base_user_id from dwd_f_userbase_baseinfo where  userid=?", userId)
		for _, vv := range *userData {
			FormatTag(vv, count)
		}
		log.Println(count)
	}
	log.Println("用户标签定时任务结束")
}

func FormatTag(data map[string]interface{}, count int) {
	registedate := common.ObjToString(data["l_registedate"])
	userId := common.ObjToString(data["userid"])
	uId := common.ObjToString(data["uid"])
	base_user_id := common.Int64All(data["base_user_id"])
	log.Println("第", count, "条：", uId, registedate, userId)
	if registedate == "" || uId == "" || userId == "" {
		log.Println("缺少信息 ", registedate, uId, userId)
		return
	}
	date, search_areass, search_wordss, product_types, subscribe_keywords, subscribe_areas := "", "", "", "", "", ""
	//访问数据
	visitData := TiDb.FindOne("dwd_f_userbase_visit_info", map[string]interface{}{"userid": userId}, "", "date desc")
	if visitData != nil && len(*visitData) > 0 {
		date = common.ObjToString((*visitData)["date"])
	}

	//搜索数据
	searchData := TiDb.Find("dwd_f_userbase_search_info", map[string]interface{}{"userid": userId}, "", "", -1, -1)
	if searchData != nil && len(*searchData) > 0 {
		search_areas, search_words := "", ""
		for k, v := range *searchData {
			search_area := common.ObjToString(v["search_area"])
			search_word := common.ObjToString(v["search_word"])
			if k == len(*searchData)-1 && search_area != "" {
				search_areas += search_area
			} else if search_area != "" {
				search_areas += search_area + ","
			}
			if k == len(*searchData)-1 && search_word != "" {
				search_words += search_word
			} else if search_word != "" {
				search_words += search_word + ","
			}
		}
		search_areas_arr, search_words_arr := []string{}, []string{}
		for _, v := range strings.Split(search_areas, ",") {
			isOk := false
			for _, vv := range search_areas_arr {
				if vv == v {
					isOk = true
				}
			}
			if !isOk && v != "" {
				search_areas_arr = append(search_areas_arr, v)
			}
		}
		for _, v := range strings.Split(search_words, ",") {
			isOk := false
			for _, vv := range search_words_arr {
				if vv == v {
					isOk = true
				}
			}
			if !isOk && v != "" {
				search_words_arr = append(search_words_arr, v)
			}
		}
		search_areass = strings.Join(search_areas_arr, ",")
		search_wordss = strings.Join(search_words_arr, ",")
	}
	//订单购买的服务
	//orderData := TiDb.Find("dwd_f_userbase_order_info", map[string]interface{}{"userid": userId, "order_status": 1, "delete_status": 0}, "", "", -1, -1)
	//orderData := TiDb.SelectBySql("select  *  from  dwd_f_userbase_order_info where   userid=? and  order_status=1 and  delete_status=0 and  payable_money > 0", userId)
	orderSql := fmt.Sprintf(`select  a.*,b.product_type as  productType from  dataexport_order  a   
    INNER JOIN   jy_order_detail   b    on   a.order_code=b.order_code and    a.user_id    = "%s"   and 
    a.order_status = 1  and  a.del_status =0 
    ORDER BY  b.service_endtime  desc `, userId)
	orderData := Mysql.SelectBySql(orderSql)
	if orderData != nil && len(*orderData) > 0 {
		product_type_arr, product_type_arrs := []string{}, []string{}
		for _, v := range *orderData {
			payable_money := gconv.Int64(v["payable_money"])
			if payable_money <= 0 {
				continue
			}
			product_type := common.ObjToString(v["product_type"])
			if product_type != "" {
				product_type_arr = append(product_type_arr, product_type)
			}
		}
		for _, v := range product_type_arr {
			isOk := false
			for _, vv := range product_type_arrs {
				if vv == v {
					isOk = true
				}
			}
			if !isOk {
				product_type_arrs = append(product_type_arrs, v)
			}
		}
		product_types = strings.Join(product_type_arrs, ",")
	}

	//订阅数据
	subscribeData := TiDb.FindOne("dwd_f_userbase_subscribe_info", map[string]interface{}{"userid": userId}, "", "updatetime desc")
	if subscribeData != nil && len(*subscribeData) > 0 {
		subscribe_keywords = common.ObjToString((*subscribeData)["subscribe_keywords"])
		subscribe_areas = common.ObjToString((*subscribeData)["subscribe_areas"])
	}

	//会员介绍页面
	start := time.Now().AddDate(0, 0, -1).Format("2006-01-02") + " 00:00:00"
	end := time.Now().Format("2006-01-02") + " 00:00:00"
	eventStr := "昨日未浏览"
	eventCount := TiDb.SelectBySql(`select * from dwd_f_userbase_event_info where userid = "` + userId + `" and eventtype = "会员介绍页面" and createtime >= "` + start + `" and createtime <= "` + end + `"`)
	if eventCount != nil && len(*eventCount) > 0 {
		eventStr = "昨日浏览过"
	}
	nowTime := time.Now().Format("2006-01-02 15:04:05")
	//聊天记录标签
	keyStr := ""
	phoneRegexp := regexp.MustCompile(`^1[0-9]{10}$`)
	keyMap := map[string]bool{}
	matchArr := Base.SelectBySql(`select  aa.name  from  socialize_keyword  aa where  aa.mold=0 and  aa.state=0 and  FIND_IN_SET(2,aa.group)  `)
	sqls := fmt.Sprintf(`select b.content from socialize_message_mailbox a LEFT JOIN socialize_message b on b.id = a.messag_id where a.type in (4,5,7) and a.own_type = 2 and a.own_id = %d and b.title = "文本" and a.send_user_type = 2 and a.create_time >= "%s"`, base_user_id, time.Now().AddDate(0, 0, -30).Format("2006-01-02 15:04:05"))
	mData := Base.SelectBySql(sqls)
	if mData != nil && *mData != nil && len(*mData) > 0 {
		for _, vv := range *mData {
			content := common.ObjToString(vv["content"])
			if phoneRegexp.MatchString(content) {
				keyMap[content] = true
			}
			for _, vvv := range *matchArr {
				key := gconv.String(vvv["name"])
				if strings.Contains(strings.ToUpper(content), strings.ToUpper(key)) {
					keyMap[key] = true
					break
				}
			}
		}
	}
	if len(keyMap) > 0 {
		keyArr := []string{}
		for k := range keyMap {
			keyArr = append(keyArr, k)
		}
		keyStr = strings.Join(keyArr, ",")
	}
	if TiDb.Count("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}) > 0 {
		TiDb.Update("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}, map[string]interface{}{
			"last_login_time":    common.If(date != "", date, nil),
			"search_areas":       search_areass,
			"search_words":       search_wordss,
			"subscribe_areas":    subscribe_areas,
			"subscribe_keywords": subscribe_keywords,
			"product_type":       product_types,
			"updatetime":         nowTime,
			"members_info":       eventStr,
			"messagekey":         common.If(keyStr == "", nil, keyStr),
		})
	} else {
		TiDb.Insert("dwd_f_crm_attribute_label", map[string]interface{}{
			"uid":                uId,
			"registedate":        registedate,
			"last_login_time":    common.If(date != "", date, nil),
			"search_areas":       search_areass,
			"search_words":       search_wordss,
			"subscribe_areas":    subscribe_areas,
			"subscribe_keywords": subscribe_keywords,
			"product_type":       product_types,
			"updatetime":         nowTime,
			"members_info":       eventStr,
			"messagekey":         common.If(keyStr == "", nil, keyStr),
		})
	}
}

func tagAddSync() {
	log.Println("注册日期、订单增量定时任务开始")
	userData := TiDb.SelectBySql(`select l_registedate,createtime,uid,userid from dwd_f_userbase_baseinfo where createtime >= "` + cfg.LastUserTime + `" order by createtime asc`)
	if userData != nil && len(*userData) > 0 {
		for k, v := range *userData {
			nowTime := time.Now().Format("2006-01-02 15:04:05")
			registedate := common.ObjToString(v["l_registedate"])
			createtime := common.ObjToString(v["createtime"])
			userId := common.ObjToString(v["userid"])
			uId := common.ObjToString(v["uid"])
			if registedate == "" || uId == "" || userId == "" {
				log.Println("缺少信息", uId, userId)
				continue
			} else {
				log.Println("新注册", uId, userId)
			}
			if k == len(*userData)-1 {
				cfg.LastUserTime = createtime
			}
			if TiDb.Count("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}) > 0 {
				TiDb.Update("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}, map[string]interface{}{
					"registedate": registedate,
					"updatetime":  nowTime,
				})
			} else {
				TiDb.Insert("dwd_f_crm_attribute_label", map[string]interface{}{
					"uid":         uId,
					"registedate": registedate,
					"updatetime":  nowTime,
				})
			}
		}
	}
	log.Println("注册日期定时任务结束")
	orderData := TiDb.SelectBySql(`select product_type,autoUpdate,uid from dwd_f_userbase_order_info where order_status = 1  and autoUpdate >= "` + cfg.LastOrderTime + `" order by autoUpdate desc`)
	if orderData != nil && len(*orderData) > 0 {
		for k, order := range *orderData {
			nowTime := time.Now().Format("2006-01-02 15:04:05")
			autoUpdate := common.ObjToString(order["autoUpdate"])
			uId := common.ObjToString(order["uid"])
			if uId == "" {
				log.Println("缺少信息")
				continue
			}
			product_types := ""
			if k == 0 {
				cfg.LastOrderTime = autoUpdate
			}
			personArr := getUserIdToUid(uId)
			orderDataSql := fmt.Sprintf(`select  a.*,b.product_type as  productType from  dataexport_order  a   
    INNER JOIN   jy_order_detail   b    on   a.order_code=b.order_code and    a.user_id    in   (%s) and 
    a.order_status = 1  and  a.del_status =0 
    ORDER BY  b.service_endtime  desc  `, strings.Join(personArr, ","))
			orderDatas := Mysql.SelectBySql(orderDataSql)
			if orderDatas != nil && len(*orderDatas) > 0 {
				product_type_arr, product_type_arrs := []string{}, []string{}
				for _, v := range *orderDatas {
					payable_money := gconv.Int64(v["pay_money"])
					if payable_money <= 0 {
						continue
					}
					product_type := common.ObjToString(v["productType"])
					log.Println("product_type ", product_type)
					if product_type != "" {
						product_type_arr = append(product_type_arr, product_type)
					}
				}
				log.Println("product_type_arr ", product_type_arr)
				for _, v := range product_type_arr {
					isOk := false
					for _, vv := range product_type_arrs {
						if vv == v {
							isOk = true
						}
					}
					if !isOk {
						product_type_arrs = append(product_type_arrs, v)
					}
				}
				product_types = strings.Join(product_type_arrs, ",")
			}
			log.Println("product_types ", uId, product_types)
			if TiDb.Count("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}) > 0 {
				TiDb.Update("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uId}, map[string]interface{}{
					"product_type": product_types,
					"updatetime":   nowTime,
				})
			} else {
				TiDb.Insert("dwd_f_crm_attribute_label", map[string]interface{}{
					"uid":          uId,
					"product_type": product_types,
					"updatetime":   nowTime,
				})
			}
		}
	}
	common.WriteSysConfig(&cfg)
	log.Println("注册日期、订单增量定时任务结束")
}

// 在线客服聊天记录30分钟一次
func messageSync() {
	log.Println("在线客服聊天记录定时任务开始")
	phoneRegexp := regexp.MustCompile(`^1[0-9]{10}$`)
	sql := fmt.Sprintf(`select * from socialize_message where title = "文本" and create_time > "%s"`, cfg.LastMessageTime)
	data := Base.SelectBySql(sql)
	if data != nil && *data != nil && len(*data) > 0 {
		for _, v := range *data {
			isOk := false
			content := common.ObjToString(v["content"])
			matchArr := Base.SelectBySql(`select  aa.name  from  socialize_keyword aa where  aa.mold=0 and  aa.state=0 and  FIND_IN_SET(2,aa.group)  `)
			for _, vv := range *matchArr {
				key := gconv.String(vv["name"])
				if strings.Contains(strings.ToUpper(content), strings.ToUpper(key)) || phoneRegexp.MatchString(content) {
					isOk = true
				}
			}
			if isOk {
				messag_id := common.Int64All(v["id"])
				mData := Base.FindOne("socialize_message_mailbox", map[string]interface{}{"messag_id": messag_id, "own_type": 2}, "", "")
				if mData != nil && len(*mData) > 0 {
					own_id := common.Int64All((*mData)["own_id"])
					if own_id > 0 {
						ok1, ok2, _ := FormatData(*mData, "message")
						if !ok1 {
							log.Println("线索卡点", "message", mData, messag_id)
						} else {
							if !ok2 {
								log.Println("用户分配已达上限", "message", mData, messag_id)
							}
						}
					}
				}
			}
			cfg.LastMessageTime = common.ObjToString(v["create_time"])
		}
	}
	common.WriteSysConfig(&cfg)
	log.Println("在线客服聊天记录定时任务结束")
}

// 客服按钮处理
func customerButton() {
	log.Println("在线客服按钮定时任务开始")
	sql := fmt.Sprintf(`select * from socialize_message where title = "点击按钮" and create_time > "%s"`, cfg.LastMessageButtonTime)
	data := Base.SelectBySql(sql)
	if data != nil && *data != nil && len(*data) > 0 {
		for _, v := range *data {
			isOk := false
			content := common.ObjToString(v["content"])
			matchArr := Base.SelectBySql(`select  aa.name  from  socialize_keyword  aa where  aa.mold=1 and  aa.state=1 and  FIND_IN_SET(2,aa.group)  `)
			for _, vv := range *matchArr {
				key := gconv.String(vv["name"])
				if strings.Contains(strings.ToUpper(content), strings.ToUpper(key)) {
					isOk = true
				}
			}
			if isOk {
				messag_id := common.Int64All(v["id"])
				mData := Base.FindOne("socialize_message_mailbox", map[string]interface{}{"messag_id": messag_id, "own_type": 2}, "", "")
				if mData != nil && len(*mData) > 0 {
					own_id := common.Int64All((*mData)["own_id"])
					if own_id > 0 {
						ok1, ok2, _ := FormatData(*mData, "message")
						if !ok1 {
							log.Println("线索卡点", "message", mData, messag_id)
						} else {
							if !ok2 {
								log.Println("用户分配已达上限", "message", mData, messag_id)
							}
						}
					}
				}
			}
			cfg.LastMessageButtonTime = common.ObjToString(v["create_time"])
		}
	}
	common.WriteSysConfig(&cfg)
	log.Println("在线客服按钮定时任务结束")
}
func LabelToClue() {
	log.Println("735需求 标签进线索开始")
	for _, v := range lableJson.ConditionConfig {
		log.Println("735标签用户sql:", v.SubName)
		dataArr := getBitmapIntersection(v.ConditionArr)
		log.Println("735标签用户sql:", v.SubName, len(dataArr))
		if len(dataArr) == 0 {
			continue
		}

		for _, baseUserId := range dataArr {
			log.Println("735标签用户:", baseUserId)
			if baseUserId == 0 {
				continue
			}
			FormatData(map[string]interface{}{
				"baseUserId": baseUserId,
				"topName":    v.TopName,
				"subName":    v.SubName,
			}, "tag")

		}
	}
	log.Println("735需求 标签进线索结束")
}

type AggStruct struct {
	Intersection []int `ch:"intersection"`
}

func getBitmapIntersection(ids []Condition) []int {
	// 构建查询的 SELECT 和 JOIN 部分
	joinParts := []string{}
	for i := len(ids); i > 0; i-- {
		joinParts = append(joinParts, fmt.Sprintf(`
            SELECT bitobj AS bit%d
            FROM dwd_d_tag
            WHERE code = '%v'`, i, ids[i-1].Code))
	}
	selectParts := ""
	for i := 0; i < len(ids); i++ {
		if i == 0 {
			if gconv.Bool(ids[i+1].Fool) {
				selectParts = fmt.Sprintf("bitmapAnd( bit%d,bit%d)", i+2, i+1)
			} else {
				selectParts = fmt.Sprintf("bitmapAndnot( bit%d,bit%d)", i+1, i+2)
			}
			i++
		} else {
			if gconv.Bool(ids[i].Fool) {
				selectParts = fmt.Sprintf("bitmapAnd( bit%d,%s)", i+1, selectParts)
			} else {
				selectParts = fmt.Sprintf("bitmapAndnot( %s,bit%d)", selectParts, i+1)
			}
		}
	}
	// 构建 SQL 查询
	query := fmt.Sprintf(`
        SELECT bitmapToArray(%s) AS intersection
        FROM (%s) AS t1`,
		selectParts, // 连接 SELECT 部分
		strings.Join(joinParts, ") AS t%d CROSS JOIN (")) // 连接 JOIN 部分
	// 修正 CROSS JOIN 部分
	for i := 1; i < len(ids); i++ {
		query = strings.Replace(query, "AS t%d", fmt.Sprintf("AS t%d", i+1), 1)
	}
	log.Println("p735标签接口:", query)
	rows, _ := ClickhouseConn.Query(context.TODO(), query)
	ass := []int{}
	for rows.Next() {
		as := AggStruct{}
		if err := rows.ScanStruct(&as); err != nil {
			logx.Error(err)
			continue
		}
		ass = as.Intersection
	}
	rows.Close()
	return ass
}