The message_thread_id Mechanism

Telegram supergroups can be divided into named Topics, each identified by a message_thread_id. When a message is sent inside a topic, Telegram attaches this integer to the message object. When sent outside any topic, the field is absent entirely — not zero, not null, but genuinely missing.

This is the exact value OpenClaw requires to route notifications to the correct topic thread. SingIDBot’s /getids command reads message_thread_id from the incoming message and reports it in plain text for direct copy-paste into your configuration.

Because the field may be absent, the code uses getattr(update.effective_message, ‘message_thread_id’, None) rather than direct attribute access, which would raise an AttributeError in non-topic contexts.

The Singlish Responder

Any free-text message (not a command) triggers the singlish_chat handler. It classifies the message into greeting, thanks, or default using keyword matching, then picks a random phrase from the corresponding pool. A guard clause returns early if the message starts with / or was sent by the bot itself, preventing an infinite echo loop.

2.1 Logging Setup

Two handlers run simultaneously: a FileHandler writing UTF-8 lines to singidbot.log, and a StreamHandler for the console. On PythonAnywhere you can watch live output in the bash session and audit the full log file afterward.

LOG_FILE = "singidbot.log"
logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    level=logging.INFO,
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)

2.2 Singlish Response Bank

SINGLISH_RESPONSES maps category keys to phrase lists. get_singlish_response() uses dict.get() with a “default” fallback so unknown categories never raise a KeyError, and random.choice() picks uniformly from the pool.

SINGLISH_RESPONSES = {
    "greeting": [
        "Wah, you here ah! Come come, sit down lah!",
        "Eh hello! Long time no see, how are you?",
        "Aiyoh, you finally came! Welcome lah!",
        "Oi! Good to see you sia. What can I help you?",
    ],
    "thanks": [
        "Aiyah, no need to thank one lah!",
        "Wah, so polite! Welcome lah, anytime!",
        "No problem one! Next time also can help you.",
        "Eh, paiseh lah - that is what I am here for!",
    ],
    "default": [
        "Hah? Can repeat or not? I blur blur lah.",
        "Wah, interesting leh. Tell me more can?",
        "Aiyoh, I also dunno how to answer you sia.",
        "Got it lah! But I not so sure about that one.",
        "Eh, you very clever leh. I learn from you!",
    ],
}

def get_singlish_response(category):
    pool = SINGLISH_RESPONSES.get(category, SINGLISH_RESPONSES["default"])
    return random.choice(pool)

2.3 log_activity() Async Helper

log_activity() is declared async to match the handler context. It resolves the user’s display identifier to @username or numeric id, and emits a structured log line with a UTC ISO-8601 timestamp.

async def log_activity(user, event_type, metadata):
    user_info = (
        f"@{user.username}" if user and user.username
        else f"id={user.id}" if user
        else "unknown"
    )
    timestamp = datetime.now(timezone.utc).isoformat()
    logger.info(
        "[%s] event=%s user=%s metadata=%s",
        timestamp, event_type, user_info, metadata
    )

2.4 /start and /help Handlers

Both handlers log the event first, then compose and send a reply. /start personalises the greeting using user.first_name with a fallback to “friend”, prepending a random Singlish greeting to make every first interaction slightly different.

async def start(update, context):
    user = update.effective_user
    await log_activity(user, "START", {"chat_id": update.effective_chat.id})
    greeting = get_singlish_response("greeting")
    name = user.first_name if user and user.first_name else "friend"
    welcome_text = (
        f"{greeting}\n\nEh {name}, I am *SingIDBot* lah!\n\nCommands:\n- /start\n- /help\n- /getids"
    )
    await update.message.reply_text(welcome_text, parse_mode="Markdown")

async def help_command(update, context):
    user = update.effective_user
    await log_activity(user, "HELP", {"chat_id": update.effective_chat.id})
    await update.message.reply_text("*SingIDBot Commands* lah!\n/start /getids /help", parse_mode="Markdown")

2.5 /getids Handler

The key line is the getattr call. Telegram only populates message_thread_id when a message arrives inside a topic thread — in a regular chat the attribute simply does not exist. getattr with a None default handles this safely without raising AttributeError.

