Add code execution with Piston

bot/bot.py

@@ -21,7 +21,7 @@
 
 class Bot(commands.Bot):
     # If not empty then only these will be loaded. Good for local debugging. If empty all found are loaded.
-    allowed_extensions = ("tortoise_dm", "github", "security", "health")
+    allowed_extensions = ("tortoise_dm", "github", "security", "health", "piston")
     banned_extensions = ("advent_of_code",)
     build_version = "mystery-build"
 

bot/cogs/piston.py

@@ -0,0 +1,156 @@
+import aiohttp
+import discord
+from discord.ext import commands
+from datetime import datetime, timedelta
+from typing import Dict
+from bot.utils.embed_handler import code_eval_embed
+
+PISTON_URL = "https://emkc.org/api/v2/piston/execute"
+
+LANG_ALIASES = {
+    "py": "python",
+    "python": "python",
+    "js": "javascript",
+    "javascript": "javascript",
+    "ts": "typescript",
+    "c": "c",
+    "cpp": "cpp",
+    "c++": "cpp",
+    "java": "java",
+    "rs": "rust",
+    "rust": "rust",
+    "go": "go",
+}
+
+
+class Piston(commands.Cog):
+    """Piston-style code execution cog with edit listening."""
+
+    def __init__(self, bot: commands.Bot):
+        self.bot = bot
+        self.session = aiohttp.ClientSession()
+
+        self.tracked: Dict[int, dict] = {}
+
+    def cog_unload(self):
+        self.bot.loop.create_task(self.session.close())
+
+    def _parse_block(self, content: str):
+        if not content.startswith("/run"):
+            return None
+
+        if "```" not in content:
+            return None
+
+        try:
+            header, block = content.split("```", 1)
+            block_content = block.split("```", 1)[0]
+
+            first_line, *rest = block_content.split("\n")
+            lang = first_line.strip().lower()
+            code = "\n".join(rest)
+
+            return lang, code
+        except Exception:
+            return None
+
+    async def _execute(self, language: str, code: str):
+        payload = {
+            "language": language,
+            "version": "*",
+            "files": [{"name": "main", "content": code}],
+            "stdin": ""
+        }
+
+        async with self.session.post(PISTON_URL, json=payload, timeout=30) as resp:
+            return await resp.json()
+
+
+    async def _send_result(self, channel: discord.TextChannel, result: dict, language: str):
+        if "run" not in result:
+            return await channel.send("Execution failed: invalid response")
+
+        run = result["run"]
+        stdout = run.get("stdout", "")
+        stderr = run.get("stderr", "")
+        output = stdout + ("\n" + stderr if stderr else "")
+
+        embed = code_eval_embed(language, output)
+        return await channel.send(embed=embed)
+
+
+    @commands.Cog.listener()
+    async def on_message(self, message: discord.Message):
+        if message.author.bot:
+            return
+
+        parsed = self._parse_block(message.content)
+        if not parsed:
+            return
+
+        lang, code = parsed
+        lang = LANG_ALIASES.get(lang, lang)
+
+        result = await self._execute(lang, code)
+        bot_msg = await self._send_result(message.channel, result, lang)
+
+        # track message for edits
+        self.tracked[message.id] = {
+            "created": datetime.utcnow(),
+            "lang": lang,
+            "bot_msg_id": bot_msg.id
+        }
+
+    @commands.Cog.listener()
+    async def on_message_edit(self, before: discord.Message, after: discord.Message):
+        if after.author.bot:
+            return
+
+        meta = self.tracked.get(after.id)
+        if not meta:
+            return
+
+        # 2 minute edit window
+        if datetime.utcnow() - meta["created"] > timedelta(minutes=2):
+            self.tracked.pop(after.id, None)
+            return
+
+        parsed = self._parse_block(after.content)
+        if not parsed:
+            return
+
+        lang, code = parsed
+        lang = LANG_ALIASES.get(lang, lang)
+
+        try:
+            result = await self._execute(lang, code)
+        except Exception:
+            return
+
+        channel = after.channel
+        try:
+            bot_msg = await channel.fetch_message(meta["bot_msg_id"])
+        except Exception:
+            return
+
+        # update embed
+        if "run" not in result:
+            return
+
+        run = result["run"]
+        stdout = run.get("stdout", "")
+        stderr = run.get("stderr", "")
+        output = stdout + ("\n" + stderr if stderr else "")
+
+        if not output:
+            output = "(no output)"
+
+        if len(output) > 1900:
+            output = output[:1900] + "\n... (truncated)"
+
+        embed = code_eval_embed(lang, output, edited=True)
+        await bot_msg.edit(embed=embed)
+
+
+async def setup(bot: commands.Bot):
+    await bot.add_cog(Piston(bot))

bot/constants.py

@@ -37,6 +37,14 @@
 general_channel_id = 577192344533598472
 staff_channel_id = 580809054067097600
 
+#Tortoise Guild channels
+leetcode_channel_id = 726403782740541470
+bot_cmd_channel_id = 581726653710073858
+project_showcase_channel_id = 581156991557304330
+resources_channel_id = 577195878620725251
+challenge_submission_channel_id = 780842875901575228
+challenge_discussion_channel_id = 781129674860003336
+
 # Roles
 muted_role_id = 707007421066772530
 verified_role_id = 599647985198039050
@@ -306,4 +314,14 @@ class SuggestionStatus(Enum):
     "rtf",
 )
 
-rate_limit_minutes = 10
+rate_limit_minutes = 10
+
+defcon_lockable_channels = [
+    general_channel_id,
+    leetcode_channel_id,
+    bot_cmd_channel_id,
+    project_showcase_channel_id,
+    resources_channel_id,
+    challenge_discussion_channel_id,
+    challenge_submission_channel_id
+]

bot/utils/embed_handler.py

@@ -401,3 +401,19 @@ def moderation_log_embed(
     embed.add_field(name="Channel", value=channel, inline=False)
     embed.add_field(name="Content", value=content[:1024], inline=False)
     return embed
+
+def code_eval_embed(language: str, output: str, *, edited: bool = False) -> Embed:
+    title = "Execution Result (edited)" if edited else "Execution Result"
+    color = Color.orange() if edited else Color.blurple()
+
+    if not output:
+        output = "(no output)"
+
+    if len(output) > 1900:
+        output = output[:1900] + "\n... (truncated)"
+
+    embed = Embed(title=title, color=color)
+    embed.add_field(name="Language", value=language, inline=True)
+    embed.add_field(name="Output", value=f"```{output}```", inline=False)
+    embed.set_footer(text="Powered by EKMC Piston")
+    return embed