ahtcg_discord_bot/arkhamdb.py

from typing import TypedDict, Optional, Union

import aiohttp

ARKHAMDB_ORIGIN = 'https://arkhamdb.adamgoldsmith.name'


class ArkhamDBDeck(TypedDict):
    id: int
    name: str
    date_creation: str
    date_update: str
    description_md: str
    user_id: Optional[str]
    investigator_code: str
    investigator_name: str
    slots: dict[str, int]
    sideSlots: Union[list[str], dict[str, int]]
    ignoreDeckLimitSlots: Optional[str]
    version: str
    xp: int
    xp_spent: int
    xp_adjustment: int
    exile_string: Optional[str]
    taboo_id: int
    meta: str
    tags: str
    previous_deck: int
    next_deck: int
    problem: Optional[str]

class ArkhamDBClient:
    _session: aiohttp.ClientSession

    origin: str

    def __init__(self, arkhamdb_origin: str = ARKHAMDB_ORIGIN) -> None:
        self._session = aiohttp.ClientSession()
        self.origin = arkhamdb_origin

    async def close(self) -> None:
        await self._session.close()

    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(
                    self.origin + 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