MissKatyPyro/misskaty/plugins/ban_user_or_chat.py

from pyrogram import filters, Client
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message

from database.users_chats_db import db
from misskaty import app
from misskaty.helper.localization import use_chat_lang
from misskaty.vars import SUPPORT_CHAT, SUDO, LOG_CHANNEL


@app.on_message(filters.private & filters.incoming, group=2)
async def ban_reply(self: Client, ctx: Message):
    ban = await db.get_ban_status(ctx.from_user.id)
    if ban.get("is_banned"):
        await ctx.reply_msg(f'I am sorry, You are banned to use Me. \nBan Reason: {ban["ban_reason"]}')


@app.on_message(filters.group & filters.incoming, group=3)
@use_chat_lang()
async def grp_bd(self: Client, ctx: Message, strings):
    chck = await db.get_chat(ctx.chat.id)
    if not chck:
        total = await self.get_chat_members_count(ctx.chat.id)
        r_j = ctx.from_user.mention if ctx.from_user else "Anonymous"
        await self.send_message(
            LOG_CHANNEL,
            strings("log_bot_added").format(ttl=ctx.chat.title, cid=ctx.chat.id, tot=total, r_j=r_j),
        )
        await db.add_chat(ctx.chat.id, ctx.chat.title)
    if chck['is_disabled']:
        buttons = [[InlineKeyboardButton("Support", url=f"https://t.me/{SUPPORT_CHAT}")]]
        reply_markup = InlineKeyboardMarkup(buttons)
        vazha = await db.get_chat(ctx.chat.id)
        try:
            k = await ctx.reply_msg(
                f"CHAT NOT ALLOWED 🐞\n\nMy owner has restricted me from working here!\nReason : <code>{vazha['reason']}</code>.",
                reply_markup=reply_markup,
            )
            await k.pin()
        except:
            pass
        await self.leave_chat(ctx.chat.id)


@Client.on_message(filters.command('banuser') & filters.user(SUDO))
async def ban_a_user(bot, message):
    if len(message.command) == 1:
        return await message.reply('Give me a user id / username')
    r = message.text.split(None)
    if len(r) > 2:
        reason = message.text.split(None, 2)[2]
        chat = message.text.split(None, 2)[1]
    else:
        chat = message.command[1]
        reason = "No reason Provided"
    try:
        chat = int(chat)
    except:
        pass
    try:
        k = await bot.get_users(chat)
    except PeerIdInvalid:
        return await message.reply("This is an invalid user, make sure i have met him before.")
    except IndexError:
        return await message.reply("This might be a channel, make sure its a user.")
    except Exception as e:
        return await message.reply(f'Error - {e}')
    else:
        jar = await db.get_ban_status(k.id)
        if jar['is_banned']:
            return await message.reply(f"{k.mention} is already banned\nReason: {jar['ban_reason']}")
        await db.ban_user(k.id, reason)
        await message.reply(f"Successfully banned this {k.mention}!! Reason: jar['ban_reason']")


    
@Client.on_message(filters.command('unbanuser') & filters.user(SUDO))
async def unban_a_user(bot, message):
    if len(message.command) == 1:
        return await message.reply('Give me a user id / username')
    r = message.text.split(None)
    if len(r) > 2:
        reason = message.text.split(None, 2)[2]
        chat = message.text.split(None, 2)[1]
    else:
        chat = message.command[1]
        reason = "No reason Provided"
    try:
        chat = int(chat)
    except:
        pass
    try:
        k = await bot.get_users(chat)
    except PeerIdInvalid:
        return await message.reply("This is an invalid user, make sure ia have met him before.")
    except IndexError:
        return await message.reply("This might be a channel, make sure its a user.")
    except Exception as e:
        return await message.reply(f'Error - {e}')
    else:
        jar = await db.get_ban_status(k.id)
        if not jar['is_banned']:
            return await message.reply(f"{k.mention} is not yet banned.")
        await db.remove_ban(k.id)
        await message.reply(f"Successfully unbanned ! Sudhrr ja babu {k.mention}")


@app.on_message(filters.command("disablechat") & filters.user(SUDO))
async def disable_chat(bot, message):
    if len(message.command) == 1:
        return await message.reply("Give me a chat id")
    r = message.text.split(None)
    if len(r) > 2:
        reason = message.text.split(None, 2)[2]
        chat = message.text.split(None, 2)[1]
    else:
        chat = message.command[1]
        reason = "No reason Provided"
    try:
        chat_ = int(chat)
    except:
        return await message.reply("Give Me A Valid Chat ID")
    cha_t = await db.get_chat(chat_)
    if not cha_t:
        return await message.reply("Chat Not Found In DB")
    if cha_t["is_disabled"]:
        return await message.reply(f"This chat is already disabled:\nReason-<code> {cha_t['reason']} </code>")
    await db.disable_chat(chat_, reason)
    await message.reply("Chat Succesfully Disabled")
    try:
        buttons = [[InlineKeyboardButton("Support", url=f"https://t.me/{SUPPORT_CHAT}")]]
        reply_markup = InlineKeyboardMarkup(buttons)
        await bot.send_message(
            chat_id=chat_,
            text=f"<b>Hello Friends, \nMy admin has told me to leave from group so i go! If you wanna add me again contact my support group.</b> \nReason : <code>{reason}</code>",
            reply_markup=reply_markup,
        )
        await bot.leave_chat(chat_)
    except Exception as e:
        await message.reply(f"Error - {e}")


@app.on_message(filters.command("enablechat") & filters.user(SUDO))
async def re_enable_chat(bot: Client, ctx: Message):
    if len(ctx.command) == 1:
        return await ctx.reply("Give me a chat id")
    chat = ctx.command[1]
    try:
        chat_ = int(chat)
    except:
        return await ctx.reply("Give Me A Valid Chat ID")
    sts = await db.get_chat(int(chat))
    if not sts:
        return await ctx.reply("Chat Not Found In DB !")
    if not sts.get("is_disabled"):
        return await ctx.reply("This chat is not yet disabled.")
    await db.re_enable_chat(chat_)
    await ctx.reply("Chat Succesfully re-enabled")