#!/usr/bin/env python3 import configargparse from datetime import datetime import json import re 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: print(f"Logged in as {self.user} (ID: {self.user.id})") print("Enabled on servers:") async for guild in self.fetch_guilds(limit=150): print(" -", guild.name) print("------") 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.*)" + self.arkhamdb_client.origin + r"/deck/view/(?P\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"] 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: print(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() validator = Validator(cards) validation_errors = list( validator.validate([deck for _, deck in latest_decks.values()]) ) if validation_errors: message_embed.add_field( name="Card overuse:", value="\n".join(validation_errors) ) 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", ) args = parser.parse_args() bot = ArkhamDBUpdater( channel_list_file=args.channel_list, command_prefix="!arkhamdb " ) bot.run(args.discord_token, reconnect=True) if __name__ == "__main__": main()