from datetime import time, timedelta, timezone, datetime, date import os from peewee import fn import logging class RandomMemoryJob(): def __init__(self, models, bot, job_queue): self.models = models self.bot = bot self.logger = logging.getLogger(self.__class__.__name__) if os.getenv("DOCKERIZED", "false") != "true": # when running locally, annoy the programmer every 60 seconds <3 job_queue.run_repeating(self.callback_memory, interval=60) self.min_age = 0 # do not filter messages: show them all else: # set the message sending time; include UTC shift +2 sending_time = time(hour=12, minute=0, second=0, tzinfo=timezone(timedelta(hours=2))) job_queue.run_daily(self.callback_memory, sending_time) self.min_age = 30 # days async def callback_memory(self, context): # last_seen of memory must be older than 10 days in past or None with self.models.db: possible_entries = self.models.JournalEntry.select().where( (self.models.JournalEntry.last_shown <= datetime.today().date() - timedelta(days=self.min_age)) | \ (self.models.JournalEntry.last_shown == None) ).order_by(fn.Random()) try: chosen_entry = possible_entries.get() except: self.logger.warning("Come back later for another memory.") return # update the last_shown of the chosen entry chosen_entry.last_shown = datetime.today().date() chosen_entry.save() chat_id = os.getenv("CHAT_ID") if chosen_entry.media_path: await self.bot.send_photo( chat_id = chat_id, photo = chosen_entry.media_path, caption = f"On {chosen_entry.date_pretty}, " f"{chosen_entry.author} wrote: \n" f"{chosen_entry.text}" ) else: await self.bot.send_message( chat_id = chat_id, text = f"On {chosen_entry.date_pretty}, " f"{chosen_entry.author} wrote: \n" f"{chosen_entry.text}" )