eval

bot.py

@@ -1,10 +1,13 @@
+import contextlib
+import io
+
 from telegram import Update, InputFile
 from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler, filters, InlineQueryHandler, \
     CallbackContext, CallbackQueryHandler
 import logging
 
 from tgbot.shit.fed import build_fediverse_inline_results, extract_content
-from tgbot.config import TOKEN, PROMPTING_USERS
+from tgbot.config import TOKEN, PROMPTING_USERS, EXEC_UIDS
 from tgbot.shit.handlers import handle_xitter, handle_red_ebalo, handle_hentai, handle_cute_button
 from tgbot.shit.prompting import gen_image
 from tgbot.shit.render import render_text_on_image, image_to_string, encode_text, decode_text
@@ -118,6 +121,51 @@ async def handle_inline(update: Update, context: ContextTypes.DEFAULT_TYPE):
 
     await handle_red_ebalo(update, context)
 
+async def handle_eval(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    user_id = update.effective_user.id
+    if user_id not in EXEC_UIDS:
+        await update.message.reply_text("Eh")
+        return
+
+    code = update.message.text.removeprefix("/eval").strip()
+    if not code:
+        await update.message.reply_text("✍️ Send some Python code to evaluate.")
+        return
+
+    # Redirect stdout to capture output
+    stdout = io.StringIO()
+
+    # Create an isolated async context for execution
+    async def run_code():
+        local_vars = {}
+        code_wrapped = f"async def __eval_func():\n"
+        for line in code.splitlines():
+            code_wrapped += "    " + line + "\n"
+
+        try:
+            exec(code_wrapped, {}, local_vars)
+            with contextlib.redirect_stdout(stdout):
+                r = await local_vars["__eval_func"]()
+            return r
+        except Exception as e:
+            return f"❌ {e}"
+
+    # Run the async function and get the output
+    result = await run_code()
+    output = stdout.getvalue()
+
+    if result is not None:
+        output += f"\n➡️ {result}"
+
+    if not output.strip():
+        output = "✅ Done."
+
+    # Send output (truncate if needed)
+    MAX_LEN = 4000
+    if len(output) > MAX_LEN:
+        output = output[:MAX_LEN] + "\n... (truncated)"
+
+    await update.message.reply_text(f"<pre>{output.strip()}</pre>", parse_mode="HTML")
 
 def main():
     application = ApplicationBuilder().token(TOKEN).build()
@@ -135,6 +183,8 @@ def main():
     txt_file_filter = filters.Document.FileExtension("txt")
     application.add_handler(MessageHandler(txt_file_filter, handle_txt_upload))
 
+    application.add_handler(CommandHandler("eval", handle_eval))
+
     application.run_polling()