95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import os
|
||
|
||
from cachetools import TTLCache
|
||
from requests import get
|
||
|
||
|
||
class Platform:
|
||
def __init__(self):
|
||
self.staff_cache = TTLCache(1000, 60)
|
||
self.exp_cache = TTLCache(1000, 60)
|
||
self.configs = TTLCache(1000, 60)
|
||
self.stage = os.getenv("STAGE", "local")
|
||
self.token = os.getenv("PLATFORM_SECURITY_TOKEN")
|
||
self.project = "РУЗ Бот"
|
||
|
||
def get_config(self, config_name):
|
||
config = self.configs.get(config_name)
|
||
if config is None:
|
||
config = get(
|
||
'https://platform.sprinthub.ru/configs/get',
|
||
headers={'X-Security-Token': self.token},
|
||
params={
|
||
'project': self.project,
|
||
'stage': self.stage,
|
||
'name': config_name
|
||
}
|
||
)
|
||
if config.status_code != 200:
|
||
return {}
|
||
config = config.json()
|
||
self.configs[config_name] = config
|
||
return config
|
||
|
||
def get_experiment(self, experiment_name):
|
||
exp = self.exp_cache.get(experiment_name)
|
||
if exp is None:
|
||
exp = get(
|
||
'https://platform.sprinthub.ru/experiments/get',
|
||
headers={'X-Security-Token': self.token},
|
||
params={
|
||
'project': self.project,
|
||
'stage': self.stage,
|
||
'name': experiment_name
|
||
}
|
||
)
|
||
if exp.status_code != 200:
|
||
return {
|
||
'enabled': False,
|
||
'condition': False
|
||
}
|
||
exp = exp.json()
|
||
self.exp_cache[experiment_name] = exp
|
||
return exp
|
||
|
||
def get_staff(self, telegram_id):
|
||
is_staff = self.staff_cache.get(telegram_id)
|
||
if is_staff is None:
|
||
exp = get(
|
||
'https://platform.sprinthub.ru/is_staff',
|
||
headers={'X-Security-Token': self.token},
|
||
params={
|
||
'telegram_id': telegram_id,
|
||
}
|
||
)
|
||
if exp.status_code != 200:
|
||
return False
|
||
data = exp.json()
|
||
is_staff = data['is_staff']
|
||
self.staff_cache[telegram_id] = is_staff
|
||
return is_staff
|
||
|
||
def experiment_enabled_for_user(self, experiment_name, telegram_id):
|
||
exp_data = self.get_experiment(experiment_name)
|
||
if not exp_data['enabled']:
|
||
return False
|
||
|
||
class User:
|
||
def __init__(self, platform, telegram_id):
|
||
self.platform = platform
|
||
self.telegram_id = telegram_id
|
||
|
||
@property
|
||
def platform_staff(self):
|
||
return self.platform.get_staff(self.telegram_id)
|
||
|
||
@property
|
||
def is_superuser(self):
|
||
return False
|
||
|
||
user = User(self, telegram_id)
|
||
return bool(eval(exp_data['condition']))
|
||
|
||
|
||
platform = Platform()
|