async def get_ids(update, context):
    user = update.effective_user
    await log_activity(user, "ID_RETRIEVAL", {"chat_id": update.effective_chat.id})
    chat_id = update.effective_chat.id
    topic_id = getattr(update.effective_message, "message_thread_id", None)
    response = f"Chat/Group ID: {chat_id}\n"
    if topic_id is not None:
        response += f"Topic ID: {topic_id}\n(This ID confirms the specific topic thread)"
    else:
        response += "Topic ID: Not available\n(Topics may not be enabled)"
    response += "\nTip: Record these IDs for configuring bot endpoints!"
    await update.message.reply_text(response, parse_mode="Markdown")
    await update.message.reply_text(f"{get_singlish_response('thanks')} Lah!")

2.6 singlish_chat Handler

Registered with filters.TEXT and ~filters.COMMAND. An internal guard also checks if the sender is the bot itself — without this, the bot’s own replies would trigger further replies in an infinite loop. Keyword classification checks thanks before greeting to avoid misclassification on mixed messages.

async def singlish_chat(update, context):
    user = update.effective_user
    await log_activity(user, "MESSAGE", {"text_preview": update.message.text[:50] if update.message else "N/A"})
    if (not update.message or update.message.text.startswith("/") or update.effective_user.id == context.bot.id):
        return
    user_text = update.message.text.lower().strip()
    if any(w in user_text for w in ["thanks", "thank", "appreciate"]):
        category = "thanks"
    elif any(w in user_text for w in ["hello", "hi", "hey", "oi", "wah", "alamak"]):
        category = "greeting"
    else:
        category = "default"
    await update.message.reply_text(get_singlish_response(category))

2.7 error_handler — Two-Tier Approach

Transient network issues (ProxyError, NetworkError, Conflict) are logged at WARNING level as a single line with no stack trace. Genuine bugs fall through to ERROR level with full exc_info traceback so they remain visible and actionable.

async def error_handler(update, context):
    error = context.error
    if isinstance(error, Exception):
        error_name = type(error).__name__
        error_msg = str(error)
        if "ProxyError" in error_msg or "NetworkError" in error_name:
            logger.warning("Network error (transient): %s", error_msg)
        elif "Conflict" in error_name or "terminated by other getUpdates" in error_msg:
            logger.warning("Conflict error (duplicate instance?): %s", error_msg)
        else:
            logger.error("Unhandled exception: %s: %s", error_name, error_msg, exc_info=context.error)

2.8 main() — Builder Chain, Timeouts, Handler Registration

All three timeout axes — connect, read, and write — are set to 30 seconds. Handlers are registered in priority order: specific commands first, then the catch-all text handler, then the error handler last.

def main():
    TOKEN = "YOUR_BOT_TOKEN_HERE"
    application = (
        Application.builder()
        .token(TOKEN)
        .connect_timeout(30)
        .read_timeout(30)
        .write_timeout(30)
        .build()
    )
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("getids", get_ids))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, singlish_chat))
    application.add_error_handler(error_handler)
    logger.info("--- SingIDBot Initializing ---")
    application.run_polling(allowed_updates=Update.ALL_TYPES)

Upload singidbot.py via the Files tab, open a Bash console, and run:

pip install python-telegram-bot --upgrade
python singidbot.py

Free-Tier Proxy Restriction

Free accounts route outbound traffic through a proxy that does not whitelist api.telegram.org. Outbound sendMessage calls can fail with httpx.ProxyError: 503 Service Unavailable. Inbound polling via getUpdates works fine — it is only the reply direction that is intermittently blocked.

In practice across four live sessions totalling over 5 hours, only 2 ProxyErrors occurred in 1800+ polling cycles. The error handler suppresses these to a single WARNING line so the log stays clean.

Avoiding Duplicate Instance Conflicts

Telegram only allows one active getUpdates poller per token. Always kill the old process before restarting:

pkill -f singidbot.py
pgrep -a -f singidbot
python singidbot.py

The bot was tested across four live sessions on PythonAnywhere’s free tier.

SingIDBot is production-stable on the free tier. The intermittent ProxyError is a platform constraint, not a code defect, and the error handler keeps it from polluting the log.