userActionEtl/user.go

package main

import (
	qu "app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/date"
	"app.yhyue.com/moapp/jybase/mongodb"
	"app.yhyue.com/moapp/jybase/redis"
	"fmt"
	"log"
	"strings"
	"time"
)

// getUserTrustedId 获取mgo中的匿名id
func getUserTrustedId(userid string) string {
	sess := Mgo_Log.GetMgoConn()
	defer Mgo_Log.DestoryMongoConn(sess)
	query := map[string]interface{}{
		"userid": userid,
	}
	it := sess.DB("qfw").C("jy_logs").Find(query).Select(map[string]interface{}{
		"trustedId": 1,
	}).Iter()
	total := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		trustedId := qu.ObjToString(tmp["trustedId"])
		if trustedId != "" {
			return MgoTrustedIdToCkTrustedId(trustedId)
		}
		tmp = make(map[string]interface{})
	}
	return ""
}

func User(st, et int64) {
	s, e := GetCurTimePiInfo(st, et)
	query := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gte": mongodb.StringTOBsonId(s),
			"$lt":  mongodb.StringTOBsonId(e),
		},
	}
	sess := Mgo_Log.GetMgoConn()
	defer Mgo_Log.DestoryMongoConn(sess)
	it := sess.DB("qfw").C("register_log").Find(query).Sort("-_id").Select(map[string]interface{}{
		"userid":      1,
		"system":      1,
		"source":      1,
		"create_time": 1,
		"phone":       1,
	}).Iter()
	total := 0

	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		userid := qu.ObjToString(tmp["userid"])
		source := qu.ObjToString(tmp["source"]) //pc wx app h5 entbase background vipgift
		create_time := qu.ObjToString(tmp["create_time"])
		t, _ := time.ParseInLocation(date.Date_Full_Layout, create_time, time.Local)
		if source == "pc" {
			if tid := getUserTrustedId(userid); tid != "" {
				sql := fmt.Sprintf(`select * from userbehavior.user_behavior_log 
         				where trusted_id='%s' and user_id ='' and date<='%s' limit 1`, tid, create_time)
				data := Ch.SelectBySql(sql)
				if data != nil && len(*data) > 0 {
					d := (*data)[0]
					ip := qu.ObjToString(d["ip"])
					area := qu.ObjToString(d["area"])
					city := qu.ObjToString(d["city"])
					os := qu.ObjToString(d["os"])
					browser := qu.ObjToString(d["browser"])
					module := qu.ObjToString(d["module"])
					href := qu.ObjToString(d["href"])
					urlname := qu.ObjToString(d["url_name"])
					refer := qu.ObjToString(d["refer"])
					refer_name := qu.ObjToString(d["refer_name"])
					refer_type := qu.ObjToString(d["refer_type"])
					phone := qu.ObjToString(d["phone"])
					urldesc := qu.ObjToString(d["desc"])
					session_id := qu.ObjToString(d["session_id"])
					page_code_id := qu.Int64All(d["page_code_id"])
					position_id := qu.ObjToString(d["position_id"])
					StrategyInfo.CacheData <- &UserBehaviorLog{
						UserID:         userid,
						ActionID:       int64(4),        // 动作ID
						TrustedID:      tid,             // 浏览器指纹
						IP:             ip,              // 用户IP地址
						Area:           area,            // 省份
						City:           city,            // 城市
						OS:             os,              // 操作系统
						Browser:        browser,         // 浏览器类型
						BrowserVersion: "",              // 浏览器版本
						Date:           t,               // 行为发生时间
						Platform:       "pc",            // 平台（如移动端、PC端）
						Subsystem:      "",              // 子系统
						Module:         module,          // 模块
						URL:            href,            // 访问的URL
						URLName:        urlname,         // URL名称
						URLElement:     "",              // 页面元素
						AddField:       "",              // 附加数据
						TimeStamp:      int32(t.Unix()), // 时间戳
						Refer:          refer,           // 引用来源
						ReferName:      refer_name,      // 引用名称
						ReferType:      refer_type,      // 引用类型
						SessionID:      session_id,      // 会话ID
						Phone:          phone,           // 手机号码
						PageName:       urlname,         // 页面名称
						Desc:           urldesc,         // 描述
						BreakData:      "",              // 额外数据
						PageCodeId:     uint64(page_code_id),
						PositionId:     position_id,
					}
				}
			}
		} else {
			StrategyInfo.CacheData <- &UserBehaviorLog{
				UserID:         userid,
				ActionID:       int64(4),                     // 动作ID
				TrustedID:      "",                           // 浏览器指纹
				IP:             "",                           // 用户IP地址
				Area:           "",                           // 省份
				City:           "",                           // 城市
				OS:             "",                           // 操作系统
				Browser:        "",                           // 浏览器类型
				BrowserVersion: "",                           // 浏览器版本
				Date:           t,                            // 行为发生时间
				Platform:       "",                           // 平台（如移动端、PC端）
				Subsystem:      "",                           // 子系统
				Module:         "",                           // 模块
				URL:            "",                           // 访问的URL
				URLName:        "",                           // URL名称
				URLElement:     "",                           // 页面元素
				AddField:       "",                           // 附加数据
				TimeStamp:      int32(t.Unix()),              // 时间戳
				Refer:          "",                           // 引用来源
				ReferName:      "",                           // 引用名称
				ReferType:      "",                           // 引用类型
				SessionID:      "",                           // 会话ID
				Phone:          qu.ObjToString(tmp["phone"]), // 手机号码
				PageName:       "",                           // 页面名称
				Desc:           "",                           // 描述
				BreakData:      "",                           // 额外数据
				PageCodeId:     0,
				PositionId:     "",
			}
		}
		tmp = make(map[string]interface{})
	}
}

