from pyrogram import Client, filters from pyrogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton from helper.helper_func import encode, get_message_id from config import LOGGER async def get_db_channels_info(client): """Get formatted database channels information with links""" db_channels = getattr(client, 'db_channels', {}) primary_db = getattr(client, 'primary_db_channel', client.db) if not db_channels: # If no additional DB channels, show primary only try: primary_chat = await client.get_chat(primary_db) if hasattr(primary_chat, 'invite_link') and primary_chat.invite_link: return f"
✦ ᴘʀɪᴍᴀʀʏ ᴅʙ ᴄʜᴀɴɴᴇʟ: {primary_chat.title}
" else: return f"
✦ ᴘʀɪᴍᴀʀʏ ᴅʙ ᴄʜᴀɴɴᴇʟ: {primary_chat.title} (`{primary_db}`)
" except: return f"
✦ ᴘʀɪᴍᴀʀʏ ᴅʙ ᴄʜᴀɴɴᴇʟ: `{primary_db}`
" # Format all DB channels with links channels_info = ["
✦ ᴀᴠᴀɪʟᴀʙʟᴇ ᴅᴀᴛᴀʙᴀsᴇ ᴄʜᴀɴɴᴇʟs:
"] for channel_id_str, channel_data in db_channels.items(): channel_name = channel_data.get('name', 'ᴜɴᴋɴᴏᴡɴ') is_primary_text = "✦ ᴘʀɪᴍᴀʀʏ" if channel_data.get('is_primary', False) else "• sᴇᴄᴏɴᴅᴀʀʏ" try: chat = await client.get_chat(int(channel_id_str)) if hasattr(chat, 'invite_link') and chat.invite_link: channels_info.append(f"{is_primary_text}: {channel_name}") else: channels_info.append(f"{is_primary_text}: {channel_name} (`{channel_id_str}`)") except: channels_info.append(f"{is_primary_text}: {channel_name} (`{channel_id_str}`)") return "\n".join(channels_info) @Client.on_message(filters.private & filters.command('batch')) async def batch(client: Client, message: Message): if message.from_user.id not in client.admins: return # Get all database channels with links db_channels_info = await get_db_channels_info(client) while True: try: first_message = await client.ask( text=f"""
ꜰᴏʀᴡᴀʀᴅ ᴛʜᴇ ꜰɪʀsᴛ ᴍᴇssᴀɢᴇ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)..
{db_channels_info}
ᴏʀ sᴇɴᴅ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴘᴏsᴛ ʟɪɴᴋ
""", chat_id=message.from_user.id, filters=(filters.forwarded | (filters.text & ~filters.forwarded)), timeout=60 ) except: return f_msg_id, source_channel_id = await get_message_id(client, first_message) if f_msg_id: break else: await first_message.reply("
✗ ᴇʀʀᴏʀ
\n\nᴛʜɪs ꜰᴏʀᴡᴀʀᴅᴇᴅ ᴘᴏsᴛ ɪs ɴᴏᴛ ꜰʀᴏᴍ ᴍʏ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴏʀ ᴛʜɪs ʟɪɴᴋ ɪs ᴛᴀᴋᴇɴ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ", quote = True) continue while True: try: second_message = await client.ask( text=f"""
ꜰᴏʀᴡᴀʀᴅ ᴛʜᴇ ʟᴀsᴛ ᴍᴇssᴀɢᴇ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)..
{db_channels_info}
ᴏʀ sᴇɴᴅ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴘᴏsᴛ ʟɪɴᴋ
""", chat_id=message.from_user.id, filters=(filters.forwarded | (filters.text & ~filters.forwarded)), timeout=60 ) except: return s_msg_id, _ = await get_message_id(client, second_message) # We only need msg_id for second message if s_msg_id: break else: await second_message.reply("
✗ ᴇʀʀᴏʀ
\n\nᴛʜɪs ꜰᴏʀᴡᴀʀᴅᴇᴅ ᴘᴏsᴛ ɪs ɴᴏᴛ ꜰʀᴏᴍ ᴍʏ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴏʀ ᴛʜɪs ʟɪɴᴋ ɪs ᴛᴀᴋᴇɴ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ", quote = True) continue # Use the source channel ID for encoding instead of default primary channel client.LOGGER(__name__, client.name).info(f"Generating batch link with source channel: {source_channel_id}, first_msg: {f_msg_id}, last_msg: {s_msg_id}") string = f"get-{f_msg_id * abs(source_channel_id)}-{s_msg_id * abs(source_channel_id)}" base64_string = await encode(string) link = f"https://t.me/{client.username}?start={base64_string}" reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("🔁 sʜᴀʀᴇ ᴜʀʟ", url=f'https://telegram.me/share/url?url={link}')]]) await second_message.reply_text(f"
✓ ʜᴇʀᴇ ɪs ʏᴏᴜʀ ʙᴀᴛᴄʜ ʟɪɴᴋ
\n\n{link}", quote=True, disable_web_page_preview=True, reply_markup=reply_markup) #===============================================================# @Client.on_message(filters.private & filters.command('genlink')) async def link_generator(client: Client, message: Message): if message.from_user.id not in client.admins: return # Get all database channels with links db_channels_info = await get_db_channels_info(client) while True: try: channel_message = await client.ask( text=f"""
ꜰᴏʀᴡᴀʀᴅ ᴍᴇssᴀɢᴇ ꜰʀᴏᴍ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)..
{db_channels_info}
ᴏʀ sᴇɴᴅ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴘᴏsᴛ ʟɪɴᴋ
""", chat_id=message.from_user.id, filters=(filters.forwarded | (filters.text & ~filters.forwarded)), timeout=60 ) except: return msg_id, source_channel_id = await get_message_id(client, channel_message) if msg_id: break else: await channel_message.reply("
✗ ᴇʀʀᴏʀ
\n\nᴛʜɪs ꜰᴏʀᴡᴀʀᴅᴇᴅ ᴘᴏsᴛ ɪs ɴᴏᴛ ꜰʀᴏᴍ ᴍʏ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴏʀ ᴛʜɪs ʟɪɴᴋ ɪs ɴᴏᴛ ᴛᴀᴋᴇɴ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ", quote = True) continue base64_string = await encode(f"get-{msg_id * abs(source_channel_id)}") link = f"https://t.me/{client.username}?start={base64_string}" reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("🔁 sʜᴀʀᴇ ᴜʀʟ", url=f'https://telegram.me/share/url?url={link}')]]) await channel_message.reply_text(f"
✓ ʜᴇʀᴇ ɪs ʏᴏᴜʀ ʟɪɴᴋ
\n\n{link}", quote=True, disable_web_page_preview=True, reply_markup=reply_markup) #===============================================================# @Client.on_message(filters.private & filters.command("nbatch")) async def nbatch(client: Client, message: Message): if message.from_user.id not in client.admins: return args = message.text.split() if len(args) < 2 or not args[1].isdigit(): await message.reply("
✗ ɪɴᴠᴀʟɪᴅ ꜰᴏʀᴍᴀᴛ!
ᴜsᴇ: /nbatch {number}") return batch_size = int(args[1]) # Get all database channels with links db_channels_info = await get_db_channels_info(client) while True: try: first_message = await client.ask( text=f"""
🚀 sᴇɴᴅ ᴅʙ ᴄʜᴀɴɴᴇʟ ꜰɪʀsᴛ ᴍᴇssᴀɢᴇ ʟɪɴᴋ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)...
{db_channels_info}""", chat_id=message.from_user.id, filters=(filters.text & ~filters.forwarded), timeout=60 ) except: return f_msg_id, source_channel_id = await get_message_id(client, first_message) if f_msg_id: break else: await first_message.reply("
😫 ɪɴᴠᴀʟɪᴅ!
sᴇɴᴅ ᴄᴏʀʀᴇᴄᴛ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴍᴇssᴀɢᴇ ʟɪɴᴋ.", quote=True) continue s_msg_id = f_msg_id + batch_size - 1 # Adding batch_size to first message ID string = f"get-{f_msg_id * abs(source_channel_id)}-{s_msg_id * abs(source_channel_id)}" base64_string = await encode(string) link = f"https://t.me/{client.username}?start={base64_string}" reply_markup = InlineKeyboardMarkup([ [InlineKeyboardButton("📫 ʏᴏᴜʀ ʙᴀᴛᴄʜ ᴜʀʟ", url=f'https://telegram.me/share/url?url={link}')] ]) await first_message.reply_text(f"
✓ ʜᴇʀᴇ ɪs ʏᴏᴜʀ ʙᴀᴛᴄʜ ʟɪɴᴋ
\n\n{link}", quote=True, disable_web_page_preview=True, reply_markup=reply_markup) #===============================================================# @Client.on_message(filters.private & filters.command("bulk")) async def bulk(client: Client, message: Message): if message.from_user.id not in client.admins: return await message.reply(client.reply_text) db_channels_info = await get_db_channels_info(client) while True: try: first_message = await client.ask( text=f"""
ꜰᴏʀᴡᴀʀᴅ ᴛʜᴇ ꜰɪʀsᴛ ᴍᴇssᴀɢᴇ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)..
\n{db_channels_info}\n\n
ᴏʀ sᴇɴᴅ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴘᴏsᴛ ʟɪɴᴋ
""", chat_id=message.from_user.id, filters=(filters.forwarded | (filters.text & ~filters.forwarded)), timeout=60 ) except: return f_msg_id, source_channel_id = await get_message_id(client, first_message) if f_msg_id: break else: await first_message.reply("
✗ ᴇʀʀᴏʀ
\n\nᴛʜɪs ꜰᴏʀᴡᴀʀᴅᴇᴅ ᴘᴏsᴛ ɪs ɴᴏᴛ ꜰʀᴏᴍ ᴍʏ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴏʀ ᴛʜɪs ʟɪɴᴋ ɪs ᴛᴀᴋᴇɴ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ", quote = True) continue while True: try: second_message = await client.ask( text=f"""
ꜰᴏʀᴡᴀʀᴅ ᴛʜᴇ ʟᴀsᴛ ᴍᴇssᴀɢᴇ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ (ᴡɪᴛʜ ǫᴜᴏᴛᴇs)..
\n{db_channels_info}\n\n
ᴏʀ sᴇɴᴅ ᴛʜᴇ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴘᴏsᴛ ʟɪɴᴋ
""", chat_id=message.from_user.id, filters=(filters.forwarded | (filters.text & ~filters.forwarded)), timeout=60 ) except: return s_msg_id, _ = await get_message_id(client, second_message) if s_msg_id: break else: await second_message.reply("
✗ ᴇʀʀᴏʀ
\n\nᴛʜɪs ꜰᴏʀᴡᴀʀᴅᴇᴅ ᴘᴏsᴛ ɪs ɴᴏᴛ ꜰʀᴏᴍ ᴍʏ ᴅʙ ᴄʜᴀɴɴᴇʟ ᴏʀ ᴛʜɪs ʟɪɴᴋ ɪs ᴛᴀᴋᴇɴ ꜰʀᴏᴍ ᴅʙ ᴄʜᴀɴɴᴇʟ", quote = True) continue if f_msg_id > s_msg_id: f_msg_id, s_msg_id = s_msg_id, f_msg_id processing_msg = await second_message.reply("⏳ Processing...", quote=True) msg_ids = list(range(f_msg_id, s_msg_id + 1)) messages = [] # fetch all messages iteratively for i in range(0, len(msg_ids), 200): try: chunk = await client.get_messages(chat_id=source_channel_id, message_ids=msg_ids[i:i+200]) for msg in chunk: if msg and not msg.empty: messages.append(msg) except: pass file_links = [] count = 1 for msg in messages: file_name = None if getattr(msg, "document", None): file_name = msg.document.file_name elif getattr(msg, "video", None): file_name = msg.video.file_name elif getattr(msg, "audio", None): file_name = msg.audio.file_name if not file_name: if msg.media: file_name = f"Unknown_Media_{msg.id}" else: continue base64_string = await encode(f"get-{msg.id * abs(source_channel_id)}") link = f"https://t.me/{client.username}?start={base64_string}" file_links.append(f"{count}. {file_name} - {link}") count += 1 db_key = str(source_channel_id)[-4:] if len(str(source_channel_id)) >= 4 else str(source_channel_id) batch_string = f"get-{f_msg_id * abs(source_channel_id)}-{s_msg_id * abs(source_channel_id)}" base64_batch = await encode(batch_string) batch_link = f"https://t.me/{client.username}?start={base64_batch}" final_text = f"📦 Bulk Upload Complete\n\nTotal Files: {len(file_links)}\nDB Key: {db_key}\n\nDownload URLs:\n\n" final_text += "\n".join(file_links) final_text += f"\n\n🔗 Batch Link: {batch_link}" if len(final_text) > 4050: import io file_bytes = io.BytesIO(final_text.encode('utf-8')) file_bytes.name = "bulk_links.txt" await second_message.reply_document(document=file_bytes, caption=f"📦 Bulk Upload Complete\nTotal Files: {len(file_links)}\nDB Key: {db_key}\n\nThe text was too long, so here is the text file.", quote=True) await processing_msg.delete() else: await processing_msg.edit_text(final_text, disable_web_page_preview=True)