128 lines
4.5 KiB
Python
Executable File
128 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from datetime import datetime
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
import discord
|
|
import aiohttp
|
|
from discord.ext import tasks
|
|
|
|
from secret import TOKEN
|
|
|
|
ArkhamDBDeck = Any # TODO: better typing
|
|
|
|
ARKHAMDB_URL = 'https://arkhamdb.adamgoldsmith.name'
|
|
|
|
# TODO: should really be a database
|
|
with open('channel_list.json') as f:
|
|
channel_list = json.load(f)
|
|
|
|
|
|
class ArkhamDBClient(discord.Client):
|
|
session: aiohttp.ClientSession
|
|
|
|
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.session = aiohttp.ClientSession()
|
|
self.arkhamdb_monitor.start()
|
|
|
|
async def gather_deck_ids(self, channel: discord.TextChannel) -> dict[int, str]:
|
|
deck_ids: dict[int, str] = {}
|
|
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 = re.finditer(r'(?P<prefix>.*)' + ARKHAMDB_URL + r'/deck/view/(?P<deck_id>\d+)', message.content)
|
|
for match in matches:
|
|
deck_ids[int(match.group('deck_id'))] = match.group('prefix')
|
|
|
|
return deck_ids
|
|
|
|
async def get_latest_deck(self, deck_id: int) -> ArkhamDBDeck:
|
|
next_deck_id = deck_id
|
|
deck = None
|
|
while deck is None or deck["next_deck"] is not None:
|
|
async with self.session.get(ARKHAMDB_URL + f"/api/public/deck/{next_deck_id}.json") as resp:
|
|
deck = await resp.json()
|
|
next_deck_id = deck["next_deck"]
|
|
|
|
return deck
|
|
|
|
async def get_latest_decks(self, deck_ids: dict[int, str]) -> dict[int, tuple[str, ArkhamDBDeck]]:
|
|
latest_decks: dict[int, tuple[str, ArkhamDBDeck]] = {}
|
|
for deck_id, prefix in deck_ids.items():
|
|
try:
|
|
deck = await self.get_latest_deck(deck_id)
|
|
latest_decks[deck['id']] = (prefix, deck)
|
|
except aiohttp.ContentTypeError:
|
|
# TODO: json was invalid, should probably alert user
|
|
pass
|
|
|
|
return latest_decks
|
|
|
|
async def update_channel_latest_decks(self, channel: discord.TextChannel) -> None:
|
|
deck_ids = await self.gather_deck_ids(channel)
|
|
latest_decks = await self.get_latest_decks(deck_ids)
|
|
|
|
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()
|
|
|
|
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']}]({ARKHAMDB_URL}/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(message.embeds) != 1 or message_text != 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 = ArkhamDBClient()
|
|
client.run(TOKEN)
|