Files
journal-bot/bot/cronjob/leaderboard.py
Remy Moll f7478fb1e3
All checks were successful
Build container / Build (pull_request) Successful in 1m10s
more refinements for the deployment
2025-07-29 15:58:34 +02:00

96 lines
3.3 KiB
Python

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
)