telegram python execution bot

telegram_python_execution_bot.py

import asyncio
import datetime
import os
import sys
import traceback
from typing import Tuple

from telethon.sync import TelegramClient, events


api_id = [int]
api_hash = "[str]"
bot_token = "[str]"
allowed_id_list = [12345, 678]

# proxy = ("http", "127.0.0.1", 1234)
proxy = None

# We have to manually call "start" if we want an explicit bot token
bot = TelegramClient("bot", api_id, api_hash, proxy=proxy).start(bot_token=bot_token)


os.makedirs("run", exist_ok=True)


async def run_command(source: str, sender_id: int) -> Tuple[str, str]:
    file_time = datetime.datetime.strftime(
        datetime.datetime.now(), "%Y-%m-%d-%H_%M_%S_%f"
    )
    file_name = f"run/{sender_id}_{file_time}.py"
    with open(file_name, "w", encoding="utf-8") as file:
        file.write(source)
    process = await asyncio.create_subprocess_exec(
        sys.executable,
        file_name,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    out, err = await process.communicate()
    await process.wait()
    return out.decode(), err.decode()


with bot:

    @bot.on(events.NewMessage(pattern="exec.*"))
    async def handler(event):
        if event.sender.id not in allowed_id_list:
            print(event.sender.id)
            return
        await event.respond("⚙️ Executing...")

        try:
            out, err = await run_command(
                event.text.replace("exec", ""), event.sender.id
            )
            await event.respond(out + err)
        except:
            await event.respond("❌ Error!\n```" + traceback.format_exc() + "```")
        else:
            await event.respond("✅ Done!")

    bot.run_until_disconnected()