129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
from typing import Tuple
|
||
|
||
import sqlalchemy as sa
|
||
import json
|
||
|
||
from fastapi import Depends
|
||
|
||
import pandas as pd
|
||
|
||
from sqlalchemy import func, or_, and_, not_
|
||
|
||
import schemas
|
||
from core.config import settings
|
||
from db.redisdb import get_redis_pool, RedisDrive
|
||
|
||
|
||
class UserAnalysis:
|
||
def __init__(self, game: str, data_in: schemas.CkQuery, rdb: RedisDrive = Depends(get_redis_pool)):
|
||
self.game = game
|
||
self.rdb = rdb
|
||
self.user_tbl = None
|
||
self.event_view = data_in.eventView
|
||
self.events = data_in.events
|
||
|
||
self.zone_time: int = 0
|
||
|
||
self.global_filters = None
|
||
self.groupby = None
|
||
self.time_particle = None
|
||
self.date_range = None
|
||
self.unit_num = None
|
||
|
||
async def init(self):
|
||
await self._init_table()
|
||
self.zone_time = self._get_zone_time()
|
||
self.time_particle = self._get_time_particle_size()
|
||
self.groupby = self._get_group_by()
|
||
self.unit_num = self._get_unit_num()
|
||
|
||
async def _init_table(self):
|
||
"""
|
||
从redis中取出表字段，构建表结构
|
||
:return:
|
||
"""
|
||
res_json = await self.rdb.get(f'{self.game}_user')
|
||
columns = json.loads(res_json).keys()
|
||
metadata = sa.MetaData(schema=self.game)
|
||
self.user_tbl = sa.Table('user_view', metadata, *[sa.Column(column) for column in columns])
|
||
|
||
def _get_time_particle_size(self):
|
||
return self.event_view.get('timeParticleSize') or 'P1D'
|
||
|
||
def _get_unit_num(self):
|
||
return self.event_view.get('unitNum')
|
||
|
||
def _get_group_by(self):
|
||
return [getattr(self.user_tbl.c, item['columnName']) for item in self.event_view.get('groupBy', [])]
|
||
|
||
def _get_zone_time(self):
|
||
return int(self.event_view.get('zone_time', 8))
|
||
|
||
def _get_filters(self, filters):
|
||
tbl = self.user_tbl
|
||
where = []
|
||
for item in filters:
|
||
col = getattr(tbl.c, item['columnName'])
|
||
|
||
comparator = item['comparator']
|
||
ftv = item['ftv']
|
||
if comparator == '==':
|
||
if len(ftv) > 1:
|
||
where.append(or_(*[col == v for v in ftv]))
|
||
else:
|
||
where.append(col == ftv[0])
|
||
elif comparator == '>=':
|
||
where.append(col >= ftv[0])
|
||
elif comparator == '<=':
|
||
where.append(col <= ftv[0])
|
||
elif comparator == '>':
|
||
where.append(col > ftv[0])
|
||
elif comparator == '<':
|
||
where.append(col < ftv[0])
|
||
|
||
elif comparator == 'is not null':
|
||
where.append(col.isnot(None))
|
||
elif comparator == 'is null':
|
||
where.append(col.is_(None))
|
||
|
||
elif comparator == '!=':
|
||
where.append(col != ftv[0])
|
||
|
||
return where
|
||
|
||
def property_model(self):
|
||
event = self.events
|
||
selectd = getattr(self.user_tbl.c, event['quota'])
|
||
qry = sa.select(selectd)
|
||
|
||
account_id_col = getattr(self.user_tbl.c, '#account_id')
|
||
binduid_col = getattr(self.user_tbl.c, '#account_id')
|
||
# 聚合方式
|
||
analysis = event['analysis']
|
||
|
||
if analysis == 'trig_user_num':
|
||
selectd = [func.count().label('values')]
|
||
elif analysis == 'distinct_count':
|
||
selectd = [
|
||
func.count(sa.distinct(getattr(self.user_tbl.c, event['quota']))).label('values')]
|
||
|
||
else:
|
||
selectd = [
|
||
func.round(getattr(func, analysis)(getattr(self.user_tbl.c, event['quota'])), 2).label(
|
||
'values')]
|
||
|
||
where = self._get_filters(event['filts'])
|
||
qry = sa.select((*self.groupby, *selectd)).where(*where)
|
||
|
||
qry = qry.group_by(*self.groupby)
|
||
qry = qry.order_by(sa.Column('values').desc())
|
||
qry = qry.limit(1000)
|
||
sql = str(qry.compile(compile_kwargs={"literal_binds": True}))
|
||
print(sql)
|
||
result = {'sql': sql,
|
||
'groupby': [i.key for i in self.groupby],
|
||
'quota':event['quota']
|
||
}
|
||
|
||
return result
|