package common import ( "app.yhyue.com/moapp/MessageCenter/entity" "app.yhyue.com/moapp/MessageCenter/rpc/internal/config" "app.yhyue.com/moapp/jybase/common" "app.yhyue.com/moapp/jybase/encrypt" "app.yhyue.com/moapp/jybase/redis" "context" "fmt" "github.com/robfig/cron/v3" "log" "strings" "time" ) var GlobMsgMap map[int]map[string]interface{} func LoadTask() { // 每隔10分钟执行一次 LoadMsgOnTime() c := cron.New(cron.WithSeconds()) c.AddFunc(config.ConfigJson.GlobMsgLoadTime, LoadMsgOnTime) c.AddFunc(config.ConfigJson.FreeIntelTime, FreeIntelUserPush) //免费用户推送 c.AddFunc(config.ConfigJson.PayIntelTime, PayIntelUserPush) //付费用户推送 go c.Start() defer c.Stop() } func LoadMsgOnTime() { fmt.Println("开始执行") msgMap := make(map[int]map[string]interface{}) m := entity.Mysql.SelectBySql("SELECT id,msg_type,title,content,send_time,menu_name,link,group_id,sign FROM message_send_log WHERE send_status = 4 AND isdel = 1 ORDER BY send_time DESC limit 2000") if m != nil && len(*m) > 0 { for _, val := range *m { msgMap[common.IntAll(val["id"])] = val } GlobMsgMap = msgMap } fmt.Println("GlobMsgMap len", len(GlobMsgMap)) } func FreeIntelUserPush() { data := messageData(config.ConfigJson.FreePushNumber) if data == nil || len(*data) == 0 { return } users := IntelUser(true) PushData(users, data) } func PayIntelUserPush() { data := messageData(config.ConfigJson.PayPushNumber) if data == nil || len(*data) == 0 { return } users := IntelUser(false) PushData(users, data) } func PushData(users []string, data *[]map[string]interface{}) { log.Printf("需推送用户:%d\n", len(users)) var ids []int64 if data != nil { for _, m := range *data { _id := common.InterfaceToStr(m["_id"]) var link []string link = append(link, fmt.Sprintf("/swordfish/page_big_pc/business_detail/%s", encrypt.EncodeArticleId2ByCheck(_id))) mobLink := fmt.Sprintf("/jy_mobile/business/detail/%s", encrypt.EncodeArticleId2ByCheck(_id)) link = append(link, mobLink) link = append(link, mobLink) link = append(link, mobLink) iData := map[string]interface{}{ "msg_type": 9, "title": "您有一条专属商机情报", "content": fmt.Sprintf("【商机情报】%s", common.ObjToString(m["title"])), "send_mode": 1, "send_time": time.Now().Format("2006-01-02 15:04:05"), "send_status": 4, "update_time": time.Now().Format("2006-01-02 15:04:05"), "createtime": time.Now().Format("2006-01-02 15:04:05"), "link": strings.Join(link, ","), "isdel": 1, "send_userid": "商机情报定时推送", "sign": 0, "group_id": 5, } id := entity.Mysql.Insert("message_send_log", iData) ids = append(ids, id) } } if len(ids) > 0 { var bits []string for _, i2 := range ids { bits = append(bits, fmt.Sprintf("toUInt64(%d)", i2)) } err := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_summary UPDATE msg_bitmap = bitmapOr(msg_bitmap,bitmapBuild([%s])) where group_id = %d`, strings.Join(bits, ","), 5)) if err != nil { log.Println("message_summary err=== ", err.Error()) return } var userIds []string for k, user := range users { userIds = append(userIds, user) if len(userIds) == 1000 || k == len(users)-1 { UpdateBatch(userIds, ids) userIds = []string{} } } } } // 用户 func IntelUser(isFree bool) []string { mUser := make(map[string]bool) data := entity.Mysql.SelectBySql(`SELECT a.phone as phone FROM entniche_user a INNER JOIN entniche_info b on b.status = 1 and a.power = 1 and a.ent_id = b.id and a.phone != ''`) if data != nil { //统计商机管理用户 for _, m := range *data { mUser[common.InterfaceToStr(m["phone"])] = true } } var uData []string switch isFree { case false: user, ok := entity.MQFW.Find("user", map[string]interface{}{ "i_appid": 2, "$or": []map[string]interface{}{ { "i_vip_status": map[string]interface{}{"$gt": 0}, }, { "i_member_status": map[string]interface{}{"$gt": 0}, }, }, }, "", `{"_id":1,"s_phone":1,"s_m_phone":1}`, false, -1, -1) if ok && user != nil && len(*user) > 0 { for _, m := range *user { phone := common.If(common.InterfaceToStr(m["s_phone"]) == "", common.InterfaceToStr(m["s_m_phone"]), common.InterfaceToStr(m["s_phone"])).(string) uData = append(uData, common.InterfaceToStr(m["_id"])) if mUser[phone] { //获取剩余商机管理用户 delete(mUser, phone) } } if len(mUser) > 0 { //统计剩余商机管理用户 var ( phones []string count int ) for phone := range mUser { count++ phones = append(phones, phone) if len(phones) == 100 || count == len(mUser) { user1, ok1 := entity.MQFW.Find("user", map[string]interface{}{ "i_appid": 2, "$or": []map[string]interface{}{ { "s_phone": map[string]interface{}{ "$in": phones, }, }, { "s_m_phone": map[string]interface{}{ "$in": phones, }, }, }, }, "", `{"_id":1}`, false, -1, -1) if ok1 && user1 != nil && len(*user1) > 0 { for _, m := range *user1 { uData = append(uData, common.InterfaceToStr(m["_id"])) } } phones = []string{} } } } } case true: sess := entity.MQFW.GetMgoConn() defer entity.MQFW.DestoryMongoConn(sess) iter := sess.DB("qfw").C("user").Find(map[string]interface{}{ "i_appid": 2, }).Select(map[string]interface{}{"i_vip_status": 1, "i_member_status": 1, "_id": 1, "s_phone": 1, "s_m_phone": 1}).Iter() for m := make(map[string]interface{}); iter.Next(&m); { if common.IntAll(m["i_vip_status"]) <= 0 && common.IntAll(m["i_member_status"]) <= 0 { phone := common.If(common.InterfaceToStr(m["s_phone"]) == "", common.InterfaceToStr(m["s_m_phone"]), common.InterfaceToStr(m["s_phone"])).(string) if !mUser[phone] { uData = append(uData, common.InterfaceToStr(m["_id"])) } } m = map[string]interface{}{} } } return uData } // 获取推送消息 func messageData(number int) *[]map[string]interface{} { now := time.Now().AddDate(0, 0, -1) starttime := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.Local).Unix() enttime := time.Date(now.Year(), now.Month(), now.Day(), 24, 0, 0, 0, time.Local).Unix() query := map[string]interface{}{ "yucetime": map[string]interface{}{"$gt": starttime, "$lt": enttime}, } data, _ := entity.Bidding.Find("project_forecast", query, `{"yucetime": -1}`, `{"title":1,"_id":1}`, false, 0, number) return data } func UpdateBatch(ids []string, msgLogId []int64) { str := fmt.Sprintf(`'%s'`, strings.Join(ids, `','`)) var bits []string for _, i2 := range msgLogId { bits = append(bits, fmt.Sprintf("toUInt64(%d)", i2)) } log.Println(fmt.Sprintf(`alter table message_user_summary UPDATE allMsg = bitmapOr(allMsg,bitmapBuild([%s])) where userId in (%s)`, strings.Join(bits, ","), str)) err1 := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_user_summary UPDATE allMsg = bitmapOr(allMsg,bitmapBuild([%s])) where userId in (%s)`, strings.Join(bits, ","), str)) if err1 != nil { log.Printf("批量更新message_user_summary出错:%s", err1) return } for _, id := range ids { keyString := fmt.Sprintf(MsgCountKey, id, 5) redis.Del(redisModule, keyString) } }