|
|
@@ -1,7 +1,6 @@
|
|
|
package main
|
|
|
|
|
|
import (
|
|
|
- "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
|
|
|
"encoding/json"
|
|
|
"fmt"
|
|
|
"github.com/robfig/cron/v3"
|
|
|
@@ -9,6 +8,7 @@ import (
|
|
|
wlog "github.com/wcc4869/common_utils/log"
|
|
|
util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
|
|
|
"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
|
|
|
+ "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
|
|
|
"net"
|
|
|
"strings"
|
|
|
"time"
|
|
|
@@ -48,7 +48,7 @@ func init() {
|
|
|
|
|
|
}
|
|
|
|
|
|
-//processUdpMsg 处理udp
|
|
|
+// processUdpMsg 处理udp
|
|
|
func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
|
|
|
switch act {
|
|
|
case udp.OP_TYPE_DATA:
|
|
|
@@ -98,7 +98,7 @@ func InitConfig() {
|
|
|
|
|
|
}
|
|
|
|
|
|
-//InitLog 初始化日志
|
|
|
+// InitLog 初始化日志
|
|
|
func InitLog() {
|
|
|
logConfig := &wlog.Options{}
|
|
|
err := viper.UnmarshalKey("log", logConfig)
|
|
|
@@ -110,7 +110,6 @@ func InitLog() {
|
|
|
}
|
|
|
|
|
|
func main() {
|
|
|
-
|
|
|
local, _ := time.LoadLocation("Asia/Shanghai")
|
|
|
c := cron.New(cron.WithLocation(local), cron.WithSeconds())
|
|
|
c.AddFunc(GF.Spec, cpUser)
|
|
|
@@ -122,7 +121,7 @@ func main() {
|
|
|
|
|
|
}
|
|
|
|
|
|
-//cpUser 处理用户订阅词，然后拷贝到新表，提供分类使用
|
|
|
+// cpUser 处理用户订阅词，然后拷贝到新表，提供分类使用
|
|
|
func cpUser() {
|
|
|
wlog.Info("cpUser", wlog.String("start", ""))
|
|
|
var nuId string
|
|
|
@@ -209,3 +208,68 @@ func cpUser() {
|
|
|
wlog.Info("处理用户订阅词", wlog.Int("over", count))
|
|
|
|
|
|
}
|
|
|
+
|
|
|
+// reAllUser 重新更新用户订阅词，打标签
|
|
|
+func reAllUser() {
|
|
|
+ sess := Mgo.GetMgoConn()
|
|
|
+ defer Mgo.DestoryMongoConn(sess)
|
|
|
+ //
|
|
|
+ query := sess.DB(GF.Mongo.DB).C(GF.Mongo.Coll).Find(nil).Select(nil).Sort("_id").Iter()
|
|
|
+ count := 0
|
|
|
+
|
|
|
+ gtid := ""
|
|
|
+ lteid := ""
|
|
|
+ //var saveUserPool = make([]map[string]interface{}, 0)
|
|
|
+ var sendMsg bool
|
|
|
+ for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
|
|
|
+ idStr := mongodb.BsonIdToSId(tmp["_id"])
|
|
|
+ if count == 0 {
|
|
|
+ gtid = idStr
|
|
|
+ } else {
|
|
|
+ lteid = idStr
|
|
|
+ }
|
|
|
+ if count%1000 == 0 {
|
|
|
+ wlog.Info("reAllUser", wlog.Int("current", count))
|
|
|
+ wlog.Info("reAllUser", wlog.Any("_id", tmp["_id"]))
|
|
|
+ }
|
|
|
+ keys := GetUserKeys(tmp)
|
|
|
+ update := map[string]interface{}{}
|
|
|
+ if len(keys) > 0 {
|
|
|
+ tags := []string{}
|
|
|
+ for _, v := range keys {
|
|
|
+ tag := toString(v)
|
|
|
+ tags = append(tags, tag)
|
|
|
+ }
|
|
|
+
|
|
|
+ //用户关键词
|
|
|
+ update["key_list"] = strings.Join(tags, ",")
|
|
|
+ update["i_appid"] = tmp["i_appid"]
|
|
|
+ update["_id"] = tmp["_id"]
|
|
|
+
|
|
|
+ err := Mgo.InsertOrUpdate(GF.Mongo.DB, GF.Mongo.SaveColl, update)
|
|
|
+ if err != nil {
|
|
|
+ wlog.Info("reAllUser", wlog.String("id 更新错误", idStr))
|
|
|
+ wlog.Error(idStr, err)
|
|
|
+ }
|
|
|
+
|
|
|
+ sendMsg = true
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ wlog.Info("reAllUser", wlog.String("gtid", gtid), wlog.String("lteid", lteid))
|
|
|
+
|
|
|
+ //调用udp 处理用户行业分类
|
|
|
+ udpData := map[string]interface{}{
|
|
|
+ "gtid": gtid,
|
|
|
+ "lteid": lteid,
|
|
|
+ "stype": "yonghuhangye",
|
|
|
+ }
|
|
|
+
|
|
|
+ if sendMsg {
|
|
|
+ wlog.Info("reAllUser", wlog.Any("udpData", udpData))
|
|
|
+ SendUdpMsg(udpData, classficationAddr)
|
|
|
+ }
|
|
|
+
|
|
|
+ wlog.Info("处理用户订阅词", wlog.Int("over", count))
|
|
|
+}
|
wcc