Begin work on Beacon (bridge protocol)

Author: greeeen-dev

shinobu/beacon/README.md

@@ -0,0 +1,2 @@
+# Shinobu Beacon
+Beacon is the cross-platform bridge framework for Shinobu.

shinobu/beacon/__init__.py

@@ -0,0 +1,26 @@
+from discord.ext import commands
+from shinobu.runtime.models import shinobu_cog
+
+class BeaconBackend(shinobu_cog.ShinobuCog):
+    def __init__(self, bot, **kwargs):
+        # Register cog metadata
+        self.setup_shinobu_cog(
+            bot,
+            **kwargs,
+            shinobu_metadata=shinobu_cog.ShinobuCogMetadata(
+                name="Beacon Backend",
+                description="Manages the Shinobu Beacon bridge protocol.",
+                visible_in_help=True
+            )
+        )
+
+    def on_entitlements_issued(self):
+        # We will initialize Beacon here
+        self.bot.shared_objects.add("beacon")
+        return
+
+def get_cog_type():
+    return BeaconBackend
+
+def setup(bot):
+    bot.add_cog(BeaconBackend(bot))

shinobu/beacon/cogs/backend.py

@@ -0,0 +1,16 @@
+class BeaconABC:
+    """A abstract base class for objects."""
+
+    def __init__(self, object_id: str, platform: str):
+        self._id: str = object_id
+        self._platform: str = platform
+
+    @property
+    def id(self) -> str:
+        """Returns the ID of an object."""
+        return self._id
+
+    @property
+    def platform(self) -> str:
+        """Returns the platform of an object."""
+        return self._platform

shinobu/beacon/models/abc.py

@@ -0,0 +1,22 @@
+from shinobu.beacon.models import messageable, server as beacon_server
+
+class BeaconChannel(messageable.BeaconMessageable):
+    """A class representing server channels."""
+
+    def __init__(self, object_id: str, platform: str, name: str, server: beacon_server.BeaconServer,
+                 nsfw: bool = False):
+        super().__init__(object_id, platform, name)
+        self._server: beacon_server.BeaconServer = server
+        self._nsfw: bool = nsfw
+
+    @property
+    def server(self) -> beacon_server.BeaconServer:
+        return self._server
+
+    @property
+    def server_id(self) -> str:
+        return self._server.id
+
+    @property
+    def nsfw(self) -> bool:
+        return self._nsfw

shinobu/beacon/models/channel.py

@@ -0,0 +1,24 @@
+from enum import Enum
+
+class BeaconContentType(Enum):
+    text = 1
+    file = 2
+
+class BeaconContentBlock:
+    """A class representing content blocks in a Beacon message."""
+
+    def __init__(self, content_type: BeaconContentType, content: dict):
+        self._type: BeaconContentType = content_type
+        self._content: dict = content
+
+    @property
+    def type(self) -> BeaconContentType:
+        return self._type
+
+    @property
+    def content(self) -> dict:
+        return self._content
+
+class BeaconContentText(BeaconContentBlock):
+    def __init__(self, content: str):
+        super().__init__(BeaconContentType.text, {"content": content})

shinobu/beacon/models/content.py

@@ -0,0 +1,82 @@
+from shinobu.beacon.models import (user as beacon_user, channel as beacon_channel, server as beacon_server,
+                                   member as beacon_member, webhook as beacon_webhook, message as beacon_message,
+                                   messageable as beacon_messageable)
+
+class BeaconDriverUnsupported(Exception):
+    def __init__(self):
+        super().__init__("Driver does not support this feature.")
+
+class BeaconDriverWebhookCache:
+    def __init__(self):
+        self._data: dict = {}
+
+    def store_webhook(self, webhook: beacon_webhook.BeaconWebhook):
+        if webhook.id in self._data:
+            raise KeyError("Webhook already in cache")
+
+        self._data.update({webhook.id: webhook})
+
+    def store_webhooks(self, webhooks: list):
+        for webhook in webhooks:
+            self.store_webhook(webhook)
+
+    def get_webhook(self, webhook_id: str):
+        return self._data.get(webhook_id)
+
+class BeaconDriver:
+    """A class representing a platform driver for the Beacon bridge protocol."""
+
+    def __init__(self, platform: str, bot):
+        self._platform: str = platform
+        self._webhooks: BeaconDriverWebhookCache = BeaconDriverWebhookCache()
+        self._bot = bot
+
+    @property
+    def platform(self) -> str:
+        """The platform the driver enables support for."""
+        return self._platform
+
+    def get_user(self, user_id: str) -> beacon_user.BeaconUser:
+        """Gets a user."""
+        raise BeaconDriverUnsupported()
+
+    async def fetch_user(self, user_id: str) -> beacon_user.BeaconUser:
+        """Fetches a user from the platform API."""
+        raise BeaconDriverUnsupported()
+
+    def get_member(self, server: beacon_server.BeaconServer, member_id: str) -> beacon_member.BeaconMember:
+        """Gets a member from a server."""
+        raise BeaconDriverUnsupported()
+
+    def get_channel(self, server: beacon_server.BeaconServer, channel_id: str) -> beacon_channel.BeaconChannel:
+        """Gets a channel from a server."""
+        raise BeaconDriverUnsupported()
+
+    def get_server(self, server_id: str) -> beacon_server.BeaconServer:
+        """Gets a server."""
+        raise BeaconDriverUnsupported()
+
+    async def fetch_server(self, server_id: str) -> beacon_server.BeaconServer:
+        """Fetches a server from the platform API."""
+        raise BeaconDriverUnsupported()
+
+    def get_webhook(self, webhook_id: str) -> beacon_webhook.BeaconWebhook:
+        """Gets a webhook."""
+
+        # Many libraries don't have a built-in webhook cache, so Beacon provides its own.
+        # However, this method can still be overwritten as needed.
+
+        return self._webhooks.get_webhook(webhook_id)
+
+    async def fetch_webhook(self, webhook_id: str) -> beacon_webhook.BeaconWebhook:
+        """Fetches a webhook from the platform API."""
+
+        # Preferrably this should call self._webhooks.store_webhook(webhook) after fetching the webhook
+        # to store the webhook to cache, unless the library has a webhook cache of its own.
+
+        raise BeaconDriverUnsupported()
+
+    async def send(self, destination: beacon_messageable.BeaconMessageable,
+                   content: beacon_message.BeaconMessageContent, send_as: beacon_user.BeaconUser):
+        """Sends a message to a given destination."""
+        raise BeaconDriverUnsupported()

