Update plugins/devtools.py

plugins/devtools.py

@@ -2,7 +2,7 @@
 # Copyright (C) 2021-2025 TeamUltroid
 #
 # This file is a part of < https://github.com/TeamUltroid/Ultroid/ >
-# Please read the GNU Affero General Public License in
+# PLease read the GNU Affero General Public License in
 # <https://www.github.com/TeamUltroid/Ultroid/blob/main/LICENSE/>.
 
 from . import get_help
@@ -12,130 +12,40 @@ __doc__ = get_help("help_devtools")
 import inspect
 import sys
 import traceback
-import time
-import json
 from io import BytesIO, StringIO
 from os import remove
 from pprint import pprint
-from random import choice
-from html import escape
 
 from telethon.utils import get_display_name
-from telethon.tl import functions
 
-# project imports (assumed available in Ultroid context)
 from pyUltroid import _ignore_eval
-from . import *  # noqa: F401,F403
-from . import upload_file as uf  # upload helper used to upload Carbon images
 
-# Optional imports: keep original fallbacks
+from . import *
+
+# Used for Formatting Eval Code, if installed
 try:
     import black
-except Exception:
+except ImportError:
     black = None
+from random import choice
 
 try:
     from yaml import safe_load
-except Exception:
-    from pyUltroid.fns.tools import safe_load  # fallback from project
-
-fn = functions
-
-pp = pprint  # alias kept from original
-bot = ultroid = ultroid_bot  # ensure these names exist in project
+except ImportError:
+    from pyUltroid.fns.tools import safe_load
 
-class u:
-    _ = ""
-
-# ---------------------------------------------------------------------
-# Helper: parse eval-friendly representation of values
-# (kept mostly as original)
-def _parse_eval(value=None):
-    if not value:
-        return value
-    if hasattr(value, "stringify"):
-        try:
-            return value.stringify()
-        except TypeError:
-            pass
-    elif isinstance(value, dict):
-        try:
-            return json.dumps(value, indent=1, default=str)
-        except BaseException:
-            pass
-    elif isinstance(value, list):
-        newlist = "["
-        for index, child in enumerate(value):
-            newlist += "\n  " + str(_parse_eval(child))
-            if index < len(value) - 1:
-                newlist += ","
-        newlist += "\n]"
-        return newlist
-    return str(value)
-
-# ---------------------------------------------------------------------
-# Exec wrapper: runs user-supplied async code in custom namespace.
-# WARNING: This runs arbitrary code via exec() — ensure only trusted users can use it.
-async def aexec(code: str, event):
-    """
-    Execute `code` (containing Python source for the body of an async function)
-    as an async function __aexec(e, client) and run it.
-
-    Returns whatever the executed function returns.
-
-    NOTE: This function intentionally exposes a number of objects to the executed
-    namespace (client, event, reply, etc.). This is powerful and dangerous — keep
-    access restricted (only owner/devs).
-    """
-    # Precompute reply message (do this before creating exec_globals)
-    try:
-        reply_msg = await event.get_reply_message()
-    except Exception:
-        reply_msg = None
-
-    # Dedicated namespace for execution. Keep builtins as in original, but this is
-    # a security risk — consider limiting builtins for untrusted usage.
-    exec_globals = {
-        "print": _stringify,
-        "p": _stringify,
-        "message": event,
-        "event": event,
-        "client": event.client,
-        "reply": reply_msg,
-        "chat": event.chat_id,
-        "u": u,
-        "__builtins__": __builtins__,
-        "__name__": __name__,
-    }
+from . import upload_file as uf
+from telethon.tl import functions
 
-    # Wrap the provided code inside an async def so we can await it.
-    # We indent every line of `code` and place it under the function body.
-    wrapped_lines = []
-    for line in code.splitlines():
-        # Ensure even empty lines are indented properly
-        if line.strip() == "":
-            wrapped_lines.append("    pass")
-        else:
-            wrapped_lines.append("    " + line)
-    wrapped_code = "async def __aexec(e, client):\n" + "\n".join(wrapped_lines)
+fn = functions
 
-    try:
-        # Execute the wrapper to create __aexec in exec_globals
-        exec(wrapped_code, exec_globals)
-        func = exec_globals["__aexec"]
-        # Call the newly created coroutine function
-        return await func(event, event.client)
-    except Exception as exc:
-        # Re-raise with clearer context (caller will handle formatting)
-        raise Exception(f"Failed to execute code: {str(exc)}")
 
-# ---------------------------------------------------------------------
-# Commands (kept behaviorally similar to original with blockquote HTML wrapping)
-@ultroid_cmd(pattern="sysinfo$")
+@ultroid_cmd(
+    pattern="sysinfo$",
+)
 async def _(e):
     xx = await e.eor(get_string("com_1"))
