This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
Telegram/classes/holo_user.py

105 lines
4.4 KiB
Python
Raw Normal View History

2022-12-12 00:31:36 +02:00
from app import app, isAnAdmin
2022-12-11 20:12:41 +02:00
from typing import Any, List, Union
2022-12-12 00:31:36 +02:00
from pyrogram.types import User, ChatMember, ChatPrivileges, Chat, Message
2022-12-11 02:31:17 +02:00
from pyrogram.client import Client
2022-12-12 00:31:36 +02:00
from pyrogram.errors import bad_request_400
from modules.database import col_users, col_context, col_warnings, col_applications, col_sponsorships, col_messages
2022-12-11 20:12:41 +02:00
from modules.logging import logWrite
2022-12-12 00:31:36 +02:00
from modules.utils import configGet, locale, should_quote
2022-12-10 13:08:30 +02:00
class UserNotFoundError(Exception):
"""HoloUser could not find user with such an ID in database"""
def __init__(self, user, user_id):
self.user = user
self.user_id = user_id
super().__init__(f"User of type {type(self.user)} with id {self.user_id} was not found")
2022-12-11 02:31:17 +02:00
class UserInvalidError(Exception):
"""Provided to HoloUser object is not supported"""
def __init__(self, user):
self.user = user
super().__init__(f"Could not find HoloUser by using {type(self.user)} as an input type")
2022-12-10 12:37:15 +02:00
class HoloUser():
2022-12-11 02:31:17 +02:00
2022-12-12 00:31:36 +02:00
def __init__(self, user: Union[List[User], User, ChatMember, int, str]) -> None:
2022-12-11 02:31:17 +02:00
# Determine input object class and extract id
if isinstance(user, list) and len(user) != 0:
self.id = user[0].id
elif isinstance(user, User):
self.id = user.id
elif isinstance(user, ChatMember):
self.id = user.user.id
elif isinstance(user, int):
self.id = user
2022-12-12 00:31:36 +02:00
elif isinstance(user, str):
try:
self.id = (app.get_users(user)).id
except bad_request_400.UsernameNotOccupied:
raise UserInvalidError(user)
except bad_request_400.PeerIdInvalid:
raise UserInvalidError(user)
2022-12-11 02:31:17 +02:00
else:
raise UserInvalidError(user)
# Find user record in DB
holo_user = col_users.find_one({"user": self.id})
if holo_user is None:
raise UserNotFoundError(user=user, user_id=self.id)
self.db_id = holo_user["_id"]
2022-12-11 20:12:41 +02:00
self.link = holo_user["link"]
2022-12-11 02:31:17 +02:00
self.label = holo_user["label"]
2022-12-11 20:12:41 +02:00
self.name = holo_user["tg_name"]
self.phone = holo_user["tg_phone"]
self.locale = holo_user["tg_locale"]
self.username = holo_user["tg_username"]
2022-12-12 00:31:36 +02:00
def set(self, key: str, value: Any) -> None:
"""Set attribute data and save it into database
### Args:
* `key` (`str`): Attribute to be changed
* `value` (`Any`): Value to set
"""
if not hasattr(self, key):
raise AttributeError()
setattr(self, key, value)
col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { key: value } }, upsert=True)
logWrite(f"Set attribute {key} of user {self.id} to {value}")
async def message(self, origin: Message, text: Union[str, None] = None, photo: Union[str, None] = None, video: Union[str, None] = None, file: Union[str, None] = None):
new_message = await app.send_message(self.id, text+locale("message_reply_notice", "message"))
await origin.reply_text(locale("message_sent", "message"), quote=should_quote(origin))
logWrite(f"Admin {origin.from_user.id} sent message '{' '.join(origin.command[2:])}' to {self.id}")
col_messages.insert_one({"origin": {"chat": origin.chat.id, "id": origin.id}, "destination": {"chat": new_message.chat.id, "id": new_message.id}})
2022-12-11 02:31:17 +02:00
2022-12-12 00:31:36 +02:00
async def set_label(self, chat: Chat, label: str):
2022-12-11 02:31:17 +02:00
"""Set label in destination group
### Args:
* app (`Client`): Pyrogram client
* label (`str`): Label you want to set
"""
self.label = label
2022-12-12 00:31:36 +02:00
self.set("label", label)
2022-12-11 02:31:17 +02:00
await app.promote_chat_member(configGet("destination_group"), self.id)
if (not await isAnAdmin(self.id)) and (chat.id == configGet("admin_group")):
await app.set_administrator_title(configGet("destination_group"), self.id, label)
2022-12-12 00:31:36 +02:00
async def reset_label(self, chat: Chat):
2022-12-11 02:31:17 +02:00
"""Reset label in destination group
### Args:
* app (`Client`): Pyrogram client
"""
self.label = ""
2022-12-12 00:31:36 +02:00
self.set("label", "")
2022-12-11 02:31:17 +02:00
await app.set_administrator_title(configGet("destination_group"), self.id, "")
if (not await isAnAdmin(self.id)) and (chat.id == configGet("admin_group")):
await app.promote_chat_member(configGet("destination_group"), self.id, privileges=ChatPrivileges(
can_manage_chat=False
2022-12-12 00:31:36 +02:00
))