ahtcg_discord_bot/ahtcg_bot.py

199 lines
6.9 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
import argparse
from datetime import datetime
import json
import re
from pathlib import Path
import discord
from discord.ext import commands, tasks
2022-03-14 00:46:20 -04:00
from arkhamdb import ArkhamDBClient, ArkhamDBDeck
from validate_decks import Validator
from secret import TOKEN
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()
2022-03-14 01:04:06 -04:00
def setup_commands(self) -> None:
2022-03-14 01:08:14 -04:00
@self.command(name="monitor")
2022-03-14 01:04:06 -04:00
async def monitor(ctx: commands.Context) -> None:
"""Watch this channel for deck links and link to latest versions."""
2022-03-14 01:08:14 -04:00
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)
2022-03-14 01:08:14 -04:00
@self.command(name="forget")
2022-03-14 01:04:06 -04:00
async def forget(ctx: commands.Context) -> None:
"""Remove this channel from the monitor list"""
2022-03-14 01:08:14 -04:00
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)
2022-03-14 01:04:06 -04:00
async def on_ready(self) -> None:
2022-03-14 01:08:14 -04:00
print(f"Logged in as {self.user} (ID: {self.user.id})")
print("Enabled on servers:")
async for guild in self.fetch_guilds(limit=150):
2022-03-14 01:08:14 -04:00
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] = {}
2021-05-02 23:26:36 -04:00
url_regex = re.compile(
2022-03-14 01:08:14 -04:00
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
2021-05-02 23:26:36 -04:00
matches = url_regex.finditer(message.content)
for match in matches:
2022-03-14 01:08:14 -04:00
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):
2022-03-14 01:08:14 -04:00
if (
message.author.id == self.user.id
and message.id != channel.last_message_id
and len(message.embeds) == 1
):
await message.delete()
2022-03-14 00:46:20 -04:00
def status_for_deck(self, prefix: str, deck: ArkhamDBDeck) -> str:
2022-03-14 01:21:53 -04:00
url = f"{self.arkhamdb_client.origin}/deck/view/{deck['id']}"
status = f"{prefix}[{deck['name']}]({url}) [{deck['id']}]"
issues = []
2022-03-14 03:00:34 -04:00
if deck["xp"] is not None:
unspent_xp = deck["xp"] - deck["xp_spent"]
if unspent_xp > 0:
issues.append(f"{unspent_xp} unspent XP")
2022-03-14 01:21:53 -04:00
if deck["problem"] is not None:
issues.append(deck["problem"])
if issues:
status += f"***{', '.join(issues)}***"
2022-03-14 00:46:20 -04:00
return status
async def update_channel_latest_decks(self, channel: discord.TextChannel) -> None:
2022-03-14 01:08:14 -04:00
print(f"Running update in channel {channel.guild} - {channel.name}")
2021-05-09 09:58:31 -04:00
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
2022-03-14 01:08:14 -04:00
message_text = "\n".join(
self.status_for_deck(prefix, deck)
for prefix, deck in latest_decks.values()
)
2021-05-09 09:58:31 -04:00
message_embed = discord.Embed(
2022-03-14 01:08:14 -04:00
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)
)
2022-03-14 01:08:14 -04:00
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
):
2021-05-09 09:58:31 -04:00
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
2022-03-14 01:08:14 -04:00
if (
message.author.id != self.user.id
and message.channel.id in self.channel_list
):
await self.update_channel_latest_decks(message.channel)
2022-03-14 01:04:06 -04:00
async def on_message(self, message: discord.Message) -> None:
await self.process_commands(message)
await self.maybe_update_channel_for_message(message)
2022-03-14 01:08:14 -04:00
async def on_message_edit(
self, before: discord.Message, after: discord.Message
) -> None:
await self.maybe_update_channel_for_message(after)
2022-03-14 01:04:06 -04:00
async def on_message_delete(self, message: discord.Message) -> None:
await self.maybe_update_channel_for_message(message)
2021-05-03 01:07:17 -04:00
@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 = argparse.ArgumentParser(
description="A Discord bot for ArkhamDB deck info."
)
parser.add_argument(
"channel_list", type=Path, help="A path to a json file to store channel info"
)
args = parser.parse_args()
bot = ArkhamDBUpdater(channel_list=args.channel_list, command_prefix="!arkhamdb ")
bot.run(TOKEN, reconnect=True)
if __name__ == "__main__":
main()