import base64 import re import asyncio from pyrogram import filters, Client from pyrogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton from pyrogram.enums import ChatMemberStatus from pyrogram.errors import UserNotParticipant, Forbidden, PeerIdInvalid, ChatAdminRequired, FloodWait from datetime import datetime, timedelta from pyrogram import errors #===============================================================# async def encode(string): string_bytes = string.encode("ascii") base64_bytes = base64.urlsafe_b64encode(string_bytes) base64_string = (base64_bytes.decode("ascii")).strip("=") return base64_string #===============================================================# async def decode(base64_string): base64_string = base64_string.strip("=") # links generated before this commit will be having = sign, hence striping them to handle padding errors. base64_bytes = (base64_string + "=" * (-len(base64_string) % 4)).encode("ascii") string_bytes = base64.urlsafe_b64decode(base64_bytes) string = string_bytes.decode("ascii") return string #===============================================================# async def get_messages(client, message_ids): messages = [] total_messages = 0 while total_messages != len(message_ids): temb_ids = message_ids[total_messages:total_messages+200] try: # Use new multi-DB channel function msgs = await get_messages_from_db_channels(client, temb_ids) except FloodWait as e: await asyncio.sleep(e.x) msgs = await get_messages_from_db_channels(client, temb_ids) except: pass total_messages += len(temb_ids) messages.extend(msgs) return messages #===============================================================# async def get_message_id(client, message): """Get message ID and source channel ID from forwarded message or link""" if message.forward_from_chat: # Check if forwarded from primary DB channel if message.forward_from_chat.id == client.db: return message.forward_from_message_id, client.db # Check against multiple DB channels db_channels = getattr(client, 'db_channels', {}) for channel_id_str in db_channels.keys(): if message.forward_from_chat.id == int(channel_id_str): return message.forward_from_message_id, int(channel_id_str) return 0, 0 elif message.forward_sender_name: return 0, 0 elif message.text: pattern = r"https://t.me/(?:c/)?(.*)/(\d+)" matches = re.match(pattern,message.text) if not matches: return 0, 0 channel_id = matches.group(1) msg_id = int(matches.group(2)) if channel_id.isdigit(): # Check primary DB channel if f"-100{channel_id}" == str(client.db): return msg_id, client.db # Check against multiple DB channels db_channels = getattr(client, 'db_channels', {}) for channel_id_str in db_channels.keys(): if f"-100{channel_id}" == channel_id_str: return msg_id, int(channel_id_str) else: # Check by username for primary DB channel if hasattr(client, 'db_channel') and channel_id == client.db_channel.username: return msg_id, client.db # Check against multiple DB channels usernames (if needed) db_channels = getattr(client, 'db_channels', {}) for channel_id_str, channel_data in db_channels.items(): try: chat = await client.get_chat(int(channel_id_str)) if hasattr(chat, 'username') and chat.username == channel_id: return msg_id, int(channel_id_str) except: continue else: return 0, 0 #===============================================================# async def get_message_id_legacy(client, message): """Legacy function for backward compatibility - returns only message ID""" msg_id, _ = await get_message_id(client, message) return msg_id #===============================================================# async def get_messages_from_db_channels(client, temb_ids): """Get messages from multiple DB channels - tries primary first, then falls back to others""" messages = [] total_messages = 0 # First try primary DB channel try: primary_db = getattr(client, 'primary_db_channel', client.db) msgs = await client.get_messages( chat_id=primary_db, message_ids=temb_ids ) # Filter out None messages (deleted/not found) valid_msgs = [msg for msg in msgs if msg is not None] messages.extend(valid_msgs) found_ids = {msg.id for msg in valid_msgs} missing_ids = [mid for mid in temb_ids if mid not in found_ids] # If we found all messages, return early if not missing_ids: return messages # Try other DB channels for missing messages db_channels = getattr(client, 'db_channels', {}) for channel_id_str, channel_data in db_channels.items(): if not channel_data.get('is_active', True): # Skip inactive channels continue if int(channel_id_str) == primary_db: # Skip primary (already tried) continue try: additional_msgs = await client.get_messages( chat_id=int(channel_id_str), message_ids=missing_ids ) valid_additional = [msg for msg in additional_msgs if msg is not None] messages.extend(valid_additional) # Update missing IDs found_additional_ids = {msg.id for msg in valid_additional} missing_ids = [mid for mid in missing_ids if mid not in found_additional_ids] # If we found all remaining messages, break if not missing_ids: break except Exception as e: client.LOGGER(__name__, client.name).warning(f"Error getting messages from DB channel {channel_id_str}: {e}") continue except FloodWait as e: await asyncio.sleep(e.x) # Retry with the same function return await get_messages_from_db_channels(client, temb_ids) except Exception as e: client.LOGGER(__name__, client.name).warning(f"Error getting messages from DB channels: {e}") return messages #===============================================================# def get_readable_time(seconds: int) -> str: count = 0 up_time = "" time_list = [] time_suffix_list = ["s", "m", "h", "days"] while count < 4: count += 1 remainder, result = divmod(seconds, 60) if count < 3 else divmod(seconds, 24) if seconds == 0 and remainder == 0: break time_list.append(int(result)) seconds = int(remainder) hmm = len(time_list) for x in range(hmm): time_list[x] = str(time_list[x]) + time_suffix_list[x] if len(time_list) == 4: up_time += f"{time_list.pop()}, " time_list.reverse() up_time += ":".join(time_list) return up_time #===============================================================# async def is_bot_admin(client, channel_id): try: bot = await client.get_chat_member(channel_id, "me") if bot.status in (ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.OWNER): if bot.privileges: required_rights = ["can_invite_users", "can_delete_messages"] missing_rights = [right for right in required_rights if not getattr(bot.privileges, right, False)] if missing_rights: return False, f"Bot is missing the following rights: {', '.join(missing_rights)}" return True, None return False, "Bot is not an admin in the channel." except errors.ChatAdminRequired: return False, "Bot lacks perminsion to access admin information in this channel." except Exception as e: return False, f"Unexpected error: {str(e)}" #===============================================================# async def check_single_subscription(client, user_id, channel_id, channel_name, channel_link, request, timer): try: # Get actual membership status first user = await client.get_chat_member(channel_id, user_id) actual_status = user.status # If user is already a member, admin, or owner if actual_status in {ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.OWNER}: await client.mongodb.update_fsub_status(user_id, channel_id, "joined") await client.mongodb.add_channel_user(channel_id, user_id) # If there was a pending join request, mark it as approved if request and await client.mongodb.has_submitted_join_request(user_id, channel_id): await client.mongodb.update_join_request_status(user_id, channel_id, "approved") return channel_id, actual_status # User is not a member - check if they left after being approved if request: # For request channels, check if user has submitted a request has_request = await client.mongodb.has_submitted_join_request(user_id, channel_id) if has_request: # User has submitted request but not yet a member request_status = await client.mongodb.get_join_request_status(user_id, channel_id) if request_status == "approved": # Request was approved but user still not in channel # This means user might have left after approval - force them to rejoin await client.mongodb.update_fsub_status(user_id, channel_id, "left") await client.mongodb.remove_join_request(user_id, channel_id) return channel_id, ChatMemberStatus.BANNED else: # Request is still pending, allow user to proceed await client.mongodb.update_fsub_status(user_id, channel_id, "request_submitted") return channel_id, ChatMemberStatus.MEMBER # Treat as subscribed for request channels else: # No request submitted yet for request channel await client.mongodb.update_fsub_status(user_id, channel_id, "not_requested") return channel_id, ChatMemberStatus.BANNED else: # Regular channel (not request), user must be a member await client.mongodb.update_fsub_status(user_id, channel_id, "left") await client.mongodb.remove_channel_user(channel_id, user_id) return channel_id, ChatMemberStatus.BANNED except UserNotParticipant: # User is not in the channel await client.mongodb.update_fsub_status(user_id, channel_id, "left") await client.mongodb.remove_channel_user(channel_id, user_id) if request: # For request channels, check if user has submitted a request has_request = await client.mongodb.has_submitted_join_request(user_id, channel_id) if has_request: # User has submitted request but not in channel - still allow access for request channels await client.mongodb.update_fsub_status(user_id, channel_id, "request_submitted") return channel_id, ChatMemberStatus.MEMBER # Treat as subscribed for request channels else: # No request submitted yet await client.mongodb.update_fsub_status(user_id, channel_id, "not_requested") return channel_id, ChatMemberStatus.BANNED else: # Regular channel, user must join return channel_id, ChatMemberStatus.BANNED except Forbidden: client.LOGGER(__name__, client.name).warning(f"Bot lacks permission for {channel_name}.") return channel_id, None except Exception as e: client.LOGGER(__name__, client.name).warning(f"Error checking {channel_name}: {e}") return channel_id, None async def check_subscription(client, user_id): """Enhanced subscription check with parallel check execution for ultimate speed.""" # Ensure user exists in database if not await client.mongodb.present_user(user_id): await client.mongodb.add_user(user_id) tasks = [] for channel_id, (channel_name, channel_link, request, timer) in client.fsub_dict.items(): tasks.append(check_single_subscription(client, user_id, channel_id, channel_name, channel_link, request, timer)) results = await asyncio.gather(*tasks) statuses = {channel_id: status for channel_id, status in results} return statuses #===============================================================# def is_user_subscribed(statuses): """Check if user is subscribed to all channels.""" return all( status in {ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.OWNER} for status in statuses.values() if status is not None ) and bool(statuses) #===============================================================# def force_sub(func): """Decorator to enforce force subscription before executing a command.""" async def wrapper(client: Client, message: Message): if not client.fsub_dict: return await func(client, message) photo = client.messages.get('FSUB_PHOTO', '') if photo: msg = await message.reply_photo( caption="ᴡᴀɪᴛ ᴀ sᴇᴄᴏɴᴅ.....", photo=photo ) else: msg = await message.reply( "ᴡᴀɪᴛ ᴀ sᴇᴄᴏɴᴅ....." ) user_id = message.from_user.id statuses = await check_subscription(client, user_id) if is_user_subscribed(statuses): await msg.delete() return await func(client, message) # User is not subscribed to all channels buttons = [] channels_message = f"{client.messages.get('FSUB', '')}\n\n" for channel_id, (channel_name, channel_link, request, timer) in client.fsub_dict.items(): status = statuses.get(channel_id, None) # Generate invite link if needed if timer > 0: expire_time = datetime.now() + timedelta(minutes=timer) try: invite = await client.create_chat_invite_link( chat_id=channel_id, expire_date=expire_time, creates_join_request=request ) channel_link = invite.invite_link except Exception as e: client.LOGGER(__name__, client.name).warning(f"Error creating invite link for {channel_name}: {e}") # Add button based on user status if status not in {ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.OWNER}: # Check if user has already submitted request for request channels if request and await client.mongodb.has_submitted_join_request(user_id, channel_id): request_status = await client.mongodb.get_join_request_status(user_id, channel_id) if request_status == "pending": # Don't add button if request is still pending continue elif request_status == "approved": # User can now join the channel button_text = f"{channel_name}" else: button_text = f"{channel_name}" else: # User hasn't submitted request or it's a regular channel if request: button_text = f"{channel_name}" else: button_text = f"{channel_name}" buttons.append(InlineKeyboardButton(button_text, url=channel_link)) # Add "Try Again" button if needed from_link = message.text.split(" ") if len(from_link) > 1: try_again_link = f"https://t.me/{client.username}/?start={from_link[1]}" buttons.append(InlineKeyboardButton("🔄 Try Again", url=try_again_link)) # Organize buttons in rows of 1 for better readability buttons_markup = InlineKeyboardMarkup([[button] for button in buttons]) buttons_markup = None if not buttons else buttons_markup # Edit message with status update and buttons try: await msg.edit_text(text=channels_message, reply_markup=buttons_markup) except Exception as e: client.LOGGER(__name__, client.name).warning(f"Error updating force sub message: {e}") # Fallback: send new message if edit fails try: await msg.delete() await message.reply(text=channels_message, reply_markup=buttons_markup) except Exception: pass return wrapper #===============================================================# #Time conversion for auto delete timer def convert_time(duration_seconds: int) -> str: periods = [ ('Yᴇᴀʀ', 60 * 60 * 24 * 365), ('Mᴏɴᴛʜ', 60 * 60 * 24 * 30), ('Day', 60 * 60 * 24), ('Hour', 60 * 60), ('Minute', 60), ('Second', 1) ] parts = [] for period_name, period_seconds in periods: if duration_seconds >= period_seconds: num_periods = duration_seconds // period_seconds duration_seconds %= period_seconds parts.append(f"{num_periods} {period_name}{'s' if num_periods > 1 else ''}") if len(parts) == 0: return "0 Sᴇᴄᴏɴᴅ" elif len(parts) == 1: return parts[0] else: return ', '.join(parts[:-1]) +' ᴀɴᴅ '+ parts[-1] #===============================================================# #.........Auto Delete Functions.......# #===============================================================# DEL_MSG = """This File is deleting automatically in {time}.. Forward in your Saved Messages..!""" #Function for provide auto delete notification message async def auto_del_notification(bot_username, msg, delay_time, transfer): temp = await msg.reply_text(DEL_MSG.format(username=bot_username, time=convert_time(delay_time)), disable_web_page_preview = True) await asyncio.sleep(delay_time) try: await temp.delete() except Exception as e: pass try: await msg.delete() except Exception as e: print(f"Error occurred on auto_del_notification() : {e}") #Function for deleteing files/Messages..... async def delete_message(msg, delay_time): await asyncio.sleep(delay_time) try: await msg.delete() except Exception as e: print(f"Error occurred on delete_message() : {e}") #===============================================================# #Function for batch auto delete - sends one notification for all files async def batch_auto_del_notification(bot_username, messages, delay_time, transfer_link, chat_id, client): """Send one notification for batch of files and delete all after timer""" if not messages: return # Send single countdown notification notification_msg = await client.send_message( chat_id=chat_id, text=DEL_MSG.format(username=bot_username, time=convert_time(delay_time)), disable_web_page_preview=True ) await asyncio.sleep(delay_time) # Delete all file messages for msg in messages: try: await msg.delete() except Exception as e: print(f"Error deleting message {getattr(msg, 'id', 'Unknown')}: {e}") try: await notification_msg.delete() except Exception as e: pass