func salesLeads_logs(st, et int64) {
	userPhoneMap := GetPhone()
	modelMap := MatchRegexModule()

	s, e := GetCurTimePiInfo(st, et)
	q := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gte": mongodb.StringTOBsonId(s),
			"$lt":  mongodb.StringTOBsonId(e),
		},
		"url": "/salesLeads/retainedCapital",
	}
	log.Println(q)
	sess := Mgo_Log.GetMgoConn()
	defer Mgo_Log.DestoryMongoConn(sess)
	it := sess.DB("qfw").C("salesLeads_logs").Find(q).Sort("-_id").Select(map[string]interface{}{
		"userid":     1,
		"ip":         1,
		"refer":      1,
		"client":     1,
		"url":        1,
		"createtime": 1,
		"date":       1,
		"trustedId":  1,
		"os":         1,
		"browse":     1,
		"positionId": 1,
	}).Iter()
	total := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		if total%50000 == 0 {
			log.Println("cur saleLeads index ", total)
		}
		userid := qu.ObjToString(tmp["userid"])
		client := qu.ObjToString(tmp["client"])
		href := qu.ObjToString(tmp["url"])

		trustedId := qu.ObjToString(tmp["trustedId"])
		ip := qu.ObjToString(tmp["ip"])
		area, city := GetIpSource(ip)
		os := qu.ObjToString(tmp["os"])
		browse := qu.ObjToString(tmp["browse"])

		refer := qu.ObjToString(tmp["refer"])

		i_createtime := qu.Int64All(tmp["date"])
		createtime := time.Unix(i_createtime, 0)

		platform := ""
		if strings.Contains(refer, "www.jianyu") {
			platform = "pc"
		} else if strings.Contains(refer, "wx.jianyu") {
			platform = "wx"
		} else {
			if strings.Contains(client, "Android") {
				platform = "android"
			}
			// 检查是否为 iOS 客户端
			if strings.Contains(client, "iPhone") || strings.Contains(client, "iPad") || strings.Contains(client, "iPod") {
				platform = "ios"
			}
		}

		hrefinfo, _ := modelMap.MatchRegexMap(Unescape(href))
		module := ""
		if hrefinfo != nil {
			module = hrefinfo.Module
		}
		referinfo, _ := modelMap.MatchRegexMap(Unescape(refer))
		refermodule := ""
		urlname := ""
		urldesc := ""
		if referinfo != nil {
			refermodule = referinfo.Module
			urlname = referinfo.Page_name
			urldesc = referinfo.Desc
		}
		referType := GetReferType(refer)
		positionid := qu.ObjToString(tmp["positionId"])
		//if !mongodb.IsObjectIdHex(userid) {
		//	userid = GetUserId(userid)
		//	positionid = userid
		//} else {
		//	positionid = gconv.String(positionMap[userid])
		//}

		StrategyInfo.CacheData <- &UserBehaviorLog{
			UserID:         userid,
			ActionID:       12,                  // 动作ID
			TrustedID:      trustedId,           // 浏览器指纹
			IP:             ip,                  // 用户IP地址
			Area:           area,                // 省份
			City:           city,                // 城市
			OS:             os,                  // 操作系统
			Browser:        browse,              // 浏览器类型
			BrowserVersion: "",                  // 浏览器版本
			Date:           createtime,          // 行为发生时间
			Platform:       platform,            // 平台（如移动端、PC端）
			Subsystem:      "",                  // 子系统
			Module:         module,              // 模块
			URL:            href,                // 访问的URL
			URLName:        urlname,             // URL名称
			URLElement:     "",                  // 页面元素
			AddField:       "",                  // 附加数据
			TimeStamp:      int32(i_createtime), // 时间戳
			Refer:          refer,               // 引用来源
			ReferName:      refermodule,         // 引用名称
			ReferType:      referType,           // 引用类型
			SessionID:      "",                  // 会话ID
			//SessionStartTime: time.Now(),           // 会话开始时间 //TODO 会话开始时间
			Phone:      userPhoneMap[userid], // 手机号码
			PageName:   urlname,              // 页面名称
			Desc:       urldesc,              // 描述
			BreakData:  "",                   // 额外数据
			PositionId: positionid,
		}
		tmp = make(map[string]interface{})

	}
	log.Println("mongodb end")
}

