Add logging, UrlTask and GenericTask

Author: bb4L

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='telegram-task-bot',
-    version='0.0.9',
+    version='0.0.10',
     license='BSD-3',
     description='rpi-radio-alarm library',
     long_description=open('README.md').read(),

telegramtaskbot/Bot.py

@@ -1,5 +1,5 @@
 import json
-import logging
+import logging.config
 import os
 from typing import List
 
@@ -8,16 +8,15 @@
 from telegram import InlineKeyboardButton, InlineKeyboardMarkup
 from telegram.ext import Updater, CommandHandler, Filters, CallbackQueryHandler
 
-logging.basicConfig(filename='telegramTaskBot.log', filemode='a',
-                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-                    level=logging.INFO)
+from telegramtaskbot.Tasks.UrlTask import UrlTask
 
 
 class TelegramTaskBot(object):
     jobs: List[telegram.ext.Job] = []
     default_button_list: List[InlineKeyboardButton] = []
     cmd_fun = {}
     job_names = {}
+    logger = logging.getLogger(__name__)
 
     def __init__(self, tasks: []):
         load_dotenv()
@@ -34,6 +33,8 @@ def __init__(self, tasks: []):
             self.cmd_fun[task.job_stop_name] = task.stop
             self.default_button_list.extend(task.get_inline_keyboard())
             if task.generic:
+                if isinstance(task, UrlTask):
+                    self.cmd_fun[task.job_actual_value] = task.get_actual_value
                 self.dispatcher.add_handler(CommandHandler(task.job_start_name, task.start_command, default_filter))
                 self.dispatcher.add_handler(CommandHandler(task.job_stop_name, task.stop_command, default_filter))
 
@@ -50,7 +51,7 @@ def get_default_filter():
         return default_filter
 
     def start(self, update, context):
-        reply_markup = InlineKeyboardMarkup(self.build_menu(self.default_button_list, n_cols=2))
+        reply_markup = InlineKeyboardMarkup(self.build_menu(self.default_button_list, n_cols=1))
         context.bot.send_message(chat_id=update.effective_chat.id, text=os.getenv('START_MESSAGE'),
                                  reply_markup=reply_markup)
 
@@ -61,7 +62,7 @@ def handle_button(self, update, context):
         query = update.callback_query
         self.cmd_fun.get(query.data)(self.jobs, update, context)
         self.save_to_json()
-        logging.info('after save')
+        self.logger.info('after save')
 
     def load_from_json(self):
         try:
@@ -71,9 +72,9 @@ def load_from_json(self):
                     for task in self.TASKS:
                         if task.job_name == job['name']:
                             task._start(self.jobs, self.updater.job_queue, job['context'])
-                logging.info(f'Loaded {len(data["jobs"])} from JSON')
+                self.logger.info(f'Loaded {len(data["jobs"])} from JSON')
         except IOError:
-            logging.info("File not accessible")
+            self.logger.info("File not accessible")
 
     def save_to_json(self):
         data = {'jobs': []}

telegramtaskbot/Tasks/GenericTask.py

@@ -0,0 +1,46 @@
+from abc import abstractmethod
+from typing import List
+
+import telegram
+from telegram.ext import JobQueue
+
+from telegramtaskbot.Tasks.Task import Task
+
+
+class GenericTask(Task):
+    job_actual_value: str
+    generic = True
+    show_subscribe_buttons = False
+
+    def __init__(self, job_queue: JobQueue = None):
+        super().__init__()
+        self.job_actual_value = 'actual_' + self.job_name
+
+    @abstractmethod
+    def callback(self, context: telegram.ext.CallbackContext):
+        pass
+
+    def start(self, jobs: List[telegram.ext.Job], update: telegram.Update, context: telegram.ext.CallbackContext):
+        self.handle_start(context, update.callback_query.message.chat_id)
+
+    def start_command(self, update: telegram.Update, context: telegram.ext.CallbackContext):
+        self.handle_start(context, update.message.chat_id)
+
+    def handle_start(self, context: telegram.ext.CallbackContext, chat_id: str):
+        context.bot.send_message(chat_id=chat_id,
+                                 text=f'Thank you for subscribing')
+        self.save_user(chat_id)
+        self.logger.debug(f'User {chat_id} subscribed')
+
+    def stop(self, jobs: List[telegram.ext.Job], update: telegram.Update, context: telegram.ext.CallbackContext):
+        self.handle_stop(context, update.callback_query.message.chat_id)
+
+    def stop_command(self, update: telegram.Update, context: telegram.ext.CallbackContext):
+        self.handle_stop(context, update.message.chat_id)
+
+    def handle_stop(self, context: telegram.ext.CallbackContext, chat_id: str):
+        users = self.load_users()
+        users.remove(chat_id)
+        self.save_to_json(users)
+        self.logger.debug(f'User {chat_id} unsubscribed')
+        context.bot.send_message(chat_id=chat_id, text=f'You succesfully unsubscribed')

telegramtaskbot/Task.py renamed to telegramtaskbot/Tasks/Task.py

@@ -1,5 +1,6 @@
 import json
 import logging
+
 from abc import abstractmethod
 from datetime import timedelta
 from typing import List
