299 lines
10 KiB
Python
Executable File
299 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from datetime import datetime
|
|
import json
|
|
import re
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import configargparse
|
|
|
|
import discord
|
|
from discord.ext import tasks
|
|
from discord import app_commands
|
|
|
|
from arkhamdb import ArkhamDBClient, ArkhamDBDeck
|
|
from validate_decks import Validator
|
|
|
|
|
|
class ArkhamDBUpdater(discord.Client):
|
|
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))
|
|
|
|
async def setup_hook(self) -> None:
|
|
self.arkhamdb_client = ArkhamDBClient()
|
|
|
|
async def close(self) -> None:
|
|
await self.arkhamdb_client.close()
|
|
await super().close()
|
|
|
|
async def setup_commands(self) -> None:
|
|
tree = app_commands.CommandTree(self)
|
|
|
|
@tree.command()
|
|
async def adbmonitor(interaction: discord.Interaction) -> None:
|
|
"""Watch this channel for deck links and link to latest versions"""
|
|
await interaction.response.send_message(
|
|
"Now monitoring this channel for deck IDs", ephemeral=True
|
|
)
|
|
logging.info(f"Now monitoring: {self.get_channel(interaction.channel_id)}")
|
|
self.channel_list.add(interaction.channel_id)
|
|
with open(self.channel_list_file, "w") as f:
|
|
json.dump(list(self.channel_list), f)
|
|
|
|
@tree.command()
|
|
@app_commands.checks.cooldown(2, 60 * 5, key=lambda i: i.guild.id)
|
|
async def adbupdate(interaction: discord.Interaction) -> None:
|
|
"""Immediately update the deck summary"""
|
|
await interaction.response.defer(ephemeral=True, thinking=True)
|
|
await self.update_channel_latest_decks(
|
|
self.get_channel(interaction.channel_id)
|
|
)
|
|
await interaction.followup.send("Done with update!", ephemeral=True)
|
|
|
|
@adbupdate.error
|
|
async def on_adbupdate_error(
|
|
interaction: discord.Interaction, error: app_commands.AppCommandError
|
|
):
|
|
if isinstance(error, app_commands.CommandOnCooldown):
|
|
await interaction.response.send_message(str(error), ephemeral=True)
|
|
|
|
@tree.command()
|
|
async def adbforget(interaction: discord.Interaction) -> None:
|
|
"""Remove this channel from the monitor list"""
|
|
await interaction.response.send_message(
|
|
"No longer monitoring this channel for deck IDs", ephemeral=True
|
|
)
|
|
logging.info(
|
|
f"Stopped monitoring: {self.get_channel(interaction.channel_id)}"
|
|
)
|
|
self.channel_list.discard(interaction.channel_id)
|
|
with open(self.channel_list_file, "w") as f:
|
|
json.dump(list(self.channel_list), f)
|
|
|
|
await tree.sync()
|
|
|
|
async def on_ready(self) -> None:
|
|
logging.info(f"Logged in as {self.user} (ID: {self.user.id})")
|
|
expected_permissions = permissions = discord.Permissions(
|
|
read_messages=True,
|
|
read_message_history=True,
|
|
send_messages=True,
|
|
use_application_commands=True,
|
|
)
|
|
logging.info(
|
|
"Invite URL: "
|
|
+ discord.utils.oauth_url(self.user.id, permissions=expected_permissions)
|
|
)
|
|
|
|
logging.info("Enabled on servers:")
|
|
async for guild in self.fetch_guilds(limit=150):
|
|
logging.info(f" - {guild.name}")
|
|
logging.info("------")
|
|
|
|
for channel_id in list(self.channel_list):
|
|
channel = self.get_channel(channel_id)
|
|
if channel is None:
|
|
logging.info(
|
|
f"channel {channel_id} does not exist, removing it from tracking!"
|
|
)
|
|
self.channel_list.remove(channel_id)
|
|
continue
|
|
permissions = channel.permissions_for(channel.guild.me)
|
|
missing_permissions = (
|
|
permissions & expected_permissions
|
|
) ^ expected_permissions
|
|
if missing_permissions.value:
|
|
logging.warning(
|
|
f"channel: {channel.guild} - {channel.name} missing permissions: {missing_permissions}"
|
|
)
|
|
else:
|
|
logging.info(f"channel {channel.guild} - {channel.name} permissions OK")
|
|
logging.debug(
|
|
f"channel {channel.guild} - {channel.name} permissions: {dict(permissions)}"
|
|
)
|
|
|
|
await self.setup_commands()
|
|
|
|
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
|
|
)
|
|
|
|
# TODO: caching
|
|
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.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)
|
|
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
|
|
bot = ArkhamDBUpdater(
|
|
channel_list_file=args.channel_list,
|
|
intents=intents,
|
|
)
|
|
bot.run(args.discord_token, reconnect=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|