rawapibot.py

This example uses only the pure, “bare-metal” API wrapper.

 1#!/usr/bin/env python
 2# pylint: disable=import-error
 3"""Simple Bot to reply to Telegram messages.
 4
 5This is built on the API wrapper, see echobot.py to see the same example built
 6on the telegram.ext bot framework.
 7This program is dedicated to the public domain under the CC0 license.
 8"""
 9import asyncio
10import contextlib
11import logging
12from typing import NoReturn
13
14from telegram import Bot, Update
15from telegram.error import Forbidden, NetworkError
16
17logging.basicConfig(
18    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
19)
20# set higher logging level for httpx to avoid all GET and POST requests being logged
21logging.getLogger("httpx").setLevel(logging.WARNING)
22
23logger = logging.getLogger(__name__)
24
25
26async def main() -> NoReturn:
27    """Run the bot."""
28    # Here we use the `async with` syntax to properly initialize and shutdown resources.
29    async with Bot("TOKEN") as bot:
30        # get the first pending update_id, this is so we can skip over it in case
31        # we get a "Forbidden" exception.
32        try:
33            update_id = (await bot.get_updates())[0].update_id
34        except IndexError:
35            update_id = None
36
37        logger.info("listening for new messages...")
38        while True:
39            try:
40                update_id = await echo(bot, update_id)
41            except NetworkError:
42                await asyncio.sleep(1)
43            except Forbidden:
44                # The user has removed or blocked the bot.
45                update_id += 1
46
47
48async def echo(bot: Bot, update_id: int) -> int:
49    """Echo the message the user sent."""
50    # Request updates after the last update_id
51    updates = await bot.get_updates(offset=update_id, timeout=10, allowed_updates=Update.ALL_TYPES)
52    for update in updates:
53        next_update_id = update.update_id + 1
54
55        # your bot can receive updates without messages
56        # and not all messages contain text
57        if update.message and update.message.text:
58            # Reply to the message
59            logger.info("Found message %s!", update.message.text)
60            await update.message.reply_text(update.message.text)
61        return next_update_id
62    return update_id
63
64
65if __name__ == "__main__":
66    with contextlib.suppress(KeyboardInterrupt):  # Ignore exception when Ctrl-C is pressed
67        asyncio.run(main())