// InitPositionId 通过用户归集获取职位id和mgoid的对应关系
func InitPositionId() map[string]int64 {
	positionIdMap := map[string]int64{}
	DataService.SelectByBath(20, func(l *[]map[string]interface{}) bool {
		for _, v := range *l {
			user_id := qu.ObjToString(v["user_id"])
			id := qu.Int64All(v["id"])
			positionIdMap[user_id] = id
		}
		return true
	}, fmt.Sprintf("SELECT userid,position_id FROM data_service.%s WHERE TYPE=0;", Sysconfig.Tidb.DataServiceCollName))
	return positionIdMap
}

// LoadUserTrustedId 获取userid和虚拟id对应关系
func LoadUserTrustedId(st, et int64) map[string]string {
	userTrusted := map[string]string{}
	starttime := time.Unix(st, 0).Format(date.Date_Full_Layout)
	endtime := time.Unix(et, 0).Format(date.Date_Full_Layout)
	Ch.SelectByBath(20, func(l *[]map[string]interface{}) bool {
		for _, v := range *l {
			userid := qu.ObjToString(v["user_id"])
			trusted_id := qu.ObjToString(v["trusted_id"])
			userTrusted[trusted_id] = userid
		}
		return true
	}, fmt.Sprintf(`select user_id,trusted_id from user_behavior_log where date>='%s' and date< '%s' and user_id !='' and trusted_id!=''`, starttime, endtime))
	return userTrusted
}

func GetPhone() map[string]string {
	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)
	q := map[string]interface{}{
		"i_appid": 2,
		"$or": []map[string]interface{}{
			map[string]interface{}{"s_phone": map[string]interface{}{"$exists": true}},
			map[string]interface{}{"s_m_phone": map[string]interface{}{"$exists": true}},
		},
	}
	userPhoneMap := map[string]string{}

	it := sess.DB("qfw").C("user").Find(q).Select(map[string]interface{}{
		"userid":    1,
		"s_phone":   1,
		"s_m_phone": 1,
	}).Iter()
	total := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		if total%50000 == 0 {
			log.Println("cur index ", total)
		}
		userid := mongodb.BsonIdToSId(tmp["_id"])
		phone := qu.ObjToString(tmp["s_phone"])
		if phone == "" {
			phone = qu.ObjToString(tmp["s_m_phone"])
		}
		if phone != "" {
			userPhoneMap[userid] = phone
		}

		tmp = make(map[string]interface{})
	}
	return userPhoneMap
}

// 获取用户id
func GetUserId(userid string) string {
	keyname := fmt.Sprintf("position_%s", userid)
	mgoUserid := redis.GetStr("useraction", keyname)
	if mgoUserid != "" {
		return mgoUserid
	}
	data := Base.SelectBySql(`select user_id from base_position where id =?`, userid)
	if data != nil && len(*data) > 0 {
		base_user_id := qu.Int64All((*data)[0]["user_id"])
		udata, _ := Mgo.FindOne("user", map[string]interface{}{
			"base_user_id": base_user_id,
		})
		if udata != nil && len(*udata) > 0 {
			mgoUserid := mongodb.BsonIdToSId((*udata)["_id"])
			redis.Put("useraction", keyname, mgoUserid, 3600*24)
			return mgoUserid
		}
	}
	return ""
}