xpggame/modules/backend/handler/statistics/keepData.go

package statistics

import (
	"fmt"
	"reflect"
	"server/common"
	"server/db"
	"server/modules/backend/app"
	"server/modules/backend/models"
	utils "server/modules/backend/util"
	"server/modules/backend/values"
	"server/util"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/liangdas/mqant/log"
	"github.com/olivere/elastic/v7"
)

// KeepData 获取留存数据
func KeepData(c *gin.Context) {
	a := app.NewApp(c)
	defer func() {
		a.Response()
	}()
	req := new(values.KeepDataReq)
	if !a.S(req) {
		return
	}
	log.Debug("keep data req:%+v", req)
	resp := values.KeepDataResp{}
	su, eu := utils.GetQueryUnix(req.Start, req.End)

	resp.List = append(resp.List, GetKeepData(su, eu, req.ActiveKeep, req.Channel)...)

	// 七天内的新增付费数据直接拉取缓存
	/*last7 := util.GetZeroTime(time.Now()).Unix() - 7*24*60*60
	if req.ActiveKeep == 4 {
		resp.List = values.GetKeepRechargeData(su, eu)
	}

	if len(resp.List) == 0 {
		resp.List = GetKeepData(su, eu, req.ActiveKeep, req.Channel)
	} else if su < last7 {
		if eu > last7 {
			eu = last7
		}
		resp.List = append(resp.List, GetKeepData(su, eu, req.ActiveKeep, req.Channel)...)
	}*/
	// total := make([]int64, 9)
	// for _, v := range resp.List {
	// 	resp.Total.NewCount += v.NewCount
	// 	total[0] += int64(utils.AddPerFloat(0, v.K1) * float64(v.NewCount) / 100)
	// 	total[1] += int64(utils.AddPerFloat(0, v.K2) * float64(v.NewCount) / 100)
	// 	total[2] += int64(utils.AddPerFloat(0, v.K3) * float64(v.NewCount) / 100)
	// 	total[3] += int64(utils.AddPerFloat(0, v.K4) * float64(v.NewCount) / 100)
	// 	total[4] += int64(utils.AddPerFloat(0, v.K5) * float64(v.NewCount) / 100)
	// 	total[5] += int64(utils.AddPerFloat(0, v.K6) * float64(v.NewCount) / 100)
	// 	total[6] += int64(utils.AddPerFloat(0, v.K7) * float64(v.NewCount) / 100)
	// 	total[7] += int64(utils.AddPerFloat(0, v.K15) * float64(v.NewCount) / 100)
	// 	total[8] += int64(utils.AddPerFloat(0, v.K30) * float64(v.NewCount) / 100)
	// }
	// if req.Channel != nil {
	// 	resp.Total.Channel = *req.Channel
	// }
	resp.Count = int64(len(resp.List))
	s := (req.Page - 1) * req.Num
	if s > len(resp.List)-1 {
		a.Code = values.CodeParam
		a.Msg = "查询参数有误"
		return
	}
	e := req.Page * req.Num
	if e > len(resp.List) {
		e = len(resp.List)
	}
	resp.List = resp.List[s:e]
	// resp.Total.K1 = utils.GetPer(total[0], resp.Total.NewCount)
	// resp.Total.K2 = utils.GetPer(total[1], resp.Total.NewCount)
	// resp.Total.K3 = utils.GetPer(total[2], resp.Total.NewCount)
	// resp.Total.K4 = utils.GetPer(total[3], resp.Total.NewCount)
	// resp.Total.K5 = utils.GetPer(total[4], resp.Total.NewCount)
	// resp.Total.K6 = utils.GetPer(total[5], resp.Total.NewCount)
	// resp.Total.K7 = utils.GetPer(total[6], resp.Total.NewCount)
	// resp.Total.K15 = utils.GetPer(total[7], resp.Total.NewCount)
	// resp.Total.K30 = utils.GetPer(total[8], resp.Total.NewCount)
	a.Data = resp
}

