DragMusic/plugins/kuri/devtools.py

"""
Devtools plugin for userbot-only development and admin commands.
Provides /sh, /xsh for shell commands and /eval, /e for Python code evaluation.
Also triggers if the message starts with 'sh', 'xsh', 'e', or 'eval' (with or without a slash or exclamation mark, and with any capitalization).
Restricted to OWNER_ID only. Usable in any chat where the userbot is present.
"""

import asyncio
import re
from pyrogram import filters
from pyrogram.types import Message
from DragMusic import userbot
from config import OWNER_ID

# Regex filters for command matching (case-insensitive)
sh_filter = filters.user(OWNER_ID) & filters.regex(r"^[!/]*x?sh( |$)", re.IGNORECASE)
eval_filter = filters.user(OWNER_ID) & filters.regex(r"^[!/]*(e|eval)( |$)", re.IGNORECASE)

@userbot.one.on_message(sh_filter)
async def shell_command_handler(client, message: Message):
    print("sh/xsh command received from OWNER_ID")  # Debug print
    parts = message.text.split(None, 1)
    if len(parts) < 2:
        await message.reply_text("Usage: sh <command>")
        return
    cmd = parts[1]
    try:
        process = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await process.communicate()
        output = (stdout + stderr).decode().strip()
        if not output:
            output = "No output."
        # Telegram message limit
        if len(output) > 4096:
            for i in range(0, len(output), 4096):
                await message.reply_text(f"<code>{output[i:i+4096]}</code>")
        else:
            await message.reply_text(f"<code>{output}</code>")
    except Exception as e:
        await message.reply_text(f"Error: <code>{e}</code>")

@userbot.one.on_message(eval_filter)
async def eval_handler(client, message: Message):
    print("eval command received from OWNER_ID")  # Debug print
    parts = message.text.split(None, 1)
    if len(parts) < 2:
        await message.reply_text("Usage: eval <python code>")
        return
    code = parts[1]
    try:
        env = {
            "client": client,
            "message": message,
            "userbot": userbot,
            "asyncio": asyncio,
        }
        exec(
            f"async def __aexec(client, message):\n"
            + "\n".join(f"    {l}" for l in code.split("\n")),
            env,
        )
        result = await env["__aexec"](client, message)
        if result is not None:
            await message.reply_text(f"<code>{result}</code>")
    except Exception as e:
        await message.reply_text(f"Error: <code>{e}</code>")

# Add more dev/admin commands below as needed