telegram-bot.py

Unified Telegram channel for the agent team — a second control/notification surface alongside WhatsApp.

#!/usr/bin/env python3
"""LankyServer Telegram Bot — unified channel for the Openclaw agent team."""
import asyncio
import logging
import os
import subprocess
import sys

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes, filters

logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    level=logging.INFO,
    stream=sys.stdout,
)
log = logging.getLogger("lanky-telegram")

TOKEN_FILE = "/home/lanky/secrets/telegram-bot-token"
ALLOWED_CHAT_FILE = "/home/lanky/secrets/telegram-allowed-chats"
ROUTER = "/home/lanky/scripts/request-router.py"
SCRIPT_DIR = "/home/lanky/scripts"

WELCOME = (
    "Hi Lanky! I'm your LankyServer assistant.\n\n"
    "I can help with:\n"
    "• Server status & agent reports\n"
    "• Restart containers\n"
    "• Check media, backups, network, SSL\n"
    "• Calendar & email (PA mode)\n\n"
    "Just ask in plain English."
)


def load_allowed_chats() -> set[int]:
    try:
        ids = set()
        for line in open(ALLOWED_CHAT_FILE).read().splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                ids.add(int(line))
        return ids
    except OSError:
        return set()


def run_router(text: str) -> str:
    try:
        result = subprocess.run(
            ["python3", ROUTER, text],
            capture_output=True,
            text=True,
            timeout=90,
        )
        out = (result.stdout + result.stderr).strip()
        return out if out else "Done — no output returned."
    except subprocess.TimeoutExpired:
        return "Request timed out (>90s). The script may still be running in the background."
    except Exception as e:
        return f"Error running request: {e}"


async def check_auth(update: Update) -> bool:
    allowed = load_allowed_chats()
    if not allowed:
        return True  # no allowlist = open (first-run mode)
    uid = update.effective_user.id
    cid = update.effective_chat.id
    if uid in allowed or cid in allowed:
        return True
    log.warning("Blocked message from user=%s chat=%s", uid, cid)
    await update.message.reply_text(
        "Sorry, you're not authorised to use this bot."
    )
    return False


async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not await check_auth(update):
        return
    await update.message.reply_text(WELCOME)


async def cmd_id(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Reply with the user's chat/user IDs — useful for setting up allowlist."""
    uid = update.effective_user.id
    cid = update.effective_chat.id
    uname = update.effective_user.username or "no username"
    await update.message.reply_text(
        f"Your user ID: {uid}\nChat ID: {cid}\nUsername: @{uname}\n\n"
        f"Add {uid} to {ALLOWED_CHAT_FILE} to authorise yourself."
    )


async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not await check_auth(update):
        return
    await update.message.reply_text("Checking agent team status...")
    reply = run_router("what is the status of all agents")
    await update.message.reply_text(reply[:4000])


async def cmd_pa(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not await check_auth(update):
        return
    result = subprocess.run(
        ["python3", f"{SCRIPT_DIR}/personal-assistant.py", "summary"],
        capture_output=True, text=True, timeout=60,
    )
    out = (result.stdout + result.stderr).strip() or "PA summary unavailable."
    await update.message.reply_text(out[:4000])


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not await check_auth(update):
        return
    text = update.message.text or ""
    if not text:
        return
    log.info("Message from %s: %s", update.effective_user.id, text[:80])
    # Show typing indicator
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id, action="typing"
    )
    reply = await asyncio.to_thread(run_router, text)
    # Telegram message limit is 4096 chars — split if needed
    for chunk in [reply[i:i+4000] for i in range(0, len(reply), 4000)]:
        await update.message.reply_text(chunk)


def main() -> None:
    try:
        token = open(TOKEN_FILE).read().strip()
    except OSError:
        log.error("Token file not found at %s", TOKEN_FILE)
        sys.exit(1)

    app = Application.builder().token(token).build()
    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("id", cmd_id))
    app.add_handler(CommandHandler("status", cmd_status))
    app.add_handler(CommandHandler("pa", cmd_pa))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    log.info("LankyServer Telegram bot starting (polling)...")
    app.run_polling(drop_pending_updates=True)


if __name__ == "__main__":
    main()

back to scripts