-    # run neofetch and strip ANSI colours; results appended to neo.txt
-    x, y = await bash("neofetch|sed 's/\\x1B\\[[0-9;\\?]*[a-zA-Z]//g' >> neo.txt")
+    x, y = await bash("neofetch|sed 's/\x1B\\[[0-9;\\?]*[a-zA-Z]//g' >> neo.txt")
     if y and y.endswith("NOT_FOUND"):
         return await xx.edit(f"Error: `{y}`")
     with open("neo.txt", "r", encoding="utf-8") as neo:
@@ -148,6 +58,7 @@ async def _(e):
         await xx.delete()
     remove("neo.txt")
 
+
 @ultroid_cmd(pattern="bash", fullsudo=True, only_devs=True)
 async def _(event):
     carb, rayso, yamlf = None, None, False
@@ -161,7 +72,6 @@ async def _(event):
             rayso = True
     except IndexError:
         return await event.eor(get_string("devs_1"), time=10)
-
     xx = await event.eor(get_string("com_1"))
     reply_to_id = event.reply_to_msg_id or event.id
     stdout, stderr = await bash(cmd, run_code=1)
@@ -170,7 +80,6 @@ async def _(event):
     if stderr:
         err = f"**• ERROR:** \n`{stderr}`\n\n"
     if stdout:
-        # When Carbon/Rayso output is requested and allowed by chat permissions:
         if (carb or udB.get_key("CARBON_ON_BASH")) and (
             event.is_private
             or event.chat.admin_rights
@@ -215,7 +124,6 @@ async def _(event):
             out = "**• OUTPUT:**"
             remove(li)
         else:
-            # treat pip-like keyed output as YAML and pretty-format
             if "pip" in cmd and all(":" in line for line in stdout.split("\n")):
                 try:
                     load = safe_load(stdout)
@@ -234,9 +142,7 @@ async def _(event):
             out = f"**• OUTPUT:**\n{stdout}"
     if not stderr and not stdout:
         out = "**• OUTPUT:**\n`Success`"
-
     OUT += err + out
-
     if len(OUT) > 4096:
         ultd = err + out
         with BytesIO(str.encode(ultd)) as out_file:
@@ -253,14 +159,40 @@ async def _(event):
 
             await xx.delete()
     else:
-        # wrap in HTML blockquote and escape unsafe chars
-        formatted_out = f"<blockquote>{escape(OUT)}</blockquote>"
-        # Telethon accepts parse_mode='html' in edit calls in recent versions
-        await xx.edit(formatted_out, parse_mode="html", link_preview=False)
+        await xx.edit(OUT, link_preview=not yamlf)
+
 
+pp = pprint  # ignore: pylint
+bot = ultroid = ultroid_bot
+
+
+class u:
+    _ = ""
+
+
+def _parse_eval(value=None):
+    if not value:
+        return value
+    if hasattr(value, "stringify"):
+        try:
+            return value.stringify()
+        except TypeError:
+            pass
+    elif isinstance(value, dict):
+        try:
+            return json_parser(value, indent=1)
+        except BaseException:
+            pass
+    elif isinstance(value, list):
+        newlist = "["
+        for index, child in enumerate(value):
+            newlist += "\n  " + str(_parse_eval(child))
+            if index < len(value) - 1:
+                newlist += ","
+        newlist += "\n]"
+        return newlist
+    return str(value)
 
-# alias kept
-pp = pprint
 
 @ultroid_cmd(pattern="eval", fullsudo=True, only_devs=True)
 async def _(event):
@@ -268,7 +200,6 @@ async def _(event):
         cmd = event.text.split(maxsplit=1)[1]
     except IndexError:
         return await event.eor(get_string("devs_2"), time=5)
-
     xx = None
     mode = ""
     spli = cmd.split()
@@ -297,18 +228,13 @@ async def _(event):
         return
     if not mode == "silent" and not xx:
         xx = await event.eor(get_string("com_1"))
-
-    # attempt to format with black if available (not critical)
     if black:
         try:
             cmd = black.format_str(cmd, mode=black.Mode())
-        except Exception:
-            # ignore formatting failures
+        except BaseException:
+            # Consider it as Code Error, and move on to be shown ahead.
             pass
-
-    reply_to_id = event.reply_to_msg_id or event.id
-
-    # SECURITY: detect and report potentially dangerous commands
+    reply_to_id = event.reply_to_msg_id or event
     if any(item in cmd for item in KEEP_SAFE().All) and (
         not (event.out or event.sender_id == ultroid_bot.uid)
     ):
@@ -320,13 +246,11 @@ async def _(event):
         return await xx.edit(
             "`Malicious Activities suspected⚠️!\nReported to owner. Aborted this request!`"
         )
-
-    # Capture stdout/stderr
     old_stderr = sys.stderr
     old_stdout = sys.stdout
     redirected_output = sys.stdout = StringIO()
     redirected_error = sys.stderr = StringIO()
-    value, exc = None, None
+    stdout, stderr, exc, timeg = None, None, None, None
     tima = time.time()
     try:
         value = await aexec(cmd, event)
@@ -338,18 +262,18 @@ async def _(event):
     stderr = redirected_error.getvalue()
     sys.stdout = old_stdout
     sys.stderr = old_stderr
-
     if value:
         try:
             if mode == "gsource":
                 exc = inspect.getsource(value)
             elif mode == "g-args":
                 args = inspect.signature(value).parameters.values()
-                name = getattr(value, "__name__", "")
+                name = ""
+                if hasattr(value, "__name__"):
+                    name = value.__name__
                 exc = f"**{name}**\n\n" + "\n ".join([str(arg) for arg in args])
         except Exception:
             exc = traceback.format_exc()
-
     evaluation = exc or stderr or stdout or _parse_eval(value) or get_string("instu_4")
     if mode == "silent":
         if exc:
@@ -359,10 +283,11 @@ async def _(event):
             if len(msg) > 4000:
                 with BytesIO(msg.encode()) as out_file:
                     out_file.name = "Eval-Error.txt"
-                return await event.client.send_message(log_chat, f"`{cmd}`", file=out_file)
+                return await event.client.send_message(
+                    log_chat, f"`{cmd}`", file=out_file
+                )
             await event.client.send_message(log_chat, msg, parse_mode="html")
         return
-
     tmt = tima * 1000
     timef = time_formatter(tmt)
     timeform = timef if not timef == "0s" else f"{tmt:.3f}ms"
@@ -385,19 +310,48 @@ async def _(event):
                 reply_to=reply_to_id,
             )
         return await xx.delete()
