update

config.json

@@ -76,5 +76,10 @@
     "source_coll": "event",
     "dest_coll": "summary_assets",
     "task_name": "summary_assets"
+  },
+  "login_repair": {
+    "source_coll": "event",
+    "dest_coll": "summary_login",
+    "task_name": "login_repair"
   }
 }

del_coll.py

@@ -0,0 +1,25 @@
+import pandas as pd
+import pymongo
+import datetime
+
+from settings import settings
+
+start_date = f'{datetime.datetime.now().year}-01-01'
+end_date = (datetime.datetime.now() - datetime.timedelta(days=150)).strftime('%Y-%m-%d')
+
+pd.date_range(start_date, end_date, freq="D")
+
+MONGO_URI = settings.local_mongo_uri
+
+myclient = pymongo.MongoClient(MONGO_URI)
+game_list = [item['id_name'] for item in
+             myclient['admin_game']['game'].find({'id_name': {'$exists': True}}, {'id_name': True})]
+
+
+for db in game_list:
+    mydb = myclient[f'game_{db}']
+    for d in pd.date_range(start_date, end_date, freq="D").to_list():
+        day = d.strftime('%Y-%m-%d')
+        event_coll = mydb[f'event_{day}']
+        event_coll.drop()
+        print(f'{db}  {day}')

settings.py

@@ -3,6 +3,7 @@ import sys
 
 
 class Config:
+    cursor_diff = -(3 * 60)
     local_mongo_uri = 'mongodb://root:iamciniao@127.0.0.1:27017/?authSource=admin&readPreference=primary&ssl=false'
 
     remote_mongo_uri = 'mongodb://root:Legu2020@dds-hp35c58764c35aa41188-pub.mongodb.huhehaote.rds.aliyuncs.com:3717,dds-hp35c58764c35aa42908-pub.mongodb.huhehaote.rds.aliyuncs.com:3717/admin?replicaSet=mgset-421510732'
@@ -20,8 +21,8 @@ class Debug(Config):
     run_model = 'debug'
     # local_mongo_uri = 'mongodb://root:iamciniao@39.104.90.204:27017/?authSource=admin&readPreference=primary&ssl=false'  # 妖灵
     # local_mongo_uri = 'mongodb://root:iamciniao@8.129.233.127:27017/?authSource=admin&readPreference=primary&ssl=false'  # 风暴
-    # local_mongo_uri = 'mongodb://root:iamciniao@42.194.225.107:27017/?authSource=admin&readPreference=primary&ssl=false'  # 生化
+    local_mongo_uri = 'mongodb://root:iamciniao@42.194.225.107:27017/?authSource=admin&readPreference=primary&ssl=false'  # 生化
-    local_mongo_uri = 'mongodb://root:iamciniao@119.29.176.224:27017/?authSource=admin&readPreference=primary&ssl=false'  # 歌手
+    # local_mongo_uri = 'mongodb://root:iamciniao@119.29.176.224:27017/?authSource=admin&readPreference=primary&ssl=false'  # 歌手
     # local_mongo_uri = 'mongodb://root:iamciniao@119.29.35.35:27017/?authSource=admin&readPreference=primary&ssl=false'  # 中古
 
 

task/first_recharge.py

