ahtcg_discord_bot/arkhamdb.py

from typing import TypedDict, Optional, Union, Any

import aiohttp

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

ArkhamDBCard = Any


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: Optional[int]
    xp_spent: Optional[int]
    xp_adjustment: int
    exile_string: Optional[str]
    taboo_id: int
    meta: str
    tags: str
    previous_deck: int
    next_deck: int
    problem: Optional[str]


# TODO: some sort of caching?
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_cards(self, encounter=False) -> list[ArkhamDBCard]:
        async with self._session.get(
            self.origin + "/api/public/cards/",
            params={"encounter": "1"} if encounter else {},
        ) as resp:
            return await resp.json()

    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.get("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