From edc4919c8da0330dae6925dccfa0b2c5354c5c61 Mon Sep 17 00:00:00 2001 From: kf_wuhao <15392746632@qq.com> Date: Sun, 7 Feb 2021 10:29:23 +0800 Subject: [PATCH] update --- config.json | 6 ++++ settings.py | 6 ++-- task/login_repair.py | 73 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 3 deletions(-) create mode 100644 task/login_repair.py diff --git a/config.json b/config.json index 2ff19bd..ee72397 100644 --- a/config.json +++ b/config.json @@ -76,5 +76,11 @@ "source_coll": "event", "dest_coll": "summary_assets", "task_name": "summary_assets" + }, + "login_repair": { + "source_coll": "event", + "dest_coll": "summary_login", + "task_name": "login_repair", + "freq": "D" } } \ No newline at end of file diff --git a/settings.py b/settings.py index 28f0605..fe146ed 100644 --- a/settings.py +++ b/settings.py @@ -3,7 +3,7 @@ import sys class Config: - cursor_diff = -(3 * 60 * 60) + 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' @@ -21,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@119.29.176.224: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.35.35:27017/?authSource=admin&readPreference=primary&ssl=false' # 中古 diff --git a/task/login_repair.py b/task/login_repair.py new file mode 100644 index 0000000..6b40701 --- /dev/null +++ b/task/login_repair.py @@ -0,0 +1,73 @@ +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: + item = self.local_db[source_coll].find_one( + {'_game_role_id': role_id, '_event_name': 'TimeSpending'}, projection) + 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'])