ahtcg_discord_bot/ahtcg_bot.py

#!/usr/bin/env python3

import configargparse
from datetime import datetime
import json
import re
import logging
from pathlib import Path

import discord
from discord.ext import commands, tasks

from arkhamdb import ArkhamDBClient, ArkhamDBDeck
from validate_decks import Validator


class ArkhamDBUpdater(commands.Bot):
    channel_list_file: Path
    channel_list: set[int]
    arkhamdb_client: ArkhamDBClient

    def __init__(self, channel_list_file: Path, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        self.channel_list_file = channel_list_file

        # TODO: should really be a database
        with open(self.channel_list_file) as f:
            self.channel_list = set(json.load(f))

        self.arkhamdb_client = ArkhamDBClient()
        self.setup_commands()

    async def close(self) -> None:
        await self.arkhamdb_client.close()
        await super().close()

    def setup_commands(self) -> None:
        @self.command(name="monitor")
        async def monitor(ctx: commands.Context) -> None:
            """Watch this channel for deck links and link to latest versions."""
            await ctx.message.reply(
                "Now monitoring this channel for deck IDs", mention_author=True
            )
            self.channel_list.add(ctx.message.channel.id)
            with open(self.channel_list_file, "w") as f:
                json.dump(list(self.channel_list), f)

        @self.command(name="forget")
        async def forget(ctx: commands.Context) -> None:
            """Remove this channel from the monitor list"""
            await ctx.message.reply(
                "No longer monitoring this channel for deck IDs", mention_author=True
            )
            self.channel_list.discard(ctx.message.channel.id)
            with open(self.channel_list_file, "w") as f:
                json.dump(list(self.channel_list), f)

    async def on_ready(self) -> None:
        logging.info(f"Logged in as {self.user} (ID: {self.user.id})")
        logging.info("Enabled on servers:")
        async for guild in self.fetch_guilds(limit=150):
            logging.info(f"  - {guild.name}")
        logging.info("------")

        self.arkhamdb_monitor.start()

    async def gather_deck_ids(self, channel: discord.TextChannel) -> dict[int, str]:
        deck_ids: dict[int, str] = {}
        url_regex = re.compile(
            r"(?P<prefix>.*)"
            + self.arkhamdb_client.origin
            + r"/deck/view/(?P<deck_id>\d+)"
        )
        async for message in channel.history(limit=200, oldest_first=True):
            # ignore the bot's messages
            if message.author.id == self.user.id:
                continue

            matches = url_regex.finditer(message.content)
            for match in matches:
                deck_ids[int(match.group("deck_id"))] = match.group("prefix")

        return deck_ids

    async def clear_old_messages(self, channel: discord.TextChannel) -> None:
        async for message in channel.history(limit=200):
            if (
                message.author.id == self.user.id
                and message.id != channel.last_message_id
                and len(message.embeds) == 1
            ):
                await message.delete()

    def status_for_deck(self, prefix: str, deck: ArkhamDBDeck) -> str:
        url = f"{self.arkhamdb_client.origin}/deck/view/{deck['id']}"
        status = f"{prefix}[{deck['name']}]({url}) [{deck['id']}]"

        issues = []
        if deck["xp"] is not None:
            unspent_xp = deck["xp"] - (deck["xp_spent"] or 0)
            if unspent_xp > 0:
                issues.append(f"{unspent_xp} unspent XP")
        if deck["problem"] is not None:
            issues.append(deck["problem"])

        if issues:
            status += f"***{', '.join(issues)}***"

        return status

    async def update_channel_latest_decks(self, channel: discord.TextChannel) -> None:
        logging.info(f"Running update in channel {channel.guild} - {channel.name}")
        async with channel.typing():
            deck_ids = await self.gather_deck_ids(channel)
            latest_decks = await self.arkhamdb_client.get_latest_decks(deck_ids)

            await self.clear_old_messages(channel)

            try:
                last_message = await channel.fetch_message(channel.last_message_id)
            except discord.NotFound:
                last_message = None

            message_text = "\n".join(
                self.status_for_deck(prefix, deck)
                for prefix, deck in latest_decks.values()
            )

            message_embed = discord.Embed(
                title=f"Updated as of {datetime.now()}", description=message_text
            )

            cards = await self.arkhamdb_client.get_cards(encounter=True)
            validator = Validator(cards)
            validation_errors = list(
                validator.validate([deck for _, deck in latest_decks.values()])
            )
            if validation_errors:
                data = "\n".join(validation_errors)
                message_embed.add_field(
                    name="Card overuse:",
                    value=(data[:1020] + "\n...") if len(data) > 1024 else data,
                )

            if (
                last_message is not None
                and last_message.author.id == self.user.id
                and len(last_message.embeds) == 1
            ):
                if (
                    message_embed.description != last_message.embeds[0].description
                    or message_embed.fields != last_message.embeds[0].fields
                ):
                    await last_message.edit(embed=message_embed)
            else:
                await channel.send(embed=message_embed)

    async def maybe_update_channel_for_message(self, message: discord.Message) -> None:
        # don't to react to the bot's changes, and only update registered channels
        if (
            message.author.id != self.user.id
            and message.channel.id in self.channel_list
        ):
            await self.update_channel_latest_decks(message.channel)

    async def on_message(self, message: discord.Message) -> None:
        await self.process_commands(message)
        await self.maybe_update_channel_for_message(message)

    async def on_message_edit(
        self, before: discord.Message, after: discord.Message
    ) -> None:
        await self.maybe_update_channel_for_message(after)

    async def on_message_delete(self, message: discord.Message) -> None:
        await self.maybe_update_channel_for_message(message)

    @tasks.loop(hours=1)
    async def arkhamdb_monitor(self) -> None:
        for channel_id in self.channel_list:
            channel = self.get_channel(channel_id)
            await self.update_channel_latest_decks(channel)


def main():
    parser = configargparse.ArgParser(
        description="A Discord bot for ArkhamDB deck info.",
        default_config_files=["./arkhamdb_discord_bot.conf"],
    )
    parser.add(
        "--channel_list",
        required=True,
        type=Path,
        help="path to a json file to store channel info",
    )
    parser.add("-c", "--config", is_config_file=True, help="config file path")
    parser.add(
        "--discord_token",
        required=True,
        help="Discord API token",
        env_var="DISCORD_TOKEN",
    )
    parser.add(
        "-d",
        "--debug",
        help="Print lots of debugging statements",
        action="store_const",
        dest="loglevel",
        const=logging.DEBUG,
        default=logging.WARNING,
    )
    parser.add_argument(
        "-v",
        "--verbose",
        help="Be verbose",
        action="store_const",
        dest="loglevel",
        const=logging.INFO,
    )
    args = parser.parse_args()

    logging.basicConfig(level=args.loglevel)

    bot = ArkhamDBUpdater(
        channel_list_file=args.channel_list, command_prefix="!arkhamdb "
    )
    bot.run(args.discord_token, reconnect=True)


if __name__ == "__main__":
    main()