# ruff: noqa: F821 from __future__ import annotations import hashlib import json import time from typing import TYPE_CHECKING if TYPE_CHECKING: from app import App from loader import loader from router import Event app: App HELPER = "0123456789abcdef0123456789abcdef" TEXT = {"text": {"type": "string", "min": 1, "max": 800, "required": True}} class Module: handler_timeout = 120 pages = [{ "id": "main", "title": {"ru": "Мастерская", "en": "Workshop"}, "elements": [ {"id": "status", "type": "text"}, {"id": "runs", "type": "metric", "label": {"ru": "Задачи", "en": "Tasks"}}, {"id": "execute", "type": "form", "action": "execute", "label": {"ru": "Проверить", "en": "Run checks"}}, {"id": "filecheck", "type": "form", "action": "filecheck", "label": {"ru": "Проверить файл", "en": "Check file"}}, {"id": "fetch", "type": "form", "action": "fetch", "label": {"ru": "Проверить адрес", "en": "Check URL"}}, {"id": "schedule", "type": "button", "action": "schedule", "label": {"ru": "Запустить расписание", "en": "Start schedule"}}, {"id": "cancel", "type": "button", "action": "cancel", "label": {"ru": "Остановить расписание", "en": "Stop schedule"}}, ], }] async def init(self) -> None: await self.publish() async def publish(self) -> None: await app.ui.set("main", { "status": await app.db.get("status", default="Ready"), "runs": await app.db.get("runs", default=0), }) async def process(self, text: str) -> dict: raw = text.encode("utf-8") result = await app.modules.call(HELPER, "digest", {"text": text}) if result["sha256"] != hashlib.sha256(raw).hexdigest(): raise ValueError("Invalid helper result") file = await app.files.upload("report.txt", raw, ttl=3600) downloaded = None try: restored = b"".join([part async for part in app.files.chunks(file["id"])]) if restored != raw: raise ValueError("File contents changed") response = await app.web.request("https://postman-echo.com/get?module=workshop") if response["status"] != 200: raise ValueError("The echo service is unavailable") if json.loads(response["body"])["args"]["module"] != "workshop": raise ValueError("Invalid echo response") uploaded = await app.web.upload("https://postman-echo.com/post", file["id"], headers={"Content-Type": "text/plain; charset=utf-8"}) if uploaded["status"] != 200 or json.loads(uploaded["body"])["data"] != text: raise ValueError("The uploaded file was not returned") downloaded = await app.web.download("https://postman-echo.com/get?file=workshop", "response.json", maximum=8192) content = b"".join([part async for part in app.files.chunks(downloaded["id"])]) if json.loads(content)["args"]["file"] != "workshop": raise ValueError("Invalid downloaded file") await app.modules.emit(HELPER, "received", {"text": text}) await app.db.set("status", f"Checked {len(raw)} bytes") await self.publish() return {"sha256": result["sha256"], "bytes": len(raw)} finally: if downloaded: await app.files.delete(downloaded["id"]) await app.files.delete(file["id"]) @loader.action(access="owner") async def cancel(self, ctx: Event) -> dict: id: str | None = await app.db.get("task") if not id: return {"state": "absent"} task = await app.tasks.cancel(id) await app.tasks.delete(id) await app.db.delete("task") return {"state": task["state"]} @loader.action(fields=TEXT, access="owner") async def execute(self, ctx: Event) -> dict: try: return await self.process(ctx.values["text"]) except (RuntimeError, ValueError, TimeoutError) as error: return {"error": str(error)} @loader.action(fields={"size": {"type": "integer", "min": 0, "max": 200000, "required": True}}, access="owner") async def filecheck(self, ctx: Event) -> dict: raw = ("Файл 🌌\n".encode() * 20000)[:ctx.values["size"]] file = await app.files.upload("chunks.bin", raw) try: restored = b"".join([part async for part in app.files.chunks(file["id"])]) if restored != raw: raise ValueError("File contents changed") listed = await app.files.list() if not any(item["id"] == file["id"] for item in listed["files"]): raise ValueError("File not listed") return {"bytes": len(restored), "sha256": hashlib.sha256(restored).hexdigest()} finally: await app.files.delete(file["id"]) @loader.action(fields={"url": {"type": "string", "min": 1, "max": 2048, "required": True}}, access="owner") async def fetch(self, ctx: Event) -> dict: try: response = await app.web.request(ctx.values["url"]) return {"status": response["status"], "body": response["body"]} except RuntimeError as error: return {"error": str(error)} @loader.callback(access="owner") async def pressed(self, ctx: Event) -> None: try: result = await self.process("Button check") await ctx.edit(f"Checked {result['bytes']} bytes") await ctx.answer("Done") except (RuntimeError, ValueError, TimeoutError): await ctx.answer("The check failed", alert=True) @loader.action(access="owner") async def schedule(self, ctx: Event) -> dict: id: str | None = await app.db.get("task") if id: tasks = await app.tasks.list() existing = next((task for task in tasks if task["id"] == id), None) if existing and existing["state"] in ("pending", "running"): return existing if existing: await app.tasks.delete(id) await app.db.delete("task") task = await app.tasks.create("tick", int(time.time()) + 3, interval=5) await app.db.set("task", task["id"]) return task @loader.task async def tick(self, ctx: Event) -> int: count: int = await app.db.get("runs", default=0) + 1 await app.db.set("runs", count) await self.publish() return count @loader.command("workshop", access="owner", source="both") async def workshop(self, ctx: Event) -> None: buttons = [[{"text": "Run checks", "callback": "pressed", "once": True}]] await app.inline.form(ctx.chat_id, "Files, network and module calls", buttons, users=[ctx.sender_id])