datatools/clueSync/subscribe.go

package main

import (
	"strings"
	"time"

	"log"

	"app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/mongodb"
)

func subscribeAddSync() {
	log.Println("订阅数据定时任务开始")
	session := MgoLog.GetMgoConn()
	lastId := cfg.LastSubscribeId
	defer func() {
		MgoLog.DestoryMongoConn(session)
	}()
	query := map[string]interface{}{}
	if lastId != "" {
		query["_id"] = map[string]interface{}{"$gt": mongodb.StringTOBsonId(lastId)}
	}
	// query["_id"] = mongodb.StringTOBsonId("64473e36c572141d78ec7a03")
	log.Println("query :", query)

	iter := session.DB(db.MgoLog.DbName).C("ovipjy_log").Find(&query).Sort("_id").Iter()
	thisData := map[string]interface{}{}

	for {
		if !iter.Next(&thisData) {
			break
		}
		cfg.LastSubscribeId = mongodb.BsonIdToSId(thisData["_id"])
		FormatSubscribeData(thisData)
	}
	common.WriteSysConfig(&cfg)
	log.Println("订阅数据定时任务结束")
}

func FormatSubscribeData(data map[string]interface{}) {
	types, stype := common.ObjToString(data["type"]), ""
	// createtime := common.Int64All(data["createtime"])
	userId := common.ObjToString(data["userid"])
	if !mongodb.IsObjectIdHex(userId) {
		userMapping := TiDb.FindOne("data_service.user_system", map[string]interface{}{"position_id": userId}, "", "")
		if userMapping != nil && len(*userMapping) > 0 {
			userId = common.ObjToString((*userMapping)["userid"])
		}
	}
	areaCodes := ""
	keywords := ""
	keyArrs := []string{}
	if types == "o_member_jy" {
		stype = "大会员订阅"
	} else if types == "o_vipjy" {
		stype = "超级订阅"
	} else if types == "o_jy" {
		stype = "免费订阅"
	}
	if types != "" {
		if jy, ok := data[types].(map[string]interface{}); ok {
			if types == "o_jy" {
				//
				if area, oks := jy["o_area"].(map[string]interface{}); oks {
					areaArr := []string{}
					for k, _ := range area {
						areaArr = append(areaArr, AreaCode[k])
					}
					if len(area) == 0 {
						areaArr = append(areaArr, AreaCode["全国"])
					}
					areaCodes = strings.Join(areaArr, ",")
				}
				//
				akey, aok := jy["a_key"].([]map[string]interface{})
				if !aok {
					akeys, _ := jy["a_key"].([]interface{})
					akey = common.ObjArrToMapArr(akeys)
				}
				for _, v := range akey {
					keysArr, asok := v["key"].([]string)
					if !asok {
						keysArr_s, _ := v["key"].([]interface{})
						keysArr = common.ObjArrToStringArr(keysArr_s)
					}
					for _, key := range keysArr {
						keyArrs = append(keyArrs, key)
					}
				}
			} else if types == "o_vipjy" {
				if area, oks := jy["area"].(map[string]interface{}); oks {
					areaArr := []string{}
					for k, _ := range area {
						areaArr = append(areaArr, AreaCode[k])
					}
					if len(area) == 0 {
						areaArr = append(areaArr, AreaCode["全国"])
					}
					areaCodes = strings.Join(areaArr, ",")
				}
				items, aok := jy["items"].([]map[string]interface{})
				if !aok {
					itemss, _ := jy["items"].([]interface{})
					items = common.ObjArrToMapArr(itemss)
				}
				for _, v := range items {
					akey, iok := v["a_key"].([]map[string]interface{})
					if !iok {
						akeys, _ := v["a_key"].([]interface{})
						akey = common.ObjArrToMapArr(akeys)
					}
					for _, v := range akey {
						keysArr, asok := v["key"].([]string)
						if !asok {
							keysArr_s, _ := v["key"].([]interface{})
							keysArr = common.ObjArrToStringArr(keysArr_s)
						}
						for _, key := range keysArr {
							keyArrs = append(keyArrs, key)
						}
					}
				}
			} else {
				if area, oks := jy["o_area"].(map[string]interface{}); oks {
					areaArr := []string{}
					for k, _ := range area {
						areaArr = append(areaArr, AreaCode[k])
					}
					if len(area) == 0 {
						areaArr = append(areaArr, AreaCode["全国"])
					}
					areaCodes = strings.Join(areaArr, ",")
				}
				if jy["o_area"] == nil {
					areaCodes = AreaCode["全国"]
				}
				items, aok := jy["a_items"].([]map[string]interface{})
				if !aok {
					itemss, _ := jy["a_items"].([]interface{})
					items = common.ObjArrToMapArr(itemss)
				}
				for _, v := range items {
					akey, iok := v["a_key"].([]map[string]interface{})
					if !iok {
						akeys, _ := v["a_key"].([]interface{})
						akey = common.ObjArrToMapArr(akeys)
					}
					for _, v := range akey {
						keysArr, asok := v["key"].([]string)
						if !asok {
							keysArr_s, _ := v["key"].([]interface{})
							keysArr = common.ObjArrToStringArr(keysArr_s)
						}
						for _, key := range keysArr {
							keyArrs = append(keyArrs, key)
						}
					}
				}
			}
			if len(keyArrs) > 0 {
				keywords = strings.Join(keyArrs, ",")
			}
		}
		start := time.Now().Format("2006-01-02") + " 00:00:00"
		end := time.Now().Format("2006-01-02") + " 23:59:59"
		nowTime := time.Now().Format("2006-01-02 15:04:05")
		subscribeData := TiDb.SelectBySql(`select * from dwd_f_userbase_subscribe_info where userid = "` + userId + `" and updatetime >= "` + start + `" and updatetime <= "` + end + `"`)
		if subscribeData != nil && len(*subscribeData) > 0 {
			TiDb.Update("dwd_f_userbase_subscribe_info", map[string]interface{}{"id": common.IntAll((*subscribeData)[0]["id"])}, map[string]interface{}{
				"updatetime":         nowTime,
				"subscribe_areas":    areaCodes,
				"subscribe_keywords": keywords,
				"member_type":        stype,
				"userid":             userId,
			})
		} else {
			TiDb.Insert("dwd_f_userbase_subscribe_info", map[string]interface{}{
				"userid":             userId,
				"updatetime":         nowTime,
				"subscribe_areas":    areaCodes,
				"subscribe_keywords": keywords,
				"member_type":        stype,
			})
		}
	}
}