package rechorder import ( "fmt" "time" "91porn-server/common/db" "91porn-server/common/log" "91porn-server/common/timeutil/timerange" "91porn-server/models/s/statordermod" "91porn-server/models/s/statrecordmod" "91porn-server/skd/skdg" ) func RunStat(defaultStatDate time.Time) { log.Info("rechorderStat stat running...") defer log.Info("rechorderStat stat finished...") now := time.Now() //从历史提交记录中获取上次记录时间 lastStatAt, err := statrecordmod.LastRecordTime(statrecordmod.RechargeOrderStatJob, defaultStatDate) //第一次从默认记录时间开始统计 if err != nil { log.Error("rechorder RunStat lastRecordTime faild!", log.E(err)) return } //过去几天的数据按一日为单位统计,帮助数据恢复 dayHead := timerange.LocDayRange(now).Head for dayHead.After(lastStatAt) { end := lastStatAt.AddDate(0, 0, 1) //防止时间超出范围 if end.After(dayHead) { end = dayHead } subTimeRange := timerange.TimeRange{ //连续子切片 Head: lastStatAt, Tail: end, } if err := statByTimeRange(subTimeRange); err != nil { log.Error("rechorder RunStat statByTimeRange faild!", log.E(err)) return } lastStatAt = end } //当天的数据按照5分钟统计 recentMinute := timerange.RecentMinute(now, statrecordmod.FiveMinuteScale) //对齐本次统计时间 Minute % frequency == 0 //拆分时间区间并依次提交,控制数据库读写压力 for recentMinute.After(lastStatAt) { end := lastStatAt.Add(statrecordmod.FiveMinuteScale * time.Minute) //防止时间超出范围 if end.After(recentMinute) { end = recentMinute } subTimeRange := timerange.TimeRange{ //连续子切片 Head: lastStatAt, Tail: end, } if err := statByTimeRange(subTimeRange); err != nil { log.Error("rechorder RunStat statByTimeRange faild!", log.E(err)) return } lastStatAt = end } } func statByTimeRange(timeRange timerange.TimeRange) error { sumDate := timerange.LocDayRange(timeRange.Head).Head //获取此时的零日点 //装配渠道统计数据 recOrderStateMap := make(RecOrderStateMap) if err := fillRecOrderStatDocMap(sumDate, timeRange, &recOrderStateMap); err != nil { log.Error("rechorder RunStat fillRecOrderStatDocMap faild!", log.E(err)) return err } //提交渠道统计数据 log.Info("rechorderStat recharge committing...", log.Any("sumDate", sumDate), log.Any("location", sumDate.Location().String())) if err := commitRec(sumDate, timeRange.Tail, recOrderStateMap); err != nil { log.Error("rechorderStat RunStat commitRec faild!", log.E(err)) return err } log.Info("rechorderStat recharge committed.", log.Any("sumDate", sumDate), log.Any("location", sumDate.Location().String())) return nil } // commit 提交订单统计数据 func commitRec(sumDate time.Time, recordTime time.Time, docMap RecOrderStateMap) error { opt := (&db.TransOpts{}).SetReEntry(10) return skdg.StatDB.Trans(func(trans *db.MongoTool) error { if len(docMap) != 0 { if err := statordermod.ChangeRechargeStatTrans(trans, sumDate, recordTime, docMap); err != nil { return fmt.Errorf("stat commitRec ChangeRechargeStatTrans faild!, err:%+v\n", err) } } //提交本次修改记录 if err := statrecordmod.UpsertOneTrans(trans, statrecordmod.RechargeOrderStatJob, recordTime); err != nil { return fmt.Errorf("stat commitRec UpsertOneTrans faild!, err:%+v\n", err) } return nil }, opt) }