func GetKeepData(s, e int64, t int, channel *int) []values.KeepData {
	q := elastic.NewBoolQuery()
	cid := 0
	if channel != nil {
		cid = *channel
	}
	var list []values.KeepData
	q.Must(elastic.NewMatchQuery("Channel", cid))
	q.Must(elastic.NewMatchQuery("Type", t))
	q.Filter(elastic.NewRangeQuery("Time").Gte(s))
	q.Filter(elastic.NewRangeQuery("Time").Lt(e))
	db.ES().QueryList(common.ESIndexBackKeepData, 0, 5000, q, &list, "Time", false)
	var oneDay int64 = 24 * 60 * 60
	now := util.GetZeroTime(time.Now()).Unix()
	flag := 0
	channelSql := ""
	if channel != nil {
		channelSql = fmt.Sprintf(" and channel_id = %v", *channel)
	}
	total := (e - s) / oneDay
	ret := make([]values.KeepData, total)
	count := 0
	group := new(sync.WaitGroup)
	group.Add(int(total))
	for j := e; j > s; j -= oneDay {
		var all int64 = 0
		start := j - oneDay
		end := j
		thisCount := count
		d := time.Unix(start, 0).Format("20060102")
		keep := &values.KeepData{Date: d, Type: t, Time: start, Channel: cid}

		var exist bool  // es中是否有该条数据
		var change bool // 该条数据是否需要写入es
		if flag < len(list) && list[flag].Time == start {
			keep = &list[flag]
			all = list[flag].NewCount
			exist = true
			flag++
		} else {
			if t == 1 {
				db.Mysql().C().Table("login_record").Where(fmt.Sprintf("date = '%v'%v", d, channelSql)).Distinct("uid").Count(&all)
			} else if t == 2 {
				all = models.GetNewPlayerCountBySql(channel, start, end)
			} else if t == 3 {
				all = models.GetPayCountBySql(channel, start, end)
			} else if t == 4 {
				all = models.GetNewPayCountBySql(channel, start, end)
			} else if t == 5 {
				all = models.GetNewPayWithdrawCountBySql(channel, start, end)
			}
			keep.NewCount = all
		}
		ref := reflect.ValueOf(keep).Elem()

		util.Go(func() {
			index := 1
			nowIndex := -1
			nowVal := ""
			for {
				k := start + int64(index)*oneDay
				if index > 30 || k > now {
					break
				}
				/*if index > 7 && index != 15 && index != 30 {
					index++
					continue
				}*/
				if index > 15 && index != 30 {
					index++
					continue
				}
				kField := ref.FieldByName(fmt.Sprintf("K%v", index))
				if kField.String() != "" {
					index++
					continue
				}
				oneD := time.Unix(k, 0).Format("20060102")
				isNow := k == now
				if isNow {
					nowIndex = index
				}
				sql := ""
				if t == 1 {
					sql = fmt.Sprintf(`SELECT count(a.uid) FROM
					(SELECT DISTINCT(uid) FROM login_record WHERE date = '%v'%v)a 
					INNER JOIN 
					(SELECT DISTINCT(uid) FROM login_record WHERE date = '%v'%v)b 
					on a.uid = b.uid `,
						d, channelSql, oneD, channelSql)
					// if k == now {
					// 	sql += " and status = 2"
					// }
				} else if t == 2 {
					sql = fmt.Sprintf("SELECT count(DISTINCT(uid)) from login_record WHERE first_time = %d and date = '%s' %s",
						start, oneD, channelSql)
				} else if t == 3 {
					sqlStr := "SELECT COUNT(DISTINCT(b.uid)) from login_record as b " +
						" INNER JOIN " +
						"(SELECT DISTINCT(uid) from recharge_order WHERE callback_time >=%d and callback_time <%d and `event` = %d and `status` = %d) as a " +
						"on a.uid = b.uid " +
						"WHERE date = '%s' %v AND a.uid = b.uid"
					sql = fmt.Sprintf(sqlStr,
						start,                        // 支付回调时间下限
						start+oneDay,                 // 支付回调时间上限
						common.CurrencyEventReCharge, // 支付事件
						common.StatusROrderPay,       // 支付状态
						oneD,                         // 统计结尾时间
						channelSql,                   // 渠道
					)
				} else if t == 4 {
					sqlStr :=
						"SELECT Count(DISTINCT(re.uid)) FROM recharge_order AS re " +
							" INNER JOIN " +
							" (SELECT DISTINCT(uid) FROM login_record WHERE date = '%s' and first_time = %d %v) as lo " +
							" ON re.uid = lo.uid " +
							" WHERE event = %d AND status = %d " +
							" AND callback_time >= %d " +
							" AND callback_time < %d %v"
					sql = fmt.Sprintf(sqlStr,
						oneD,                         // 统计登录时间
						start,                        // 注册时间下限
						channelSql,                   // 渠道
						common.CurrencyEventReCharge, // 支付事件
						common.StatusROrderPay,       // 支付状态
						start,                        // 支付回调时间下限
						start+oneDay,                 // 支付回调时间上限
						channelSql,                   // 渠道
					)
				} else if t == 5 {
					sqlStr :=
						"SELECT Count(DISTINCT(re.uid)) FROM withdraw_order AS re " +
							" INNER JOIN " +
							" (SELECT DISTINCT(uid) FROM login_record WHERE date = '%s' and first_time = %d %v) as lo " +
							" ON re.uid = lo.uid " +
							" WHERE event = %d AND status = %d " +
							" AND callback_time >= %d " +
							" AND callback_time < %d %v"
					sql = fmt.Sprintf(sqlStr,
						oneD,                         // 统计登录时间
						start,                        // 注册时间下限
						channelSql,                   // 渠道
						common.CurrencyEventReCharge, // 支付事件
						common.StatusROrderPay,       // 支付状态
						start,                        // 支付回调时间下限
						start+oneDay,                 // 支付回调时间上限
						channelSql,                   // 渠道
					)
				}
				var next int64
				if err := db.Mysql().C().Raw(sql).Scan(&next).Error; err != nil {
					log.Error("err:%v", err)
				}

				// 当天的数据会变动，不写入es
				// log.Debug("k:%v,now:%v", k, now)
				if isNow {
					nowVal = utils.GetPer(next, all)
				} else {
					change = true
					kField.SetString(utils.GetPer(next, all))
				}
				index++
			}
			if change {
				id := fmt.Sprintf("%v_%v_%v", d, cid, t)
				if exist {
					db.ES().DeleteByID(common.ESIndexBackKeepData, id)
				}
				db.ES().InsertToESByID(common.ESIndexBackKeepData, id, keep)
			}
			// 赋值当天的留存数据
			if nowIndex > 0 {
				ref.FieldByName(fmt.Sprintf("K%v", nowIndex)).SetString(nowVal)
			}
			ret[thisCount] = *keep
			group.Done()
		})
		count++
	}
	group.Wait()
	return ret
}

