feat: Simple twitch component.

profiles/twitch/component.py

@@ -0,0 +1,81 @@
+from typing import Optional
+import asyncio
+
+import twitchio
+from twitchio.ext import commands as cmds
+
+from meta import Bot
+from meta.logger import log_wrap
+from utils.lib import utc_now
+
+from . import logger
+
+from ..data import ProfilesData, UserProfile, TwitchProfileLink, Community, TwitchCommunityLink
+from ..profiles import ProfilesRegistry
+
+
+class ProfilesComponent(cmds.Component):
+    def __init__(self, bot: Bot):
+        self.bot = bot
+
+        self.data = bot.dbconn.load_registry(ProfilesData())
+        self.profiles = ProfilesRegistry(self.data)
+
+    # ----- API -----
+    async def component_load(self):
+        await self.data.init()
+
+    async def component_teardown(self):
+        pass
+
+    # ------ Commands -----
+    @log_wrap(isolate=True, action="Fetch Profile")
+    async def fetch_profile(
+        self,
+        user: twitchio.PartialUser,
+        touch: bool = False,
+    ) -> UserProfile:
+        """
+        Fetch or create the profile for the given user.
+        """
+        userid = str(user.id)
+        async with self.bot.dbconn.connection() as conn:
+            self.bot.dbconn.conn = conn 
+            async with conn.transaction():
+                profile = await self.profiles.get_profile_twitch(userid)
+                if profile is None:
+                    args = {}
+                    try:
+                        user = await user.user()
+                        args['nickname'] = user.display_name
+                        args['avatar'] = user.profile_image.url
+                    except twitchio.HTTPException:
+                        pass
+                    profile = await UserProfile.create(**args)
+                    await TwitchProfileLink.create(profileid=profile.profileid, userid=userid)
+                elif touch:
+                    await profile.update(last_seen=utc_now())
+
+        return profile
+
+    @log_wrap(isolate=True, action="Fetch Community")
+    async def fetch_community(
+        self,
+        channel: twitchio.PartialUser,
+        touch: bool = False,
+    ) -> Community:
+        """
+        Fetch or create the community for this channel.
+        """
+        chanid = channel.id
+        async with self.bot.dbconn.connection() as conn:
+            self.bot.dbconn.conn = conn
+            async with conn.transaction():
+                comm = await self.profiles.get_community_twitch(chanid)
+                if comm is None:
+                    comm = await Community.create()
+                    await TwitchCommunityLink.create(channelid=chanid, communityid=comm.communityid)
+                elif touch:
+                    await comm.update(last_seen=utc_now())
+        return comm
+