from traceback import print_exc from app import app, isAnAdmin import asyncio from ftfy import fix_text from pyrogram import filters from pyrogram.types import ( Message, ForceReply, InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardRemove, ) from pyrogram.client import Client from classes.holo_user import HoloUser from modules.utils import configGet, logWrite, locale, all_locales from modules.database import col_messages, col_spoilers from modules import custom_filters async def message_involved(msg: Message) -> bool: message = col_messages.find_one( { "destination.id": msg.reply_to_message.id, "destination.chat": msg.reply_to_message.chat.id, } ) if message is not None: return True return False async def message_context(msg: Message) -> tuple: message = col_messages.find_one( { "destination.id": msg.reply_to_message.id, "destination.chat": msg.reply_to_message.chat.id, } ) if message is not None: return message["origin"]["chat"], message["origin"]["id"] return 0, 0 # Any other input ============================================================================================================== @app.on_message( ~filters.scheduled & (filters.private | filters.chat(configGet("admin", "groups"))) & ~custom_filters.banned ) async def any_stage(app: Client, msg: Message): if msg.via_bot is None: holo_user = HoloUser(msg.from_user) if (msg.reply_to_message is not None) and (await message_involved(msg)): context = await message_context(msg) context_message = await app.get_messages(context[0], context[1]) destination_user = HoloUser(context_message.from_user) if destination_user is None: return await destination_user.message( origin=context_message, context=msg, text=str(msg.text), caption=msg.caption, photo=msg.photo, video=msg.video, file=msg.document, animation=msg.animation, voice=msg.voice, adm_origin=await isAnAdmin(context_message.from_user.id), adm_context=await isAnAdmin(msg.from_user.id), ) return if msg.chat.id == configGet("admin", "groups"): return if msg.text is not None: if configGet("enabled", "features", "applications") is True: await holo_user.application_next(str(msg.text), msg=msg) if configGet("enabled", "features", "sponsorships") is True: await holo_user.sponsorship_next(str(msg.text), msg) if msg.photo is not None: await holo_user.sponsorship_next(str(msg.text), msg=msg, photo=msg.photo) if ( holo_user.application_state()[0] != "fill" and holo_user.sponsorship_state()[0] != "fill" ): if configGet("enabled", "features", "spoilers") is False: return spoiler = col_spoilers.find_one( {"user": msg.from_user.id, "completed": False} ) if spoiler is None: return if spoiler["category"] is None: found = False # Find category in all locales for lc in all_locales("spoiler_categories", "message"): for key in lc: if lc[key] == str(msg.text): found = True category = key if found is False: await msg.reply_text( locale( "spoiler_incorrect_category", "message", locale=msg.from_user, ) ) return col_spoilers.find_one_and_update( {"_id": spoiler["_id"]}, {"$set": {"category": category}} ) await msg.reply_text( locale("spoiler_send_description", "message", locale=msg.from_user), reply_markup=ForceReply( placeholder=locale( "spoiler_description", "force_reply", locale=msg.from_user ) ), ) return if spoiler["description"] is None and ( spoiler["photo"] is None and spoiler["video"] is None and spoiler["audio"] is None and spoiler["animation"] is None and spoiler["text"] is None ): # for lc in all_locales("spoiler_description", "keyboard"): # if msg.text == lc[-1][0]: # await msg.reply_text(locale("spoiler_description_enter", "message", locale=msg.from_user), reply_markup=ForceReply(placeholder=locale("spoiler_description", "force_reply", locale=msg.from_user))) # return if str(msg.text) != "-": msg.text = fix_text(str(msg.text)) if len(str(msg.text)) > 1024: await msg.reply_text( locale( "spoiler_description_too_long", "message", locale=msg.from_user, ), reply_markup=ForceReply( placeholder=locale( "spoiler_description", "force_reply", locale=msg.from_user, ) ), ) return col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, {"$set": {"description": msg.text}}, ) else: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, {"$set": {"description": ""}}, ) logWrite( f"Adding description '{str(msg.text)}' to {msg.from_user.id}'s spoiler" ) await msg.reply_text( locale( "spoiler_using_description", "message", locale=msg.from_user ).format(msg.text), reply_markup=ForceReply( placeholder=locale( "spoiler_content", "force_reply", locale=msg.from_user ) ), ) return ready = False if msg.photo is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, { "$set": { "photo": msg.photo.file_id, "caption": msg.caption, "completed": True, } }, ) logWrite( f"Adding photo with id {msg.photo.file_id} to {msg.from_user.id}'s spoiler" ) ready = True if msg.video is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, { "$set": { "video": msg.video.file_id, "caption": msg.caption, "completed": True, } }, ) logWrite( f"Adding audio with id {msg.video.file_id} to {msg.from_user.id}'s spoiler" ) ready = True if msg.audio is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, { "$set": { "audio": msg.audio.file_id, "caption": msg.caption, "completed": True, } }, ) logWrite( f"Adding video with id {msg.audio.file_id} to {msg.from_user.id}'s spoiler" ) ready = True if msg.animation is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, { "$set": { "animation": msg.animation.file_id, "caption": msg.caption, "completed": True, } }, ) logWrite( f"Adding animation with id {msg.animation.file_id} to {msg.from_user.id}'s spoiler" ) ready = True if msg.document is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, { "$set": { "document": msg.document.file_id, "caption": msg.caption, "completed": True, } }, ) logWrite( f"Adding document with id {msg.document.file_id} to {msg.from_user.id}'s spoiler" ) ready = True if ( spoiler["photo"] is None and spoiler["video"] is None and spoiler["audio"] is None and spoiler["animation"] is None and spoiler["document"] is None and spoiler["text"] is None ): if msg.text is not None: col_spoilers.find_one_and_update( {"user": msg.from_user.id, "completed": False}, {"$set": {"text": str(msg.text), "completed": True}}, ) logWrite( f"Adding text '{str(msg.text)}' to {msg.from_user.id}'s spoiler" ) ready = True if ready is True: await msg.reply_text( locale("spoiler_ready", "message", locale=msg.from_user), reply_markup=ReplyKeyboardRemove(), ) if configGet("allow_external", "features", "spoilers") is True: await msg.reply_text( locale("spoiler_send", "message", locale=msg.from_user), reply_markup=InlineKeyboardMarkup( [ [ InlineKeyboardButton( locale( "spoiler_preview", "button", locale=msg.from_user, ), callback_data=f"sid_{spoiler['_id'].__str__()}", ) ], [ InlineKeyboardButton( locale( "spoiler_send_chat", "button", locale=msg.from_user, ), callback_data=f"shc_{spoiler['_id'].__str__()}", ) ], [ InlineKeyboardButton( locale( "spoiler_send_other", "button", locale=msg.from_user, ), switch_inline_query=f"spoiler:{spoiler['_id'].__str__()}", ) ], ] ), ) else: await msg.reply_text( locale("spoiler_send", "message", locale=msg.from_user), reply_markup=InlineKeyboardMarkup( [ [ InlineKeyboardButton( locale( "spoiler_preview", "button", locale=msg.from_user, ), callback_data=f"sid_{spoiler['_id'].__str__()}", ) ], [ InlineKeyboardButton( locale( "spoiler_send_chat", "button", locale=msg.from_user, ), callback_data=f"shc_{spoiler['_id'].__str__()}", ) ], ] ), ) else: await msg.reply_text( locale("spoiler_incorrect_content", "message", locale=msg.from_user) ) @app.on_message(~filters.scheduled & filters.group) async def message_in_group(app: Client, msg: Message): if (msg.chat is not None) and (msg.via_bot is not None): if (msg.via_bot.id == (await app.get_me()).id) and ( msg.chat.id == configGet("users", "groups") ): if str(msg.text).startswith( locale("spoiler_described", "message").split()[0] ) or str(msg.text).startswith( locale("spoiler_empty", "message").split()[0] ): logWrite(f"User {msg.from_user.id} sent spoiler to user's group") try: logWrite("Forwarding spoiler to admin's group") await msg.copy( configGet("admin", "groups"), disable_notification=True ) except Exception as exp: logWrite( f"Could not forward spoiler to admin's group due to '{exp}': {print_exc()}" ) return if configGet("remove_application_time") > 0: logWrite( f"User {msg.from_user.id} requested application in destination group, removing in {configGet('remove_application_time')} minutes" ) await asyncio.sleep(configGet("remove_application_time") * 60) await msg.delete() logWrite( f"Removed application requested by {msg.from_user.id} in destination group" ) # ==============================================================================================================================