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/modules/commands/sponsorship.py

326 lines
11 KiB
Python

from datetime import datetime, timedelta
from typing import Union
from app import app
from pyrogram import filters
from pyrogram.types import (
InlineKeyboardMarkup,
InlineKeyboardButton,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
ForceReply,
Message,
)
from pyrogram.client import Client
from classes.holo_user import HoloUser
from modules import custom_filters
from modules.handlers.confirmation import confirm_yes
from modules.utils import all_locales, locale, should_quote
from modules.database import col_sponsorships, col_tmp
from convopyro import listen_message
def is_none_or_cancel(message: Union[Message, None]) -> bool:
if (
message is None
or message.text is not None
and message.text.lower() == "/cancel"
):
return True
return False
@app.on_message(
custom_filters.enabled_sponsorships
& ~filters.scheduled
& filters.command(["sponsorship"], prefixes=["/"])
& ~custom_filters.banned
& (custom_filters.allowed | custom_filters.admin)
)
async def cmd_sponsorship(app: Client, msg: Message):
holo_user = HoloUser(msg.from_user)
if holo_user.application_state()[0] == "fill":
await msg.reply_text(
locale("finish_application", "message", locale=msg.from_user),
quote=should_quote(msg),
)
return
if holo_user.spoiler_state() is True:
await msg.reply_text(locale("spoiler_in_progress", "message", locale=holo_user))
return
existent = col_sponsorships.find_one(
{
"user": msg.from_user.id,
"sponsorship.expires": {"$gt": datetime.now() - timedelta(days=1)},
}
)
if existent is None:
await msg.reply_text(
locale("sponsorship_apply", "message", locale=msg.from_user),
reply_markup=InlineKeyboardMarkup(
[
[
InlineKeyboardButton(
text=str(
locale("sponsor_apply", "button", locale=msg.from_user)
),
callback_data=f"sponsor_apply_{msg.from_user.id}",
)
]
]
),
quote=should_quote(msg),
)
return
await msg.reply_text(
locale("sponsor_resubmit", "message", locale=msg.from_user).format(
existent["sponsorship"]["streamer"]
),
reply_markup=ReplyKeyboardMarkup(
locale("sponsorship_restore", "keyboard", locale=msg.from_user),
resize_keyboard=True,
one_time_keyboard=True,
),
)
answer_decision = await listen_message(app, msg.chat.id)
if is_none_or_cancel(answer_decision):
return
input_streamer = existent["sponsorship"]["streamer"]
values_keep = []
for pattern in all_locales("sponsorship_restore", "keyboard"):
values_keep.append(pattern[0][0].lower())
values_new = []
for pattern in all_locales("sponsorship_restore", "keyboard"):
values_new.append(pattern[1][0].lower())
if answer_decision.text.lower() in values_keep:
await answer_decision.reply_text(
locale("sponsor2", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor2",
"force_reply",
locale=msg.from_user,
)
)
),
)
while True:
answer_date = await listen_message(app, msg.chat.id)
if is_none_or_cancel(answer_date):
return
try:
input_dt = datetime.strptime(answer_date.text, "%d.%m.%Y")
if datetime.now() >= input_dt:
await msg.reply_text(
locale("sponsor2_past", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale("sponsor2", "force_reply", locale=msg.from_user)
)
),
)
continue
break
except ValueError:
await answer_date.reply_text(
locale(f"sponsor2_invalid", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor2",
"force_reply",
locale=msg.from_user,
)
)
),
)
continue
await answer_date.reply_text(
locale("sponsor3", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor3",
"force_reply",
locale=msg.from_user,
)
)
),
)
while True:
answer_proof = await listen_message(app, msg.chat.id)
if is_none_or_cancel(answer_proof):
return
if answer_proof.photo is None:
await answer_proof.reply_text(
locale("sponsor3", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor3",
"force_reply",
locale=msg.from_user,
)
)
),
)
continue
input_proof = answer_proof.photo.file_id
break
await msg.reply_text(
locale("sponsor4_resubmit", "message", locale=msg.from_user).format(
existent["sponsorship"]["label"]
),
reply_markup=ReplyKeyboardMarkup(
locale("sponsorship_restore_label", "keyboard", locale=msg.from_user),
resize_keyboard=True,
one_time_keyboard=True,
),
)
while True:
answer_label_decision = await listen_message(app, msg.chat.id)
if is_none_or_cancel(answer_label_decision):
return
if answer_label_decision.text is None:
await answer_label_decision.reply_text(
"Please, choose a valid option.",
reply_markup=ReplyKeyboardMarkup(
locale(
"sponsorship_restore_label",
"keyboard",
locale=msg.from_user,
),
resize_keyboard=True,
one_time_keyboard=True,
),
)
continue
values_keep = []
for pattern in all_locales("sponsorship_restore_label", "keyboard"):
values_keep.append(pattern[0][0].lower())
values_new = []
for pattern in all_locales("sponsorship_restore_label", "keyboard"):
values_new.append(pattern[1][0].lower())
if answer_label_decision.text.lower() in values_keep:
input_label = existent["sponsorship"]["label"]
elif answer_label_decision.text.lower() in values_new:
await answer_label_decision.reply_text(
locale("sponsor4", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor4",
"force_reply",
locale=msg.from_user,
)
)
),
)
while True:
answer_label = await listen_message(app, msg.chat.id)
if is_none_or_cancel(answer_label_decision):
return
if answer_label.text is None:
await answer_label.reply_text(
locale("label_too_long", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor4",
"force_reply",
locale=msg.from_user,
)
)
),
)
continue
elif len(answer_label.text) > 16:
await answer_label.reply_text(
locale("label_too_long", "message", locale=msg.from_user),
reply_markup=ForceReply(
placeholder=str(
locale(
f"sponsor4",
"force_reply",
locale=msg.from_user,
)
)
),
)
continue
input_label = answer_label.text
break
col_tmp.find_one_and_delete(
{"user": msg.from_user.id, "type": "sponsorship"}
)
col_tmp.insert_one(
{
"user": msg.from_user.id,
"type": "sponsorship",
"complete": True,
"sent": False,
"state": "fill",
"stage": 4,
"sponsorship": {
"streamer": input_streamer,
"expires": input_dt,
"proof": input_proof,
"label": input_label,
},
}
)
await confirm_yes(app, msg, kind="sponsorship")
return
elif answer_decision.text.lower() in values_new:
await msg.reply_text(
locale("sponsorship_apply", "message", locale=msg.from_user),
reply_markup=InlineKeyboardMarkup(
[
[
InlineKeyboardButton(
text=str(
locale("sponsor_apply", "button", locale=msg.from_user)
),
callback_data=f"sponsor_apply_{msg.from_user.id}",
)
]
]
),
quote=should_quote(msg),
)
return
else:
await answer_decision.reply_text(
locale("sponsor_resubmit_invalid_option", "message", locale=msg.from_user),
reply_markup=ReplyKeyboardRemove(),
)
return
# else:
# await msg.reply_text(locale("sponsorship_application_empty", "message"))