+    await xx.edit(final_output)
 
-    # wrap eval output in HTML blockquote and escape unsafe chars
-    formatted_eval = f"<blockquote>{escape(final_output)}</blockquote>"
-    await xx.edit(formatted_eval, parse_mode="html")
 
-# Small helper used by exec namespace
 def _stringify(text=None, *args, **kwargs):
     if text:
         u._ = text
         text = _parse_eval(text)
     return print(text, *args, **kwargs)
 
-# ---------------------------------------------------------------------
+
+async def aexec(code, event):
+    # Create a dedicated namespace for execution
+    exec_globals = {
+        'print': _stringify,
+        'p': _stringify,
+        'message': event,
+        'event': event,
+        'client': event.client,
+        'reply': await event.get_reply_message(),
+        'chat': event.chat_id,
+        'u': u,
+        '__builtins__': __builtins__,
+        '__name__': __name__
+    }
+    
+    # Format the async function definition
+    wrapped_code = (
+        'async def __aexec(e, client):\n' +
+        '\n'.join(f'    {line}' for line in code.split('\n'))
+    )
+    
+    try:
+        # Execute the wrapped code in our custom namespace
+        exec(wrapped_code, exec_globals)
+        # Get the defined async function
+        func = exec_globals['__aexec']
+        # Execute it with proper parameters
+        return await func(event, event.client)
+    except Exception as e:
+        raise Exception(f"Failed to execute code: {str(e)}")
+
+
 DUMMY_CPP = """#include <iostream>
 using namespace std;
 
@@ -406,6 +360,7 @@ int main(){
 }
 """
 
+
 @ultroid_cmd(pattern="cpp", only_devs=True)
 async def doie(e):
     match = e.text.split(" ", maxsplit=1)
@@ -423,9 +378,9 @@ async def doie(e):
     if m[1]:
         o_cpp += f"\n\n**• Error :**\n`{m[1]}`"
         if len(o_cpp) > 3000:
-            remove("cpp-ultroid.cpp")
+            os.remove("cpp-ultroid.cpp")
             if os.path.exists("CppUltroid"):
-                remove("CppUltroid")
+                os.remove("CppUltroid")
             with BytesIO(str.encode(o_cpp)) as out_file:
                 out_file.name = "error.txt"
                 return await msg.reply(f"`{match}`", file=out_file)
@@ -440,8 +395,6 @@ async def doie(e):
             out_file.name = "eval.txt"
             await msg.reply(f"`{match}`", file=out_file)
     else:
-        # wrap cpp output in HTML blockquote and escape unsafe chars
-        formatted_cpp = f"<blockquote>{escape(o_cpp)}</blockquote>"
-        await eor(msg, formatted_cpp, parse_mode="html")
-    remove("CppUltroid")
-    remove("cpp-ultroid.cpp")
+        await eor(msg, o_cpp)
+    os.remove("CppUltroid")
+    os.remove("cpp-ultroid.cpp")