| | from pyrogram import Client, filters, enums |
| | from pyrogram.errors import ChatAdminRequired |
| | from pyrogram.types import Message, InlineKeyboardButton, InlineKeyboardMarkup, CallbackQuery |
| | from pyrogram.errors.exceptions.bad_request_400 import MessageTooLong, PeerIdInvalid, UserNotParticipant, MediaEmpty, PhotoInvalidDimensions, WebpageMediaEmpty |
| |
|
| | from info import ADMINS, LOG_CHANNEL, SUPPORT_CHAT, WELCOM_PIC, WELCOM_TEXT, IMDB_TEMPLATE |
| | from utils import get_size, temp, extract_user, get_file_id, get_poster, humanbytes |
| | from database.users_chats_db import db |
| | from database.ia_filterdb import Media |
| | from datetime import datetime |
| | from Script import script |
| | import logging, re, asyncio, time, shutil, psutil, os, sys |
| |
|
| | logger = logging.getLogger(__name__) |
| | logger.setLevel(logging.ERROR) |
| |
|
| | @Client.on_message(filters.command('leave') & filters.user(ADMINS)) |
| | async def leave_a_chat(bot, message): |
| | """ |
| | Command to make the bot leave a chat. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'leave' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Cʜᴀᴛ Iᴅ') |
| | chat = message.command[1] |
| | try: |
| | chat = int(chat) |
| | except: |
| | logger.info("Invalid chat ID provided.") |
| | chat = chat |
| | try: |
| | buttons = [[InlineKeyboardButton('Sᴜᴩᴩᴏʀᴛ', url=f'https://t.me/{SUPPORT_CHAT}')]] |
| | await bot.send_message(chat_id=chat, text='<b>Hᴇʟʟᴏ Fʀɪᴇɴᴅs, \nMʏ Aᴅᴍɪɴ Hᴀs Rᴇsᴛʀɪᴄᴛᴇᴅ Mᴇ Fʀᴏᴍ Wᴏʀᴋɪɴɢ Hᴇʀᴇ ! Iғ Yᴏᴜ Wᴀɴɴᴀ Aᴅᴅ Mᴇ Aɢᴀɪɴ Cᴏɴᴛᴀᴄᴛ Sᴜᴘᴘᴏʀᴛ</b>', reply_markup=InlineKeyboardMarkup(buttons)) |
| | logger.info(f"Bot left chat {chat} at user {message.from_user.id}'s request.") |
| | try: |
| | await k.pin() |
| | except: |
| | logger.info("Failed to pin message in chat {chat}.") |
| | pass |
| | return await bot.leave_chat(message.chat.id) |
| | except Exception as e: |
| | logger.error(f"Error leaving chat {chat}: {e}") |
| | return await message.reply(f'Eʀʀᴏʀ: {e}') |
| |
|
| | @Client.on_message(filters.command('disable') & filters.user(ADMINS)) |
| | async def disable_chat(bot, message): |
| | """ |
| | Command to disable a chat. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'disable' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Cʜᴀᴛ Iᴅ') |
| | 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: |
| | logger.info("Invalid chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Vᴀʟɪᴅ Cʜᴀᴛ ID') |
| | cha_t = await db.get_chat(int(chat_)) |
| | if not cha_t: |
| | logger.info(f"Chat {chat_} not found in DB.") |
| | return await message.reply("Cʜᴀᴛ Nᴏᴛ Fᴏᴜɴᴅ Iɴ DB") |
| | if cha_t['is_disabled']: |
| | logger.info(f"Chat {chat_} is already disabled.") |
| | return await message.reply(f"Tʜɪꜱ Cʜᴀᴛ Is Aʟʀᴇᴅʏ Dɪꜱᴀʙʟᴇᴅ:\nRᴇᴀꜱᴏɴ: <code> {cha_t['reason']} </code>") |
| | await db.disable_chat(int(chat_), reason) |
| | temp.BANNED_CHATS.append(int(chat_)) |
| | await message.reply('Cʜᴀᴛ Sᴜᴄᴄᴇꜱꜰᴜʟʟʏ Dɪꜱᴀʙʟᴇᴅ') |
| | logger.info(f"Chat {chat_} disabled successfully.") |
| | try: |
| | buttons = [[InlineKeyboardButton('Sᴜᴩᴩᴏʀᴛ', url=f'https://t.me/{SUPPORT_CHAT}')]] |
| | await bot.send_message(chat_id=chat_, text=f'<b>Hᴇʟʟᴏ Fʀɪᴇɴᴅs, \nᴍʏ Aᴅᴍɪɴ Hᴀs Tᴏʟᴅ Mᴇ Tᴏ Lᴇᴀᴠᴇ Fʀᴏᴍ Gʀᴏᴜᴘ Sᴏ I Gᴏ! Iғ Yᴏᴜ Wᴀɴɴᴀ Aᴅᴅ Mᴇ Aɢᴀɪɴ Cᴏɴᴛᴀᴄᴛ Mʏ Sᴜᴘᴘᴏʀᴛ Gʀᴏᴜᴘ.</b> \nRᴇᴀꜱᴏɴ : <code>{reason}</code>', reply_markup=InlineKeyboardMarkup(buttons)) |
| | await bot.leave_chat(chat_) |
| | logger.info(f"Bot left chat {chat_} after disabling.") |
| | except Exception as e: |
| | logger.error(f"Error leaving chat {chat_}: {e}") |
| | await message.reply(f"Eʀʀᴏʀ: {e}") |
| |
|
| | @Client.on_message(filters.command('enable') & filters.user(ADMINS)) |
| | async def re_enable_chat(bot, message): |
| | """ |
| | Command to re-enable a chat. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'enable' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Cʜᴀᴛ Iᴅ') |
| | chat = message.command[1] |
| | try: |
| | chat_ = int(chat) |
| | except: |
| | logger.info("Invalid chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Vᴀʟɪᴅ Cʜᴀᴛ ID') |
| | sts = await db.get_chat(int(chat)) |
| | if not sts: |
| | logger.info(f"Chat {chat_} not found in DB.") |
| | return await message.reply("Cʜᴀᴛ Nᴏᴛ Fᴏᴜɴᴅ Iɴ DB") |
| | if not sts.get('is_disabled'): |
| | logger.info(f"Chat {chat_} is not disabled.") |
| | return await message.reply('Tʜɪꜱ Cʜᴀᴛ Iꜱ Nᴏᴛ Yᴇᴛ Dɪꜱᴀʙʟᴇᴅ') |
| | await db.re_enable_chat(int(chat_)) |
| | temp.BANNED_CHATS.remove(int(chat_)) |
| | await message.reply("Cʜᴀᴛ Sᴜᴄᴄᴇꜱꜰᴜʟʟʏ Rᴇ-Eɴᴀʙʟᴇᴅ") |
| | logger.info(f"Chat {chat_} re-enabled successfully.") |
| |
|
| | @Client.on_message(filters.command('stats') & filters.incoming) |
| | async def get_ststs(bot, message): |
| | """ |
| | Command to get bot statistics. |
| | Accessible to both admins and users. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'stats' command.") |
| | rju = await message.reply('<b>Pʟᴇᴀꜱᴇ Wᴀɪᴛ...</b>') |
| | total_users = await db.total_users_count() |
| | totl_chats = await db.total_chat_count() |
| | files = await Media.count_documents() |
| | size = await db.get_db_size() |
| | free = 536870912 - size |
| | size = get_size(size) |
| | free = get_size(free) |
| | await rju.edit(script.STATUS_TXT.format(files, total_users, totl_chats, size, free)) |
| | logger.info(f"Bot statistics sent to user {message.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command('invite') & filters.user(ADMINS)) |
| | async def gen_invite(bot, message): |
| | """ |
| | Command to generate an invite link for a chat. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'invite' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Cʜᴀᴛ Iᴅ') |
| | chat = message.command[1] |
| | try: |
| | chat_ = int(chat) |
| | except: |
| | logger.info("Invalid chat ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Vᴀʟɪᴅ Cʜᴀᴛ ID') |
| | try: |
| | link = await bot.create_chat_invite_link(chat_) |
| | logger.info(f"Invite link generated for chat {chat_}.") |
| | except ChatAdminRequired: |
| | logger.error(f"Bot is not an admin in chat {chat_}.") |
| | return await message.reply("Iɴᴠɪᴛᴇ Lɪɴᴋ Gᴇɴᴇʀᴀᴛɪᴏɴ Fᴀɪʟᴇᴅ, Iᴀᴍ Nᴏᴛ Hᴀᴠɪɴɢ Sᴜғғɪᴄɪᴇɴᴛ Rɪɢʜᴛꜱ") |
| | except Exception as e: |
| | logger.error(f"Error generating invite link for chat {chat_}: {e}") |
| | return await message.reply(f'Eʀʀᴏʀ: {e}') |
| | await message.reply(f'Hᴇʀᴇ Iꜱ Yᴏᴜʀ Iɴᴠɪᴛᴇ Lɪɴᴋ: {link.invite_link}') |
| | logger.info(f"Invite link sent to user {message.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command('ban_user') & filters.user(ADMINS)) |
| | async def ban_a_user(bot, message): |
| | """ |
| | Command to ban a user. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'ban_user' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no user ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Uꜱᴇʀ Iᴅ / Uꜱᴇʀɴᴀᴍᴇ') |
| | 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) |
| | logger.info(f"User {chat} found: {k.mention}.") |
| | except PeerIdInvalid: |
| | logger.error(f"Peer ID invalid for user {chat}.") |
| | return await message.reply("Tʜɪs Is Aɴ Iɴᴠᴀʟɪᴅ Usᴇʀ, Mᴀᴋᴇ Sᴜʀᴇ Iᴀ Hᴀᴠᴇ Mᴇᴛ Hɪᴍ Bᴇғᴏʀᴇ") |
| | except IndexError: |
| | logger.error(f"Index error for user {chat}.") |
| | return await message.reply("Tʜɪs Mɪɢʜᴛ Bᴇ A Cʜᴀɴɴᴇʟ, Mᴀᴋᴇ Sᴜʀᴇ Iᴛs A Usᴇʀ.") |
| | except Exception as e: |
| | logger.error(f"Error getting user {chat}: {e}") |
| | return await message.reply(f'Eʀʀᴏʀ: {e}') |
| | else: |
| | jar = await db.get_ban_status(k.id) |
| | if jar['is_banned']: |
| | logger.info(f"User {k.id} is already banned.") |
| | return await message.reply(f"{k.mention} Iꜱ Aʟʀᴇᴅʏ Bᴀɴɴᴇᴅ\nRᴇᴀꜱᴏɴ: {jar['ban_reason']}") |
| | await db.ban_user(k.id, reason) |
| | temp.BANNED_USERS.append(k.id) |
| | await message.reply(f"Sᴜᴄᴄᴇꜱꜰᴜʟʟʏ Bᴀɴɴᴇᴅ {k.mention}") |
| | logger.info(f"User {k.id} banned successfully by user {message.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command('unban_user') & filters.user(ADMINS)) |
| | async def unban_a_user(bot, message): |
| | """ |
| | Command to unban a user. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'unban_user' command.") |
| | if len(message.command) == 1: |
| | logger.info("Command incomplete, no user ID provided.") |
| | return await message.reply('Gɪᴠᴇ Mᴇ A Uꜱᴇʀ Iᴅ / Uꜱᴇʀɴᴀᴍᴇ') |
| | 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) |
| | logger.info(f"User {chat} found: {k.mention}.") |
| | except PeerIdInvalid: |
| | logger.error(f"Peer ID invalid for user {chat}.") |
| | return await message.reply("Tʜɪs Is Aɴ Iɴᴠᴀʟɪᴅ Usᴇʀ, Mᴀᴋᴇ Sᴜʀᴇ Iᴀ Hᴀᴠᴇ Mᴇᴛ Hɪᴍ Bᴇғᴏʀᴇ") |
| | except IndexError: |
| | logger.error(f"Index error for user {chat}.") |
| | return await message.reply("Tʜɪs Mɪɢʜᴛ Bᴇ A Cʜᴀɴɴᴇʟ, Mᴀᴋᴇ Sᴜʀᴇ Iᴛs A Usᴇʀ.") |
| | except Exception as e: |
| | logger.error(f"Error getting user {chat}: {e}") |
| | return await message.reply(f'Eʀʀᴏʀ: {e}') |
| | else: |
| | jar = await db.get_ban_status(k.id) |
| | if not jar['is_banned']: |
| | logger.info(f"User {k.id} is not banned.") |
| | return await message.reply(f"{k.mention} Iꜱ Nᴏᴛ Yᴇᴛ Bᴀɴɴᴇᴅ") |
| | await db.remove_ban(k.id) |
| | temp.BANNED_USERS.remove(k.id) |
| | await message.reply(f"Sᴜᴄᴄᴇꜱꜰᴜʟʟʏ Uɴʙᴀɴɴᴇᴅ {k.mention}") |
| | logger.info(f"User {k.id} unbanned successfully by user {message.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command('users') & filters.user(ADMINS)) |
| | async def list_users(bot, message): |
| | """ |
| | Command to list all users. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'users' command.") |
| | sps = await message.reply('Gᴇᴛᴛɪɴɢ Lɪꜱᴛ Oꜰ Uꜱᴇʀꜱ') |
| | users = await db.get_all_users() |
| | out = "Uꜱᴇʀꜱ Sᴀᴠᴇᴅ Iɴ DB Aʀᴇ:\n\n" |
| | async for user in users: |
| | out += f"<a href=tg://user?id={user['id']}>{user['name']}</a>\n" |
| | try: |
| | await sps.edit_text(out) |
| | logger.info(f"User list sent to user {message.from_user.id}.") |
| | except MessageTooLong: |
| | with open('users.txt', 'w+') as outfile: |
| | outfile.write(out) |
| | await message.reply_document('users.txt', caption="Lɪꜱᴛ Oꜰ Uꜱᴇʀꜱ") |
| | logger.info(f"User list too long, sent as document to user {message.from_user.id}.") |
| | os.remove('users.txt') |
| |
|
| | @Client.on_message(filters.command('chats') & filters.user(ADMINS)) |
| | async def list_chats(bot, message): |
| | """ |
| | Command to list all chats. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'chats' command.") |
| | sps = await message.reply('Gᴇᴛᴛɪɴɢ Lɪꜱᴛ Oꜰ Cʜᴀᴛꜱ') |
| | chats = await db.get_all_chats() |
| | out = "Cʜᴀᴛꜱ Sᴀᴠᴇᴅ Iɴ DB Aʀᴇ:\n\n" |
| | async for chat in chats: |
| | username = chat['username'] |
| | username = "private" if not username else "@" + username |
| | out += f"**- Tɪᴛʟᴇ:** `{chat['title']}`\n**- ID:** `{chat['id']}`\n**Uꜱᴇʀɴᴀᴍᴇ:** {username}\n" |
| | try: |
| | await sps.edit_text(out) |
| | logger.info(f"Chat list sent to user {message.from_user.id}.") |
| | except MessageTooLong: |
| | with open('chats.txt', 'w+') as outfile: |
| | outfile.write(out) |
| | await message.reply_document('chats.txt', caption="Lɪꜱᴛ Oꜰ Cʜᴀᴛꜱ") |
| | logger.info(f"Chat list too long, sent as document to user {message.from_user.id}.") |
| | os.remove('chats.txt') |
| |
|
| | @Client.on_message(filters.command(["id"])) |
| | async def show_id(client, message): |
| | """ |
| | Command to show user/group ID. |
| | Accessible to both admins and users. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'id' command.") |
| | chat_type = message.chat.type |
| | if chat_type == enums.ChatType.PRIVATE: |
| | user_id = message.chat.id |
| | first = message.from_user.first_name |
| | last = message.from_user.last_name or "" |
| | username = message.from_user.username |
| | dc_id = message.from_user.dc_id or "" |
| | await message.reply_text(f"<b>➲ ꜰɪʀꜱᴛ ɴᴀᴍᴇ:</b> {first}\n<b>➲ ʟᴀꜱᴛ ɴᴀᴍᴇ:</b> {last}\n<b>➲ ᴜꜱᴇʀɴᴀᴍᴇ:</b> {username}\n<b>➲ ᴛᴇʟᴇɢʀᴀᴍ ɪᴅ:</b> <code>{user_id}</code>\n<b>➲ ᴅᴄ ɪᴅ:</b> <code>{dc_id}</code>", quote=True) |
| | logger.info(f"ID information sent to user {message.from_user.id}.") |
| | elif chat_type in [enums.ChatType.GROUP, enums.ChatType.SUPERGROUP]: |
| | _id = "" |
| | _id += f"<b>➲ ᴄʜᴀᴛ ɪᴅ</b>: <code>{message.chat.id}</code>\n" |
| | if message.reply_to_message: |
| | _id += ( |
| | "<b>➲ ᴜꜱᴇʀ ɪᴅ</b>: " |
| | f"<code>{message.from_user.id if message.from_user else 'Anonymous'}</code>\n" |
| | "<b>➲ ʀᴇᴩʟɪᴇᴅ ᴜꜱᴇʀ ɪᴅ</b>: " |
| | f"<code>{message.reply_to_message.from_user.id if message.reply_to_message.from_user else 'Anonymous'}</code>\n" |
| | ) |
| | file_info = get_file_id(message.reply_to_message) |
| | else: |
| | _id += ( |
| | "<b>➲ ᴜꜱᴇʀ ɪᴅ</b>: " |
| | f"<code>{message.from_user.id if message.from_user else 'Anonymous'}</code>\n" |
| | ) |
| | file_info = get_file_id(message) |
| | if file_info: |
| | _id += ( |
| | f"<b>{file_info.message_type}</b>: " |
| | f"<code>{file_info.file_id}</code>\n" |
| | ) |
| | await message.reply_text(_id, quote=True) |
| | logger.info(f"ID information sent to user {message.from_user.id} in chat {message.chat.id}.") |
| |
|
| | @Client.on_message(filters.command(["info"])) |
| | async def user_info(client, message): |
| | """ |
| | Command to get detailed user information. |
| | Accessible to both admins and users. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'info' command.") |
| | status_message = await message.reply_text("`ᴩʟᴇᴀꜱᴇ ᴡᴀɪᴛ....`") |
| | from_user = None |
| | from_user_id, _ = extract_user(message) |
| | try: |
| | from_user = await client.get_users(from_user_id) |
| | logger.info(f"User {from_user_id} found: {from_user.first_name}.") |
| | except Exception as error: |
| | logger.error(f"Error getting user {from_user_id}: {error}") |
| | return await status_message.edit(str(error)) |
| | if from_user is None: |
| | logger.info(f"User {from_user_id} not found.") |
| | return await status_message.edit("ɴᴏ ᴠᴀʟɪᴅ ᴜꜱᴇʀ_ɪᴅ / ᴍᴇꜱꜱᴀɢᴇ sᴘᴇᴄɪꜰɪᴇᴅ") |
| | message_out_str = "" |
| | message_out_str += f"<b>➲ꜰɪʀꜱᴛ ɴᴀᴍᴇ:</b> {from_user.first_name}\n" |
| | last_name = from_user.last_name or "<b>ɴᴏɴᴇ</b>" |
| | message_out_str += f"<b>➲ʟᴀꜱᴛ ɴᴀᴍᴇ:</b> {last_name}\n" |
| | message_out_str += f"<b>➲ᴛɢ-ɪᴅ:</b> <code>{from_user.id}</code>\n" |
| | username = from_user.username or "<b>ɴᴏɴᴇ</b>" |
| | dc_id = from_user.dc_id or "[ᴜꜱᴇʀ ᴅᴏꜱᴇ'ᴛ ʜᴀᴠᴇ ᴀ ᴠᴀʟɪᴅ ᴅᴩ]" |
| | message_out_str += f"<b>➲ᴅᴄ-ɪᴅ:</b> <code>{dc_id}</code>\n" |
| | message_out_str += f"<b>➲ᴜꜱᴇʀɴᴀᴍᴇ:</b> @{username}\n" |
| | message_out_str += f"<b>➲ᴜꜱᴇʀ ʟɪɴᴋ:</b> <a href='tg://user?id={from_user.id}'><b>ᴄʟɪᴄᴋ ʜᴇʀᴇ</b></a>\n" |
| | if message.chat.type in ((enums.ChatType.SUPERGROUP, enums.ChatType.CHANNEL)): |
| | try: |
| | chat_member_p = await message.chat.get_member(from_user_id) |
| | joined_date = (chat_member_p.joined_date or datetime.now()).strftime("%Y.%m.%d %H:%M:%S") |
| | message_out_str += f"<b>➲ᴊᴏɪɴᴇᴅ ᴛʜɪꜱ ᴄʜᴀᴛ ᴏɴ:</b> <code>{joined_date}</code>\n" |
| | logger.info(f"User {from_user_id} joined chat {message.chat.id} on {joined_date}.") |
| | except UserNotParticipant: |
| | logger.info(f"User {from_user_id} is not a participant in chat {message.chat.id}.") |
| | pass |
| | chat_photo = from_user.photo |
| | if chat_photo: |
| | local_user_photo = await client.download_media(message=chat_photo.big_file_id) |
| | buttons = [[InlineKeyboardButton('ᴄʟᴏꜱᴇ ✘', callback_data='close_data')]] |
| | await message.reply_photo( |
| | photo=local_user_photo, |
| | quote=True, |
| | reply_markup=InlineKeyboardMarkup(buttons), |
| | caption=message_out_str, |
| | parse_mode=enums.ParseMode.HTML, |
| | disable_notification=True |
| | ) |
| | os.remove(local_user_photo) |
| | logger.info(f"User photo sent to user {message.from_user.id}.") |
| | else: |
| | buttons = [[InlineKeyboardButton('ᴄʟᴏꜱᴇ ✘', callback_data='close_data')]] |
| | await message.reply_text( |
| | text=message_out_str, |
| | reply_markup=InlineKeyboardMarkup(buttons), |
| | quote=True, |
| | parse_mode=enums.ParseMode.HTML, |
| | disable_notification=True |
| | ) |
| | logger.info(f"User information sent to user {message.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command(["imdb", 'search'])) |
| | async def imdb_search(client, message): |
| | """ |
| | Command to search for a movie on IMDb. |
| | Accessible to both admins and users. |
| | """ |
| | logger.info(f"User {message.from_user.id} issued 'imdb' or 'search' command.") |
| | if ' ' in message.text: |
| | k = await message.reply('ꜱᴇᴀʀᴄʜɪɴɢ ɪᴍᴅʙ..') |
| | r, title = message.text.split(None, 1) |
| | movies = await get_poster(title, bulk=True) |
| | if not movies: |
| | logger.info(f"No IMDb results found for title: {title}.") |
| | return await message.reply("ɴᴏ ʀᴇꜱᴜʟᴛ ꜰᴏᴜɴᴅ") |
| | btn = [[InlineKeyboardButton(f"{movie.get('title')} - {movie.get('year')}", callback_data=f"imdb#{movie.movieID}")] for movie in movies ] |
| | await k.edit('Hᴇʀᴇ Is Wʜᴀᴛ I Fᴏᴜɴᴅ Oɴ Iᴍᴅʙ', reply_markup=InlineKeyboardMarkup(btn)) |
| | logger.info(f"IMDb results found and sent to user {message.from_user.id}.") |
| | else: |
| | logger.info(f"Command 'imdb' or 'search' issued without a title.") |
| | await message.reply('Gɪᴠᴇ Mᴇ A Mᴏᴠɪᴇ / Sᴇʀɪᴇꜱ Nᴀᴍᴇ') |
| |
|
| | @Client.on_callback_query(filters.regex('^imdb')) |
| | async def imdb_callback(bot: Client, quer_y: CallbackQuery): |
| | """ |
| | Callback to handle IMDb search results. |
| | """ |
| | logger.info(f"Callback query received for IMDb: {quer_y.data}") |
| | i, movie = quer_y.data.split('#') |
| | imdb = await get_poster(query=movie, id=True) |
| | btn = [[InlineKeyboardButton(f"{imdb.get('title')}", url=imdb['url'])]] |
| | message = quer_y.message.reply_to_message or quer_y.message |
| | if imdb: |
| | caption = IMDB_TEMPLATE.format( |
| | query = imdb['title'], |
| | title = imdb['title'], |
| | votes = imdb['votes'], |
| | aka = imdb["aka"], |
| | seasons = imdb["seasons"], |
| | box_office = imdb['box_office'], |
| | localized_title = imdb['localized_title'], |
| | kind = imdb['kind'], |
| | imdb_id = imdb["imdb_id"], |
| | cast = imdb["cast"], |
| | runtime = imdb["runtime"], |
| | countries = imdb["countries"], |
| | certificates = imdb["certificates"], |
| | languages = imdb["languages"], |
| | director = imdb["director"], |
| | writer = imdb["writer"], |
| | producer = imdb["producer"], |
| | composer = imdb["composer"], |
| | cinematographer = imdb["cinematographer"], |
| | music_team = imdb["music_team"], |
| | distributors = imdb["distributors"], |
| | release_date = imdb['release_date'], |
| | year = imdb['year'], |
| | genres = imdb['genres'], |
| | poster = imdb['poster'], |
| | plot = imdb['plot'], |
| | rating = imdb['rating'], |
| | url = imdb['url'], |
| | **locals() |
| | ) |
| | logger.info(f"IMDb caption generated for movie: {imdb['title']}.") |
| | else: |
| | caption = "ɴᴏ ʀᴇꜱᴜʟᴛꜱ" |
| | logger.info(f"No IMDb results found for movie ID: {movie}.") |
| |
|
| | if imdb.get('poster'): |
| | try: |
| | await quer_y.message.reply_photo(photo=imdb['poster'], caption=caption, reply_markup=InlineKeyboardMarkup(btn)) |
| | logger.info(f"IMDb poster sent to user {quer_y.from_user.id}.") |
| | except (MediaEmpty, PhotoInvalidDimensions, WebpageMediaEmpty): |
| | pic = imdb.get('poster') |
| | poster = pic.replace('.jpg', "._V1_UX360.jpg") |
| | await quer_y.message.reply_photo(photo=poster, caption=caption, reply_markup=InlineKeyboardMarkup(btn)) |
| | logger.info(f"IMDb poster (fallback) sent to user {quer_y.from_user.id}.") |
| | except Exception as e: |
| | logger.error(f"Error sending IMDb poster: {e}") |
| | await quer_y.message.reply(caption, reply_markup=InlineKeyboardMarkup(btn), disable_web_page_preview=False) |
| | logger.info(f"IMDb caption sent to user {quer_y.from_user.id}.") |
| | await quer_y.message.delete() |
| | logger.info(f"Original message deleted after sending IMDb results.") |
| | else: |
| | await quer_y.message.edit(caption, reply_markup=InlineKeyboardMarkup(btn), disable_web_page_preview=False) |
| | logger.info(f"IMDb caption edited in message for user {quer_y.from_user.id}.") |
| |
|
| | @Client.on_message(filters.command('logs') & filters.user(ADMINS)) |
| | async def log_file(bot, msg): |
| | """ |
| | Command to send bot logs. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {msg.from_user.id} issued 'logs' command.") |
| | try: |
| | await msg.reply_document('BotLog.txt') |
| | logger.info(f"Bot logs sent to user {msg.from_user.id}.") |
| | except Exception as e: |
| | logger.error(f"Error sending bot logs: {e}") |
| | await msg.reply(str(e)) |
| |
|
| | @Client.on_message(filters.command("restart") & filters.user(ADMINS)) |
| | async def restart_bot(bot, msg): |
| | """ |
| | Command to restart the bot. |
| | Only accessible to admins. |
| | """ |
| | logger.info(f"User {msg.from_user.id} issued 'restart' command.") |
| | await msg.reply("Rᴇꜱᴛᴀᴛɪɴɢ........") |
| | await asyncio.sleep(2) |
| | await sts.delete() |
| | os.execl(sys.executable, sys.executable, *sys.argv) |
| | logger.info(f"Bot restarted by user {msg.from_user.id}.") |
