refactor

.gitignore

@@ -1,5 +1,6 @@
 home
 snapshots
+.env
 
 .DS_Store
 

config/discord/leaderboard/messages/ShowUsers

@@ -1,17 +1,17 @@
-{% if user_discord_mention %}
+{% if user_to_mention %}
 ✅ **Utilizadores registados**
-{% for (user, mention) in user_discord_mention.items() %}
+{% for (user, mention) in user_to_mention.items() %}
 {{user}} {{mention}}
 {% endfor %}
 {% endif %}
 
-{% if members_not_user %}
+{% if unregistered_mentions %}
 ❌ **Utilizadores não registados**
-{% for mention in members_not_user %} {{mention}}{% endfor %}
+{% for mention in unregistered_mentions %} {{mention}}{% endfor %}
 {% endif %}
 
 
-{% if users_not_member %}
+{% if unregistered_users %}
 ✨ **Utilizadores disponíveis**
-{% for user in users_not_member %}{{user}} {% endfor %}
+{% for user in unregistered_users %}{{user}} {% endfor %}
 {% endif %}

docker-compose.yml

@@ -2,15 +2,12 @@
 version: "3.7"
 services:
   gago:
-    tty: true
-    stdin_open: true
     build:
       context: .
     ports:
       - "2222:22"
     environment:
       - PASS=${PASSWORD}
-      - DISCORD_BOT_TOKEN=${DISCORD_BOT_TOKEN}
       - TZ=Europe/London
     volumes:
       - ./home:/home
@@ -22,4 +19,16 @@ services:
       - ./etc/group:/opt/etc/group
       - ./etc/gshadow:/opt/etc/gshadow
       - ./snapshots:/snapshots
-    restart: always
+    restart: unless-stopped
+  discord:
+    build:
+      context: .
+    working_dir: /leaderboard
+    entrypoint: python3 -m bot.discrd
+    environment:
+      - DISCORD_BOT_TOKEN=${DISCORD_BOT_TOKEN}
+      - TZ=Europe/London
+    volumes:
+      - ./leaderboard:/leaderboard
+      - ./home:/home
+      - ./config:/config

leaderboard/bot/bot.py

@@ -0,0 +1,138 @@
+from dataclasses import dataclass
+from typing import Any, Dict, List
+
+from exercises.score import is_valid_exercise
+from notify import pull_notifications
+from utils import cancel_gen
+
+
+@dataclass
+class InvalidUser:
+    user: Any
+
+
+@dataclass
+class AlreadySetUser:
+    user: Any
+
+
+@dataclass
+class UserSet:
+    user: Any
+
+
+@dataclass
+class ShowUsers:
+    user_to_mention: Dict
+    unregistered_users: List
+    unregistered_mentions: List
+
+
+@dataclass
+class AlreadyRunning:
+    pass
+
+
+@dataclass
+class Go:
+    pass
+
+
+@dataclass
+class NotRunning:
+    pass
+
+
+@dataclass
+class Stop:
+    exercise: Any
+
+
+@dataclass
+class NoExercise:
+    pass
+
+
+@dataclass
+class InvalidExercise:
+    exercise: Any
+
+
+@dataclass
+class Chart:
+    exercise: Any
+
+
+class Session:
+    def __init__(self, users):
+        self.users = set(users)
+        self.user_to_member = {}
+        self.notifications = None
+        self.exercise = None
+
+    @property
+    def running(self):
+        return bool(self.notifications)
+
+    @property
+    def registered_users(self):
+        return set(self.user_to_member)
+
+    def register(self, user, member):
+        if user not in self.users:
+            return InvalidUser(user=user)
+
+        elif user in self.registered_users:
+            return AlreadySetUser(user=user)
+
+        else:
+            self.user_to_member[user] = member
+            return UserSet(user=user)
+
+    def get_member(self, user):
+        return self.user_to_member.get(user)
+
+    def get_users_status(self, online_members):
+        unregistered_users = sorted(self.users - self.registered_users)
+        registered_members = set(self.user_to_member.values())
+        unregistered_mentions = sorted(
+            member.mention for member in set(online_members) - registered_members
+        )
+        user_to_mention = {
+            user: member.mention for user, member in self.user_to_member.items()
+        }
+
+        return ShowUsers(
+            user_to_mention=user_to_mention,
+            unregistered_users=unregistered_users,
+            unregistered_mentions=unregistered_mentions,
+        )
+
+    async def start(self, exercise):
+        if self.running:
+            yield AlreadyRunning()
+        else:
+            self.notifications = pull_notifications(exercise)
+            self.exercise = exercise
+
+            yield Go()
+
+            async for notification in self.notifications:
+                yield notification
+
+    async def stop(self):
+        if not self.running:
+            return NotRunning()
+
+        await cancel_gen(self.notifications)
+        return Stop(exercise=self.exercise)
+
+    def chart(self, exercise=None):
+        chart_exercise = self.exercise or exercise
+
+        if not chart_exercise:
+            return NoExercise()
+        elif not is_valid_exercise(chart_exercise):
+            return InvalidExercise(exercise=chart_exercise)
+        else:
+            return Chart(exercise=chart_exercise)

