more refinements for the deployment

bot/cronjob/leaderboard.py

@@ -0,0 +1,95 @@
+import os
+from telegram.ext import ExtBot
+from telegram.constants import ParseMode
+import logging
+from datetime import time, timedelta, timezone, datetime, date
+from peewee import fn
+import models
+from telegram.ext import JobQueue
+
+
+RANKING_TEMPLATE = """
+<b>Journal Leaderboard</b>
+This week: 📈{week_leader_name} - {week_leader_count} 📉{week_last_name} - {week_last_count}
+This month: 📈{month_leader_name} - {month_leader_count} 📉{month_last_name} - {month_last_count}
+This year: 📈{year_leader_name} - {year_leader_count} 📉{year_last_name} - {year_last_count}
+
+🏆 Leader: {leader_name}
+"""
+
+
+
+def get_author_ranking(since_days):
+    """Returns the query for the top authors by counting their journal entries. An additional field for the count is added."""
+
+    cutoff_date = date.today() - timedelta(days=since_days)
+    with models.db:
+        return models.JournalEntry.select(
+            models.JournalEntry.author,
+            fn.Count(models.JournalEntry.id).alias('message_count')
+        ).where(
+            models.JournalEntry.date >= cutoff_date
+        ).group_by(
+            models.JournalEntry.author
+        ).order_by(
+            fn.Count(models.JournalEntry.id).desc()
+        )
+
+
+
+
+class SendLeaderboard():
+    def __init__(self, bot: ExtBot, job_queue: JobQueue):
+        self.bot = bot
+        self.logger = logging.getLogger(self.__class__.__name__)
+
+        if not models.IS_PRODUCTION:
+            # when running locally, just run once after 10 seconds
+            job_queue.run_once(self.callback_leaderboard, when=10)
+        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_leaderboard, when=sending_time, day=-1)
+
+
+    async def callback_leaderboard(self, context):
+        """Send a weakly leaderboard to the chat."""
+        if date.today().weekday() != 1:
+            self.logger.info("Today is not Monday, skipping leaderboard.")
+            return
+
+        # get the top contributions of the past week, month and year:
+        ranking_week = get_author_ranking(7)
+        ranking_month = get_author_ranking(30)
+        ranking_year = get_author_ranking(365)
+
+        week_leader, week_last = ranking_week.first(n=2)
+        month_leader, month_last = ranking_month.first(n=2)
+        year_leader, year_last = ranking_year.first(n=2)
+
+        leader = year_leader
+
+        message_text = RANKING_TEMPLATE.format(
+            week_leader_name=week_leader.author,
+            week_leader_count=week_leader.message_count,
+            week_last_name=week_last.author,
+            week_last_count=week_last.message_count,
+            month_leader_name=month_leader.author,
+            month_leader_count=month_leader.message_count,
+            month_last_name=month_last.author,
+            month_last_count=month_last.message_count,
+            year_leader_name=year_leader.author,
+            year_leader_count=year_leader.message_count,
+            year_last_name=year_last.author,
+            year_last_count=year_last.message_count,
+            leader_name=leader.author
+        )
+
+        print(message_text)
+
+        chat_id = os.getenv("CHAT_ID")
+        await self.bot.send_message(
+            chat_id = chat_id,
+            text = message_text,
+            parse_mode=ParseMode.HTML
+        )