You've already forked Curso-lenguaje-python
Add movies trivial Bot for Telegram
This commit is contained in:
225
catch-all/06_bots_telegram/07_movie2_bot/server/Server.py
Normal file
225
catch-all/06_bots_telegram/07_movie2_bot/server/Server.py
Normal file
@@ -0,0 +1,225 @@
|
||||
import logging
|
||||
from argparse import ArgumentParser
|
||||
from telegram import Update
|
||||
from telegram.constants import ParseMode
|
||||
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext
|
||||
|
||||
import conf
|
||||
import interfaces
|
||||
import player
|
||||
import quiz
|
||||
import session
|
||||
from translations.required import *
|
||||
|
||||
|
||||
class Server:
|
||||
logger = logging.getLogger(__name__)
|
||||
SESSIONS = dict()
|
||||
|
||||
def __init__(self):
|
||||
self.config_instance = self.config_init()
|
||||
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
level=logging.DEBUG,
|
||||
handlers=[
|
||||
logging.FileHandler(self.config_instance.LOG_FILE),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
self.application = Application.builder().token(
|
||||
self.config_instance.TELEGRAM_BOT_API).build()
|
||||
|
||||
# Add handlers
|
||||
self.application.add_handler(MessageHandler(
|
||||
filters.TEXT & ~filters.COMMAND, self.command_check_resps)) # Add handler for non-command messages
|
||||
self.application.add_handler(
|
||||
CommandHandler('start', self.command_start))
|
||||
self.application.add_handler(CommandHandler('roll', self.command_roll))
|
||||
self.application.add_handler(CommandHandler(
|
||||
"leaderboard", self.command_leaderboard))
|
||||
self.application.add_handler(
|
||||
CommandHandler("repeat", self.command_repeat))
|
||||
self.application.add_handler(CommandHandler("cut", self.command_cut))
|
||||
self.application.add_handler(CommandHandler("stop", self.command_stop))
|
||||
self.application.add_error_handler(self.error)
|
||||
|
||||
# Set up job queue
|
||||
self.application.job_queue.run_repeating(
|
||||
self.update_all_timers, interval=60)
|
||||
|
||||
self.logger.info("Iniciando...")
|
||||
|
||||
# Start the Bot
|
||||
self.application.run_polling()
|
||||
|
||||
def config_init(self):
|
||||
self.logger.info("Config init...")
|
||||
|
||||
arg_parser = ArgumentParser(description="Movie2 Bot")
|
||||
|
||||
arg_parser.add_argument(
|
||||
"-e", "--env", metavar="env", type=str, default="prod",
|
||||
help="Environment to run the bot: dev, test or prod"
|
||||
)
|
||||
|
||||
arg_parser.add_argument(
|
||||
"-v", "--verbose", metavar="verbose", type=bool, default=False,
|
||||
help="Print information about running bot"
|
||||
)
|
||||
|
||||
args = arg_parser.parse_args()
|
||||
|
||||
if args.env == "prod":
|
||||
return conf.ProductionConfig()
|
||||
elif args.env == "dev":
|
||||
return conf.DevelopmentConfig()
|
||||
else:
|
||||
return conf.TestingConfig()
|
||||
|
||||
async def error(self, update: Update, context: CallbackContext):
|
||||
self.logger.warning(
|
||||
f'Update "{update}" caused error "{context.error}"')
|
||||
|
||||
async def update_all_timers(self, context: CallbackContext):
|
||||
self.logger.info("Updating all timers...")
|
||||
for sess in self.SESSIONS.values():
|
||||
sess.update_timers()
|
||||
if hasattr(sess, 'quiz'):
|
||||
sess.quiz.check_expiration()
|
||||
else:
|
||||
self.logger.error("Quiz class not found in session")
|
||||
|
||||
async def command_start(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command start...")
|
||||
chat_id = update.message.chat_id
|
||||
await update.message.reply_text('¡Hola! El comando /start ha sido recibido.')
|
||||
|
||||
if chat_id not in self.SESSIONS:
|
||||
self.messenger = interfaces.TelegramMessenger(
|
||||
context.bot, self.logger
|
||||
)
|
||||
self.SESSIONS[chat_id] = session.Session(
|
||||
chat_id, self.config_instance, self.logger
|
||||
)
|
||||
self.SESSIONS[chat_id].set_messenger(self.messenger)
|
||||
|
||||
# Crear una instancia de la clase Quiz
|
||||
try:
|
||||
self.SESSIONS[chat_id].quiz = quiz.Quiz(self.SESSIONS[chat_id])
|
||||
except AttributeError as e:
|
||||
self.logger.error(f"Error creating Quiz instance: {e}")
|
||||
await self.messenger.send_msg(
|
||||
chat_id, _("error_creating_quiz_instance")
|
||||
)
|
||||
logging.error(f"Error creating Quiz instance: {e}")
|
||||
|
||||
async def command_roll(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command roll...")
|
||||
chat_id = update.message.chat_id
|
||||
args = context.args
|
||||
rand_type = args[0] if args else None
|
||||
await self.SESSIONS[chat_id].messenger.send_msg(
|
||||
chat_id, _("searching_movies"))
|
||||
if hasattr(self.SESSIONS[chat_id], 'quiz'):
|
||||
await self.SESSIONS[chat_id].quiz.show(update, rand_type)
|
||||
else:
|
||||
self.logger.error("Quiz instance not found in session")
|
||||
|
||||
async def command_leaderboard(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command leaderboard...")
|
||||
chat_id = update.message.chat_id
|
||||
sess = self.SESSIONS.get(chat_id)
|
||||
if sess:
|
||||
try:
|
||||
await sess.messenger.send_msg(
|
||||
chat_id, _("leader_board_title"), 'highlights'
|
||||
)
|
||||
ldb = sess.get_leaderboard()
|
||||
await sess.messenger.send_msg(chat_id, ldb)
|
||||
except ValueError as e:
|
||||
await sess.messenger.send_msg(
|
||||
chat_id, f'{update.message.from_user.first_name} {e.args[0]}'
|
||||
)
|
||||
else:
|
||||
await update.message.reply_text(_("session_not_found"))
|
||||
|
||||
async def command_action(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command action...")
|
||||
chat_id = update.message.chat_id
|
||||
try:
|
||||
player = player.Player(update.message.from_user.id)
|
||||
player.name = f'{update.message.from_user.first_name} {update.message.from_user.last_name}'
|
||||
self.SESSIONS[chat_id].player_add(player)
|
||||
await self.SESSIONS[chat_id].messenger.send(
|
||||
update, f'{player.name} {_("joined_the_game")}'
|
||||
)
|
||||
except ValueError as e:
|
||||
await self.SESSIONS[chat_id].messenger.send(
|
||||
update, f'{update.message.from_user.first_name} {e.args[0]}'
|
||||
)
|
||||
|
||||
async def command_repeat(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command repeat...")
|
||||
chat_id = update.message.chat_id
|
||||
if chat_id in self.SESSIONS:
|
||||
movie_img = self.SESSIONS[chat_id].quiz.get_question()
|
||||
await self.SESSIONS[chat_id].messenger.send_msg(
|
||||
chat_id, _("repeting")
|
||||
)
|
||||
await self.SESSIONS[chat_id].messenger.send_photo(
|
||||
chat_id=chat_id, photo=movie_img,
|
||||
caption=_("what_is_the_movie_series_name")
|
||||
)
|
||||
await self.SESSIONS[chat_id].messenger.send_msg(
|
||||
chat_id, "==========================="
|
||||
)
|
||||
else:
|
||||
await update.message.reply_text(_("session_not_found"))
|
||||
|
||||
async def command_cut(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command cut...")
|
||||
chat_id = update.message.chat_id
|
||||
if chat_id in self.SESSIONS:
|
||||
try:
|
||||
player = player.Player(update.message.from_user.id)
|
||||
self.SESSIONS[chat_id].player_quit(player)
|
||||
await self.SESSIONS[chat_id].messenger.send(
|
||||
update, f'{player.name} {_("quit_the_game")}'
|
||||
)
|
||||
except ValueError as e:
|
||||
await self.SESSIONS[chat_id].messenger.send(
|
||||
update, f'{update.message.from_user.first_name} {e.args[0]}'
|
||||
)
|
||||
else:
|
||||
await update.message.reply_text(_("session_not_found"))
|
||||
|
||||
async def command_stop(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Command stop...")
|
||||
chat_id = update.message.chat_id
|
||||
if chat_id in self.SESSIONS:
|
||||
self.SESSIONS[chat_id].stop()
|
||||
await self.SESSIONS[chat_id].messenger.send(
|
||||
update, _("game_stopped")
|
||||
)
|
||||
else:
|
||||
await update.message.reply_text(_("session_not_found"))
|
||||
|
||||
# Implement the command_check_resps method
|
||||
async def command_check_resps(self, update: Update, context: CallbackContext):
|
||||
self.logger.info("Checking responses...")
|
||||
chat_id = update.message.chat_id
|
||||
if chat_id in self.SESSIONS:
|
||||
user_response = update.message.text
|
||||
sess = self.SESSIONS[chat_id]
|
||||
# Add logic to handle user responses here
|
||||
await sess.messenger.send_msg(chat_id, f"Received: {user_response}")
|
||||
else:
|
||||
await update.message.reply_text(_("session_not_found"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
Server()
|
||||
@@ -0,0 +1 @@
|
||||
from server.Server import Server
|
||||
Reference in New Issue
Block a user