first commit
This commit is contained in:
commit
5ee04c855d
4 changed files with 141 additions and 0 deletions
1
.gitignore
vendored
Normal file
1
.gitignore
vendored
Normal file
|
@ -0,0 +1 @@
|
|||
__pycache__
|
78
bot.py
Normal file
78
bot.py
Normal file
|
@ -0,0 +1,78 @@
|
|||
#!/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", Butler.help_command))
|
||||
application.add_handler(CommandHandler("stop", self.stop))
|
||||
application.add_handler(CommandHandler("echo", Butler.echo))
|
||||
application.add_handler(CommandHandler("who", Butler.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).
|
||||
"""
|
||||
|
||||
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(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Send a message when the command /help is issued."""
|
||||
await update.message.reply_text(Butler.help_msg())
|
||||
|
||||
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Echo the user message."""
|
||||
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(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
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)
|
42
main.py
Normal file
42
main.py
Normal file
|
@ -0,0 +1,42 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
import logging
|
||||
import argparse
|
||||
|
||||
from bot import Butler
|
||||
|
||||
|
||||
coloredlogsError = None
|
||||
try:
|
||||
import coloredlogs
|
||||
except ImportError as e:
|
||||
coloredlogsError = e
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Vylion\'s butler bot.')
|
||||
parser.add_argument('token', metavar='TOKEN',
|
||||
help='The Bot Token to work with the Telegram Bot API')
|
||||
parser.add_argument('-c', '--chats', nargs='*', default=[], metavar='[CHATS]',
|
||||
help='A whitelist of allowed chats. Empty enables all.')
|
||||
args = parser.parse_args()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
bot = Butler(logger, args.token, args.chats)
|
||||
|
||||
# Enable logging
|
||||
log_format = "[{}][%(asctime)s]%(name)s::%(levelname)s: %(message)s".format(bot.get_username().upper())
|
||||
|
||||
if coloredlogsError:
|
||||
logging.basicConfig(format=log_format, level=logging.INFO)
|
||||
logger.warning("Unable to load coloredlogs:")
|
||||
logger.warning(coloredlogsError)
|
||||
else:
|
||||
coloredlogs.install(level=logging.INFO, fmt=log_format)
|
||||
|
||||
logger.info("Bot started up!")
|
||||
bot.run()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
20
timezone.py
Normal file
20
timezone.py
Normal file
|
@ -0,0 +1,20 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from datetime import datetime, tzinfo, timedelta
|
||||
|
||||
|
||||
class TZMadrid(tzinfo):
|
||||
def utcoffset(self, dt):
|
||||
return timedelta(hours=1) + self.dst(dt)
|
||||
|
||||
def dst(self, dt):
|
||||
dston = datetime.date(dt.year, 4, 1)
|
||||
dstoff = datetime.date(dt.year, 10, 27)
|
||||
# Code to set dston and dstoff to the time zone's DST
|
||||
# transition times based on the input dt.year, and expressed
|
||||
# in standard local time.
|
||||
|
||||
if dston <= dt.replace(tzinfo=None) < dstoff:
|
||||
return timedelta(hours=1)
|
||||
else:
|
||||
return timedelta(0)
|
Loading…
Reference in a new issue