File size: 5,504 Bytes
5d668bd 3867e4d 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd 18ce7e4 5d668bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import os
from pyrogram import Client, filters
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup, CallbackQuery
from pyrogram.errors import UserNotParticipant
from config import Config
FORCE_SUB_CHANNELS = Config.FORCE_SUB_CHANNELS
IMAGE_URL = "https://graph.org/file/a27d85469761da836337c.jpg"
async def not_subscribed(_, __, message):
# Force sub is completely disabled as requested
return False
# Anti-spam cache mapping to prevent 100+ error spams when forwarding folders
force_sub_spam_cache = {}
@Client.on_message(filters.private & filters.create(not_subscribed))
async def forces_sub(client, message):
not_joined_channels = []
admin_error = None
for channel in FORCE_SUB_CHANNELS:
try:
peer = int(channel) if str(channel).lstrip('-').isdigit() else channel
user = await client.get_chat_member(peer, message.from_user.id)
if user.status in {"kicked", "left"}:
not_joined_channels.append(channel)
except UserNotParticipant:
not_joined_channels.append(channel)
except Exception as e:
# Captures ANY Pyrogram validation crash (PeerIdInvalid, ChatAdminRequired, etc)
admin_error = f"[{type(e).__name__}] {str(e)}"
not_joined_channels.append(channel)
# Anti-spam logic: don't bombard user with 100 errors instantly
user_id = message.from_user.id
import time
last_sent = force_sub_spam_cache.get(user_id, 0)
if time.time() - last_sent < 15:
# Ignore this message if we already warned them within the last 15 seconds
return
force_sub_spam_cache[user_id] = time.time()
if admin_error:
text = (
"⚠️ **Bot Configuration Error!** ⚠️\n\n"
f"The bot is failing to verify the Force-Sub channel due to an API error:\n\n"
f"`{admin_error}`\n\n"
"**Fixes:**\n"
"1. Make sure the bot is an Admin.\n"
"2. If it's a private channel, you MUST forward at least 1 message from the channel to the bot so it can learn the channel ID!"
)
return await message.reply_text(text)
buttons = [
[
InlineKeyboardButton(
text=f"• ᴊᴏɪɴ Cʜᴀɴɴᴇʟ •", url=f"https://t.me/{str(channel).replace('@', '')}" if not str(channel).lstrip('-').isdigit() else "https://t.me/codeflix_bots"
)
]
for channel in not_joined_channels
]
buttons.append(
[
InlineKeyboardButton(
text="• ᴊᴏɪɴᴇᴅ •", callback_data="check_subscription"
)
]
)
text = "**ʙᴀᴋᴋᴀ!!, ʏᴏᴜ'ʀᴇ ɴᴏᴛ ᴊᴏɪɴᴇᴅ ᴛᴏ ᴀʟʟ ʀᴇǫᴜɪʀᴇᴅ ᴄʜᴀɴɴᴇʟs, ᴊᴏɪɴ ᴛʜᴇ ᴜᴘᴅᴀᴛᴇ ᴄʜᴀɴɴᴇʟs ᴛᴏ ᴄᴏɴᴛɪɴᴜᴇ**"
await message.reply_photo(
photo=IMAGE_URL,
caption=text,
reply_markup=InlineKeyboardMarkup(buttons)
)
@Client.on_callback_query(filters.regex("check_subscription"))
async def check_subscription(client, callback_query: CallbackQuery):
user_id = callback_query.from_user.id
not_joined_channels = []
admin_error = False
for channel in FORCE_SUB_CHANNELS:
try:
peer = int(channel) if str(channel).lstrip('-').isdigit() else channel
user = await client.get_chat_member(peer, user_id)
if user.status in {"kicked", "left"}:
not_joined_channels.append(channel)
except UserNotParticipant:
not_joined_channels.append(channel)
except Exception as e:
if "CHAT_ADMIN_REQUIRED" in str(e):
admin_error = True
not_joined_channels.append(channel)
if admin_error:
return await callback_query.answer("⚠️ The bot is not an admin in the Force Sub channel!", show_alert=True)
if not not_joined_channels:
new_text = "**ʏᴏᴜ ʜᴀᴠᴇ ᴊᴏɪɴᴇᴅ ᴀʟʟ ᴛʜᴇ ʀᴇǫᴜɪʀᴇᴅ ᴄʜᴀɴɴᴇʟs. ᴛʜᴀɴᴋ ʏᴏᴜ! 😊 /start ɴᴏᴡ**"
if callback_query.message.caption != new_text:
await callback_query.message.edit_caption(
caption=new_text,
reply_markup=InlineKeyboardMarkup([
[InlineKeyboardButton("• ɴᴏᴡ ᴄʟɪᴄᴋ ʜᴇʀᴇ •", callback_data='help')]
])
)
else:
buttons = [
[
InlineKeyboardButton(
text=f"• ᴊᴏɪɴ Cʜᴀɴɴᴇʟ •",
url=f"https://t.me/{str(channel).replace('@', '')}" if not str(channel).lstrip('-').isdigit() else "https://t.me/codeflix_bots",
)
]
for channel in not_joined_channels
]
buttons.append(
[
InlineKeyboardButton(
text="• ᴊᴏɪɴᴇᴅ •", callback_data="check_subscription"
)
]
)
text = "**ʏᴏᴜ ʜᴀᴠᴇ ᴊᴏɪɴᴇᴅ ᴀʟʟ ᴛʜᴇ ʀᴇǫᴜɪʀᴇᴅ ᴄʜᴀɴɴᴇʟs. ᴘʟᴇᴀsᴇ ᴊᴏɪɴ ᴛʜᴇ ᴜᴘᴅᴀᴛᴇ ᴄʜᴀɴɴᴇʟs ᴛᴏ ᴄᴏɴᴛɪɴᴜᴇ**"
if callback_query.message.caption != text:
await callback_query.message.edit_caption(
caption=text,
reply_markup=InlineKeyboardMarkup(buttons)
)
|