leaderboard/bot/discrd.py

@@ -0,0 +1,118 @@
+import io
+import os
+from dataclasses import dataclass
+
+import discord
+from discord.ext.commands import Bot, Group, Command
+
+from bot.bot import Session as BotSession, Chart
+from chart import get_scores as get_chart_scores, convert_svg_png
+from render.discord import DiscordEnv as DiscordRenderEnv, DiscordTextMessage
+from render.svg_chart import SVGChartEnv as ChartRenderEnv
+from users import get_users as get_os_users
+from utils import run_in_executor
+
+users = get_os_users()
+session = BotSession(users)
+
+discord_render_env = DiscordRenderEnv(session)
+chart_render_env = ChartRenderEnv()
+
+
+@dataclass
+class SubcommandNotFound:
+    subcommand: str
+
+
+def create_chart_file(exercise):
+    scores = get_chart_scores(session.registered_users, exercise)
+    svg_text = chart_render_env.render(scores=scores)
+    png_bytes = convert_svg_png(svg_text)
+
+    return discord.File(io.BytesIO(png_bytes), filename="chart.png")
+
+
+@run_in_executor
+def format_message(notification):
+    if isinstance(notification, Chart):
+        return create_chart_file(notification.exercise)
+
+    else:
+        return DiscordTextMessage(discord_render_env, notification)
+
+
+async def gago(ctx, subcommand):
+    message = await format_message(SubcommandNotFound(subcommand))
+    await ctx.reply(message)
+
+
+async def start(ctx, exercise=None):
+    async for notification in session.start(exercise):
+        message = await format_message(notification)
+
+        await ctx.send(message)
+
+
+async def stop(ctx):
+    notification = await session.stop()
+    message = await format_message(notification)
+
+    await ctx.reply(message)
+
+
+async def set_user(ctx, user=None):
+    member = ctx.message.author
+    notification = session.register(user, member)
+    message = await format_message(notification)
+
+    await ctx.reply(message)
+
+
+async def show_users(ctx):
+    async with ctx.typing():
+        members = ctx.guild.members
+        notification = session.get_users_status(members)
+        message = await format_message(notification)
+
+        await ctx.reply(message)
+
+
+async def chart(ctx, exercise=None):
+    notification = session.chart(exercise)
+    message = await format_message(notification)
+
+    if isinstance(message, discord.File):
+        await ctx.reply(file=message)
+    else:
+        await ctx.reply(message)
+
+
+intents = discord.Intents.default()
+intents.members = True
+bot = Bot(intents=intents, command_prefix=("$", "!", "/"))
+
+group = Group(gago, invoke_without_command=True)
+
+
+def is_admin(ctx):
+    channel = ctx.channel
+    permissions = channel.permissions_for(ctx.author)
+    return permissions.administrator
+
+
+group.add_command(Command(start, checks=[is_admin]))
+group.add_command(Command(stop, checks=[is_admin]))
+group.add_command(Command(set_user, name="user", aliases=("setuser",)))
+group.add_command(Command(show_users, name="users"))
+group.add_command(Command(chart, aliases=("scores", "leaderboard")))
+
+bot.add_command(group)
+
+if __name__ == "__main__":
+    import logging
+
+    logging.basicConfig(level=logging.DEBUG)
+
+    TOKEN = os.environ["DISCORD_BOT_TOKEN"]
+
+    bot.run(TOKEN)

leaderboard/chart.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass
+from operator import itemgetter
+from typing import Any
+
+import cairosvg
+
+from exercises.score import score as get_score
+
+
+@dataclass
+class Score:
+    user: Any
+    xp: int
+    place: int
+
+
+def _center_scores(dsc_scores):
+    asc_scores = dsc_scores[::-1]
+    offset = 1 if len(asc_scores) % 2 == 0 else 0
+    centered_scores = asc_scores[1::2] + dsc_scores[offset::2]
+
+    return centered_scores
+
+
+def get_scores(users, exercise):
+    if len(users) == 0:
+        raise ValueError("No users to score")
+
+    dsc_scores = sorted(get_score(exercise).items(), key=itemgetter(1), reverse=True)
+    dsc_placed_users = [
+        Score(user=user, xp=score, place=place)
+        for place, (user, score) in enumerate(dsc_scores, 1)
+        if user in users
+    ]
+
+    return _center_scores(dsc_placed_users)
+
+
+def convert_svg_png(svg_text):
+    png_bytes = cairosvg.svg2png(bytestring=bytes(svg_text, encoding="utf-8"))
+
+    return png_bytes