ahtcg_discord_bot/ahtcg_bot.py

107 lines
3.7 KiB
Python
Executable File

#!/usr/bin/env python3
from datetime import datetime
import json
import re
import discord
from discord.ext import tasks
from arkhamdb import ArkhamDBClient
from secret import TOKEN
# TODO: should really be a database
with open('channel_list.json') as f:
channel_list = json.load(f)
class ArkhamDBUpdater(discord.Client):
arkhamdb_client: ArkhamDBClient
async def on_ready(self):
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_client = ArkhamDBClient()
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:
await message.delete()
async def update_channel_latest_decks(self, channel: discord.TextChannel) -> None:
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(
f"{prefix}[{deck['name']}]({self.arkhamdb_client.origin}/deck/view/{deck['id']}) [{deck['id']}]"
for prefix, deck in latest_decks.values())
message_embed = discord.Embed(
title=f'Updated as of {datetime.now()}',
description=message_text)
if last_message is not None and last_message.author.id == self.user.id:
if len(last_message.embeds) == 1 and message_text != last_message.embeds[0].description:
await last_message.edit(embed=message_embed)
else:
await channel.send(embed=message_embed)
async def on_message(self, message: discord.Message):
# we do not want the bot to reply to itself
if message.author.id == self.user.id:
return
if message.content.startswith('!arkhamdb_monitor'):
await message.reply('Hello!', mention_author=True)
channel_list.append(message.channel.id)
with open('channel_list.json', 'w') as f:
json.dump(channel_list, f)
if message.channel.id in channel_list:
await self.update_channel_latest_decks(message.channel)
async def on_message_edit(self, before: discord.Message, after: discord.Message):
# we do not want the bot to reply to itself
if after.author.id == self.user.id:
return
if after.channel.id in channel_list:
await self.update_channel_latest_decks(after.channel)
@tasks.loop(seconds=20)
async def arkhamdb_monitor(self) -> None:
for channel_id in channel_list:
channel = self.get_channel(channel_id)
await self.update_channel_latest_decks(channel)
client = ArkhamDBUpdater()
client.run(TOKEN)