@@ -33,7 +33,7 @@ class FirstRecharge(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/login_repair.py

@@ -0,0 +1,76 @@
+import traceback
+
+from pymongo import UpdateOne
+from pydantic import Field
+import pandas as pd
+
+from .task import Task
+from utils import *
+from model import GBaseModel
+
+
+class LoginRepair(Task):
+    """
+    每日活跃
+    """
+
+    class Model(GBaseModel):
+        cdate: int = Field(..., title='当天0点')
+        first_device_id: str = Field(None, title='第一次登录设备id', alias='_first_device_id')
+        manufacturer: str = Field(None, title='设备品牌', alias='_manufacturer')
+        model: str = Field(None, title='型号', alias='_model')
+        os_version: str = Field(None, title='系统版本', alias='_os_version')
+        app_name: str = Field(None, title='游戏版本', alias='_app_name')
+        screen_width: int = Field(None, title='屏幕宽', alias='_screen_width')
+        screen_height: int = Field(None, title='屏幕高', alias='_screen_height')
+
+    def cleaning(self, cursor_list):
+        for cursor in cursor_list:  # type:dict
+            for source_coll, ts in cursor.items():  # type:str,dict
+                if ts['cursor_st'] == ts['cursor_et']:
+                    continue
+                logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
+                where = {
+                    '_event_time': {
+                        '$gte': ts['cursor_st'],
+                        '$lt': ts['cursor_et'],
+                    },
+                }
+
+                projection = self.Model.get_fields()
+                bulk_data = []
+                # 当前活跃role id
+                role_list = self.local_db[source_coll].distinct('_game_role_id', where)
+                cdate = int(pd.Timestamp(ts['cursor_st'], unit='s', tz=self.timezone) \
+                            .normalize().timestamp())
+                exists_role_list = self.local_db[self.dest_coll].distinct('_game_role_id', {'cdate': cdate})
+
+                # 还没有记录的
+                role_set = set(role_list) - set(exists_role_list)
+                for role_id in role_set:
+                    # logger.debug(f'add {role_id}')
+                    item = self.local_db['user'].find_one(
+                        {'_game_role_id': role_id}, projection)
+                    if not item:
+                        continue
+                    try:
+                        item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \
+                                            .normalize().timestamp())
+                        model = self.Model(**item)
+                        data = model.dict(by_alias=True)
+                        bulk_data.append(
+                            UpdateOne({'cdate': item['cdate'], '_game_role_id': data['_game_role_id']}, {'$set': data},
+                                      upsert=True))
+                    except Exception as e:
+                        del item['_id']
+                        bulk_data.append(
+                            UpdateOne({'cdate': item['cdate'], '_game_role_id': item['_game_role_id']}, {'$set': item},
+                                      upsert=True))
+                        msg = traceback.format_exc()
+                        ddsend_msg(f'{self.game_name}.{source_coll}字段验证异常 还是写入了 {msg}\n{e.json()}')
+                        logger.error(repr(e))
+                        # pass
+                if bulk_data:
+                    self.local_db[self.dest_coll].bulk_write(bulk_data, ordered=False)
+                    self.remote_db[self.dest_coll].bulk_write(bulk_data, ordered=False)
+                self.set_cursor(cursor_st=ts['cursor_st'], cursor_et=ts['cursor_et'])

task/summary3.py

@@ -260,12 +260,18 @@ class Summary3(Task):
                 self.local_db[f'summary_{cat}'].update_one({'_id': data['_id']}, {'$set': data})
                 self.remote_db[f'summary_{cat}'].update_one({'_id': data['_id']}, {'$set': data})
             # 计算累计充值
