101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from telegram import Update
|
|
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
|
|
|
|
|
|
class Butler(object):
|
|
def __init__(self, logger, token, whitelist=[]):
|
|
self.logger = logger
|
|
self.token = token
|
|
self.whitelist = whitelist
|
|
|
|
# Create the Application and pass it your bot's token.
|
|
application = Application.builder().token(self.token).build()
|
|
|
|
# on different commands - answer in Telegram
|
|
application.add_handler(CommandHandler("start", Butler.start))
|
|
application.add_handler(CommandHandler("help", self.help_command))
|
|
application.add_handler(CommandHandler("stop", self.stop))
|
|
application.add_handler(CommandHandler("echo", self.echo))
|
|
application.add_handler(CommandHandler("who", self.who))
|
|
|
|
# on non command i.e message - echo the message on Telegram
|
|
# application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.echo))
|
|
|
|
# bot_user = asyncio.run(application.bot.get_me())
|
|
self.username = 'Vylbot' # bot_user.username
|
|
self.application = application
|
|
|
|
def get_username(self):
|
|
return self.username
|
|
|
|
def help_msg():
|
|
return """Respondo a los siguientes comandos:
|
|
|
|
/echo - Repite el texto.
|
|
/who - Información sobre ti y tu mensaje (para debugging).
|
|
"""
|
|
|
|
def valid_ids(self, update: Update):
|
|
user = update.effective_user
|
|
chat = update.effective_chat
|
|
chat_name = chat.effective_name
|
|
ids = [user.id, chat.id]
|
|
|
|
if not self.whitelist:
|
|
return True
|
|
for id in ids:
|
|
if str(id) in self.whitelist:
|
|
return True
|
|
|
|
self.logger.warning("User {} [{}] or chat {} [{}] are not in the whitelist."
|
|
.format(user.username, user.id, chat_name, chat.id))
|
|
return False
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Send a message when the command /start is issued."""
|
|
user = update.effective_user
|
|
await update.message.reply_html(rf"Hi {user.mention_html()}!")
|
|
|
|
async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Send a message when the command /help is issued."""
|
|
if not self.valid_ids(update):
|
|
return
|
|
await update.message.reply_text(Butler.help_msg())
|
|
|
|
async def echo(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Echo the user message."""
|
|
if not self.valid_ids(update):
|
|
return
|
|
echo_text = update.message.text.split(maxsplit=1)
|
|
await update.message.reply_text(echo_text[1])
|
|
|
|
async def stop(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
user = update.effective_user
|
|
self.logger.warning("I got blocked by user {} [{}]".format(user.username, user.id))
|
|
|
|
async def who(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
if not self.valid_ids(update):
|
|
return
|
|
|
|
mssg = update.message
|
|
user = update.effective_user
|
|
chat = update.effective_chat
|
|
name = chat.effective_name
|
|
date = mssg.date.astimezone()
|
|
|
|
answer = ("You're **{name}**, with username `{username}`, and"
|
|
" id `{uid}`.\nYou're messaging in the {ctype} chat named"
|
|
" __{cname}__, with id `{cid}`, and timestamp `{tstamp}`."
|
|
).format(name=user.full_name, username=user.username,
|
|
uid=user.id, cname=name, cid=chat.id,
|
|
ctype=chat.type, tstamp=str(date))
|
|
|
|
await mssg.reply_markdown(answer)
|
|
|
|
def run(self) -> None:
|
|
"""Start the bot."""
|
|
|
|
# Run the bot until the user presses Ctrl-C
|
|
self.application.run_polling(allowed_updates=Update.ALL_TYPES)
|