""" Implementación Bot Telegram API """ import logging import random from telegram.ext import (Updater, CommandHandler, MessageHandler, Filters) import fetchip import config import constantes from sesion_colgado import SesionUsuario # Credenciales desde archivo de configuración TOKEN = config.TOKEN ADMIN = config.ADMIN MSG_INICIO = ' 🦀 @BlastardBot iniciado 🤖 ' LOG_FILE = 'bot.log' user_ses = SesionUsuario() # Creación objeto Updater updater = Updater(token=TOKEN, use_context=True) def main(): """ Agrega maneadores de funciones callback a updater, inicia log y bot. Bot queda a la espera de recibir algun mensaje. """ # Ejecuta start al recibir cualquier mensaje del usuario #updater.dispatcher.add_handler(MessageHandler(Filters.regex('.*'),start)) # Ejecuta start al iniciar chat con bot, al recibir comando '/start' updater.dispatcher.add_handler(CommandHandler('start', start)) # Ejecuta start al recibir cualquier mensaje, excepto '/comandos' #updater.dispatcher.add_handler(MessageHandler(Filters.text & (~Filters.command), start)) updater.dispatcher.add_handler(CommandHandler('huhuu', huhuu)) updater.dispatcher.add_handler(CommandHandler('allyb', allyb)) # Ejecuta desconocido la recibir un comando desconocido '/comando_desconocido' updater.dispatcher.add_handler(MessageHandler(Filters.command, desconocido)) inicia_log() updater.start_polling() # Iniciar Bot updater.bot.send_message(chat_id=ADMIN, text=MSG_INICIO) updater.idle() # En espera def start(update, context): """ Crea diccionario usuario, registro en log y envía mensaje de inicio """ user_id = update.message.from_user.id user_name = update.message.from_user.username user_log = f"USER [{user_name}] - ID [{user_id}] ha inicado el bot" logging.info(user_log) # Manejador para start con cualquier palabra que no empiece con "/" updater.dispatcher.add_handler(MessageHandler(Filters.text & (~Filters.command), jugar)) jugar(update, context) def jugar(update, context): reply = user_ses.accion(update.message.text) update.message.reply_text(reply["msg"], reply_markup=reply["reply_markup"]) def huhuu(update, context): """ Envia respuesta de audio (archivo) """ context.bot.send_audio(chat_id=update.effective_chat.id, audio=open('./media/huhuu.wav', 'rb')) def allyb(update, context): """ Envia img como respuesta """ context.bot.send_photo(chat_id=update.effective_chat.id, photo=open('./media/allyb.png', 'rb')) def desconocido(update, context): """ Envía respuesta de texto si no reconoce el comando recibido """ resp_desc = random.choice(constantes.RANDOM_RESP) context.bot.send_message(chat_id=update.effective_chat.id, text=resp_desc) def inicia_log(): """ Config. modulo de logging """ logging.basicConfig( filename = LOG_FILE, format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s', level = logging.INFO ) logging.info(MSG_INICIO) # Cola de trabajos jobs = updater.job_queue # Alerta cambio ip pública, cada 30 minutos def ip_alert(context: telegram.ext.CallbackContext): """ Alerta, registra e informa cambio de ip pública """ try: with open('./conf/last_ipp', 'r') as last_ipp_file: last_ipp = last_ipp_file.read() # No shell user # real_ipp = os.popen('curl -s ifconfig.me').readline() real_ipp = fetchip.get_public_ip() if real_ipp != last_ipp: context.bot.send_message(chat_id=ADMIN, text=real_ipp) with open('./conf/last_ipp', 'w') as archivo: archivo.write(real_ipp) except ConnectionError as ex: report = 'Error IP-alert :' + ex context.bot.send_message(chat_id=ADMIN, text=report) job_ipp = jobs.run_repeating(ip_alert, interval=1800, first=0) if __name__ == "__main__": main()