+            type_map = {
+                'role': '_game_role_id',
+                'account': '_channel_uid',
+                'device': '_device_id',
+            }
+
             for id_, doc in update_doc.items():
                 pipeline = [
                     {
                         '$match': {
                             "cdate": {'$gte': doc['cdate'], '$lte': cdate},
-                            f'_game_{cat}_id': {'$in': doc[f'now_pay_{cat}_list']},
+                            type_map[cat]: {'$in': doc[f'now_pay_{cat}_list']},
                         }
                     },
                     {'$group': {'_id': None, 'sum_money': {'$sum': '$money'}}}

task/summary_assets.py

@@ -57,7 +57,7 @@ class SummaryAssets(Task):
                     '$or': [{'prize.a': {'$in': list(a)}, 'prize.t': {'$in': list(t)}},
                             {'need.a': {'$in': list(a)}, 'need.t': {'$in': list(t)}}],
                     '_event_name': 'res',
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/summary_func.py

@@ -30,7 +30,7 @@ class SummaryFunc(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': 'Func',
+                    '_event_name': 'Func',
                     '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],

task/summary_funnel.py

@@ -33,7 +33,7 @@ class SummaryFunnel(Task):
                 where = {
                     '_event_name': 'Guide',
                     'step': {'$in': step_list},
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/summary_join_hd.py

@@ -41,7 +41,7 @@ class SummaryJoinHd(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': 'res',
+                    '_event_name': 'res',
                     'function': 'hdgetprize',
                     '_event_time': {
                         '$gte': ts['cursor_st'],

task/summary_login.py

@@ -31,7 +31,7 @@ class SummaryLogin(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }
@@ -41,25 +41,25 @@ class SummaryLogin(Task):
                 bulk_data = []
                 # 当前活跃role id
                 role_list = self.local_db[source_coll].distinct('_game_role_id', where)
-                # 当日已记录的活跃role
+                date = int(pd.Timestamp(ts['cursor_st'], unit='s', tz=self.timezone) \
-                cdate = int(pd.Timestamp(ts['cursor_st'], unit='s', tz=self.timezone) \
                            .normalize().timestamp())
-                exists_role_list = self.local_db[self.dest_coll].distinct('_game_role_id', {'cdate': cdate})
+                exists_role_list = self.local_db[self.dest_coll].distinct('_game_role_id', {'cdate': date})
 
                 # 还没有记录的
                 role_set = set(role_list) - set(exists_role_list)
                 for item in self.local_db[source_coll].find({'_game_role_id': {'$in': list(role_set)}}, projection):
                     try:
-                        item['cdate'] = cdate
+                        item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \
+                                            .normalize().timestamp())
                         model = self.Model(**item)
                         data = model.dict(by_alias=True)
                         bulk_data.append(
-                            UpdateOne({'cdate': cdate, '_game_role_id': data['_game_role_id']}, {'$set': data},
+                            UpdateOne({'cdate': item['cdate'], '_game_role_id': data['_game_role_id']}, {'$set': data},
                                       upsert=True))
                     except ValidationError as e:
                         del item['_id']
                         bulk_data.append(
-                            UpdateOne({'cdate': cdate, '_game_role_id': item['_game_role_id']}, {'$set': item},
+                            UpdateOne({'cdate': item['cdate'], '_game_role_id': item['_game_role_id']}, {'$set': item},
                                       upsert=True))
                         msg = traceback.format_exc()
                         ddsend_msg(f'{self.game_name}.{source_coll}字段验证异常 还是写入了 {msg}\n{e.json()}')

task/summary_open_hd.py

@@ -29,7 +29,7 @@ class SummaryOpenHd(Task):
                     '_event_name': 'Activity',
                     'act': 'click',
                     'htype': {"$exists": 1},
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/summary_pay.py

@@ -39,7 +39,7 @@ class SummaryPay(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/summary_shopbuy.py

@@ -21,7 +21,7 @@ class SummaryShopbuy(Task):
         prize: List[dict] = Field(None, title='奖励')
         need: List[dict] = Field(None, title='消耗')
         user_name: str = Field(None, title='昵称')
-        stype: IntStr = Field(None, title='商店id')
+        stype: int = Field(None, title='商店id')
         needa: str = Field(...)
         needn: int = Field(...)
         needt: str = Field(...)
@@ -35,16 +35,28 @@ class SummaryShopbuy(Task):
                 if ts['cursor_st'] == ts['cursor_et']:
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
+                projection = self.Model.get_fields()
+
+                if self.game_name in ['shenghuajiyuan']:
+                    where = {
+                        "_event_name": "shopBuy",
+                        '_event_time': {
+                            '$gte': ts['cursor_st'],
+                            '$lt': ts['cursor_et'],
+                        }
+                    }
+                    projection.append('shopType')
+
+                else:
                     where = {
                         '_event_name': 'Shop',
                         "act": "buy",
-                    '_ut': {
+                        '_event_time': {
                             '$gte': ts['cursor_st'],
                             '$lt': ts['cursor_et'],
                         }
                     }
 
-                projection = self.Model.get_fields()
                 bulk_data = []
                 for item in self.local_db[source_coll].find(where, projection):
                     try:
@@ -56,6 +68,7 @@ class SummaryShopbuy(Task):
                         item['prizea'] = item['prize'][0]['a']
                         item['prizet'] = item['prize'][0]['t']
                         item['prizen'] = item['prize'][0]['n']
+                        item['stype'] = item.get('stype') or item.get('shopType', '255')
                         model = self.Model(**item)
                         data = model.dict(by_alias=True)
                         bulk_data.append(UpdateOne({'_id': data['_id']}, {'$set': data}, upsert=True))

task/sync_user.py

@@ -17,7 +17,7 @@ class SyncUser(Task):
                     continue
                 logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
                 where = {
-                    '_ut': {
+                    '_event_time': {
                         '$gte': ts['cursor_st'],
                         '$lt': ts['cursor_et'],
                     }

task/task.py

@@ -75,10 +75,11 @@ class Task(metaclass=abc.ABCMeta):
             self.cursor_st = self.task_info.get('cursor_et')
 
         if not self.cursor_st:
-            self.cursor_st = int(pd.Timestamp(time.time(), unit='s', tz=self.timezone).normalize().timestamp())
+            self.cursor_st = int(
+                pd.Timestamp(time.time(), unit='s', tz=self.timezone).normalize().timestamp()) + settings.cursor_diff
 
         if not self.cursor_et:
-            self.cursor_et = int(time.time())
+            self.cursor_et = int(time.time()) + settings.cursor_diff
 
     def set_cursor(self, **kwargs):
         """

test.py

@@ -1 +1 @@
-print('test')
+print('test  master')