shinobu/beacon/models/driver.py

@@ -0,0 +1,17 @@
+from shinobu.beacon.models import user, server as beacon_server
+
+class BeaconMember(user.BeaconUser):
+    """A class representing server members."""
+
+    def __init__(self, object_id: str, platform: str, name: str, server: beacon_server.BeaconServer,
+                 display_name: str | None = None, avatar_url: str | None = None):
+        super().__init__(object_id, platform, name, display_name=display_name, avatar_url=avatar_url)
+        self._server: beacon_server.BeaconServer = server
+
+    @property
+    def server(self) -> beacon_server.BeaconServer:
+        return self._server
+
+    @property
+    def server_id(self) -> str:
+        return self._server.id

shinobu/beacon/models/member.py

@@ -0,0 +1,72 @@
+from shinobu.beacon.models import (content as beacon_content, abc, user as beacon_user, channel as beacon_channel,
+                                   server as beacon_server, webhook as beacon_webhook)
+
+class BeaconMessageContent:
+    def __init__(self):
+        self._blocks: dict = {}
+
+    def to_plaintext(self) -> str:
+        components: list = []
+        for block in self._blocks:
+            block_obj: beacon_content.BeaconContentBlock = self._blocks[block]
+
+            # We'll restrict blocks to BeaconContentText only here
+            if type(block_obj) is not beacon_content.BeaconContentBlock:
+                continue
+
+            components.append(block_obj.content)
+
+        return "\n".join(components)
+
+class BeaconMessage(abc.BeaconABC):
+    def __init__(self, message_id: str, platform: str, author: beacon_user.BeaconUser | beacon_webhook.BeaconWebhook,
+                 server: beacon_server.BeaconServer | None = None, channel: beacon_channel.BeaconChannel | None = None,
+                 content: str | dict | None = None):
+        super().__init__(message_id, platform)
+        self._author: beacon_user.BeaconUser | beacon_webhook.BeaconWebhook = author
+        self._server: beacon_server.BeaconServer | None = server
+        self._channel: beacon_channel.BeaconChannel | None = channel
+        self._content: str | dict | None = content
+
+    @property
+    def author(self) -> beacon_user.BeaconUser | beacon_webhook.BeaconWebhook:
+        return self._author
+
+    @property
+    def webhook(self) -> beacon_webhook.BeaconWebhook | None:
+        if type(self._author) is not beacon_webhook.BeaconWebhook:
+            return None
+
+        return self._author
+
+    @property
+    def server(self) -> beacon_server.BeaconServer | None:
+        return self._server
+
+    @property
+    def channel(self) -> beacon_channel.BeaconChannel | None:
+        return self._channel
+
+    @property
+    def content(self) -> str | dict | None:
+        return self._content
+
+    def edit_content(self, new_content: str | dict | None):
+        self._content = new_content
+
+    def to_dict(self, include_content: bool = False) -> dict:
+        """Returns a BeaconMessage object as a dictionary. Should be used for backing up
+        Beacon state."""
+        converted = {
+            "id": self.id,
+            "platform": self.platform,
+            "author_id": self.author.id,
+            "server_id": self.server.id if self.server else None,
+            "channel_id": self.channel.id if self.channel else None,
+            "webhook_id": self.webhook.id if self.webhook else None
+        }
+
+        if include_content:
+            converted.update({"content": self.content})
+
+        return converted

shinobu/beacon/models/message.py

@@ -0,0 +1,12 @@
+from shinobu.beacon.models import abc
+
+class BeaconMessageable(abc.BeaconABC):
+    """A base class for messageable objects (e.g. users, channels)."""
+
+    def __init__(self, object_id: str, platform: str, name: str):
+        super().__init__(object_id, platform)
+        self._name: str = name
+
+    @property
+    def name(self) -> str:
+        return self._name