| import html |
| from pyrogram import Client, filters, enums |
| from pyrogram.types import Message |
| from helper.database import codeflixbots as db |
| from utils.channel_resolver import resolve_routing_channel |
|
|
| @Client.on_message(filters.command("add_channel") & filters.private) |
| async def add_channel_cmd(client: Client, message: Message): |
| await client.stop_listening(chat_id=message.chat.id, user_id=message.from_user.id) |
| user_id = message.from_user.id |
| if not await db.is_user_authorized(user_id): |
| return await message.reply_text("❌ You are not authorized to use this bot.") |
|
|
| chat = None |
| hint = None |
|
|
| |
| forwarded_message = message.reply_to_message |
| forwarded_chat = getattr(forwarded_message, "forward_from_chat", None) if forwarded_message else None |
| if forwarded_chat is None and forwarded_message: |
| origin = getattr(forwarded_message, "forward_origin", None) |
| forwarded_chat = getattr(origin, "chat", None) if origin else None |
|
|
| if forwarded_chat is not None: |
| if getattr(getattr(forwarded_chat, "type", None), "name", None) != "CHANNEL": |
| await message.reply_text("The forwarded message is not from a channel.") |
| return |
| try: |
| chat = await resolve_routing_channel(client, forwarded_chat=forwarded_chat) |
| except Exception as e: |
| return await message.reply_text(f"Could not use the forwarded channel. {e}") |
| if len(message.command) > 1: |
| hint = message.text.split(None, 1)[1].strip() |
| else: |
| text = message.text.strip() |
| parts = text.split() |
| if len(parts) < 2: |
| await message.reply_text( |
| "Usage:\n" |
| "• `/add_channel <hint> <channel ID/@username/t.me link>`\n" |
| "• Or reply to a forwarded channel post with `/add_channel <hint>`.\n" |
| "Private invite links alone are not enough; forward a post instead." |
| ) |
| return |
|
|
| args = parts[1:] |
| channel_id_or_username = None |
| hint_words = [] |
| for arg in args: |
| if arg.startswith("-") or arg.isdigit() or arg.startswith("@") or "t.me/" in arg: |
| channel_id_or_username = arg |
| else: |
| hint_words.append(arg) |
|
|
| if not channel_id_or_username: |
| channel_id_or_username = args[0] |
| hint_words = args[1:] |
| hint = " ".join(hint_words).strip() if hint_words else None |
| try: |
| chat = await resolve_routing_channel(client, channel_id_or_username) |
| except Exception as e: |
| return await message.reply_text(f"Could not resolve channel. {e}") |
|
|
| if getattr(getattr(chat, "type", None), "name", None) != "CHANNEL": |
| await message.reply_text("The resolved chat is not a channel.") |
| return |
|
|
| title = chat.title or str(chat.id) |
| if chat.username: |
| link = f"https://t.me/{chat.username}" |
| else: |
| link = chat.invite_link |
| if not link: |
| try: |
| link = await client.export_chat_invite_link(chat.id) |
| except Exception: |
| link = "" |
| if not hint: |
| hint = title |
|
|
| photo_id = None |
| if forwarded_message and forwarded_message.photo: |
| photo_id = forwarded_message.photo.file_id |
| await db.add_channel(user_id, chat.id, title, link, hint, photo_id=photo_id) |
| await message.reply_text( |
| f"✅ <b>Channel Added</b>\n\nTitle: {html.escape(title)}\nID: <code>{chat.id}</code>\n" |
| f"Hint: <code>{html.escape(hint)}</code>\nLink: {html.escape(link or 'No invite link available')}", |
| parse_mode=enums.ParseMode.HTML |
| ) |
|
|
| @Client.on_message(filters.command("view_channel") & filters.private) |
| async def view_channel_cmd(client: Client, message: Message): |
| await client.stop_listening(chat_id=message.chat.id, user_id=message.from_user.id) |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("❌ You are not authorized to use this bot.") |
|
|
| channels = await db.get_channels(message.from_user.id) |
| if not channels: |
| await message.reply_text("No channels added yet.") |
| return |
| |
| text = "**Your Saved Channels:**\n\n" |
| for i, ch in enumerate(channels, 1): |
| text += f"{i}. **{ch['title']}**\n" |
| text += f" - ID: `{ch['channel_id']}`\n" |
| text += f" - Hint: `{ch['hint']}`\n" |
| text += f" - Link: {ch['link']}\n" |
| text += f" - DB ID: `{ch['_id']}`\n\n" |
| |
| await message.reply_text(text, disable_web_page_preview=True) |
|
|
| @Client.on_message(filters.command("remove_channel") & filters.private) |
| async def remove_channel_cmd(client: Client, message: Message): |
| await client.stop_listening(chat_id=message.chat.id, user_id=message.from_user.id) |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("❌ You are not authorized to use this bot.") |
|
|
| args = message.text.split(" ") |
| if len(args) < 2: |
| await message.reply_text("Usage: `/remove_channel <db_id>`\nGet DB ID from /view_channel.") |
| return |
| |
| db_id = args[1] |
| try: |
| await db.remove_channel(db_id) |
| await message.reply_text(f"Channel removed successfully.") |
| except Exception as e: |
| await message.reply_text(f"Error removing channel. Is the ID correct?\nError: {e}") |
|
|
