|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field, asdict
|
|
|
|
import typing
|
|
|
|
import time
|
|
|
|
|
|
|
|
import telebot
|
|
|
|
|
|
|
|
from community import Community
|
|
|
|
from telebot.types import ChatMember
|
|
|
|
|
|
|
|
if typing.TYPE_CHECKING:
|
|
|
|
from bot import Bot
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
|
class User:
|
|
|
|
user_id: int
|
|
|
|
chat_id: int = 0
|
|
|
|
locale: str = ''
|
|
|
|
start_timestamp: int = field(default_factory=lambda: int(time.time()))
|
|
|
|
last_action_timestamp: int = field(default_factory=lambda: int(time.time()))
|
|
|
|
# Store the chats which were checked
|
|
|
|
# (so that we send a message for communities
|
|
|
|
# where the user is a member of the chat but not a member of the RT community)
|
|
|
|
checked_chats: typing.List[int] = field(default_factory=list)
|
|
|
|
|
|
|
|
def check_chat(self, chat_id: int, bot: Bot) -> bool:
|
|
|
|
"""
|
|
|
|
Check if the user should be notified of this chat
|
|
|
|
"""
|
|
|
|
if chat_id in self.checked_chats:
|
|
|
|
return False
|
|
|
|
self.checked_chats.append(chat_id)
|
|
|
|
try:
|
|
|
|
chat_member_info: ChatMember = bot.get_chat_member(chat_id, self.user_id)
|
|
|
|
except telebot.apihelper.ApiTelegramException: # user not found
|
|
|
|
return False
|
|
|
|
return chat_member_info.is_member or chat_member_info.status in ['creator', 'administrator', 'admin', 'member']
|
|
|
|
|
|
|
|
def dict(self) -> dict:
|
|
|
|
data = asdict(self)
|
|
|
|
data['last_action_timestamp'] = int(time.time())
|
|
|
|
return data
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def from_dict(cls, data: dict) -> User:
|
|
|
|
data = getattr(data, '__dict__', data)
|
|
|
|
data_ = {key: data[key] for key in data.keys() if
|
|
|
|
key in ['user_id', 'chat_id', 'locale', 'start_timestamp', 'last_action_timestamp', 'checked_chats']}
|
|
|
|
self = cls(**data_)
|
|
|
|
return self
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def by_id(cls, user_id: int, bot) -> typing.Optional[User]:
|
|
|
|
data = bot.get_user_from_db({'user_id': user_id})
|
|
|
|
if data is None:
|
|
|
|
return None
|
|
|
|
return data
|
|
|
|
|
|
|
|
@property
|
|
|
|
def id(self):
|
|
|
|
return self._id
|
|
|
|
|
|
|
|
def is_admin(self) -> bool:
|
|
|
|
return self.user_id in [218952152]
|
|
|
|
|
|
|
|
def get_communities(self, bot: Bot) -> typing.Iterable[Community]:
|
|
|
|
"""
|
|
|
|
Returns a list of community ids for which the user is a member
|
|
|
|
|
|
|
|
Perform a MongoDB query to get the list of communities for which the user is a member
|
|
|
|
"""
|
|
|
|
for community in bot.db.communities.find({'members': self.user_id}):
|
|
|
|
yield Community.from_dict(community)
|