wuhao/data_cleaning
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Compare commits

base: wuhao:master
Branches Tags
wuhao:master
wuhao:ck
..
compare: wuhao:ck
Branches Tags
wuhao:master
wuhao:ck

No commits in common. "master" and "ck" have entirely different histories.
master ... ck

17 changed files with 33 additions and 160 deletions
Show Stats Expand all files Collapse all files

5
config.json
View File

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

25
del_coll.py
View File

@ -1,25 +0,0 @@
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}')

5
settings.py
View File

@ -3,7 +3,6 @@ import sys
class Config: class Config:
cursor_diff = -(3 * 60)
local_mongo_uri = 'mongodb://root:iamciniao@127.0.0.1:27017/?authSource=admin&readPreference=primary&ssl=false' 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' 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 +20,8 @@ class Debug(Config):
run_model = 'debug' 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@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@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' # 中古 # local_mongo_uri = 'mongodb://root:iamciniao@119.29.35.35:27017/?authSource=admin&readPreference=primary&ssl=false' # 中古

2
task/first_recharge.py
View File

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

76
task/login_repair.py
View File

@ -1,76 +0,0 @@
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'])

8
task/summary3.py
View File

@ -260,18 +260,12 @@ class Summary3(Task):
self.local_db[f'summary_{cat}'].update_one({'_id': data['_id']}, {'$set': data}) 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}) 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(): for id_, doc in update_doc.items():
pipeline = [ pipeline = [
{ {
'$match': { '$match': {
"cdate": {'$gte': doc['cdate'], '$lte': cdate}, "cdate": {'$gte': doc['cdate'], '$lte': cdate},
type_map[cat]: {'$in': doc[f'now_pay_{cat}_list']}, f'_game_{cat}_id': {'$in': doc[f'now_pay_{cat}_list']},
} }
}, },
{'$group': {'_id': None, 'sum_money': {'$sum': '$money'}}} {'$group': {'_id': None, 'sum_money': {'$sum': '$money'}}}

2
task/summary_assets.py
View File

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

2
task/summary_func.py
View File

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

2
task/summary_funnel.py
View File

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

2
task/summary_join_hd.py
View File

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

16
task/summary_login.py
View File

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

2
task/summary_open_hd.py
View File

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

2
task/summary_pay.py
View File

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

35
task/summary_shopbuy.py
View File

@ -21,7 +21,7 @@ class SummaryShopbuy(Task):
prize: List[dict] = Field(None, title='奖励') prize: List[dict] = Field(None, title='奖励')
need: List[dict] = Field(None, title='消耗') need: List[dict] = Field(None, title='消耗')
user_name: str = Field(None, title='昵称') user_name: str = Field(None, title='昵称')
stype: int = Field(None, title='商店id') stype: IntStr = Field(None, title='商店id')
needa: str = Field(...) needa: str = Field(...)
needn: int = Field(...) needn: int = Field(...)
needt: str = Field(...) needt: str = Field(...)
@ -35,40 +35,27 @@ class SummaryShopbuy(Task):
if ts['cursor_st'] == ts['cursor_et']: if ts['cursor_st'] == ts['cursor_et']:
continue continue
logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}') logger.info(f'开始处理{self.game_name} 处理 {source_coll} 游标 {ts}')
where = {
'_event_name': 'Shop',
"act": "buy",
'_ut': {
'$gte': ts['cursor_st'],
'$lt': ts['cursor_et'],
}
}
projection = self.Model.get_fields() 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",
'_event_time': {
'$gte': ts['cursor_st'],
'$lt': ts['cursor_et'],
}
}
bulk_data = [] bulk_data = []
for item in self.local_db[source_coll].find(where, projection): for item in self.local_db[source_coll].find(where, projection):
try: try:
item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \ item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \
.normalize().timestamp()) .normalize().timestamp())
item['needa'] = item['need'][0]['a'] item['needa'] = item['need'][0]['a']
item['needt'] = item['need'][0]['t'] item['needt'] = item['need'][0]['t']
item['needn'] = item['need'][0]['n'] item['needn'] = item['need'][0]['n']
item['prizea'] = item['prize'][0]['a'] item['prizea'] = item['prize'][0]['a']
item['prizet'] = item['prize'][0]['t'] item['prizet'] = item['prize'][0]['t']
item['prizen'] = item['prize'][0]['n'] item['prizen'] = item['prize'][0]['n']
item['stype'] = item.get('stype') or item.get('shopType', '255')
model = self.Model(**item) model = self.Model(**item)
data = model.dict(by_alias=True) data = model.dict(by_alias=True)
bulk_data.append(UpdateOne({'_id': data['_id']}, {'$set': data}, upsert=True)) bulk_data.append(UpdateOne({'_id': data['_id']}, {'$set': data}, upsert=True))

2
task/sync_user.py
View File

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

5
task/task.py
View File

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

2
test.py
View File

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