feat(modules) add piston to run code snippets (#263)

locales/bot.pot

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PROJECT VERSION\n"
 "Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
-"POT-Creation-Date: 2024-09-07 12:44-0300\n"
+"POT-Creation-Date: 2024-09-07 14:24-0300\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <[email protected]>\n"
@@ -697,6 +697,50 @@ msgstr ""
 msgid "Failed to download the media."
 msgstr ""
 
+#: src/korone/modules/piston/handlers/langs.py:17
+#: src/korone/modules/piston/handlers/run.py:30
+msgid "Failed to fetch the available languages."
+msgstr ""
+
+#: src/korone/modules/piston/handlers/langs.py:20
+msgid "<b>Supported languages</b>:\n"
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:21
+msgid ""
+"You need to provide a command to run. Example: <code>/piston python "
+"print('Hello, World!')</code>"
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:35
+msgid ""
+"Invalid language. Use <code>/pistonlangs</code> to see the available "
+"languages."
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:44
+msgid "An error occurred while running the code."
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:47
+msgid ""
+"<b>Code</b>:\n"
+"<pre language='{lang}'>{code}</pre>\n"
+"\n"
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:52
+msgid ""
+"<b>Output</b>:\n"
+"<pre language='bash'>{output}</pre>\n"
+msgstr ""
+
+#: src/korone/modules/piston/handlers/run.py:57
+msgid ""
+"<b>Compiler Output</b>:\n"
+"<pre language='bash'>{output}</pre>"
+msgstr ""
+
 #: src/korone/modules/pm_menu/handlers/about.py:28
 msgid ""
 "Korone is a comprehensive and cutting-edge Telegram bot that offers a "

news/263.feature.rst

@@ -0,0 +1 @@
+Added the `/piston` command to run the Piston code evaluator. This command allows users to evaluate code snippets in various languages directly in the chat. For example, use `/piston python print("Hello, World!")` to run a Python snippet. Supported languages include Python, JavaScript, Ruby, and more (see `/pistonlangs` for a full list).

src/korone/modules/piston/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>

src/korone/modules/piston/handlers/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>

src/korone/modules/piston/handlers/langs.py

@@ -0,0 +1,22 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+from hydrogram import Client
+from hydrogram.types import Message
+
+from korone.decorators import router
+from korone.filters.command import Command
+from korone.modules.piston.utils.api import get_languages
+from korone.utils.i18n import gettext as _
+
+
+@router.message(Command("pistonlangs"))
+async def langs_handler(client: Client, message: Message) -> None:
+    languages = await get_languages()
+    if not languages:
+        await message.reply(_("Failed to fetch the available languages."))
+        return
+
+    text = _("<b>Supported languages</b>:\n")
+    text += "\n".join(f"- <code>{lang}</code>" for lang in languages)
+    await message.reply(text)

src/korone/modules/piston/handlers/run.py

@@ -0,0 +1,61 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+from hairydogm.chat_action import ChatActionSender
+from hydrogram import Client
+from hydrogram.enums import ChatAction
+from hydrogram.types import Message
+
+from korone.decorators import router
+from korone.filters import Command, CommandObject
+from korone.modules.piston.utils.api import create_request, get_languages, run_code
+from korone.utils.i18n import gettext as _
+
+
+@router.message(Command("piston"))
+async def piston_command(client: Client, message: Message) -> None:
+    command = CommandObject(message).parse()
+
+    if not command.args:
+        await message.reply(
+            _(
+                "You need to provide a command to run. "
+                "Example: <code>/piston python print('Hello, World!')</code>"
+            )
+        )
+        return
+
+    languages = await get_languages()
+    if not languages:
+        await message.reply(_("Failed to fetch the available languages."))
+        return
+
+    if command.args.split()[0] not in languages:
+        await message.reply(
+            _("Invalid language. Use <code>/pistonlangs</code> to see the available languages.")
+        )
+        return
+
+    async with ChatActionSender(client=client, chat_id=message.chat.id, action=ChatAction.TYPING):
+        request = create_request(command.args)
+        response = await run_code(request)
+
+        if response.result == "error":
+            await message.reply(_("An error occurred while running the code."))
+            return
+
+        text = _("<b>Code</b>:\n<pre language='{lang}'>{code}</pre>\n\n").format(
+            lang=command.args.split()[0], code=request.code
+        )
+
+        if response.output:
+            text += _("<b>Output</b>:\n<pre language='bash'>{output}</pre>\n").format(
+                output=response.output
+            )
+
+        if response.compiler_output:
+            text += _("<b>Compiler Output</b>:\n<pre language='bash'>{output}</pre>").format(
+                output=response.compiler_output
+            )
+
+        await message.reply(text, disable_web_page_preview=True)

src/korone/modules/piston/utils/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>

src/korone/modules/piston/utils/api.py

@@ -0,0 +1,78 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+import re
+from contextlib import asynccontextmanager
+from datetime import timedelta
+
+import httpx
+import orjson
+from cashews import NOT_NONE
+
+from korone.utils.caching import cache
+from korone.utils.logging import logger
+
+from .types import RunRequest, RunResponse
+
+STDIN_PATTERN = re.compile(r"\s/stdin\b")
+
+
+@asynccontextmanager
+async def get_async_client():
+    async with httpx.AsyncClient(http2=True, timeout=20) as client:
+        yield client
+
+
+@cache(ttl=timedelta(weeks=1), condition=NOT_NONE)
+async def get_languages() -> list[str] | None:
+    url = "https://emkc.org/api/v2/piston/runtimes"
+    async with get_async_client() as client:
+        try:
+            response = await client.get(url)
+            response.raise_for_status()
+            languages_map = response.json()
+            language_set = {entry["language"] for entry in languages_map}
+            return sorted(language_set)
+        except (httpx.HTTPStatusError, Exception) as err:
+            await logger.aerror("[Piston] Error fetching languages: %s", err)
+            return None
+
+
+def create_request(text: str) -> RunRequest:
+    text = text.strip()
+    try:
+        lang, code = text.split(" ", 1)
+    except ValueError as err:
+        msg = "Input must contain both language and code."
+        raise ValueError(msg) from err
+
+    code = code.lstrip()
+    stdin = None
+
+    if stdin_match := STDIN_PATTERN.search(code):
+        start, end = stdin_match.span()
+        stdin = code[end + 1 :].strip()
+        code = code[:start].strip()
+
+    if not code:
+        msg = "Bad query: Code is empty."
+        raise ValueError(msg)
+
+    return RunRequest(language=lang, code=code, stdin=stdin)
+
+
+@cache(ttl=timedelta(weeks=1), condition=NOT_NONE)
+async def run_code(request: RunRequest) -> RunResponse:
+    url = "https://emkc.org/api/v2/piston/execute"
+    json_body = orjson.dumps(request.to_dict())
+    async with get_async_client() as client:
+        try:
+            response = await client.post(
+                url, content=json_body, headers={"Content-Type": "application/json"}
+            )
+            response.raise_for_status()
+            data = response.json()
+            return RunResponse.from_api_response(data)
+        except (httpx.HTTPStatusError, Exception) as err:
+            await logger.aerror("[Piston] Error running code: %s", err)
+            return RunResponse(result="error", output=str(err))

src/korone/modules/piston/utils/types.py

@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class RunRequest:
+    language: str
+    code: str | None = None
+    stdin: str | None = None
+    version: str = "*"
+
+    def to_dict(self) -> dict:
+        return {
+            "language": self.language,
+            "version": self.version,
+            "files": self.code,
+            "stdin": self.stdin or "",
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class RunResponse:
+    result: str
+    output: str | None = None
+    compiler_output: str | None = None
+
+    @staticmethod
+    def from_api_response(data: dict) -> RunResponse:
+        return RunResponse(
+            result="success",
+            output=data.get("run", {}).get("output", None),
+            compiler_output=data.get("compile", {}).get("output", None),
+        )