| """Tiny TG bot: reacts to /start (and a couple of helper commands). |
| |
| Pure stdlib \u2014 no aiogram/python-telegram-bot dependency. The panel |
| spawns it as a subprocess (sees run.py) and injects BOT_TOKEN. |
| |
| Why stdlib only: the panel's requirements.txt does not include |
| aiogram, and a multi-tenant subprocess model means each bot would |
| otherwise need its own venv. ``urllib.request`` + json is enough |
| for the Bot API. Long-polling handles the event loop. |
| """ |
| import json |
| import os |
| import sys |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
|
|
|
|
| TOKEN = (os.environ.get("BOT_TOKEN") or "").strip() |
| if not TOKEN: |
| sys.stderr.write("FATAL: BOT_TOKEN env var is empty\n") |
| sys.exit(2) |
|
|
| API = "https://api.telegram.org/bot" + TOKEN |
|
|
|
|
| def api(method: str, **params): |
| """Minimal Bot API wrapper. Returns parsed JSON dict. |
| |
| Raises on HTTP errors so the loop's except can log + back off. |
| """ |
| data = urllib.parse.urlencode(params).encode("utf-8") |
| req = urllib.request.Request(API + "/" + method, data=data, |
| headers={"Content-Type": "application/x-www-form/x-www-form-urlencoded"}) |
| with urllib.request.urlopen(req, timeout=35) as resp: |
| body = resp.read() |
| return json.loads(body) |
|
|
|
|
| def reply(chat_id: int, text: str): |
| try: |
| return api("sendMessage", chat_id=chat_id, text=text, |
| parse_mode="HTML", disable_web_page_preview=True) |
| except urllib.error.HTTPError as exc: |
| |
| sys.stderr.write(f"sendMessage HTTP {exc.code}: {exc.read()[:200]!r}\n") |
| return None |
|
|
|
|
| def help_text() -> str: |
| return ( |
| "<b>MMaxAI micro bot</b>\n" |
| "/start \u2014 say hi\n" |
| "/help \u2014 this list\n" |
| "/ping \u2014 pong\n" |
| "/echo \u2014 echo back what you type\n" |
| "/id \u2014 show chat id + msg id\n" |
| ) |
|
|
|
|
| def handle(text: str, chat_id: int, msg_id: int, sender: str | None) -> None: |
| cmd = text.split(maxsplit=1)[0].lower() if text else "" |
| if cmd == "/start": |
| reply(chat_id, f"\U0001f44b Привіт, <b>{sender or 'друже'}</b>!\n" + help_text()) |
| elif cmd == "/help": |
| reply(chat_id, help_text()) |
| elif cmd == "/ping": |
| reply(chat_id, "pong \U0001f3d3") |
| elif cmd == "/id": |
| reply(chat_id, f"chat_id = <code>{chat_id}</code>\nmsg_id = <code>{msg_id}</code>") |
| elif cmd == "/echo": |
| body = text.split(maxsplit=1, )[1] if " " in text else "" |
| reply(chat_id, body or "(empty)") |
| else: |
| reply(chat_id, "Unknown command. Try /help.") |
|
|
|
|
| def main() -> int: |
| sys.stdout.write(f"micro-bot starting (chat link hidden, " |
| f"token length {len(TOKEN)})\n") |
| sys.stdout.flush() |
| offset = 0 |
| failures = 0 |
| while True: |
| try: |
| data = api("getUpdates", offset=offset, timeout=25, |
| allowed_updates="message") |
| for upd in data.get("result", []): |
| offset = max(offset, upd["update_id"] + 1) |
| msg = upd.get("message") or {} |
| chat = msg.get("chat") or {} |
| sender = (msg.get("from") or {}).get("first_name") |
| text = (msg.get("text") or "").strip() |
| if not text.startswith("/"): |
| continue |
| handle(text, chat["id"], msg["message_id"], sender) |
| failures = 0 |
| except urllib.error.HTTPError as exc: |
| failures += 1 |
| sys.stderr.write(f"getUpdates HTTP {exc.code} (attempt {failures})\n") |
| sys.stderr.flush() |
| time.sleep(min(30, 2 * failures)) |
| except (urllib.error.URLError, TimeoutError, OSError) as exc: |
| failures += 1 |
| sys.stderr.write(f"getUpdates net err {exc!r} (attempt {failures})\n") |
| sys.stderr.flush() |
| time.sleep(min(30, 2 * failures)) |
| except Exception as exc: |
| failures += 1 |
| sys.stderr.write(f"getUpdates unhandled {exc!r}\n") |
| sys.stderr.flush() |
| time.sleep(5) |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
