|
|
@@ -1,9 +1,7 @@
|
|
|
package entity
|
|
|
|
|
|
import (
|
|
|
- "app.yhyue.com/moapp/jybase/common"
|
|
|
"fmt"
|
|
|
- "go.mongodb.org/mongo-driver/bson"
|
|
|
"log"
|
|
|
"regexp"
|
|
|
"strings"
|
|
|
@@ -12,32 +10,36 @@ import (
|
|
|
"telemarketingEtl/util"
|
|
|
"time"
|
|
|
|
|
|
+ "app.yhyue.com/moapp/jybase/common"
|
|
|
+ "go.mongodb.org/mongo-driver/bson"
|
|
|
+
|
|
|
"app.yhyue.com/moapp/jybase/date"
|
|
|
- "app.yhyue.com/moapp/jybase/mongodb"
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
|
"github.com/gogf/gf/v2/util/gconv"
|
|
|
)
|
|
|
|
|
|
-var USERPOOL = 20
|
|
|
-var locks = make([]*sync.Mutex, USERPOOL)
|
|
|
-
|
|
|
-func lock(userid string) *sync.Mutex {
|
|
|
- n := 0
|
|
|
- for _, v := range userid {
|
|
|
- n += int(v)
|
|
|
- }
|
|
|
- return locks[n%USERPOOL]
|
|
|
-}
|
|
|
+var (
|
|
|
+ wxReg = regexp.MustCompile("MicroMessenger")
|
|
|
+ contentReg = regexp.MustCompile(".*article/content/(.*)\\.html")
|
|
|
+ portrait1Reg = regexp.MustCompile(".*/swordfish/page_big_pc/unit_portrayal/(.*)")
|
|
|
+ portrait2Reg = regexp.MustCompile(".*swordfish/page_big_pc/(.*)/ent_ser_portrait")
|
|
|
+ searchReg = regexp.MustCompile(".*jybx/core/(.*)/searchList")
|
|
|
+)
|
|
|
|
|
|
-func init() {
|
|
|
- for i := 0; i < USERPOOL; i++ {
|
|
|
- locks[i] = &sync.Mutex{}
|
|
|
- }
|
|
|
+type VisitInfo struct {
|
|
|
+ number int
|
|
|
+ contentNum int
|
|
|
+ portraitNum int
|
|
|
+ searchNum int
|
|
|
+ platform int
|
|
|
+ lastLoginTime string
|
|
|
+ uId string
|
|
|
}
|
|
|
|
|
|
// 查看事件
|
|
|
func VisitInfoAdd(start, end int64) {
|
|
|
+ idMapping := LoadIdMapping()
|
|
|
s_id := util.GetObjectId(start)
|
|
|
e_id := util.GetObjectId(end)
|
|
|
query := map[string]interface{}{
|
|
|
@@ -46,117 +48,118 @@ func VisitInfoAdd(start, end int64) {
|
|
|
"$lt": e_id,
|
|
|
},
|
|
|
}
|
|
|
- RegWx, _ := regexp.Compile("MicroMessenger")
|
|
|
-
|
|
|
log.Println("task run", start, end, gtime.NewFromTimeStamp(start), query)
|
|
|
- //regWx, _ := regexp.Compile("MicroMessenger")
|
|
|
tables := g.Cfg().MustGet(ctx, "logTables").Strings()
|
|
|
- log.Println("analy tables:", tables)
|
|
|
+ log.Println("需要遍历日志表", tables)
|
|
|
+ session := config.MgoLog.GetMgoConn()
|
|
|
+ defer config.MgoLog.DestoryMongoConn(session)
|
|
|
+ all := map[string]map[string]*VisitInfo{}
|
|
|
+ lastLogin := map[string]string{}
|
|
|
for _, table := range tables {
|
|
|
- /*go*/ func(table string) {
|
|
|
- session := config.MgoLog.GetMgoConn()
|
|
|
- iter := session.DB("qfw").C(table).Find(query).Iter()
|
|
|
- count := 0
|
|
|
- for thisData := map[string]interface{}{}; iter.Next(&thisData); {
|
|
|
- // sPool <- true
|
|
|
- // sWait.Add(1)
|
|
|
- /*go*/
|
|
|
- func(thisData map[string]interface{}) {
|
|
|
- // defer func() {
|
|
|
- // <-sPool
|
|
|
- // sWait.Done()
|
|
|
- // }()
|
|
|
- userid, uid := gconv.String(thisData["userid"]), ""
|
|
|
- if userid == "" {
|
|
|
- return
|
|
|
- }
|
|
|
- if !mongodb.IsObjectIdHex(userid) {
|
|
|
- userid, uid = GetUserIdByPositionId(userid)
|
|
|
- } else {
|
|
|
- idmdata := config.JianyuSubjectdb.SelectBySql(`select uid from dwd_f_userbase_id_mapping where userid = "` + userid + `"`)
|
|
|
- if idmdata != nil && len(*idmdata) > 0 {
|
|
|
- uid = gconv.String((*idmdata)[0]["uid"])
|
|
|
- }
|
|
|
- }
|
|
|
- if userid == "" {
|
|
|
- return
|
|
|
- }
|
|
|
- url_ := gconv.String(thisData["url"])
|
|
|
- reg := regexp.MustCompile(".*article/content/(.*)\\.html")
|
|
|
- portrait1reg := regexp.MustCompile(".*/swordfish/page_big_pc/unit_portrayal/(.*)")
|
|
|
- portrait2reg := regexp.MustCompile(".*swordfish/page_big_pc/(.*)/ent_ser_portrait")
|
|
|
- searchreg := regexp.MustCompile(".*jybx/core/(.*)/searchList")
|
|
|
- contentnum := 0
|
|
|
- portraitnum := 0
|
|
|
- searchnum := 0
|
|
|
- if reg.MatchString(url_) {
|
|
|
- contentnum = 1
|
|
|
- }
|
|
|
- if portrait1reg.MatchString(url_) || portrait2reg.MatchString(url_) {
|
|
|
- portraitnum = 1
|
|
|
- }
|
|
|
- if searchreg.MatchString(url_) {
|
|
|
- searchnum = 1
|
|
|
- }
|
|
|
- createtime := gconv.Int64(thisData["date"])
|
|
|
- starttime, endtime := getToday(createtime)
|
|
|
- craetetimeStr := time.Unix(createtime, 0).Format(date.Date_Full_Layout)
|
|
|
- platform := APP
|
|
|
- if table == "jy_logs" {
|
|
|
- platform = PC
|
|
|
- if client := gconv.String(thisData["client"]); client != "" {
|
|
|
- if RegWx.MatchString(client) {
|
|
|
- platform = WX
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- lock(userid).Lock()
|
|
|
- defer lock(userid).Unlock()
|
|
|
- if config.JianyuSubjectdb.CountBySql(`select count(1) from dwd_f_userbase_visit_info where userid = ? and createtime>= ? and createtime <?`, userid, starttime, endtime) > 0 {
|
|
|
- if contentnum != 0 {
|
|
|
- config.JianyuSubjectdb.UpdateOrDeleteBySql(`update dwd_f_userbase_visit_info set number = number+1,contentnum = contentnum + 1,date =?,platform=? where userid = ? and createtime>= ? and createtime <?`, craetetimeStr, platform, userid, starttime, endtime)
|
|
|
- } else if portraitnum != 0 {
|
|
|
- config.JianyuSubjectdb.UpdateOrDeleteBySql(`update dwd_f_userbase_visit_info set number = number+1,portraitnum = contentnum + 1,date =?,platform=? where userid = ? and createtime>= ? and createtime <?`, craetetimeStr, platform, userid, starttime, endtime)
|
|
|
- } else if searchnum != 0 {
|
|
|
- config.JianyuSubjectdb.UpdateOrDeleteBySql(`update dwd_f_userbase_visit_info set number = number+1,searchnum = contentnum + 1,date =?,platform=? where userid = ? and createtime>= ? and createtime <?`, craetetimeStr, platform, userid, starttime, endtime)
|
|
|
- } else {
|
|
|
- config.JianyuSubjectdb.UpdateOrDeleteBySql(`update dwd_f_userbase_visit_info set number = number+1,date =?,platform=? where userid = ? and createtime>= ? and createtime <?`, craetetimeStr, platform, userid, starttime, endtime)
|
|
|
- }
|
|
|
- } else {
|
|
|
- config.JianyuSubjectdb.InsertBySql(`INSERT INTO dwd_f_userbase_visit_info
|
|
|
- (userid,DATE, number, platform,createtime,contentnum,portraitnum,searchnum)
|
|
|
- VALUES (?,?,?,?,?,?,?,?)`, userid, craetetimeStr, 1, platform, craetetimeStr, contentnum, portraitnum, searchnum)
|
|
|
- }
|
|
|
- if uid != "" {
|
|
|
- if config.JianyuSubjectdb.CountBySql(`select count(1) from dwd_f_crm_attribute_label where uid = ?`, uid) > 0 {
|
|
|
- config.JianyuSubjectdb.Update("dwd_f_crm_attribute_label", map[string]interface{}{"uid": uid}, map[string]interface{}{"last_login_time": craetetimeStr})
|
|
|
- } else {
|
|
|
- config.JianyuSubjectdb.Insert("dwd_f_crm_attribute_label", map[string]interface{}{
|
|
|
- "uid": uid,
|
|
|
- "last_login_time": craetetimeStr,
|
|
|
- "updatetime": time.Now().Format(time.DateTime),
|
|
|
- })
|
|
|
- }
|
|
|
+ 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)
|
|
|
+ for thisData := map[string]interface{}{}; iter.Next(&thisData); {
|
|
|
+ userId := gconv.String(thisData["userid"])
|
|
|
+ if userId == "" || idMapping[userId] == nil {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ userId = idMapping[userId].UserId
|
|
|
+ createtime := gtime.New(thisData["date"])
|
|
|
+ ymd := createtime.Format("Y-m-d")
|
|
|
+ if all[ymd] == nil {
|
|
|
+ all[ymd] = map[string]*VisitInfo{}
|
|
|
+ }
|
|
|
+ vi := all[ymd][userId]
|
|
|
+ if vi == nil {
|
|
|
+ vi = &VisitInfo{
|
|
|
+ uId: idMapping[userId].Uid,
|
|
|
+ }
|
|
|
+ }
|
|
|
+ vi.number++
|
|
|
+ vi.lastLoginTime = createtime.String()
|
|
|
+ lastLogin[userId] = vi.lastLoginTime
|
|
|
+ url_ := gconv.String(thisData["url"])
|
|
|
+ if contentReg.MatchString(url_) {
|
|
|
+ vi.contentNum++
|
|
|
+ } else if portrait1Reg.MatchString(url_) || portrait2Reg.MatchString(url_) {
|
|
|
+ vi.portraitNum++
|
|
|
+ } else if searchReg.MatchString(url_) {
|
|
|
+ vi.searchNum++
|
|
|
+ }
|
|
|
+ vi.platform = APP
|
|
|
+ if table == "jy_logs" {
|
|
|
+ vi.platform = PC
|
|
|
+ if client := gconv.String(thisData["client"]); client != "" {
|
|
|
+ if wxReg.MatchString(client) {
|
|
|
+ vi.platform = WX
|
|
|
}
|
|
|
- }(thisData)
|
|
|
- count++
|
|
|
- if count%5000 == 0 {
|
|
|
- log.Printf("%s已完成%d条数据\n", table, count)
|
|
|
}
|
|
|
- thisData = map[string]interface{}{}
|
|
|
}
|
|
|
- log.Printf("%s一共完成%d条数据\n", table, count)
|
|
|
- log.Println("end!")
|
|
|
- // sWait.Wait()
|
|
|
- }(table)
|
|
|
+ all[ymd][userId] = vi
|
|
|
+ thisData = map[string]interface{}{}
|
|
|
+ count++
|
|
|
+ if count%50000 == 0 {
|
|
|
+ log.Printf("遍历日志表", table, count)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.Println("遍历日志表结束。。。", table, count)
|
|
|
}
|
|
|
-}
|
|
|
-
|
|
|
-func getToday(createtime int64) (start, end string) {
|
|
|
- now := time.Unix(createtime, 0)
|
|
|
- startOfDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
|
|
|
- endOfDay := time.Date(now.Year(), now.Month(), now.Day(), 23, 59, 59, 0, now.Location())
|
|
|
- return startOfDay.Format(date.Date_Full_Layout), endOfDay.Format(date.Date_Full_Layout)
|
|
|
+ pool := make(chan bool, g.Cfg().MustGet(ctx, "visitInfoSavePool").Int())
|
|
|
+ wait := &sync.WaitGroup{}
|
|
|
+ index := 0
|
|
|
+ log.Println("开始保存统计结果。。。")
|
|
|
+ for k, v := range all {
|
|
|
+ t := gtime.New(k)
|
|
|
+ startTime, endTime := t.String(), t.AddDate(0, 0, 1).String()
|
|
|
+ for kk, vv := range v {
|
|
|
+ index++
|
|
|
+ if index%500 == 0 {
|
|
|
+ log.Println("保存统计结果", index)
|
|
|
+ }
|
|
|
+ //
|
|
|
+ pool <- true
|
|
|
+ wait.Add(1)
|
|
|
+ go func(userId string, vi *VisitInfo) {
|
|
|
+ defer func() {
|
|
|
+ <-pool
|
|
|
+ wait.Done()
|
|
|
+ }()
|
|
|
+ if list := config.JianyuSubjectdb.SelectBySql(`select id from dwd_f_userbase_visit_info where userid=? and createtime>=? and createtime<? limit 1`, userId, startTime, endTime); list != nil && len(*list) == 1 {
|
|
|
+ config.JianyuSubjectdb.UpdateOrDeleteBySql(`update dwd_f_userbase_visit_info set number=number+`+fmt.Sprint(vi.number)+`,contentnum=contentnum+`+fmt.Sprint(vi.contentNum)+`,portraitnum=portraitnum+`+fmt.Sprint(vi.portraitNum)+`,searchnum=searchnum+`+fmt.Sprint(vi.searchNum)+`,date=?,platform=? where id=?`, vi.lastLoginTime, vi.platform, (*list)[0]["id"])
|
|
|
+ } else {
|
|
|
+ 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.ExecBySql("update dwd_f_crm_attribute_label set last_login_time=? where uid=?", vi.uId, lastLogin[userId])
|
|
|
+ } else {
|
|
|
+ config.JianyuSubjectdb.InsertBatch("dwd_f_crm_attribute_label", []string{"uid", "last_login_time", "updatetime"}, []interface{}{vi.uId, lastLogin[userId], gtime.Now().String()})
|
|
|
+ }
|
|
|
+ }(kk, vv)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.Println("保存统计结果结束。。。", index)
|
|
|
+ wait.Wait()
|
|
|
+ //
|
|
|
+ log.Println("开始更新不活跃重新活跃用户。。。")
|
|
|
+ sTime := gtime.Now().StartOfDay()
|
|
|
+ day30, today, nowFormat, tomorrow := sTime.AddDate(0, 0, -30).String(), sTime.String(), gtime.Now().String(), sTime.AddDate(0, 0, 1).String()
|
|
|
+ config.JianyuSubjectdb.SelectByBath(1, func(l *[]map[string]interface{}) bool {
|
|
|
+ userid := gconv.String((*l)[0]["userid"])
|
|
|
+ datas := config.JianyuMaindb.SelectBySql(`select act_again_date from bi_service.freeClubSign where mogUserId=? limit 1`, userid)
|
|
|
+ if datas == nil || len(*datas) == 0 || gtime.New((*datas)[0]["act_again_date"]).Before(sTime) {
|
|
|
+ log.Println("更新不活跃重新活跃用户", userid)
|
|
|
+ config.JianyuMaindb.ExecBySql(`INSERT INTO bi_service.freeClubSign (mogUserId,act_again_date,create_time) VALUES (?,?,?) ON DUPLICATE KEY UPDATE act_again_date=?`, userid, nowFormat, nowFormat, nowFormat)
|
|
|
+ }
|
|
|
+ return true
|
|
|
+ }, `SELECT userid FROM dwd_f_userbase_visit_info a WHERE a.date>? AND a.date<? AND a.contentnum>=5 and not exists(
|
|
|
+ select 1 from dwd_f_userbase_visit_info b where b.date>? and b.date<? and b.contentnum>=5 and a.userid=b.userid
|
|
|
+ ) and not exists (SELECT 1 FROM dwd_f_data_equity_info b WHERE b.endtime>? and a.userid=b.userid)`, today, tomorrow, day30, today, nowFormat)
|
|
|
+ log.Println("更新不活跃重新活跃用户结束。。。")
|
|
|
}
|
|
|
|
|
|
type UserVisit struct {
|
|
|
@@ -256,6 +259,7 @@ func CountMaxVisit() {
|
|
|
})
|
|
|
}
|
|
|
}
|
|
|
+ mData = make(map[string]*UserVisit)
|
|
|
}
|
|
|
|
|
|
// @Author jianghan
|
|
|
@@ -311,8 +315,8 @@ func Count3DaysSearch() {
|
|
|
countSql = "SELECT COUNT(v.id) FROM dwd_f_userbase_search_info v WHERE DATE(v.search_time) BETWEEN DATE(NOW() - INTERVAL 3 DAY) AND DATE(NOW())"
|
|
|
} else {
|
|
|
aTime := time.Now().Add(-1 * time.Hour)
|
|
|
- sql = "SELECT v.userid, b.uid, v.search_time FROM dwd_f_userbase_search_info v LEFT JOIN dwd_f_userbase_baseinfo b ON v.userid = b.userid WHERE v.search_time >= " + aTime.Format(time.DateTime) + " ORDER BY v.date ASC"
|
|
|
- countSql = "SELECT COUNT(v.id) FROM dwd_f_userbase_search_info v WHERE v.search_time >= " + aTime.Format(time.DateTime)
|
|
|
+ sql = fmt.Sprintf("SELECT v.userid, b.uid, v.search_time FROM dwd_f_userbase_search_info v LEFT JOIN dwd_f_userbase_baseinfo b ON v.userid = b.userid WHERE v.search_time >= '%s' ORDER BY v.search_time ASC", aTime.Format(time.DateTime))
|
|
|
+ countSql = fmt.Sprintf("SELECT COUNT(v.id) FROM dwd_f_userbase_search_info v WHERE v.search_time >= '%s'", aTime.Format(time.DateTime))
|
|
|
}
|
|
|
total = config.JianyuSubjectdb.CountBySql(countSql)
|
|
|
log.Println("Count3DaysSearch total---", total)
|
|
|
@@ -364,6 +368,7 @@ func Count3DaysSearch() {
|
|
|
})
|
|
|
}
|
|
|
}
|
|
|
+ mSearchMap = make(map[string]int)
|
|
|
}
|
|
|
|
|
|
func Count3DaysDetail() {
|
|
|
@@ -391,8 +396,8 @@ func Count3DaysDetail() {
|
|
|
countSql = "SELECT COUNT(v.id) FROM dwd_f_userbase_visit_info v WHERE DATE(date) BETWEEN DATE(NOW() - INTERVAL 3 DAY) AND DATE(NOW())"
|
|
|
} else {
|
|
|
aTime := time.Now().Add(-1 * time.Hour)
|
|
|
- sql = "SELECT v.userid, b.uid, v.date, v.contentnum FROM dwd_f_userbase_visit_info v LEFT JOIN dwd_f_userbase_baseinfo b ON v.userid = b.userid WHERE v.date >= " + aTime.Format(time.DateTime) + " ORDER BY v.date ASC"
|
|
|
- countSql = "SELECT COUNT(v.id) FROM dwd_f_userbase_visit_info v WHERE v.date >= " + aTime.Format(time.DateTime)
|
|
|
+ sql = fmt.Sprintf("SELECT v.userid, b.uid, v.date, v.contentnum FROM dwd_f_userbase_visit_info v LEFT JOIN dwd_f_userbase_baseinfo b ON v.userid = b.userid WHERE v.date >= '%s' ORDER BY v.date ASC", aTime.Format(time.DateTime))
|
|
|
+ countSql = fmt.Sprintf("SELECT COUNT(v.id) FROM dwd_f_userbase_visit_info v WHERE v.date >= '%s'", aTime.Format(time.DateTime))
|
|
|
}
|
|
|
total = config.JianyuSubjectdb.CountBySql(countSql)
|
|
|
log.Println("Count3DaysDetail total---", total)
|
|
|
@@ -413,7 +418,7 @@ func Count3DaysDetail() {
|
|
|
}
|
|
|
err = rows.Scan(scanArgs...)
|
|
|
if err != nil {
|
|
|
- log.Println("CountMaxVisit---", err)
|
|
|
+ log.Println("Count3DaysDetail---", err)
|
|
|
break
|
|
|
}
|
|
|
for i, col := range values {
|
|
|
@@ -424,7 +429,7 @@ func Count3DaysDetail() {
|
|
|
}
|
|
|
}
|
|
|
count++
|
|
|
- if count%2000 == 0 {
|
|
|
+ if count%5000 == 0 {
|
|
|
log.Println("Count3DaysDetail current-------", count)
|
|
|
}
|
|
|
uid := common.ObjToString(ret["uid"])
|
|
|
@@ -445,6 +450,7 @@ func Count3DaysDetail() {
|
|
|
})
|
|
|
}
|
|
|
}
|
|
|
+ mDetailMap = make(map[string]int)
|
|
|
}
|
|
|
|
|
|
func VipExpire() {
|
|
|
@@ -460,8 +466,8 @@ func VipExpire() {
|
|
|
config.JianyuSubjectdb.ExecBySql("update dwd_f_crm_attribute_label set supersub = 0 where supersub = 1")
|
|
|
config.JianyuSubjectdb.ExecBySql("update dwd_f_crm_attribute_label set areasubpkg = 0 where areasubpkg = 1")
|
|
|
|
|
|
- sql = fmt.Sprintf("select uid, product_type from dwd_f_data_equity_info WHERE product_type = '省份订阅包' or product_type = '超级订阅' AND endtime > %s", now.Format(date.Date_Short_Layout))
|
|
|
- countSql = fmt.Sprintf("select count(id) from dwd_f_data_equity_info WHERE product_type = '省份订阅包' or product_type = '超级订阅' AND endtime > %s", now.Format(date.Date_Short_Layout))
|
|
|
+ sql = fmt.Sprintf("select uid, product_type from dwd_f_data_equity_info WHERE product_type = '省份订阅包' or product_type = '超级订阅' AND endtime > '%s'", now.Format(date.Date_Short_Layout))
|
|
|
+ countSql = fmt.Sprintf("select count(id) from dwd_f_data_equity_info WHERE product_type = '省份订阅包' or product_type = '超级订阅' AND endtime > '%s'", now.Format(date.Date_Short_Layout))
|
|
|
|
|
|
total = config.JianyuSubjectdb.CountBySql(countSql)
|
|
|
log.Println("VipExpire total---", total)
|
|
|
@@ -493,7 +499,7 @@ func VipExpire() {
|
|
|
}
|
|
|
}
|
|
|
count++
|
|
|
- if count%2000 == 0 {
|
|
|
+ if count%5000 == 0 {
|
|
|
log.Println("VipExpire current-------", count)
|
|
|
}
|
|
|
|
