feat: add api wrapper (#1)

* add api

* pre-commit improvements

* add tobfd as author

* fix authors

* do self.header and self.session only use in class

* import all classes

* update documentation

* improvements

* Remove double underscores

* Rename models

* remove raise_error

* remove return None

* remove _close_session and error descriptions

* update README.md

* split get_member_activity

* Add title and license

* Close session properly

* Shorten code

* Use get method and change route names

* Convert dates

* Improve doc strings

* Add image routes

* Add overload

* Load key from env

* Add async context manager

* Fix user error

* Add default days

* Convert activity dates and change route

* Remove mypy args

---------

Co-authored-by: Timo <35654063+tibue99@users.noreply.github.com>

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 .idea/
 venv/
 .env
+tests
 
 # Packaging
 build/

.pre-commit-config.yaml

@@ -1,5 +1,5 @@
 ci:
-    autoupdate_schedule: quarterly
+    autoupdate_schedule: monthly
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
@@ -26,6 +26,7 @@ repos:
     rev: v1.10.0
     hooks:
       - id: mypy
+        args: [--disable-error-code=union-attr]
 
   - repo: https://github.com/asottile/pyupgrade
     rev: v3.15.2

README.md

@@ -1,4 +1,28 @@
 # CookieBot API
+Official wrapper for the [CookieBot](https://cookie-bot.xyz) API.
+
+## Installation
+Python 3.9 or higher is required
 ```
 pip install cookiebot
 ```
+
+## Example Usage
+```python
+import asyncio
+from cookiebot import CookieAPI
+
+api = CookieAPI(api_key="[YOUR_API_KEY]")
+
+async def main():
+    user_stats = await api.get_user_stats(123456789)  # Replace with user ID
+    await api.close()
+
+asyncio.run(main())
+```
+You can also use an asynchronous context manager (recommended)
+```python
+async def main():
+    async with api as con:
+        user_stats = await con.get_user_stats(123456789)  # Replace with user ID
+```

cookiebot/__init__.py

@@ -1,3 +1,7 @@
+__title__ = "cookiebot"
+__license__ = "MIT"
 __version__ = "0.0.1"
 
 from .api import CookieAPI
+from .errors import *
+from .models import *

cookiebot/api.py

@@ -1,2 +1,223 @@
+import os
+from datetime import date, datetime
+from typing import overload
+
+import aiohttp
+from dotenv import load_dotenv
+
+from .errors import GuildNotFound, InvalidAPIKey, NoGuildAccess, NotFound, UserNotFound
+from .models import GuildActivity, MemberActivity, MemberStats, UserStats
+
+DEFAULT_DAYS = 14
+
+
+def _stats_dict(data: dict[str, int]) -> dict[date, int]:
+    return {datetime.strptime(d, "%Y-%m-%d").date(): count for d, count in data.items()}
+
+
 class CookieAPI:
-    pass
+    """A class to interact with the Cookie Bot API.
+
+    Parameters
+    ----------
+    api_key:
+        The API key to use. If no key is provided, ``COOKIE_KEY`` is loaded from the environment.
+    """
+
+    def __init__(self, api_key: str | None = None):
+        self._session: aiohttp.ClientSession | None = None
+
+        if api_key is None:
+            load_dotenv()
+            api_key = os.getenv("COOKIE_KEY")
+            if api_key is None:
+                raise InvalidAPIKey(
+                    "Please provide an API key or set the COOKIE_KEY environment variable."
+                )
+
+        self._header = {"key": api_key, "accept": "application/json"}
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        await self.close()
+
+    async def _setup(self):
+        if self._session is None:
+            self._session = aiohttp.ClientSession()
+
+    async def close(self):
+        await self._session.close()
+
+    @overload
+    async def _get(self, endpoint: str) -> dict: ...
+
+    @overload
+    async def _get(self, endpoint: str, stream: bool) -> bytes: ...
+
+    async def _get(self, endpoint: str, stream: bool = False):
+        async with self._session.get(
+            f"https://api.cookie-bot.xyz/v1/{endpoint}", headers=self._header
+        ) as response:
+            if response.status == 401:
+                raise InvalidAPIKey()
+            elif response.status == 403:
+                raise NoGuildAccess()
+            elif response.status == 404:
+                response = await response.json()
+                message = response.get("detail")
+                if "user" in message.lower() or "member" in message.lower():
+                    raise UserNotFound()
+                elif "guild" in message.lower():
+                    raise GuildNotFound()
+                raise NotFound()
+
+            if stream:
+                return await response.read()
+
+            return await response.json()
+
+    async def get_member_count(self, guild_id: int, days: int = DEFAULT_DAYS) -> dict[date, int]:
+        """Get the history of the guild member count for the provided number of days.
+
+        Parameters
+        ----------
+        guild_id:
+            The guild's ID
+        days:
+            The number of days. Defaults to ``14``.
+
+        Raises
+        ------
+        GuildNotFound:
+            The guild was not found.
+        """
+        await self._setup()
+        message_data = await self._get(f"member_count/{guild_id}?days={days}")
+
+        return _stats_dict(message_data)
+
+    async def get_user_stats(self, user_id: int) -> UserStats:
+        """Get the user's level stats.
+
+        Parameters
+        ----------
+        user_id:
+            The user's ID.
+
+        Raises
+        ------
+        UserNotFound:
+            The user was not found.
+        """
+        await self._setup()
+        data = await self._get(f"stats/user/{user_id}")
+        return UserStats(user_id, **data)
+
+    async def get_member_stats(self, user_id: int, guild_id: int) -> MemberStats:
+        """Get the member's level stats.
+
+        Parameters
+        ----------
+        user_id:
+            The user's ID.
+        guild_id:
+            The guild's ID.
+
+        Raises
+        ------
+        UserNotFound:
+            The user was not found.
+        """
+        await self._setup()
+        data = await self._get(f"stats/member/{user_id}/{guild_id}")
+        return MemberStats(user_id, guild_id, **data)
+
+    async def get_member_activity(
+        self, user_id: int, guild_id: int, days: int = 14
+    ) -> MemberActivity:
+        """Get the member's activity for the provided number of days.
+
+        Parameters
+        ----------
+        user_id:
+            The user's ID.
+        guild_id:
+            The guild's ID.
+        days:
+            The number of days. Defaults to ``14``.
+
+        Raises
+        ------
+        UserNotFound:
+            The user was not found.
+        """
+        await self._setup()
+        data = await self._get(f"activity/member/{user_id}/{guild_id}?days={days}")
+        msg_activity = _stats_dict(data.pop("msg_activity"))
+        voice_activity = _stats_dict(data.pop("voice_activity"))
+        return MemberActivity(days, user_id, guild_id, msg_activity, voice_activity, **data)
+
+    async def get_guild_activity(self, guild_id: int, days: int = DEFAULT_DAYS) -> GuildActivity:
+        """Get the guild's activity for the provided number of days.
+
+        Parameters
+        ----------
+        guild_id:
+            The guild's ID.
+        days:
+            The number of days. Defaults to ``14``.
+
+        Raises
+        ------
+        GuildNotFound:
+            The guild was not found.
+        """
+        await self._setup()
+        data = await self._get(f"activity/guild/{guild_id}?days={days}")
+        msg_activity = _stats_dict(data.pop("msg_activity"))
+        voice_activity = _stats_dict(data.pop("voice_activity"))
+        return GuildActivity(days, guild_id, msg_activity, voice_activity, **data)
+
+    async def get_guild_image(self, guild_id: int, days: int = DEFAULT_DAYS) -> bytes:
+        """Get the guild's activity image for the provided number of days.
+
+        Parameters
+        ----------
+        guild_id:
+            The guild's ID.
+        days:
+            The number of days. Defaults to ``14``.
+
+        Raises
+        ------
+        GuildNotFound:
+            The guild was not found.
+        """
+        await self._setup()
+        return await self._get(f"activity/guild/{guild_id}/image?days={days}", stream=True)
+
+    async def get_member_image(
+        self, user_id: int, guild_id: int, days: int = DEFAULT_DAYS
+    ) -> bytes:
+        """Get the member's activity image for the provided number of days.
+
+        Parameters
+        ----------
+        user_id:
+            The user's ID.
+        guild_id:
+            The guild's ID.
+        days:
+            The number of days. Defaults to ``14``.
+
+        Raises
+        ------
+        UserNotFound:
+            The user was not found.
+        """
+        await self._setup()
+        return await self._get(
+            f"activity/member/{user_id}/{guild_id}/image?days={days}", stream=True
+        )

cookiebot/errors.py

@@ -0,0 +1,26 @@
+class CookieError(Exception):
+    pass
+
+
+class InvalidAPIKey(CookieError):
+    def __init__(self, msg: str | None = None):
+        super().__init__(msg or "Invalid API key.")
+
+
+class NotFound(CookieError):
+    pass
+
+
+class UserNotFound(NotFound):
+    def __init__(self):
+        super().__init__("Could not find the user ID.")
+
+
+class GuildNotFound(NotFound):
+    def __init__(self):
+        super().__init__("Could not find the guild ID.")
+
+
+class NoGuildAccess(CookieError):
+    def __init__(self):
+        super().__init__("You are not a member of this guild.")

cookiebot/models.py

@@ -0,0 +1,59 @@
+from dataclasses import dataclass
+from datetime import date
+
+
+@dataclass
+class UserStats:
+    user_id: int
+    max_streak: int
+    streak: int
+    cookies: int
+    career: str
+    total_shifts: int
+    job: str
+
+
+@dataclass
+class MemberStats:
+    user_id: int
+    guild_id: int
+    level: int
+    xp: int
+    msg_count: int
+    voice_min: int
+    voice_xp: int
+    voice_level: int
+    current_level_progress: int
+    current_level_end: int
+    msg_rank: int
+    msg_total_members: int
+    voice_rank: int
+    voice_total_members: int
+
+
+@dataclass
+class MemberActivity:
+    days: int
+    user_id: int
+    guild_id: int
+    msg_activity: dict[date, int]
+    voice_activity: dict[date, int]
+    msg_count: int
+    voice_min: int
+    msg_rank: int
+    voice_rank: int
+    current_voice_min: int
+
+
+@dataclass
+class GuildActivity:
+    days: int
+    guild_id: int
+    msg_activity: dict[date, int]
+    voice_activity: dict[date, int]
+    msg_count: int
+    voice_min: int
+    top_channel: int
+    top_channel_messages: int
+    most_active_user_day: int | None
+    most_active_user_hour: int | None

pyproject.toml

@@ -6,7 +6,7 @@ license = {text = "MIT"}
 readme = "README.md"
 keywords = ["cookie-bot", "cookie", "api"]
 authors = [
-    { name = "tibue99" }
+    { name = "tibue99" }, { name="tobfd" }
 ]
 classifiers = [
     "Development Status :: 4 - Beta",

requirements.txt

@@ -1 +1,2 @@
 aiohttp
+python-dotenv