synthesis.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419
  1. package service
  2. import (
  3. "context"
  4. "errors"
  5. "log"
  6. "math"
  7. "time"
  8. "github.com/mhaya/game/game_cluster/internal/constant"
  9. "github.com/mhaya/game/game_cluster/internal/mdb"
  10. "github.com/mhaya/game/game_cluster/internal/mdb/models"
  11. "github.com/mhaya/game/game_cluster/nodes/webadmin/entity"
  12. "github.com/spf13/cast"
  13. "go.mongodb.org/mongo-driver/bson"
  14. "go.mongodb.org/mongo-driver/bson/primitive"
  15. "go.mongodb.org/mongo-driver/mongo"
  16. "go.mongodb.org/mongo-driver/mongo/options"
  17. )
  18. type Synthesis struct {
  19. db *mongo.Database
  20. }
  21. func NewSynthesis() *Synthesis {
  22. return &Synthesis{
  23. db: mdb.MDB,
  24. }
  25. }
  26. func (s *Synthesis) FindMDBUserLogDaily(req *entity.UserLogDailyReq) ([]entity.UserLogDailyResp, error) {
  27. // 定义上下文
  28. ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
  29. defer cancel()
  30. // 指定集合
  31. collection := mdb.MDB.Collection("playerDailyRecord")
  32. // 构建查询条件 - 如果查询值为空那就不添加查询条件
  33. filter := bson.M{}
  34. if req.StartTime != 0 {
  35. filter["daily"] = bson.M{
  36. "$gte": req.StartTime,
  37. "$lte": req.EndTime,
  38. }
  39. }
  40. if req.Platform != "" && req.Platform != "all" {
  41. filter["platform"] = req.Platform
  42. }
  43. if req.Channel != "" {
  44. filter["channel"] = req.Channel
  45. }
  46. // 分页参数
  47. skip := (req.Page - 1) * req.Size
  48. req.Total = 0
  49. // 计算总数
  50. count, err := collection.CountDocuments(ctx, filter)
  51. if err != nil {
  52. return nil, err
  53. }
  54. req.Total = count
  55. // 执行查询
  56. opts := options.Find()
  57. opts.SetSkip(int64(skip))
  58. opts.SetLimit(int64(req.Size))
  59. opts.SetSort(bson.D{{"daily", -1}})
  60. cursor, err := collection.Find(ctx, filter, opts)
  61. if err != nil {
  62. return nil, err
  63. }
  64. defer cursor.Close(ctx)
  65. // 解析查询结果
  66. var results []entity.UserLogDailyResp
  67. for cursor.Next(ctx) {
  68. var result entity.UserLogDailyResp
  69. err := cursor.Decode(&result)
  70. if err != nil {
  71. return nil, err
  72. }
  73. results = append(results, result)
  74. }
  75. // 同一天的数据全部放到platform= ALl 且数据累加
  76. // 如果没有platform=all 的那就新增一个 同一个时间段只能有一个 platform=all
  77. // 将同一天的数据累加到platform=all的记录中
  78. allPlatformRecordMap := make(map[int64]*entity.UserLogDailyResp)
  79. for _, result := range results {
  80. allPlatformRecordMap[result.Timestamp] = &entity.UserLogDailyResp{
  81. Timestamp: result.Timestamp,
  82. Platform: "all",
  83. }
  84. }
  85. for _, v := range results {
  86. if v.Timestamp == allPlatformRecordMap[v.Timestamp].Timestamp {
  87. allPlatformRecordMap[v.Timestamp].Registered += v.Registered
  88. allPlatformRecordMap[v.Timestamp].LoggedIn += v.LoggedIn
  89. allPlatformRecordMap[v.Timestamp].NewActive += v.NewActive
  90. allPlatformRecordMap[v.Timestamp].OldActive += v.OldActive
  91. allPlatformRecordMap[v.Timestamp].TotalPoints += v.TotalPoints
  92. allPlatformRecordMap[v.Timestamp].UProduced += v.UProduced
  93. allPlatformRecordMap[v.Timestamp].UCashout += v.UCashout
  94. allPlatformRecordMap[v.Timestamp].NewLogin += v.NewLogin
  95. allPlatformRecordMap[v.Timestamp].OldLogin += v.OldLogin
  96. }
  97. }
  98. // 替换原有结果
  99. for _, record := range allPlatformRecordMap {
  100. results = append(results, *record)
  101. }
  102. return results, nil
  103. }
  104. // FindWithdrawal 根据请求查询提现记录
  105. func (s *Synthesis) FindWithdrawal(req *entity.UserWithdrawalReq) ([]*entity.UserWithdrawalResp, int64, error) {
  106. ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
  107. defer cancel()
  108. collection := mdb.MDB.Collection(constant.CNameCashOutRecord)
  109. // 构建过滤器
  110. filter := bson.M{}
  111. if req.UserName != "" {
  112. filter["userName"] = req.UserName
  113. }
  114. if req.NickName != "" {
  115. filter["nickName"] = req.NickName
  116. }
  117. if req.StartTime != 0 {
  118. filter["createAt"] = bson.M{
  119. "$gte": req.StartTime,
  120. "$lte": req.EndTime,
  121. }
  122. }
  123. // 设置分页选项
  124. findOptions := options.Find()
  125. findOptions.SetSkip(int64((req.Page - 1) * req.Size))
  126. findOptions.SetLimit(int64(req.Size))
  127. findOptions.SetSort(bson.D{{"createAt", -1}})
  128. // 获取总数total
  129. count, err := collection.CountDocuments(ctx, filter)
  130. if err != nil {
  131. return nil, 0, err
  132. }
  133. // 查询数据
  134. var results []*entity.UserWithdrawalResp
  135. cursor, err := collection.Find(ctx, filter, findOptions)
  136. if err != nil {
  137. return nil, 0, err
  138. }
  139. defer cursor.Close(ctx)
  140. // 解析结果
  141. for cursor.Next(ctx) {
  142. var result entity.UserWithdrawalResp
  143. if err := cursor.Decode(&result); err != nil {
  144. return nil, 0, err
  145. }
  146. results = append(results, &result)
  147. }
  148. if err := cursor.Err(); err != nil {
  149. return nil, 0, err
  150. }
  151. return results, count, nil
  152. }
  153. // WithdrawalStatus 更新提现状态
  154. func (s *Synthesis) WithdrawalStatus(req *entity.UserWithdrawalStatus) error {
  155. // ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
  156. // defer cancel()
  157. collection := mdb.MDB.Collection(constant.CNameCashOutRecord)
  158. // 更新条件
  159. updateCondition := bson.M{"userName": req.UserName}
  160. // 更新内容
  161. updateContent := bson.M{"$set": bson.M{"status": req.Status}}
  162. // 设置更新选项
  163. updateOptions := options.Update() // 设置 upsert 选项
  164. // 执行更新操作
  165. _, err := collection.UpdateOne(context.TODO(), updateCondition, updateContent, updateOptions)
  166. if err != nil {
  167. return err
  168. }
  169. return nil
  170. }
  171. // WithdrawalStatusBatch 更新提现状态
  172. func (s *Synthesis) WithdrawalStatusBatch(req *entity.UserWithdrawalStatusBatch) error {
  173. collection := mdb.MDB.Collection(constant.CNameCashOutRecord)
  174. if len(req.ID) == 0 {
  175. return errors.New("id 不能为空")
  176. }
  177. for _, id := range req.ID {
  178. objID := primitive.ObjectID{}
  179. objID, _ = primitive.ObjectIDFromHex(id)
  180. updateCondition := bson.M{"_id": objID}
  181. updateContent := bson.M{}
  182. withdrawal := models.CashOutRecord{}
  183. err := collection.FindOne(context.TODO(), updateCondition).Decode(&withdrawal)
  184. if err != nil {
  185. log.Printf("查询提现记录失败: %v", err)
  186. continue
  187. }
  188. if req.Withdrawal != 0 {
  189. if withdrawal.Status == 1 {
  190. updateContent = bson.M{"$set": bson.M{"withdrawal": req.Withdrawal}}
  191. } else {
  192. continue
  193. }
  194. }
  195. if req.Status > 0 {
  196. if withdrawal.Status != 0 {
  197. continue
  198. }
  199. updateContent = bson.M{"$set": bson.M{"status": req.Status}}
  200. }
  201. updateOptions := options.Update().SetUpsert(true)
  202. _, err = collection.UpdateOne(context.TODO(), updateCondition, updateContent, updateOptions)
  203. if err != nil {
  204. continue
  205. }
  206. }
  207. return nil
  208. }
  209. // FindUserCountryCount 查询用户国家分布
  210. //
  211. // 返回值为 UserCountryResp 的切片和错误。
  212. func (s *Synthesis) FindUserCountryCount() ([]*entity.UserCountryResp, error) {
  213. // 选择数据库和集合
  214. collection := mdb.MDB.Collection("playerCountryStat")
  215. // 定义聚合管道
  216. // 定义聚合管道
  217. pipeline := []bson.D{
  218. {
  219. {Key: "$project", Value: bson.D{
  220. {Key: "playerRegisterCountry", Value: bson.D{{Key: "$objectToArray", Value: "$playerRegisterCountry"}}},
  221. }},
  222. },
  223. {
  224. {Key: "$unwind", Value: "$playerRegisterCountry"},
  225. },
  226. {
  227. {Key: "$group", Value: bson.D{
  228. {Key: "_id", Value: "$playerRegisterCountry.k"},
  229. {Key: "totalValue", Value: bson.D{{Key: "$sum", Value: "$playerRegisterCountry.v"}}},
  230. }},
  231. },
  232. {
  233. {Key: "$project", Value: bson.D{
  234. {Key: "_id", Value: 0},
  235. {Key: "countryKey", Value: "$_id"},
  236. {Key: "totalValue", Value: 1},
  237. }},
  238. },
  239. }
  240. // 执行聚合查询
  241. cursor, err := collection.Aggregate(context.TODO(), pipeline)
  242. if err != nil {
  243. return nil, err
  244. }
  245. defer cursor.Close(context.TODO())
  246. // 遍历查询结果
  247. var results []bson.M
  248. if err := cursor.All(context.TODO(), &results); err != nil {
  249. return nil, err
  250. }
  251. var totalIPCount int64
  252. var data []*entity.UserCountryResp
  253. // 将结果转换为 UserCountryResp
  254. for _, r := range results {
  255. var resp entity.UserCountryResp
  256. resp.Country = r["countryKey"].(string)
  257. resp.IPCount = int(r["totalValue"].(int32))
  258. totalIPCount += int64(resp.IPCount)
  259. data = append(data, &resp)
  260. }
  261. for _, v := range data {
  262. // 保留小数点后两位
  263. v.Percentage = math.Round(float64(v.IPCount)/float64(totalIPCount)*10000) / 100
  264. }
  265. // 根据阈值过滤结果
  266. otherCount := 0
  267. otherPercentage := 0.00
  268. filteredResults := make([]*entity.UserCountryResp, 0)
  269. threshold := 1.00
  270. for _, r := range data {
  271. if r.Percentage >= threshold {
  272. filteredResults = append(filteredResults, r)
  273. // 保留小数点后两位
  274. r.Percentage = math.Round(r.Percentage*100) / 100
  275. otherPercentage += r.Percentage
  276. } else {
  277. otherCount += r.IPCount
  278. }
  279. }
  280. // 将其他国家添加到过滤后的结果中
  281. if otherCount > 0 {
  282. p := 100.00 - math.Round(otherPercentage*100)/100
  283. filteredResults = append(filteredResults, &entity.UserCountryResp{
  284. Country: "other",
  285. IPCount: otherCount,
  286. Percentage: math.Round(p*100) / 100,
  287. })
  288. }
  289. return filteredResults, nil
  290. }
  291. // FindUserRetention UserRetentionResp 用户留存率
  292. // 1. 获取指定日期范围内的注册用户数量
  293. // 2. 获取指定日期范围内的活跃用户数量
  294. // 3. 计算留存率
  295. // 4. 返回结果
  296. func (s *Synthesis) FindUserRetention(req *entity.UserRetentionReq) ([]*entity.UserRetentionResp, error) {
  297. playerPreserve := models.GetPlayerPreserve(req.StartTime, req.EndTime)
  298. // 查询数据
  299. var results []*entity.UserRetentionResp
  300. for key, v := range playerPreserve {
  301. var resp entity.UserRetentionResp
  302. var retention entity.Retention
  303. for _, vv := range v {
  304. if vv.ID == 1 {
  305. retention.Day1 = entity.DayRetention{
  306. LoggedIn: vv.Ratio,
  307. LoginDate: key,
  308. }
  309. }
  310. if vv.ID == 3 {
  311. retention.Day3 = entity.DayRetention{
  312. LoggedIn: vv.Ratio,
  313. LoginDate: key,
  314. }
  315. }
  316. if vv.ID == 7 {
  317. retention.Day7 = entity.DayRetention{
  318. LoggedIn: vv.Ratio,
  319. LoginDate: key,
  320. }
  321. }
  322. if vv.ID == 14 {
  323. retention.Day14 = entity.DayRetention{
  324. LoggedIn: vv.Ratio,
  325. LoginDate: key,
  326. }
  327. }
  328. if vv.ID == 30 {
  329. retention.Day30 = entity.DayRetention{
  330. LoggedIn: vv.Ratio,
  331. LoginDate: key,
  332. }
  333. }
  334. }
  335. resp.RegistrationDate = key
  336. resp.RetentionData = retention
  337. results = append(results, &resp)
  338. }
  339. return results, nil
  340. }
  341. // FindUserLevel 用户等级统计
  342. func (s *Synthesis) FindUserLevel() ([]*entity.UserLevelCountResp, error) {
  343. ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
  344. defer cancel()
  345. collection := mdb.MDB.Collection("playerLevelStat")
  346. // 查询所有文档
  347. cursor, err := collection.Find(ctx, bson.M{})
  348. if err != nil {
  349. return nil, err
  350. }
  351. defer cursor.Close(ctx)
  352. var results []*entity.UserLevelCountResp
  353. var reData []*models.PlayerLevelStat
  354. for cursor.Next(ctx) {
  355. var result models.PlayerLevelStat
  356. if err := cursor.Decode(&result); err != nil {
  357. return nil, err
  358. }
  359. reData = append(reData, &result)
  360. }
  361. if err := cursor.Err(); err != nil {
  362. return nil, err
  363. }
  364. dataMap := make(map[string]int)
  365. for _, v := range reData {
  366. for key, v1 := range v.ServerLevel {
  367. dataMap[key] += v1
  368. }
  369. }
  370. for k, v := range dataMap {
  371. results = append(results, &entity.UserLevelCountResp{
  372. Level: cast.ToInt(k),
  373. UserCount: v,
  374. })
  375. }
  376. return results, nil
  377. }