// 活跃玩家留存
func getActiveKeep(s, e int64, channel *int) ([]values.KeepData, []int64) {
	var ret []values.KeepData
	var oneDay int64 = 24 * 60 * 60

	now := time.Now().Unix()
	total := make([]int64, 9)
	for j := e; j > s; j -= oneDay {
		var index int64 = 1
		var all int64 = 0
		i := j - oneDay
		d := time.Unix(i, 0).Format("20060102")
		sql := fmt.Sprintf("date = '%v'", d)
		keep := values.KeepData{Date: d}
		if channel != nil {
			keep.Channel = *channel
			sql += fmt.Sprintf(" and channel_id = %v", *channel)
		}
		all = db.Mysql().Count(&common.LoginRecord{}, sql)
		keep.NewCount = all
		if all == 0 {
			ret = append(ret, keep)
			continue
		}
		// log.Debug("t:%v,i:%v,j:%v,all:%v", t, i, j, all)
		for {
			k := i + index*oneDay
			if index > 30 || k > now {
				break
			}
			if index <= 7 || index == 15 || index == 30 {
				oneD := time.Unix(k, 0).Format("20060102")
				channelSql := ""
				if channel != nil {
					channelSql = fmt.Sprintf(" and channel_id = %v", *channel)
				}
				sql := fmt.Sprintf("SELECT count(b.uid) FROM (SELECT uid FROM login_record WHERE date = '%v'%v)a LEFT JOIN (SELECT uid FROM login_record WHERE date = '%v'%v)b on a.uid=b.uid", d, channelSql, oneD, channelSql)
				var next int64
				if err := db.Mysql().C().Raw(sql).Scan(&next).Error; err != nil {
					log.Error("err:%v", err)
				}
				switch index {
				case 1:
					keep.K1 = utils.GetPer(next, all)
					total[0] += next
				case 2:
					keep.K2 = utils.GetPer(next, all)
					total[1] += next
				case 3:
					keep.K3 = utils.GetPer(next, all)
					total[2] += next
				case 4:
					keep.K4 = utils.GetPer(next, all)
					total[3] += next
				case 5:
					keep.K5 = utils.GetPer(next, all)
					total[4] += next
				case 6:
					keep.K6 = utils.GetPer(next, all)
					total[5] += next
				case 7:
					keep.K7 = utils.GetPer(next, all)
					total[6] += next
				case 15:
					keep.K15 = utils.GetPer(next, all)
					total[7] += next
				case 30:
					keep.K30 = utils.GetPer(next, all)
					total[8] += next
				}
			}
			index++
		}

		log.Debug("keep:%+v", keep)
		ret = append(ret, keep)
	}
	return ret, total
}

