Bump to v1.5 (Migration to Telethon)

bot/__init__.py

@@ -1,19 +1,15 @@
-from pyrogram import Client
+from telethon import TelegramClient
 from logging import getLogger
 from logging.config import dictConfig
 from .config import Telegram, LOGGER_CONFIG_JSON
 
 dictConfig(LOGGER_CONFIG_JSON)
 
-version = 1.4
+version = 1.5
 logger = getLogger('bot')
 
-TelegramBot = Client(
- name="bot",
- api_id = Telegram.API_ID,
- api_hash = Telegram.API_HASH,
- bot_token = Telegram.BOT_TOKEN,
- plugins={"root": "bot/plugins"},
- workers = Telegram.BOT_WORKERS,
- max_concurrent_transmissions=1000
+TelegramBot = TelegramClient(
+ session='bot',
+ api_id=Telegram.API_ID,
+ api_hash=Telegram.API_HASH
 )

bot/__main__.py

@@ -1,7 +1,22 @@
+from importlib import import_module
+from pathlib import Path
 from bot import TelegramBot, logger
+from bot.config import Telegram
 from bot.server import server
 
+def load_plugins():
+ count = 0
+ for path in Path('bot/plugins').rglob('*.py'):
+ import_module(f'bot.plugins.{path.stem}')
+ count += 1
+ logger.info(f'Loaded {count} {"plugins" if count > 1 else "plugin"}.')
+
 if __name__ == '__main__':
- logger.info('Initializing...')
+ logger.info('initializing...')
  TelegramBot.loop.create_task(server.serve())
- TelegramBot.run()
+ TelegramBot.start(bot_token=Telegram.BOT_TOKEN)
+ logger.info('Telegram client is now started.')
+ logger.info('Loading bot plugins...')
+ load_plugins()
+ logger.info('Bot is now ready!')
+ TelegramBot.run_until_disconnected()

bot/config.py

@@ -8,7 +8,6 @@ class Telegram:
  BOT_USERNAME = env.get("TELEGRAM_BOT_USERNAME", "BotFather")
  BOT_TOKEN = env.get("TELEGRAM_BOT_TOKEN", "1234:abcd")
  CHANNEL_ID = int(env.get("TELEGRAM_CHANNEL_ID", -1001234567890))
- BOT_WORKERS = int(env.get("BOT_WORKERS", 10))
  SECRET_CODE_LENGTH = int(env.get("SECRET_CODE_LENGTH", 12))
 
 class Server:

bot/modules/decorators.py

@@ -1,16 +1,22 @@
-from pyrogram import Client 
-from pyrogram.types import Message, CallbackQuery
-from typing import Union, Callable
+from telethon.events import NewMessage, CallbackQuery
+from telethon.tl.custom import Message
+from typing import Callable
 from functools import wraps
 from bot.config import Telegram
+from bot.modules.static import *
 
-def verify_user(func: Callable):
+def verify_user(private: bool = False):
 
- @wraps(func)
- async def decorator(client: Client, update: Union[Message, CallbackQuery]):
- chat_id = str(update.from_user.id if update.from_user else update.chat.id)
+ def decorator(func: Callable):
+ @wraps(func)
+ async def wrapper(update: NewMessage.Event | CallbackQuery.Event):
+ if private and not update.is_private:
+ return
 
- if not Telegram.ALLOWED_USER_IDS or chat_id in Telegram.ALLOWED_USER_IDS:
- return await func(client, update)
-
+ chat_id = str(update.chat_id)
+
+ if not Telegram.ALLOWED_USER_IDS or chat_id in Telegram.ALLOWED_USER_IDS:
+ return await func(update)
+
+ return wrapper
  return decorator

bot/modules/static.py

@@ -9,6 +9,21 @@
 - /log to get bot logs. (admin only!)
 """
 
+UserInfoText = \
+"""
+**First Name:**
+`{sender.first_name}`
+
+**Last Name:**
+`{sender.last_name}`
+
+**User ID:**
+`{sender.id}`
+
+**Username:**
+`@{sender.username}`
+"""
+
 FileLinksText = \
 """
 **Download Link:**
@@ -45,4 +60,9 @@
 InvalidPayloadText = \
 """
 Invalid payload.
-"""
+"""
+
+MediaTypeNotSupportedText = \
+"""
+Sorry, this media type is not supported.
+"""

bot/modules/telegram.py

@@ -1,56 +1,68 @@
-from pyrogram.types import Message
+from telethon.events import NewMessage
+from telethon.tl.custom import Message
 from datetime import datetime
 from mimetypes import guess_type
 from bot import TelegramBot
 from bot.config import Telegram
 from bot.server.error import abort
 
-async def get_message(message_id: int):
+async def get_message(message_id: int) -> Message | None:
  message = None
 
  try:
- message = await TelegramBot.get_messages(
- chat_id=Telegram.CHANNEL_ID,
- message_ids=message_id
- )
- if message.empty: message = None
+ message = await TelegramBot.get_messages(Telegram.CHANNEL_ID, ids=message_id)
  except Exception:
  pass
 
  return message
 
-async def get_file_properties(msg: Message):
- attributes = (
- 'document',
- 'video',
- 'audio',
- 'voice',
- 'photo',
- 'video_note'
+async def send_message(message:Message, send_to:int = Telegram.CHANNEL_ID) -> Message:
+ message.forward
+ return await TelegramBot.send_message(
+ entity=send_to,
+ message=message
  )
 
- for attribute in attributes:
- media = getattr(msg, attribute, None)
- if media:
- file_type = attribute
- break
-
- if not media: abort(400, 'Unknown file type.')
+def filter_files(update: NewMessage.Event | Message):
+ return bool(
+ (
+ update.document
+ or update.photo
+ or update.video
+ or update.video_note
+ or update.audio
+ or update.gif
+ )
+ and not update.sticker
+ )
 
- file_name = getattr(media, 'file_name', None)
- file_size = getattr(media, 'file_size', 0)
+def get_file_properties(message: Message):
+ file_name = message.file.name
+ file_size = message.file.size or 0
+ mime_type = message.file.mime_type
 
  if not file_name:
- file_format = {
+ attributes = {
  'video': 'mp4',
  'audio': 'mp3',
  'voice': 'ogg',
  'photo': 'jpg',
  'video_note': 'mp4'
- }.get(attribute)
- date = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
- file_name = f'{file_type}-{date}.{file_format}'
+ }
 
- mime_type = guess_type(file_name)[0] or 'application/octet-stream'
+ for attribute in attributes:
+ media = getattr(message, attribute, None)
+ if media:
+ file_type, file_format = attribute, attributes[attribute]
+ break
+
+ if not media:
+ abort(400, 'Invalid media type.')
 
+ date = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+ file_name = f'{file_type}-{date}.{file_format}'
+
+ if not mime_type:
+ mime_type = guess_type(file_name)[0] or 'application/octet-stream'
+
  return file_name, file_size, mime_type

bot/plugins/callback.py

@@ -1,27 +1,22 @@
-from pyrogram.types import CallbackQuery
+from telethon.events import CallbackQuery
 from bot import TelegramBot
-from bot.modules.static import *
 from bot.modules.decorators import verify_user
+from bot.modules.static import *
 from bot.modules.telegram import get_message
 
-@TelegramBot.on_callback_query()
-@verify_user
-async def manage_callback(bot, q: CallbackQuery):
- query = q.data
- if query.startswith('rm_'):
- sq = query.split('_')
+@TelegramBot.on(CallbackQuery(pattern=r'^rm_'))
+@verify_user(private=True)
+async def delete_file(event: CallbackQuery.Event):
+ query_data = event.query.data.decode().split('_')
+
+ if len(query_data) != 3:
+ return await event.answer(InvalidQueryText, alert=True)
+
+ message = await get_message(int(query_data[1]))
+
+ if not message:
+ return await event.answer(MessageNotExist, alert=True)
 
- if len(sq) != 3:
- return await q.answer(InvalidQueryText, show_alert=True)
-
- message = await get_message(int(sq[1]))
-
- if not message:
- return await q.answer(MessageNotExist, show_alert=True)
- if sq[2] != message.caption:
- return await q.answer(InvalidQueryText, show_alert=True)
+ await message.delete()
 
- await message.delete()
- await q.answer(LinkRevokedText, show_alert=True)
- else:
- await q.answer(InvalidQueryText, show_alert=True)
+ return await event.answer(LinkRevokedText, alert=True)

bot/plugins/commands.py

@@ -1,45 +1,28 @@
-from pyrogram import filters
-from pyrogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton
-from aiofiles import open as async_open
-from aiofiles.os import remove as async_rm
-from bot import TelegramBot, logger
+from telethon import Button
+from telethon.events import NewMessage
+from telethon.tl.custom.message import Message
+from bot import TelegramBot
 from bot.config import Telegram
 from bot.modules.static import *
-from .deeplinks import deeplinks
 from bot.modules.decorators import verify_user
 
-@TelegramBot.on_message(filters.command('start') & filters.private)
-@verify_user
-async def start(_, msg: Message):
- if len(msg.command) != 1:
- return await deeplinks(msg, msg.command[1])
-
- await msg.reply(
- text=WelcomeText % {'first_name': msg.from_user.first_name},
- quote=True,
- reply_markup=InlineKeyboardMarkup(
+@TelegramBot.on(NewMessage(incoming=True, pattern=r'^/start$'))
+@verify_user(private=True)
+async def welcome(event: NewMessage.Event | Message):
+ await event.reply(
+ message=WelcomeText % {'first_name': event.sender.first_name},
+ buttons=[
  [
- [
- InlineKeyboardButton('Add to Channel', url=f'https://t.me/{Telegram.BOT_USERNAME}?startchannel&admin=post_messages+edit_messages+delete_messages')
- ]
+ Button.url('Add to Channel', f'https://t.me/{Telegram.BOT_USERNAME}?startchannel&admin=post_messages+edit_messages+delete_messages')
  ]
- )
+ ]
  )
 
-@TelegramBot.on_message(filters.command('info') & filters.private)
-@verify_user
-async def user_info(_, msg: Message):
- await msg.reply(text=f'`{msg.from_user}`', quote=True)
-
- filename = f'{msg.from_user.id}.json'
- async with async_open(filename, "w") as file:
- await file.write(f'{msg.from_user}')
-
- await msg.reply_document(filename)
- await async_rm(filename)
-
-@TelegramBot.on_message(filters.private & filters.command('log') & filters.user(Telegram.OWNER_ID))
-async def send_log(_, msg: Message):
- await msg.reply_document('event-log.txt', quote=True)
+@TelegramBot.on(NewMessage(incoming=True, pattern=r'^/info$'))
+@verify_user(private=True)
+async def user_info(event: Message):
+ await event.reply(UserInfoText.format(sender=event.sender))
 
-logger.info('Bot is now started!')
+@TelegramBot.on(NewMessage(chats=Telegram.OWNER_ID, incoming=True, pattern=r'^/log$'))
+async def send_log(event: NewMessage.Event | Message):
+ await event.reply(file='event-log.txt')

bot/plugins/deeplinks.py

@@ -1,21 +1,22 @@
-from pyrogram.types import Message
+from telethon.events import NewMessage
+from telethon.tl.custom import Message
+from bot import TelegramBot
+from bot.modules.decorators import verify_user
+from bot.modules.telegram import get_message, send_message
 from bot.modules.static import *
-from bot.modules.telegram import get_message
 
-async def deeplinks(msg: Message, payload: str):
- if payload.startswith('file_'):
- sp = payload.split('_')
+@TelegramBot.on(NewMessage(incoming=True, pattern=r'^/start file_'))
+@verify_user(private=True)
+async def send_file(event: NewMessage.Event | Message):
+ payload = event.raw_text.split()[-1].split('_')
 
-  if len(sp) != 3:
-  return await msg.reply(InvalidPayloadText, quote=True)
-  
-  message = await get_message(int(sp[1]))
+ if len(payload) != 3:
+ return await event.reply(InvalidPayloadText)
+
+ message = await get_message(int(payload[1]))
 
- if not message:
- return await msg.reply(MessageNotExist)
- if sp[2] != message.caption:
- return await msg.reply(InvalidPayloadText, quote=True)
-
- await message.copy(chat_id=msg.from_user.id, caption="")
- else:
- await msg.reply(InvalidPayloadText, quote=True)
+ if not message:
+ return await event.reply(MessageNotExist)
+
+ message.raw_text = ''
+ await send_message(message, send_to=event.chat_id)