@@ -13,24 +14,27 @@ class Task(object):
     job_name: str
     job_start_name: str
     job_stop_name: str
+    disable_notifications: bool = True
     generic: bool = False  # defines if the task looks the same for each user
     first_time = 0
     repeat_time: timedelta = timedelta(seconds=5)
     filename: str = ''
+    logger = logging.getLogger(__name__)
 
-    def __init__(self, job_queue: JobQueue):
-        pass
+    def __init__(self, job_queue: JobQueue = None):
+        self.job_start_name = 'start_' + self.job_name
+        self.job_stop_name = 'stop_' + self.job_name
 
     def start(self, jobs: List[telegram.ext.Job], update: telegram.Update, context: telegram.ext.CallbackContext):
         context.bot.send_message(chat_id=update.callback_query.message.chat_id,
                                  text=f'Setting the job {self.job_name} up')
         self._start(jobs, context.job_queue, update.callback_query.message.chat_id)
 
     def start_command(self, update: telegram.Update, context: telegram.ext.CallbackContext):
-        raise NotImplementedError
+        pass
 
     def stop_command(self, update: telegram.Update, context: telegram.ext.CallbackContext):
-        raise NotImplementedError
+        pass
 
     def _start(self, jobs: List[telegram.ext.Job], job_queue: JobQueue, chat_id):
         new_job = job_queue.run_repeating(self.callback, self.repeat_time, context=chat_id, first=self.first_time)
@@ -47,7 +51,7 @@ def stop(self, jobs: List[telegram.ext.Job], update: telegram.Update, context: t
             count += 1
             idx = jobs.index(job_to_stop)
             jobs.pop(idx)
-        logging.info(f' stopped {count} of {num_jobs} jobs for chat_id={chat_id}')
+        self.logger.info(f' stopped {count} of {num_jobs} jobs for chat_id={chat_id}')
 
     @abstractmethod
     def callback(self, context: telegram.ext.CallbackContext):
@@ -68,7 +72,7 @@ def save_to_json(self, users):
         data = {'users': users}
         with open(self.filename + '.json', 'w') as outfile:
             json.dump(data, outfile)
-        logging.info('Saved User')
+        self.logger.debug('Saved User')
 
     def load_users(self):
         try:
@@ -77,6 +81,6 @@ def load_users(self):
                 users = data['users']
         except IOError:
             users = []
-            logging.info("File not accessible")
+            self.logger.error("File not accessible")
         finally:
             return users

telegramtaskbot/Tasks/UrlTask.py

@@ -0,0 +1,58 @@
+import time as timer
+from abc import abstractmethod
+
+import requests
+import telegram
+from requests import Response
+from telegram import InlineKeyboardButton
+
+from telegramtaskbot.Tasks.GenericTask import GenericTask
+
+
+class UrlTask(GenericTask):
+    disable_notifications = True
+    url: str
+
+    def callback(self, context: telegram.ext.CallbackContext):
+        self.logger.info(f'Run {self.job_name}')
+        users = self.load_users()
+        response_message = self.get_data()
+        self.logger.info(f'Notifying {len(users)} users for {self.job_name}')
+        for user in users:
+            context.bot.send_message(chat_id=user, text=response_message,
+                                     disable_notification=self.disable_notifications)
+
+    def get_actual_value(self, joblist: [], update: telegram.Update, context: telegram.ext.CallbackContext):
+        self.logger.debug(f'Get actual value from {self.job_name} for {update.callback_query.message.chat_id}')
+        data: str = self.get_data()
+        context.bot.send_message(chat_id=update.callback_query.message.chat_id, text=data)
+        self.logger.debug(
+            f'Send message to {update.callback_query.message.chat_id} with content: \"{" ".join(data.splitlines())}\"')
+
+    def get_data(self):
+        return self.handle_response(self.get_response())
+
+    @abstractmethod
+    def handle_response(self, response: Response):
+        pass
+
+    def get_response(self):
+        count = 0
+        response = requests.get(self.url)
+        if response.status_code != 200:
+            while response.status_code != 200:
+                timer.sleep(2)
+                resp = requests.get(self.url)
+                count += 1
+                response = resp
+        self.logger.debug(f'{self.job_name} tried for {count} times')
+        return response
+
+    def get_inline_keyboard(self):
+        buttons = [
+            InlineKeyboardButton(f"Get actual Value for {self.job_name}", callback_data=self.job_actual_value),
+        ]
+        if self.show_subscribe_buttons:
+            buttons.append(InlineKeyboardButton(f"Subscribe for {self.job_name}", callback_data=self.job_start_name))
+            buttons.append(InlineKeyboardButton(f"Unsubscribe for {self.job_name}", callback_data=self.job_stop_name))
+        return buttons

telegramtaskbot/Tasks/__init__.py

@@ -1,2 +1,8 @@
-from .Task import Task
+import logging
+
 from .Bot import TelegramTaskBot
+from .Tasks.GenericTask import GenericTask
+from .Tasks.Task import Task
+from .Tasks.UrlTask import UrlTask
+
+logging.getLogger(__name__).addHandler(logging.NullHandler())