1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
from ..config import ConfigUnit
from typing import Optional, Union
from abc import ABC
from enum import Enum
class TelegramUserListType(Enum):
USERS = 'users'
NOTIFY = 'notify_users'
class TelegramUserIdsConfig(ConfigUnit):
NAME = 'telegram_user_ids'
@classmethod
def schema(cls) -> Optional[dict]:
return {
'roottype': 'dict',
'type': 'integer'
}
_user_ids_config = TelegramUserIdsConfig()
def _user_id_mapper(user: Union[str, int]) -> int:
if isinstance(user, int):
return user
return _user_ids_config[user]
class TelegramChatsConfig(ConfigUnit):
NAME = 'telegram_chats'
@classmethod
def schema(cls) -> Optional[dict]:
return {
'type': 'dict',
'schema': {
'id': {'type': 'string', 'required': True},
'token': {'type': 'string', 'required': True},
}
}
class TelegramBotConfig(ConfigUnit, ABC):
@classmethod
def schema(cls) -> Optional[dict]:
return {
'bot': {
'type': 'dict',
'schema': {
'token': {'type': 'string', 'required': True},
TelegramUserListType.USERS.value: {**TelegramBotConfig._userlist_schema(), 'required': True},
TelegramUserListType.NOTIFY.value: TelegramBotConfig._userlist_schema(),
}
}
}
@staticmethod
def _userlist_schema() -> dict:
return {'type': 'list', 'schema': {'type': ['string', 'integer']}}
@staticmethod
def custom_validator(data):
for ult in TelegramUserListType:
users = data['bot'][ult.value]
for user in users:
if isinstance(user, str):
if user not in _user_ids_config:
raise ValueError(f'user {user} not found in {TelegramUserIdsConfig.NAME}')
def get_user_ids(self,
ult: TelegramUserListType = TelegramUserListType.USERS) -> list[int]:
try:
return list(map(_user_id_mapper, self['bot'][ult.value]))
except KeyError:
return []
|
