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

954 lines
39 KiB
Python
Raw Permalink Normal View History

2022-12-14 14:56:58 +02:00
from datetime import datetime
2023-01-06 22:27:47 +02:00
from asyncio import sleep
2023-01-10 14:41:08 +02:00
from ftfy import fix_text
from traceback import format_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
2023-03-09 17:25:06 +02:00
from pyrogram.types import (
User,
ChatMember,
ChatPrivileges,
Chat,
Message,
Photo,
Video,
Document,
Animation,
Voice,
ForceReply,
ReplyKeyboardMarkup,
)
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
2022-12-28 19:56:13 +02:00
from classes.errors.geo import PlaceNotFoundError
2023-03-09 17:25:06 +02:00
from classes.errors.holo_user import (
UserInvalidError,
UserNotFoundError,
LabelTooLongError,
LabelSettingError,
)
from classes.templates import DefaultApplicationTemp, DefaultSponsorshipTemp
2023-03-09 17:25:06 +02:00
from modules.database import (
col_tmp,
col_users,
col_applications,
col_sponsorships,
col_messages,
col_spoilers,
)
2022-12-11 20:12:41 +02:00
from modules.logging import logWrite
from modules.utils import configGet, find_location, locale, should_quote
2022-12-10 13:08:30 +02:00
2023-03-09 17:25:06 +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.
2023-03-09 17:25:06 +02:00
"""
2022-12-11 02:31:17 +02:00
2022-12-15 15:30:21 +02:00
def __init__(self, user: Union[User, List[User], ChatMember, int]) -> None:
2022-12-13 15:45:10 +02:00
"""A user of Holo bot. Used to simplify DB interaction.
### Args:
2022-12-15 15:30:21 +02:00
* user (`Union[User, List[User], ChatMember, int]`): Any possible way to identify the user. Pass `User` object, user's telegram ID, etc...
2022-12-13 15:45:10 +02:00
### Raises:
* UserInvalidError: Provided to `HoloUser` object is not supported
* UserNotFoundError: `HoloUser` could not find user with such an ID in database
2023-03-09 17:25:06 +02:00
"""
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-15 15:30:21 +02:00
# elif isinstance(user, str):
# try:
# get_users = async_to_sync(app.get_users)
# self.id = get_users(user).id # this line requires testing though
# 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-21 16:22:14 +02:00
if isinstance(user, User):
2023-03-09 17:25:06 +02:00
if (
(self.name != user.first_name)
and hasattr(user, "first_name")
and (user.first_name is not None)
):
2022-12-22 16:05:27 +02:00
self.set("name", user.first_name, db_key="tg_name")
2023-03-09 17:25:06 +02:00
if (
(self.phone != user.phone_number)
and hasattr(user, "phone")
and (user.phone_number is not None)
):
2022-12-22 16:05:27 +02:00
self.set("phone", user.phone_number, db_key="tg_phone")
2023-03-09 17:25:06 +02:00
if (
(self.locale != user.language_code)
and hasattr(user, "locale")
and (user.language_code is not None)
):
2022-12-22 16:05:27 +02:00
self.set("locale", user.language_code, db_key="tg_locale")
2022-12-21 16:22:14 +02:00
2023-03-09 17:25:06 +02:00
if (
(self.username != user.username)
and hasattr(user, "username")
and (user.username is not None)
):
2022-12-22 16:05:27 +02:00
self.set("username", user.username, db_key="tg_username")
2023-03-09 17:25:06 +02:00
2022-12-22 16:05:27 +02:00
def set(self, key: str, value: Any, db_key: Union[str, None] = None) -> None:
2022-12-12 00:31:36 +02:00
"""Set attribute data and save it into database
### Args:
* `key` (`str`): Attribute to be changed
* `value` (`Any`): Value to set
2023-03-09 17:25:06 +02:00
"""
2022-12-12 00:31:36 +02:00
if not hasattr(self, key):
raise AttributeError()
setattr(self, key, value)
2022-12-22 16:05:27 +02:00
db_key = key if db_key is None else db_key
2023-03-09 17:25:06 +02:00
col_users.update_one(
filter={"_id": self.db_id}, update={"$set": {db_key: value}}, upsert=True
)
2022-12-12 00:31:36 +02:00
logWrite(f"Set attribute {key} of user {self.id} to {value}")
2023-03-09 17:25:06 +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,
2023-03-09 17:25:06 +02:00
adm_context: bool = False,
2022-12-14 15:05:09 +02:00
) -> None:
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)
2022-12-14 15:57:39 +02:00
* 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.
2023-03-09 17:25:06 +02:00
"""
2022-12-13 15:24:31 +02:00
2023-01-10 14:41:08 +02:00
if text is not None:
text = fix_text(text)
elif caption is not None:
caption = fix_text(caption)
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:
2023-03-09 17:25:06 +02:00
logWrite(
f"{context.from_user.id} sent message '{text}' to {self.id} (source message: {context.id})"
)
2022-12-13 15:24:31 +02:00
elif caption is not None:
2023-03-09 17:25:06 +02:00
logWrite(
f"{context.from_user.id} sent message '{caption}' to {self.id} (source message: {context.id})"
)
2022-12-13 15:24:31 +02:00
else:
2023-03-09 17:25:06 +02:00
logWrite(
f"{context.from_user.id} sent message to {self.id} (source message: {context.id})"
)
2022-12-13 15:24:31 +02:00
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:
2023-03-09 17:25:06 +02:00
text = (
locale("message_from", "message").format(
context.from_user.first_name, context.from_user.id
)
+ text
+ locale("message_reply_notice", "message")
)
2022-12-13 15:24:31 +02:00
else:
text = locale("message_reply_notice", "message")
if caption is not None:
if adm_context:
caption += locale("message_reply_notice", "message")
elif adm_origin:
2023-03-09 17:25:06 +02:00
caption = (
locale("message_from", "message").format(
context.from_user.first_name, context.from_user.id
)
+ caption
+ locale("message_reply_notice", "message")
)
2022-12-13 15:24:31 +02:00
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
2023-03-09 17:25:06 +02:00
new_message = await origin.reply_cached_media(
photo, caption=caption, quote=True
)
2022-12-13 15:33:22 +02:00
elif video is not None:
if isinstance(video, Video):
video = video.file_id
2023-03-09 17:25:06 +02:00
new_message = await origin.reply_cached_media(
video, caption=caption, quote=True
)
2022-12-13 15:33:22 +02:00
elif file is not None:
if isinstance(file, Document):
file = file.file_id
2023-03-09 17:25:06 +02:00
new_message = await origin.reply_cached_media(
file, caption=caption, quote=True
)
2022-12-13 15:33:22 +02:00
elif animation is not None:
if isinstance(animation, Animation):
animation = animation.file_id
2023-03-09 17:25:06 +02:00
new_message = await origin.reply_cached_media(
animation, caption=caption, quote=True
)
2022-12-13 15:33:22 +02:00
elif voice is not None:
if isinstance(voice, Voice):
voice = voice.file_id
2023-03-09 17:25:06 +02:00
new_message = await origin.reply_voice(
voice, caption=caption, quote=True
)
2022-12-13 15:33:22 +02:00
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
2023-03-09 17:25:06 +02:00
new_message = await app.send_cached_media(
self.id, photo, caption=caption
)
2022-12-13 15:33:22 +02:00
elif video is not None:
if isinstance(video, Video):
video = video.file_id
2023-03-09 17:25:06 +02:00
new_message = await app.send_cached_media(
self.id, video, caption=caption
)
2022-12-13 15:33:22 +02:00
elif file is not None:
if isinstance(file, Document):
file = file.file_id
2023-03-09 17:25:06 +02:00
new_message = await app.send_cached_media(
self.id, file, caption=caption
)
2022-12-13 15:33:22 +02:00
elif animation is not None:
if isinstance(animation, Animation):
animation = animation.file_id
2023-03-09 17:25:06 +02:00
new_message = await app.send_cached_media(
self.id, animation, caption=caption
)
2022-12-13 15:33:22 +02:00
elif voice is not None:
if isinstance(voice, Voice):
voice = voice.file_id
2023-03-09 17:25:06 +02:00
new_message = await app.send_cached_media(
self.id, voice, caption=caption
)
2022-12-13 15:33:22 +02:00
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
2023-03-09 17:25:06 +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:33:22 +02:00
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:
2023-03-09 17:25:06 +02:00
logWrite(
f"Exception {exp} happened as {context.from_user.id} tried to send message to {self.id}. Traceback:\n{format_exc()}"
)
2022-12-13 15:50:21 +02:00
try:
2023-03-09 17:25:06 +02:00
await app.send_message(
configGet("owner"),
locale("message_traceback", "message").format(
context.from_user.id, self.id, exp, format_exc()
),
)
2022-12-13 15:50:21 +02:00
except bad_request_400.PeerIdInvalid:
2023-03-09 17:25:06 +02:00
logWrite(
f"Could not notify admin about failure when sending message! Admin has never interacted with bot!"
)
await context.reply_text(
locale("message_error", "message"), quote=should_quote(context)
)
2022-12-11 02:31:17 +02:00
async def label_set(self, chat: Chat, label: str) -> None:
2022-12-11 02:31:17 +02:00
"""Set label in destination group
### Args:
2022-12-14 15:19:20 +02:00
* chat (`Chat`): Telegram chat
2022-12-11 02:31:17 +02:00
* label (`str`): Label you want to set
2023-03-09 17:25:06 +02:00
"""
if len(label) > 16:
raise LabelTooLongError(label)
2022-12-11 02:31:17 +02:00
self.label = label
2023-01-02 15:19:35 +02:00
try:
2023-03-09 17:25:06 +02:00
await app.promote_chat_member(
configGet("users", "groups"),
self.id,
privileges=ChatPrivileges(
can_pin_messages=True, can_manage_video_chats=True
),
)
2023-01-02 15:19:35 +02:00
if not await isAnAdmin(self.id):
2023-01-06 22:27:47 +02:00
await sleep(0.5)
2023-03-09 17:25:06 +02:00
await app.set_administrator_title(
configGet("users", "groups"), self.id, label
)
2023-01-02 15:19:35 +02:00
self.set("label", label)
except Exception as exp:
logWrite(f"Could not set {self.id}'s title to '{self.label}' due to {exp}")
raise LabelSettingError(exp, format_exc())
2022-12-11 02:31:17 +02:00
async def label_reset(self, chat: Chat) -> None:
2022-12-11 02:31:17 +02:00
"""Reset label in destination group
### Args:
2022-12-14 15:19:20 +02:00
* chat (`Chat`): Telegram chat
2023-03-09 17:25:06 +02:00
"""
2022-12-11 02:31:17 +02:00
self.label = ""
2022-12-12 00:31:36 +02:00
self.set("label", "")
2023-01-04 20:59:09 +02:00
await app.set_administrator_title(configGet("users", "groups"), self.id, "")
2022-12-15 15:30:21 +02:00
if not await isAnAdmin(self.id):
2023-03-09 17:25:06 +02:00
await app.promote_chat_member(
configGet("users", "groups"),
self.id,
privileges=ChatPrivileges(
can_manage_chat=False,
can_pin_messages=False,
can_manage_video_chats=False,
),
)
def application_state(
self,
) -> tuple[Literal["none", "fill", "approved", "rejected"], bool]:
2022-12-14 15:05:09 +02:00
"""Check the current state of application in tmp collection
### Returns:
* `tuple[Literal["none", "fill", "approved", "rejected"], bool]`: First element is an enum of a state and the second one is whether application is complete.
2023-03-09 17:25:06 +02:00
"""
2022-12-14 14:56:58 +02:00
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"]
2022-12-16 12:27:56 +02:00
def application_approved(self) -> bool:
"""Check whether user has a completed application and it got approved
### Returns:
* `bool`: `True` if yes and `False` if no
2023-03-09 17:25:06 +02:00
"""
return (
True if col_applications.find_one({"user": self.id}) is not None else False
)
2022-12-16 12:27:56 +02:00
2022-12-23 13:54:51 +02:00
def application_restart(self, reapply: bool = False) -> None:
2023-03-09 17:25:06 +02:00
"""Reset application of a user in tmp collection and replace it with an empty one"""
2022-12-14 14:56:58 +02:00
if col_tmp.find_one({"user": self.id, "type": "application"}) is None:
2023-03-09 17:25:06 +02:00
col_tmp.insert_one(
document=DefaultApplicationTemp(self.id, reapply=reapply).dict
)
2023-01-11 13:28:58 +02:00
else:
2023-03-09 17:25:06 +02:00
col_tmp.find_one_and_replace(
{"user": self.id, "type": "application"},
DefaultApplicationTemp(self.id, reapply=reapply).dict,
)
2022-12-14 14:56:58 +02:00
2022-12-14 15:05:09 +02:00
async def application_next(self, query: str, msg: Message) -> None:
"""Move on filling application of user
### Args:
* query (`str`): Some kind of input
* msg (`Message`): Message that should receive replies
2023-03-09 17:25:06 +02:00
"""
2022-12-14 14:56:58 +02:00
2023-01-03 14:01:46 +02:00
# if col_tmp.find_one({"user": self.id, "type": "application"}) is None:
2022-12-14 14:56:58 +02:00
2023-01-03 14:01:46 +02:00
if self.sponsorship_state()[0] == "fill":
return
2023-01-02 15:19:35 +02:00
2023-01-04 21:13:29 +02:00
if self.spoiler_state() is True:
return
2023-01-03 14:01:46 +02:00
# col_tmp.insert_one(
# document=DefaultApplicationTemp(self.id).dict
# )
2022-12-14 14:56:58 +02:00
progress = col_tmp.find_one({"user": self.id, "type": "application"})
2022-12-28 19:56:13 +02:00
if progress is None:
return
2022-12-14 14:56:58 +02:00
stage = progress["stage"]
2023-03-09 17:25:06 +02:00
2023-01-03 14:01:46 +02:00
# if self.sponsorship_state()[0] == "fill":
# await msg.reply_text(locale("finish_sponsorship", "message"), quote=should_quote(msg))
# return
2022-12-14 14:56:58 +02:00
2022-12-23 13:54:51 +02:00
if progress["state"] == "fill" and progress["sent"] is False:
2023-01-10 14:41:08 +02:00
if msg.text is not None:
msg.text = fix_text(str(msg.text))
2023-01-10 14:41:08 +02:00
2022-12-14 14:56:58 +02:00
if stage == 2:
try:
input_dt = datetime.strptime(query, "%d.%m.%Y")
except ValueError:
2023-03-09 17:25:06 +02:00
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", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-14 14:56:58 +02:00
return
2023-03-09 17:25:06 +02:00
if (datetime.now() <= input_dt) or (
(datetime.now() - input_dt).days
) > (
(
datetime.now()
- datetime.now().replace(
year=datetime.now().year - configGet("age_maximum")
)
).days
):
logWrite(
f"User {msg.from_user.id} failed stage {stage} due to joking"
)
await msg.reply_text(
locale("question2_joke", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale("question2", "force_reply", locale=self.locale)
)
),
)
2022-12-14 14:56:58 +02:00
return
2023-03-09 17:25:06 +02:00
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", locale=self.locale
).format(str(configGet("age_allowed"))),
reply_markup=ForceReply(
placeholder=str(
locale("question2", "force_reply", locale=self.locale)
)
),
)
2022-12-14 14:56:58 +02:00
return
else:
progress["application"][str(stage)] = input_dt
2023-03-09 17:25:06 +02:00
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", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage+1}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-14 14:56:58 +02:00
elif stage == 3:
try:
2022-12-28 19:56:13 +02:00
progress["application"][str(stage)] = find_location(query)
2023-03-09 17:25:06 +02:00
if (
"lat" in progress["application"][str(stage)]
and "lng" in progress["application"][str(stage)]
):
progress["application"][str(stage)]["location"] = [
float(progress["application"][str(stage)]["lng"]),
float(progress["application"][str(stage)]["lat"]),
]
2022-12-28 19:56:13 +02:00
del progress["application"][str(stage)]["lat"]
del progress["application"][str(stage)]["lng"]
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "application"}},
{
"$set": {
"application": progress["application"],
"stage": progress["stage"] + 1,
}
},
)
await msg.reply_text(
locale("question3_found", "message", locale=self.locale).format(
progress["application"][str(stage)]["name"],
progress["application"][str(stage)]["adminName1"],
)
)
await msg.reply_text(
locale(f"question{stage+1}", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage+1}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-28 19:56:13 +02:00
except PlaceNotFoundError:
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("question3_invalid", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-14 14:56:58 +02:00
return
except Exception as exp:
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("question3_error", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-14 14:56:58 +02:00
try:
2023-03-09 17:25:06 +02:00
await app.send_message(
configGet("owner"),
locale(
"question3_traceback", "message", locale=self.locale
).format(query, exp, format_exc()),
)
2022-12-14 14:56:58 +02:00
except bad_request_400.PeerIdInvalid:
2023-03-09 17:25:06 +02:00
logWrite(
f"Could not notify admin about failure when sending message! Admin has never interacted with bot!"
)
2022-12-14 14:56:58 +02:00
return
elif stage == 10:
2023-01-10 14:41:08 +02:00
if len(query) > 1024:
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("question10_too_long", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2023-01-10 14:41:08 +02:00
return
2022-12-14 14:56:58 +02:00
progress["application"][str(stage)] = query
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "application"}},
{
"$set": {
"application": progress["application"],
"complete": True,
}
},
)
2022-12-14 14:56:58 +02:00
application_content = []
i = 1
for question in progress["application"]:
if i == 2:
2023-03-09 17:25:06 +02:00
age = relativedelta(
datetime.now(), progress["application"]["2"]
)
application_content.append(
f"{locale('question'+str(i), 'message', 'question_titles', locale=self.locale)} {progress['application']['2'].strftime('%d.%m.%Y')} ({age.years} р.)"
)
2022-12-14 14:56:58 +02:00
elif i == 3:
2023-03-09 17:25:06 +02:00
if progress["application"]["3"]["countryCode"] == "UA":
application_content.append(
f"{locale('question'+str(i), 'message', 'question_titles', locale=self.locale)} {progress['application']['3']['name']} ({progress['application']['3']['adminName1']})"
)
2022-12-14 14:56:58 +02:00
else:
2023-03-09 17:25:06 +02:00
application_content.append(
f"{locale('question'+str(i), 'message', 'question_titles', locale=self.locale)} {progress['application']['3']['name']} ({progress['application']['3']['adminName1']}, {progress['application']['3']['countryName']})"
)
2022-12-14 14:56:58 +02:00
else:
2023-03-09 17:25:06 +02:00
application_content.append(
f"{locale('question'+str(i), 'message', 'question_titles', locale=self.locale)} {progress['application'][question]}"
)
2022-12-14 14:56:58 +02:00
i += 1
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("confirm", "message", locale=self.locale).format(
"\n".join(application_content)
),
reply_markup=ReplyKeyboardMarkup(
locale("confirm", "keyboard", locale=self.locale),
resize_keyboard=True,
),
)
2022-12-14 14:56:58 +02:00
else:
2023-01-10 14:41:08 +02:00
if len(query) > 256:
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("question_too_long", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2023-01-10 14:41:08 +02:00
return
2022-12-14 14:56:58 +02:00
progress["application"][str(stage)] = query
2023-03-09 17:25:06 +02:00
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", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"question{stage+1}", "force_reply", locale=self.locale
)
)
),
)
2022-12-14 14:56:58 +02:00
2022-12-21 16:22:14 +02:00
logWrite(f"User {self.id} completed stage {stage} of application")
else:
2022-12-22 16:05:27 +02:00
return
2023-03-09 17:25:06 +02:00
def sponsorship_state(
self,
) -> tuple[Literal["none", "fill", "approved", "rejected"], bool]:
2022-12-22 16:05:27 +02:00
"""Check the current state of sponsorship in tmp collection
### Returns:
* `tuple[Literal["none", "fill", "approved", "rejected"], bool]`: First element is an enum of a state and the second one is whether sponsorship application is complete.
2023-03-09 17:25:06 +02:00
"""
2022-12-22 16:05:27 +02:00
tmp_sponsorship = col_tmp.find_one({"user": self.id, "type": "sponsorship"})
if tmp_sponsorship is None:
return "none", False
else:
return tmp_sponsorship["state"], tmp_sponsorship["complete"]
def sponsorship_valid(self) -> bool:
"""Check whether user has a valid sponsorship
### Returns:
* `bool`: `True` if yes and `False` if no
2023-03-09 17:25:06 +02:00
"""
return (
True
if col_sponsorships.find_one(
{"user": self.id, "expires": {"$gt": datetime.now()}}
)
is not None
else False
)
2022-12-22 16:05:27 +02:00
def sponsorship_restart(self) -> None:
2023-03-09 17:25:06 +02:00
"""Reset sponsorship of a user in tmp collection and replace it with an empty one"""
2022-12-22 16:05:27 +02:00
if col_tmp.find_one({"user": self.id, "type": "sponsorship"}) is None:
col_tmp.insert_one(document=DefaultSponsorshipTemp(self.id).dict)
else:
col_tmp.delete_one({"user": self.id, "type": "sponsorship"})
col_tmp.insert_one(document=DefaultSponsorshipTemp(self.id).dict)
2023-03-09 17:25:06 +02:00
async def sponsorship_next(
self, query: str, msg: Message, photo: Union[Photo, None] = None
) -> None:
2022-12-22 16:05:27 +02:00
"""Move on filling sponsorship of user
### Args:
* query (`str`): Some kind of input
* msg (`Message`): Message that should receive replies
2023-03-09 17:25:06 +02:00
"""
2022-12-22 16:05:27 +02:00
progress = col_tmp.find_one({"user": self.id, "type": "sponsorship"})
2022-12-22 16:05:27 +02:00
if progress is not None:
2022-12-22 16:05:27 +02:00
stage = progress["stage"]
2023-01-10 14:41:08 +02:00
if msg.text is not None:
msg.text = fix_text(str(msg.text))
2023-01-10 14:41:08 +02:00
elif msg.caption is not None:
msg.caption = fix_text(msg.caption)
2022-12-23 13:54:51 +02:00
if progress["state"] == "fill" and progress["sent"] is False:
2022-12-22 16:05:27 +02:00
if stage == 1:
2023-01-10 14:41:08 +02:00
if len(query) > 240:
2023-03-09 17:25:06 +02:00
logWrite(
f"User {msg.from_user.id} failed stage {stage} due to sending invalid date format"
)
await msg.reply_text(
locale(f"sponsor1_invalid", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2023-01-10 14:41:08 +02:00
return
2022-12-22 16:05:27 +02:00
progress["sponsorship"]["streamer"] = query
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "sponsorship"}},
{
"$set": {
"sponsorship": progress["sponsorship"],
"stage": progress["stage"] + 1,
}
},
)
await msg.reply_text(
locale(f"sponsor{stage+1}", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor{stage+1}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-22 16:05:27 +02:00
elif stage == 2:
try:
input_dt = datetime.strptime(query, "%d.%m.%Y")
except ValueError:
2023-03-09 17:25:06 +02:00
logWrite(
f"User {msg.from_user.id} failed stage {stage} due to sending invalid date format"
)
await msg.reply_text(
locale(f"sponsor2_invalid", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor{stage}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-22 16:05:27 +02:00
return
if datetime.now() >= input_dt:
2023-03-09 17:25:06 +02:00
logWrite(
f"User {msg.from_user.id} failed stage {stage} due to sending date in the past"
)
await msg.reply_text(
locale("sponsor2_past", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
"sponsor2", "force_reply", locale=self.locale
)
)
),
)
2022-12-22 16:05:27 +02:00
return
else:
progress["sponsorship"]["expires"] = input_dt
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "sponsorship"}},
{
"$set": {
"sponsorship": progress["sponsorship"],
"stage": progress["stage"] + 1,
}
},
)
await msg.reply_text(
locale(f"sponsor{stage+1}", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor{stage+1}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-22 16:05:27 +02:00
elif stage == 3:
if photo is not None:
progress["sponsorship"]["proof"] = photo.file_id
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "sponsorship"}},
{
"$set": {
"sponsorship": progress["sponsorship"],
"stage": progress["stage"] + 1,
}
},
)
await msg.reply_text(
locale(f"sponsor{stage+1}", "message", locale=self.locale),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor{stage+1}",
"force_reply",
locale=self.locale,
)
)
),
)
2022-12-22 16:05:27 +02:00
elif stage == 4:
2022-12-22 23:11:15 +02:00
if len(query) > 16:
2023-03-09 17:25:06 +02:00
await msg.reply_text(
locale("label_too_long", "message"),
reply_markup=ForceReply(
placeholder=str(
locale(
"sponsor4", "force_reply", locale=self.locale
)
)
),
)
2022-12-22 23:11:15 +02:00
return
2022-12-22 16:05:27 +02:00
progress["sponsorship"]["label"] = query
2023-03-09 17:25:06 +02:00
col_tmp.update_one(
{"user": {"$eq": self.id}, "type": {"$eq": "sponsorship"}},
{
"$set": {
"sponsorship": progress["sponsorship"],
"complete": True,
}
},
)
await msg.reply_cached_media(
progress["sponsorship"]["proof"],
2023-03-09 17:25:06 +02:00
caption=locale(
"sponsor_confirm", "message", locale=self.locale
).format(
2023-01-03 14:01:46 +02:00
progress["sponsorship"]["streamer"],
progress["sponsorship"]["expires"].strftime("%d.%m.%Y"),
2023-03-09 17:25:06 +02:00
progress["sponsorship"]["label"],
),
reply_markup=ReplyKeyboardMarkup(
locale("confirm", "keyboard", locale=self.locale),
resize_keyboard=True,
2023-01-03 14:01:46 +02:00
),
2023-03-09 17:25:06 +02:00
)
2022-12-22 16:05:27 +02:00
else:
return
logWrite(f"User {self.id} completed stage {stage} of sponsorship")
else:
2023-01-04 20:58:54 +02:00
return
def spoiler_state(self) -> bool:
"""Check if user has any started but not finished spoilers
### Returns:
* `bool`: `True` if any not finished spoilers available and `False` if none.
2023-03-09 17:25:06 +02:00
"""
return (
False
if col_spoilers.find_one({"user": self.id, "completed": False}) is None
else True
)