feat:不活跃变活跃

telemarketingEtl/entity/dwd_f_userbase_visit_info.go

@@ -56,9 +56,13 @@ func VisitInfoAdd(start, end int64) {
 	all := map[string]map[string]*VisitInfo{}
 	lastLogin := map[string]string{}
 	for _, table := range tables {
-		iter := session.DB("qfw").C(table).Find(query).Sort("_id").Iter()
+		iter := session.DB("qfw").C(table).Find(query).Select(map[string]interface{}{
+			"userid": 1,
+			"date":   1,
+			"url":    1,
+		}).Sort("_id").Iter()
 		count := 0
-		log.Println("遍历日志表结束。。。", table)
+		log.Println("开始遍历日志表。。。", table)
 		for thisData := map[string]interface{}{}; iter.Next(&thisData); {
 			userId := gconv.String(thisData["userid"])
 			if userId == "" || idMapping[userId] == nil {
@@ -66,7 +70,7 @@ func VisitInfoAdd(start, end int64) {
 			}
 			userId = idMapping[userId].UserId
 			createtime := gtime.New(thisData["date"])
-			ymd := createtime.Format(date.Date_Short_Layout)
+			ymd := createtime.Format("Y-m-d")
 			if all[ymd] == nil {
 				all[ymd] = map[string]*VisitInfo{}
 			}
@@ -77,7 +81,7 @@ func VisitInfoAdd(start, end int64) {
 				}
 			}
 			vi.number++
-			vi.lastLoginTime = createtime.Format(date.Date_Full_Layout)
+			vi.lastLoginTime = createtime.String()
 			lastLogin[userId] = vi.lastLoginTime
 			url_ := gconv.String(thisData["url"])
 			if contentReg.MatchString(url_) {
@@ -107,16 +111,12 @@ func VisitInfoAdd(start, end int64) {
 	}
 	pool := make(chan bool, g.Cfg().MustGet(ctx, "visitInfoSavePool").Int())
 	wait := &sync.WaitGroup{}
-	newAll := map[string]string{}
 	index := 0
 	log.Println("开始保存统计结果。。。")
 	for k, v := range all {
 		t := gtime.New(k)
-		startTime, endTime := t.Format(date.Date_Full_Layout), t.AddDate(0, 0, 1).Format(date.Date_Full_Layout)
+		startTime, endTime := t.String(), t.AddDate(0, 0, 1).String()
 		for kk, vv := range v {
-			if idMapping[kk] != nil && (newAll[idMapping[kk].Uid] == "" || gtime.New(vv.lastLoginTime).After(gtime.New(newAll[idMapping[kk].Uid]))) {
-				newAll[idMapping[kk].Uid] = vv.lastLoginTime
-			}
 			index++
 			if index%500 == 0 {
 				log.Println("保存统计结果", index)
@@ -135,13 +135,9 @@ func VisitInfoAdd(start, end int64) {
 					config.JianyuSubjectdb.InsertBySql(`INSERT INTO dwd_f_userbase_visit_info (userid,date, number, platform,createtime,contentnum,portraitnum,searchnum) VALUES (?,?,?,?,?,?,?,?)`, userId, vi.lastLoginTime, 1, vi.platform, vi.lastLoginTime, vi.contentNum, vi.portraitNum, vi.searchNum)
 				}
 				if config.JianyuSubjectdb.CountBySql(`select count(1) from dwd_f_crm_attribute_label where uid=?`, vi.uId) > 0 {
-					config.JianyuSubjectdb.Update("dwd_f_crm_attribute_label", map[string]interface{}{"uid": vi.uId}, map[string]interface{}{"last_login_time": vi.lastLoginTime})
+					config.JianyuSubjectdb.ExecBySql("update dwd_f_crm_attribute_label set last_login_time=? where uid=?", vi.uId, lastLogin[userId])
 				} else {
-					config.JianyuSubjectdb.Insert("dwd_f_crm_attribute_label", map[string]interface{}{
-						"uid":             vi.uId,
-						"last_login_time": lastLogin[userId],
-						"updatetime":      time.Now().Format(time.DateTime),
-					})
+					config.JianyuSubjectdb.InsertBatch("dwd_f_crm_attribute_label", []string{"uid", "last_login_time", "updatetime"}, []interface{}{vi.uId, lastLogin[userId], gtime.Now().String()})
 				}
 			}(kk, vv)
 		}
@@ -150,8 +146,8 @@ func VisitInfoAdd(start, end int64) {
 	wait.Wait()
 	//
 	log.Println("开始更新不活跃重新活跃用户。。。")
-	sTime := gtime.New(start)
-	day30, today, nowFormat, tomorrow := sTime.AddDate(0, 0, -30).Format(time.DateTime), sTime.Format(time.DateTime), sTime.Format(time.DateTime), sTime.AddDate(0, 0, 1).Format(time.DateTime)
+	sTime := gtime.Now().AddDate(0, 0, -1).StartOfDay()
+	day30, today, nowFormat, tomorrow := sTime.AddDate(0, 0, -30).String(), sTime.String(), sTime.String(), sTime.AddDate(0, 0, 1).String()
 	config.JianyuSubjectdb.SelectByBath(1, func(l *[]map[string]interface{}) bool {
 		userid := gconv.String((*l)[0]["userid"])
 		log.Println("更新不活跃重新活跃用户", userid)

telemarketingEtl/entity/entity.go

@@ -36,9 +36,9 @@ type IdMp struct {
 
 //加载所有的uid
 func LoadIdMapping() map[string]*IdMp {
-	index := 0
+	log.Println("开始加载dwd_f_userbase_id_mapping。。。")
 	m := map[string]*IdMp{}
-	log.Println("开始加载dwd_f_userbase_id_mapping。。。", index)
+	index := 0
 	config.JianyuSubjectdb.SelectByBath(1, func(l *[]map[string]interface{}) bool {
 		index++
 		if index%5000 == 0 {