// 新增玩家留存
func getPlayerKeep(s, e int64, channel *int) ([]values.KeepData, []int64) {
	var ret []values.KeepData
	var oneDay int64 = 24 * 60 * 60

	now := time.Now().Unix()
	total := make([]int64, 16)
	for j := e; j > s; j -= oneDay {
		var index int64 = 1
		var all int64 = 0
		i := j - oneDay

		all = models.GetNewPlayerCountBySql(channel, i, i)
		t := time.Unix(i, 0).Format("20060102")
		keep := values.KeepData{Date: t, NewCount: all}
		if channel != nil {
			keep.Channel = *channel
		}
		if all == 0 {
			ret = append(ret, keep)
			continue
		}
		// log.Debug("t:%v,i:%v,j:%v,all:%v", t, i, j, all)
		for {
			k := i + index*oneDay
			if index > 30 || k > now {
				break
			}
			Kt := time.Unix(k, 0).Format("20060102")
			if index <= 15 || index == 30 {
				var str string
				if channel != nil {
					str = fmt.Sprintf("SELECT COUNT(DISTINCT(l.uid)) FROM login_record AS l INNER JOIN (SELECT * FROM users WHERE birth >= %d AND birth < %d AND channel_id = %d) AS b ON l.uid = b.id WHERE l.date = %s AND l.channel_id = %d", i, i+oneDay, *channel, Kt, *channel)
				} else {
					str = fmt.Sprintf("SELECT COUNT(DISTINCT(l.uid)) FROM login_record AS l INNER JOIN (SELECT * FROM users WHERE birth >= %d AND birth < %d) AS b ON l.uid = b.id WHERE l.date = %s ", i, i+oneDay, Kt)
				}

				var next int64
				err := db.Mysql().QueryBySql(str, &next)
				if err != nil {
					log.Error(err.Error())
				}

				switch index {
				case 1:
					keep.K1 = utils.GetPer(next, all)
					total[0] += next
				case 2:
					keep.K2 = utils.GetPer(next, all)
					total[1] += next
				case 3:
					keep.K3 = utils.GetPer(next, all)
					total[2] += next
				case 4:
					keep.K4 = utils.GetPer(next, all)
					total[3] += next
				case 5:
					keep.K5 = utils.GetPer(next, all)
					total[4] += next
				case 6:
					keep.K6 = utils.GetPer(next, all)
					total[5] += next
				case 7:
					keep.K7 = utils.GetPer(next, all)
					total[6] += next
				case 8:
					keep.K8 = utils.GetPer(next, all)
					total[7] += next
				case 9:
					keep.K9 = utils.GetPer(next, all)
					total[8] += next
				case 10:
					keep.K10 = utils.GetPer(next, all)
					total[9] += next
				case 11:
					keep.K11 = utils.GetPer(next, all)
					total[10] += next
				case 12:
					keep.K12 = utils.GetPer(next, all)
					total[11] += next
				case 13:
					keep.K13 = utils.GetPer(next, all)
					total[12] += next
				case 14:
					keep.K14 = utils.GetPer(next, all)
					total[13] += next
				case 15:
					keep.K15 = utils.GetPer(next, all)
					total[14] += next
				case 30:
					keep.K30 = utils.GetPer(next, all)
					total[15] += next
				}
			}
			index++
		}

		log.Debug("keep:%+v", keep)
		ret = append(ret, keep)
	}
	return ret, total
}

