Simple utilities bot for Telegram. /ping, /whoami and such.
1import logging
2
3from telegram import Update
4from telegram.ext import (
5 Application,
6 CommandHandler,
7 ContextTypes,
8 MessageHandler,
9 filters,
10)
11
12logger = logging.getLogger(__name__)
13
14
15def _help_msg():
16 return """Respondo a los siguientes comandos:
17
18/ping - Pong.
19/who - Información sobre ti y tu mensaje (para debugging).
20"""
21
22
23class PingBot(object):
24 def __init__(self, bot_token: str, bot_name: str) -> None:
25 self.token = bot_token
26 self.name = bot_name
27
28 # Create the Application and pass it your bot's token.
29 application = (
30 Application.builder()
31 .token(bot_token)
32 .concurrent_updates(True)
33 .build()
34 )
35
36 # on different commands - answer in Telegram
37 application.add_handler(CommandHandler("start", PingBot.start))
38 application.add_handler(CommandHandler("help", PingBot.help))
39 application.add_handler(CommandHandler("stop", PingBot.stop))
40 application.add_handler(CommandHandler("ping", PingBot.ping))
41 application.add_handler(CommandHandler("whoami", self.who))
42 application.add_handler(
43 MessageHandler(filters.TEXT & ~filters.COMMAND, PingBot.echo)
44 )
45
46 self.app = application
47
48 @staticmethod
49 async def start(
50 update: Update, _context: ContextTypes.DEFAULT_TYPE
51 ) -> None:
52 """Welcome message to a new user."""
53
54 assert update.message is not None
55 await update.message.reply_text("Hi! I'm a simple utilities bot.")
56
57 @staticmethod
58 async def help(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
59 """Send a message when the command /help is issued."""
60
61 assert update.message is not None
62 await update.message.reply_text(_help_msg())
63
64 @staticmethod
65 async def echo(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
66 """Repeats any message that is not a command."""
67
68 assert update.message is not None
69 await update.message.reply_text(update.message.text or "")
70
71 @staticmethod
72 async def ping(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
73 """Pongs back."""
74
75 assert update.message is not None
76 await update.message.reply_text("Pong!")
77
78 @staticmethod
79 async def stop(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
80 user = update.effective_user
81 assert user is not None
82 logger.warning(
83 "I got blocked by user {} [{}]".format(user.username, user.id)
84 )
85
86 async def who(
87 self, update: Update, _context: ContextTypes.DEFAULT_TYPE
88 ) -> None:
89 msg = update.message
90 user = update.effective_user
91 chat = update.effective_chat
92
93 assert msg is not None
94 assert user is not None
95 assert chat is not None
96
97 name = chat.effective_name
98 date = msg.date.astimezone()
99
100 answer = (
101 "You're <b>{name}</b>, with username <code>{username}"
102 "</code> and id <code>{uid}</code>.\nWe are on a ctype "
103 "'<i>{ctype}</i>' chat named <u>{cname}</u>, with chat id "
104 "<code>{cid}</code>, and the time right now is "
105 "<code>{tstamp}</code>."
106 ).format(
107 name=user.full_name,
108 username=user.username,
109 uid=user.id,
110 cname=name,
111 cid=chat.id,
112 ctype=chat.type,
113 tstamp=str(date),
114 )
115
116 await msg.reply_html(answer)
117
118 async def start_app(self) -> None:
119 """Starts the bot app."""
120
121 await self.app.initialize()
122 await self.app.start()
123 await self.app.updater.start_polling(allowed_updates=Update.ALL_TYPES)
124 logger.info("%s started up!", self.name)
125 # await self.notify_admins(f"Telegram Bot started up!\n{whitelist_str}")
126
127 async def stop_app(self) -> None:
128 """Stops the bot app."""
129
130 # await self.notify_admins("Shutting down!")
131 await self.app.updater.stop()
132 await self.app.stop()
133 await self.app.shutdown()
134 logger.info("%s stopped", self.name)