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() await self.bot.version_check(*self.data.VERSION) await self.profiles.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) # TODO: 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 except IndexError: 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 # TODO: 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