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

375 lines
19 KiB
Python
Raw Normal View History

2022-12-14 14:56:58 +02:00
from datetime import datetime
from requests import get
2022-12-13 15:33:22 +02:00
from traceback import print_exc
2022-12-12 00:31:36 +02:00
from app import app, isAnAdmin
2022-12-14 14:56:58 +02:00
from typing import Any, List, Literal, Union
from pyrogram.types import User, ChatMember, ChatPrivileges, Chat, Message, Photo, Video, Document, Animation, Voice, ForceReply, ReplyKeyboardMarkup
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
2022-12-14 14:56:58 +02:00
from dateutil.relativedelta import relativedelta
from modules.database import col_tmp, 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-13 15:45:10 +02:00
"""This object represents a user of HoloChecker bot.
It is primarily used to interact with a database in a more python-friendly way,
as well as provide better programming experience in case of adding new features, etc.
"""
2022-12-11 02:31:17 +02:00
2022-12-13 15:45:10 +02:00
def __init__(self, user: Union[User, List[User], ChatMember, int, str]) -> None:
"""A user of Holo bot. Used to simplify DB interaction.
### Args:
* user (`Union[User, List[User], ChatMember, int, str]`): Any possible way to identify the user. Pass `User` object, user's telegram ID, username or something...
### Raises:
* UserInvalidError: Provided to `HoloUser` object is not supported
* UserNotFoundError: `HoloUser` could not find user with such an ID in database
"""
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:
2022-12-12 00:32:04 +02:00
self.id = (app.get_users(user)).id # this line requires testing though
2022-12-12 00:31:36 +02:00
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}")
2022-12-13 15:24:31 +02:00
async def message(self,
2022-12-13 15:33:22 +02:00
context: Message,
2022-12-13 15:24:31 +02:00
origin: Union[Message, None] = None,
text: Union[str, None] = None,
caption: Union[str, None] = None,
photo: Union[str, Photo, None] = None,
video: Union[str, Video, None] = None,
file: Union[str, Document, None] = None,
animation: Union[str, Animation, None] = None,
voice: Union[str, Voice, None] = None,
adm_origin: bool = False,
adm_context: bool = False
2022-12-13 15:38:38 +02:00
):
"""Send a message to user
### Args:
* context (`Message`): Context (mostly the message where this method is called)
* origin (`Union[Message, None]`, optional): Origin message where to refer. None if called in a command. Defaults to None.
* text (`Union[str, None]`, optional): Text if this is a simple text message. Defaults to None.
* caption (`Union[str, None]`, optional): Text if this is a media message. Defaults to None.
* photo (`Union[str, Photo, None]`, optional): Photo as a photo object or file_id as a string. Defaults to None.
* video (`Union[str, Video, None]`, optional): Video as a video object or file_id as a string. Defaults to None.
* file (`Union[str, Document, None]`, optional): File as a document object or file_id as a string. Defaults to None.
* animation (`Union[str, Animation, None]`, optional): Animation as an animation object or file_id as a string. Defaults to None.
* voice (`Union[str, Voice, None]`, optional): Voice as a voice object or file_id as a string. Defaults to None.
2022-12-13 15:45:10 +02:00
* adm_origin (`bool`, *optional*): Whether origin sender is an admin. Defaults to False.
* adm_context (`bool`, *optional*): Whether context sender is an admin. Defaults to False.
2022-12-13 15:38:38 +02:00
"""
2022-12-13 15:24:31 +02:00
2022-12-13 15:50:21 +02:00
# Check if any text available and log message sending
2022-12-13 15:24:31 +02:00
if text is not None:
logWrite(f"{context.from_user.id} sent message '{text}' to {self.id}")
elif caption is not None:
logWrite(f"{context.from_user.id} sent message '{caption}' to {self.id}")
else:
logWrite(f"{context.from_user.id} sent message to {self.id}")
2022-12-13 15:50:21 +02:00
# Add notices for admin or user
2022-12-13 15:24:31 +02:00
if text is not None:
if adm_context:
text += locale("message_reply_notice", "message")
elif adm_origin:
text = locale("message_from", "message").format(context.from_user.first_name, context.from_user.id) + text + locale("message_reply_notice", "message")
else:
text = locale("message_reply_notice", "message")
if caption is not None:
if adm_context:
caption += locale("message_reply_notice", "message")
elif adm_origin:
caption = locale("message_from", "message").format(context.from_user.first_name, context.from_user.id) + caption + locale("message_reply_notice", "message")
else:
caption = locale("message_reply_notice", "message")
2022-12-13 15:50:21 +02:00
# Try sending the message
2022-12-13 15:33:22 +02:00
try:
2022-12-13 15:50:21 +02:00
# Check if origin message exists
# This check decides whether we send_ a message or reply_ to one
2022-12-13 15:33:22 +02:00
if origin is not None:
if photo is not None:
if isinstance(photo, Photo):
photo = photo.file_id
new_message = await origin.reply_photo(photo, caption=caption, quote=True)
elif video is not None:
if isinstance(video, Video):
video = video.file_id
new_message = await origin.reply_video(video, caption=caption, quote=True)
elif file is not None:
if isinstance(file, Document):
file = file.file_id
new_message = await origin.reply_document(file, caption=caption, quote=True)
elif animation is not None:
if isinstance(animation, Animation):
animation = animation.file_id
new_message = await origin.reply_animation(animation, caption=caption, quote=True)
elif voice is not None:
if isinstance(voice, Voice):
voice = voice.file_id
new_message = await origin.reply_voice(voice, caption=caption, quote=True)
else:
new_message = await origin.reply_text(text, quote=True)
2022-12-13 15:24:31 +02:00
else:
2022-12-13 15:33:22 +02:00
if photo is not None:
if isinstance(photo, Photo):
photo = photo.file_id
new_message = await app.send_photo(self.id, photo, caption=caption)
elif video is not None:
if isinstance(video, Video):
video = video.file_id
new_message = await app.send_video(self.id, video, caption=caption)
elif file is not None:
if isinstance(file, Document):
file = file.file_id
new_message = await app.send_document(self.id, file, caption=caption)
elif animation is not None:
if isinstance(animation, Animation):
animation = animation.file_id
new_message = await app.send_animation(animation, caption=caption, quote=True)
elif voice is not None:
if isinstance(voice, Voice):
voice = voice.file_id
new_message = await app.send_voice(voice, caption=caption, quote=True)
else:
new_message = await app.send_message(self.id, text)
2022-12-13 15:50:21 +02:00
# Acknowledge sending a message and save entry into DB
2022-12-13 15:33:22 +02:00
await context.reply_text(locale("message_sent", "message"), quote=should_quote(context))
col_messages.insert_one({"origin": {"chat": context.chat.id, "id": context.id}, "destination": {"chat": new_message.chat.id, "id": new_message.id}})
2022-12-13 15:50:21 +02:00
# Report to admin and to sender about message sending failure
2022-12-13 15:33:22 +02:00
except Exception as exp:
2022-12-13 15:50:21 +02:00
logWrite(f"Exception {exp} happened as {context.from_user.id} tried to send message to {self.id}. Traceback:\n{print_exc()}")
try:
await app.send_message(configGet("owner"), locale("message_traceback", "message").format(context.from_user.id, self.id, exp, print_exc()))
except bad_request_400.PeerIdInvalid:
logWrite(f"Could not notify admin about failure when sending message! Admin has never interacted with bot!")
2022-12-13 15:33:22 +02:00
await context.reply_text(locale("message_error", "message"), quote=should_quote(context))
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-14 14:56:58 +02:00
))
def application_state(self) -> tuple[Literal["none", "fill", "approved", "rejected"], bool]:
tmp_application = col_tmp.find_one({"user": self.id, "type": "application"})
if tmp_application is None:
return "none", False
else:
return tmp_application["state"], tmp_application["complete"]
def application_restart(self):
doc = {
"user": self.id,
"type": "application",
"complete": False,
"sent": False,
"state": "fill",
"reapply": False,
"stage": 1,
"application": {
"1": None,
"2": None,
"3": None,
"4": None,
"5": None,
"6": None,
"7": None,
"8": None,
"9": None,
"10": None
}
}
if col_tmp.find_one({"user": self.id, "type": "application"}) is None:
col_tmp.insert_one(document=doc)
else:
col_tmp.delete_one({"user": self.id, "type": "application"})
col_tmp.insert_one(document=doc)
async def application_next(self, query: str, msg: Union[Message, None] = None):
if col_tmp.find_one({"user": self.id, "type": "application"}) is None:
col_tmp.insert_one(
document={
"user": self.id,
"type": "application",
"complete": False,
"sent": False,
"state": "fill",
"reapply": False,
"stage": 1,
"application": {
"1": None,
"2": None,
"3": None,
"4": None,
"5": None,
"6": None,
"7": None,
"8": None,
"9": None,
"10": None
}
}
)
progress = col_tmp.find_one({"user": self.id, "type": "application"})
stage = progress["stage"]
if progress["state"] == "fill":
if stage == 2:
try:
input_dt = datetime.strptime(query, "%d.%m.%Y")
except ValueError:
logWrite(f"User {msg.from_user.id} failed stage {stage} due to sending invalid date format")
await msg.reply_text(locale(f"question2_invalid", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage}", "force_reply"))))
return
if datetime.now() <= input_dt:
logWrite(f"User {msg.from_user.id} failed stage {stage} due to joking")
await msg.reply_text(locale("question2_joke", "message"), reply_markup=ForceReply(placeholder=str(locale("question2", "force_reply"))))
return
elif ((datetime.now() - input_dt).days) < ((datetime.now() - datetime.now().replace(year=datetime.now().year - configGet("age_allowed"))).days):
logWrite(f"User {msg.from_user.id} failed stage {stage} due to being underage")
await msg.reply_text(locale("question2_underage", "message").format(str(configGet("age_allowed"))), reply_markup=ForceReply(placeholder=str(locale("question2", "force_reply"))))
return
else:
print(f'Look: {((datetime.now() - input_dt).days)} > {(datetime.now() - datetime.now().replace(year=datetime.now().year - configGet("age_allowed"))).days}')
progress["application"][str(stage)] = input_dt
col_tmp.update_one({"user": {"$eq": self.id}, "type": {"$eq": "application"}}, {"$set": {"application": progress["application"], "stage": progress["stage"]+1}})
await msg.reply_text(locale(f"question{stage+1}", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage+1}", "force_reply"))))
elif stage == 3:
try:
result = (get(f"http://api.geonames.org/searchJSON?q={query}&maxRows=1&countryBias=UA&lang=uk&orderby=relevance&featureClass=P&featureClass=A&username={configGet('username', 'geocoding')}")).json()
progress["application"][str(stage)] = result["geonames"][0]
col_tmp.update_one({"user": {"$eq": self.id}, "type": {"$eq": "application"}}, {"$set": {"application": progress["application"], "stage": progress["stage"]+1}})
await msg.reply_text(locale(f"question3_found", "message").format(result["geonames"][0]["name"], result["geonames"][0]["adminName1"]))
await msg.reply_text(locale(f"question{stage+1}", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage+1}", "force_reply"))))
except (ValueError, KeyError, IndexError):
await msg.reply_text(locale(f"question3_invalid", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage}", "force_reply"))))
return
except Exception as exp:
await msg.reply_text(locale("question3_error", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage}", "force_reply"))))
try:
await app.send_message(configGet("owner"), locale("question3_traceback", "message").format(query, exp, print_exc()))
except bad_request_400.PeerIdInvalid:
logWrite(f"Could not notify admin about failure when sending message! Admin has never interacted with bot!")
return
elif stage == 10:
progress["application"][str(stage)] = query
col_tmp.update_one({"user": {"$eq": self.id}, "type": {"$eq": "application"}}, {"$set": {"application": progress["application"], "complete": True}})
application_content = []
i = 1
for question in progress["application"]:
if i == 2:
age = relativedelta(datetime.now(), progress['application']['2'])
application_content.append(f"{locale('question'+str(i), 'message', 'question_titles')} {progress['application']['2'].strftime('%d.%m.%Y')} ({age.years} р.)")
elif i == 3:
if progress['application']['3']['countryCode'] == "UA":
application_content.append(f"{locale('question'+str(i), 'message', 'question_titles')} {progress['application']['3']['name']} ({progress['application']['3']['adminName1']})")
else:
application_content.append(f"{locale('question'+str(i), 'message', 'question_titles')} {progress['application']['3']['name']} ({progress['application']['3']['adminName1']}, {progress['application']['3']['countryName']})")
else:
application_content.append(f"{locale('question'+str(i), 'message', 'question_titles')} {progress['application'][question]}")
i += 1
await msg.reply_text(locale("confirm", "message").format("\n".join(application_content)), reply_markup=ReplyKeyboardMarkup(locale("confirm", "keyboard"), resize_keyboard=True))
else:
progress["application"][str(stage)] = query
col_tmp.update_one({"user": {"$eq": self.id}, "type": {"$eq": "application"}}, {"$set": {"application": progress["application"], "stage": progress["stage"]+1}})
await msg.reply_text(locale(f"question{stage+1}", "message"), reply_markup=ForceReply(placeholder=str(locale(f"question{stage+1}", "force_reply"))))
logWrite(f"User {self.id} completed stage {stage} of application")