// 活跃付费存留
func getActivePayKeep(s, e int64, channel *int) ([]values.KeepData, []int64) {
	var ret []values.KeepData
	var oneDay int64 = 24 * 60 * 60

	now := time.Now().Unix()
	total := make([]int64, 9)

	var channelStr string
	if channel != nil {
		channelStr = fmt.Sprintf(" AND channel_id = %v ", *channel)
	}

	sqlStr := " SELECT COUNT(DISTINCT(re.uid)) FROM recharge_order re " +
		" LEFT JOIN " +
		" ( " +
		" SELECT uid FROM login_record WHERE " +
		" date = %s " +
		" %v " +
		" GROUP BY uid " +
		" ) lo " +
		" ON re.uid = lo.uid " +
		" WHERE  re.uid = lo.uid " +
		" AND re.event = %d " +
		" AND re.status = %d " +
		" AND re.callback_time >= %d " +
		" AND re.callback_time < %d " +
		" %v "

	for j := e; j > s; j -= oneDay {
		i := j - oneDay
		var index int64 = 1
		var all int64 = 0
		var keep values.KeepData
		var d = time.Unix(i, 0).Format("20060102")
		keep.Date = d
		if channel != nil {
			keep.Channel = *channel
		}
		all = models.GetPayCountBySql(channel, i, j)
		keep.NewCount = all
		if all == 0 {
			ret = append(ret, keep)
			continue
		}

		for {
			k := i + index*oneDay
			if index > 30 || k > now {
				break
			}
			if index <= 7 || index == 15 || index == 30 {
				oneD := time.Unix(k, 0).Format("20060102")

				sql := fmt.Sprintf(sqlStr,
					oneD,                         // 统计结尾时间
					channelStr,                   // 渠道
					common.CurrencyEventReCharge, // 支付事件
					common.StatusROrderPay,       // 支付状态
					i,                            // 支付回调时间下限
					i+24*60*60,                   // 支付回调时间上限
					channelStr,                   // 渠道
				)

				var next int64
				if err := db.Mysql().C().Raw(sql).Scan(&next).Error; err != nil {
					log.Error("err:%v", err)
				}
				switch index {
				case 1:
					keep.K1 = utils.GetPer(next, all)
					total[0] += next
				case 2:
					keep.K2 = utils.GetPer(next, all)
					total[1] += next
				case 3:
					keep.K3 = utils.GetPer(next, all)
					total[2] += next
				case 4:
					keep.K4 = utils.GetPer(next, all)
					total[3] += next
				case 5:
					keep.K5 = utils.GetPer(next, all)
					total[4] += next
				case 6:
					keep.K6 = utils.GetPer(next, all)
					total[5] += next
				case 7:
					keep.K7 = utils.GetPer(next, all)
					total[6] += next
				case 15:
					keep.K15 = utils.GetPer(next, all)
					total[7] += next
				case 30:
					keep.K30 = utils.GetPer(next, all)
					total[8] += next
				}
			}
			index++
		}

		ret = append(ret, keep)
	}
	return ret, total
}

