51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
import pymongo
|
|
from pydantic import ValidationError
|
|
import pandas as pd
|
|
from model import *
|
|
|
|
__all__ = ('select_db', 'check_data')
|
|
|
|
|
|
def select_db(game, date, mongo_uri):
|
|
myclient = pymongo.MongoClient(mongo_uri)
|
|
factory = {
|
|
'user': {'coll': None, 'models': [TUserModel, UserModel]},
|
|
'paylist': {'coll': None, 'models': [TPaylistModel, PaylistModel]},
|
|
'event': {'coll': None, 'models': [TEventModel, EventModel]}
|
|
}
|
|
mydb = myclient[f"game_{game}"]
|
|
user_coll = mydb['user']
|
|
paylist_coll = mydb['paylist']
|
|
event_coll = mydb[f'event_{date}']
|
|
|
|
factory['user']['coll'] = user_coll
|
|
factory['paylist']['coll'] = paylist_coll
|
|
factory['event']['coll'] = event_coll
|
|
|
|
return factory
|
|
|
|
|
|
def check_data(name, where, factory):
|
|
conf = factory[name]
|
|
coll = conf['coll']
|
|
models = conf['models']
|
|
res = []
|
|
for item in coll.find(where):
|
|
tmp = {}
|
|
for m in models:
|
|
try:
|
|
m(**item)
|
|
except ValidationError as e:
|
|
for error in e.errors():
|
|
tmp[error['loc'][0]] = error['msg']
|
|
tmp['id'] = str(item['_id'])
|
|
res.append(tmp)
|
|
df = pd.DataFrame(res)
|
|
if df.shape == (0, 0):
|
|
return None
|
|
|
|
df.set_index('id', inplace=True)
|
|
df.drop_duplicates(inplace=True)
|
|
df.fillna('ok', inplace=True)
|
|
return df.to_html()
|