// 新增付费存留
func getPlayerPayKeep(s, e int64, channel *int) ([]values.KeepData, []int64) {
	var ret []values.KeepData
	var oneDay int64 = 24 * 60 * 60

	now := time.Now().Unix()
	total := make([]int64, 9)

	sqlStr := " SELECT COUNT(DISTINCT(id)) FROM users " +
		" LEFT JOIN " +
		" ( " +
		" SELECT uid FROM recharge_order WHERE " +
		" event = %d " +
		" AND status = %d " +
		" AND callback_time >= %d" +
		" AND callback_time < %d" +
		" %v " +
		" GROUP BY uid" +
		" ) " +
		" AS re ON id = re.uid " +
		" LEFT JOIN " +
		" ( SELECT uid FROM login_record WHERE " +
		" date = %s " +
		" %v " +
		" GROUP BY uid " +
		" ) " +
		" AS lo ON id = lo.uid" +
		" WHERE " +
		" id = re.uid " +
		" AND id = lo.uid " +
		" AND birth >= %d " +
		" AND birth < %d " +
		" %v "

	var channelStr string
	if channel != nil {
		channelStr = fmt.Sprintf(" AND channel_id = %v ", *channel)
	}

	for j := e; j > s; j -= oneDay {
		var index int64 = 1
		var all int64 = 0
		i := j - oneDay

		all = models.GetNewPayCountBySql(channel, i, i)
		t := time.Unix(i, 0).Format("20060102")
		keep := values.KeepData{Date: t, NewCount: all}
		if channel != nil {
			keep.Channel = *channel
		}
		if all == 0 {
			ret = append(ret, keep)
			continue
		}
		for {
			k := i + index*oneDay
			if index > 30 || k > now {
				break
			}
			Kt := time.Unix(k, 0).Format("20060102")
			if index <= 7 || index == 15 || index == 30 {
				str := fmt.Sprintf(sqlStr,
					common.CurrencyEventReCharge, // 支付事件
					common.StatusROrderPay,       // 支付状态
					i,                            // 支付回调时间下限
					i+oneDay,                     // 支付回调时间上限
					channelStr,                   // 渠道
					Kt,                           // 统计登录时间
					channelStr,                   // 渠道
					i,                            // 注册时间下限
					i+oneDay,                     // 注册时间上限
					channelStr,                   // 渠道
				)

				var next int64
				err := db.Mysql().QueryBySql(str, &next)
				if err != nil {
					log.Error(err.Error())
				}

				switch index {
				case 1:
					keep.K1 = utils.GetPer(next, all)
					total[0] += next
				case 2:
					keep.K2 = utils.GetPer(next, all)
					total[1] += next
				case 3:
					keep.K3 = utils.GetPer(next, all)
					total[2] += next
				case 4:
					keep.K4 = utils.GetPer(next, all)
					total[3] += next
				case 5:
					keep.K5 = utils.GetPer(next, all)
					total[4] += next
				case 6:
					keep.K6 = utils.GetPer(next, all)
					total[5] += next
				case 7:
					keep.K7 = utils.GetPer(next, all)
					total[6] += next
				case 15:
					keep.K15 = utils.GetPer(next, all)
					total[7] += next
				case 30:
					keep.K30 = utils.GetPer(next, all)
					total[8] += next
				}
			}
			index++
		}

		log.Debug("keep:%+v", keep)
		ret = append(ret, keep)
	}
	return ret, total
}