From 8421c5359dec177434addda99e4da8aca43ff68a Mon Sep 17 00:00:00 2001 From: Interitio Date: Fri, 6 Jun 2025 00:05:41 +1000 Subject: [PATCH] (WIP) Add user profile module. --- src/modules/profiles/__init__.py | 8 + src/modules/profiles/cog.py | 436 ++++++++++++++++++++++++++++++ src/modules/profiles/community.py | 123 +++++++++ src/modules/profiles/data.py | 158 +++++++++++ src/modules/profiles/profile.py | 138 ++++++++++ 5 files changed, 863 insertions(+) create mode 100644 src/modules/profiles/__init__.py create mode 100644 src/modules/profiles/cog.py create mode 100644 src/modules/profiles/community.py create mode 100644 src/modules/profiles/data.py create mode 100644 src/modules/profiles/profile.py diff --git a/src/modules/profiles/__init__.py b/src/modules/profiles/__init__.py new file mode 100644 index 0000000..67decbf --- /dev/null +++ b/src/modules/profiles/__init__.py @@ -0,0 +1,8 @@ +import logging + +logger = logging.getLogger(__name__) + +from .cog import ProfileCog + +async def setup(bot): + await bot.add_cog(ProfileCog(bot)) diff --git a/src/modules/profiles/cog.py b/src/modules/profiles/cog.py new file mode 100644 index 0000000..6a56df5 --- /dev/null +++ b/src/modules/profiles/cog.py @@ -0,0 +1,436 @@ +import asyncio +from enum import Enum +from typing import Optional, overload +from datetime import timedelta + +import discord +from discord import app_commands as appcmds +from discord.ext import commands as cmds +from twitchAPI.helper import first +from twitchAPI.type import AuthScope +import twitchio +from twitchio.ext import commands +from twitchio import User +from twitchAPI.object.api import TwitchUser + + +from data.queries import ORDER +from meta import LionCog, LionBot, LionContext +from meta.logger import log_wrap +from utils.lib import utc_now +from . import logger +from .data import ProfileData +from .profile import UserProfile +from .community import Community + + +class ProfileCog(LionCog): + def __init__(self, bot: LionBot): + self.bot = bot + + self.data = bot.db.load_registry(ProfileData()) + + self._profile_migrators = {} + self._comm_migrators = {} + + async def cog_load(self): + await self.data.init() + + async def cog_check(self, ctx): + return True + + # Profile API + def add_profile_migrator(self, migrator, name=None): + name = name or migrator.__name__ + self._profile_migrators[name or migrator.__name__] = migrator + + logger.info( + f"Added user profile migrator {name}: {migrator}" + ) + return migrator + + def del_profile_migrator(self, name: str): + migrator = self._profile_migrators.pop(name, None) + + logger.info( + f"Removed user profile migrator {name}: {migrator}" + ) + + @log_wrap(action="profile migration") + async def migrate_profile(self, source_profile, target_profile) -> list[str]: + logger.info( + f"Beginning user profile migration from {source_profile!r} to {target_profile!r}" + ) + results = [] + # Wrap this in a transaction so if something goes wrong with migration, + # we roll back safely (although this may mess up caches) + async with self.bot.db.connection() as conn: + self.bot.db.conn = conn + async with conn.transaction(): + for name, migrator in self._profile_migrators.items(): + try: + result = await migrator(source_profile, target_profile) + if result: + results.append(result) + except Exception: + logger.exception( + f"Unexpected exception running user profile migrator {name} " + f"migrating {source_profile!r} to {target_profile!r}." + ) + raise + + # Move all Discord and Twitch profile references over to the new profile + discord_rows = await self.data.DiscordProfileRow.table.update_where( + profileid=source_profile.profileid + ).set(profileid=target_profile.profileid) + results.append(f"Migrated {len(discord_rows)} attached discord account(s).") + + twitch_rows = await self.data.TwitchProfileRow.table.update_where( + profileid=source_profile.profileid + ).set(profileid=target_profile.profileid) + results.append(f"Migrated {len(twitch_rows)} attached twitch account(s).") + + # And then mark the old profile as migrated + await source_profile.profile_row.update(migrated=target_profile.profileid) + results.append("Marking old profile as migrated.. finished!") + return results + + async def fetch_profile_by_id(self, profile_id: int) -> UserProfile: + """ + Fetch a UserProfile by the given id. + """ + return await UserProfile.fetch(self.bot, profile_id=profile_id) + + async def fetch_profile_discord(self, user: discord.Member | discord.User) -> UserProfile: + """ + Fetch or create a UserProfile from the provided discord account. + """ + profile = await UserProfile.fetch_from_discordid(self.bot, user.id) + if profile is None: + profile = await UserProfile.create_from_discord(self.bot, user) + return profile + + async def fetch_profile_twitch(self, user: twitchio.User) -> UserProfile: + """ + Fetch or create a UserProfile from the provided twitch account. + """ + profile = await UserProfile.fetch_from_twitchid(self.bot, user.id) + if profile is None: + profile = await UserProfile.create_from_twitch(self.bot, user) + return profile + + # Community API + def add_community_migrator(self, migrator, name=None): + name = name or migrator.__name__ + self._comm_migrators[name or migrator.__name__] = migrator + + logger.info( + f"Added community migrator {name}: {migrator}" + ) + return migrator + + def del_community_migrator(self, name: str): + migrator = self._comm_migrators.pop(name, None) + + logger.info( + f"Removed community migrator {name}: {migrator}" + ) + + @log_wrap(action="community migration") + async def migrate_community(self, source_comm, target_comm) -> list[str]: + logger.info( + f"Beginning community migration from {source_comm!r} to {target_comm!r}" + ) + results = [] + # Wrap this in a transaction so if something goes wrong with migration, + # we roll back safely (although this may mess up caches) + async with self.bot.db.connection() as conn: + self.bot.db.conn = conn + async with conn.transaction(): + for name, migrator in self._comm_migrators.items(): + try: + result = await migrator(source_comm, target_comm) + if result: + results.append(result) + except Exception: + logger.exception( + f"Unexpected exception running community migrator {name} " + f"migrating {source_comm!r} to {target_comm!r}." + ) + raise + + # Move all Discord and Twitch community preferences over to the new profile + discord_rows = await self.data.DiscordCommunityRow.table.update_where( + profileid=source_comm.communityid + ).set(communityid=target_comm.communityid) + results.append(f"Migrated {len(discord_rows)} attached discord guilds.") + + twitch_rows = await self.data.TwitchCommunityRow.table.update_where( + communityid=source_comm.communityid + ).set(communityid=target_comm.communityid) + results.append(f"Migrated {len(twitch_rows)} attached twitch channel(s).") + + # And then mark the old community as migrated + await source_comm.update(migrated=target_comm.communityid) + results.append("Marking old community as migrated.. finished!") + return results + + async def fetch_community_by_id(self, community_id: int) -> Community: + """ + Fetch a Community by the given id. + """ + return await Community.fetch(self.bot, community_id=community_id) + + async def fetch_community_discord(self, guild: discord.Guild) -> Community: + """ + Fetch or create a Community from the provided discord guild. + """ + comm = await Community.fetch_from_discordid(self.bot, guild.id) + if comm is None: + comm = await Community.create_from_discord(self.bot, guild) + return comm + + async def fetch_community_twitch(self, user: twitchio.User) -> Community: + """ + Fetch or create a Community from the provided twitch account. + """ + community = await Community.fetch_from_twitchid(self.bot, user.id) + if community is None: + community = await Community.create_from_twitch(self.bot, user) + return community + + # ----- Profile Commands ----- + @cmds.hybrid_group( + name='profiles', + description="Base comand group for user profiles." + ) + async def profiles_grp(self, ctx: LionContext): + ... + + @profiles_grp.group( + name='link', + description="Base command group for linking profiles" + ) + async def profiles_link_grp(self, ctx: LionContext): + ... + + @profiles_link_grp.command( + name='twitch', + description="Link a twitch account to your current profile." + ) + async def profiles_link_twitch_cmd(self, ctx: LionContext): + if not ctx.interaction: + return + + await ctx.interaction.response.defer(ephemeral=True) + + # Ask the user to go through auth to get their userid + auth_cog = self.bot.get_cog('TwitchAuthCog') + flow = await auth_cog.start_auth() + message = await ctx.reply( + f"Please [click here]({flow.auth.return_auth_url()}) to link your profile " + "to Twitch." + ) + authrow = await flow.run() + await message.edit( + content="Authentication Complete! Beginning profile merge..." + ) + + # results = await self.crocbot.fetch_users(ids=[authrow.userid]) + # if not results: + # logger.error( + # f"User {authrow} obtained from Twitch authentication does not exist." + # ) + # await ctx.error_reply("Sorry, something went wrong. Please try again later!") + # return + + # user = results[0] + try: + user = await first(self.bot.twitch.get_users(user_ids=[str(authrow.userid)])) + except Exception: + logger.error( + f"Looking up user {authrow} from Twitch authentication flow raised an error.", + exc_info=True + ) + await ctx.error_reply("Sorry, something went wrong. Please try again later!") + return + + if user is None: + logger.error( + f"User {authrow} obtained from Twitch authentication does not exist." + ) + await ctx.error_reply("Sorry, something went wrong. Please try again later!") + return + + + # Retrieve author's profile if it exists + author_profile = await UserProfile.fetch_from_discordid(self.bot, ctx.author.id) + + # Check if the twitch-side user has a profile + source_profile = await UserProfile.fetch_from_twitchid(self.bot, user.id) + + if author_profile and source_profile is None: + # All we need to do is attach the twitch row + await author_profile.attach_twitch(user.id) + await message.edit( + content=f"Successfully added Twitch account **{user.display_name}**! There was no profile data to merge." + ) + elif source_profile and author_profile is None: + # Attach the discord row to the profile + await source_profile.attach_discord(ctx.author.id) + await message.edit( + content=f"Successfully connected to Twitch profile **{user.display_name}**! There was no profile data to merge." + ) + elif source_profile is None and author_profile is None: + profile = await UserProfile.create_from_discord(self.bot, ctx.author) + await profile.attach_twitch(user.id) + + await message.edit( + content=f"Opened a new user profile for you and linked Twitch account **{user.display_name}**." + ) + elif author_profile.profileid == source_profile.profileid: + await message.edit( + content=f"The Twitch account **{user.display_name}** is already linked to your profile!" + ) + else: + # Migrate the existing profile data to the new profiles + try: + results = await self.migrate_profile(source_profile, author_profile) + except Exception: + await ctx.error_reply( + "An issue was encountered while merging your account profiles!\n" + "Migration rolled back, no data has been lost.\n" + "The developer has been notified. Please try again later!" + ) + raise + + content = '\n'.join(( + "## Connecting Twitch account and merging profiles...", + *results, + "**Successfully linked account and merged profile data!**" + )) + await message.edit(content=content) + + # ----- Community Commands ----- + @cmds.hybrid_group( + name='community', + description="Base comand group for community profiles." + ) + async def community_grp(self, ctx: LionContext): + ... + + @community_grp.group( + name='link', + description="Base command group for linking communities" + ) + async def community_link_grp(self, ctx: LionContext): + ... + + @community_link_grp.command( + name='twitch', + description="Link a twitch account to this community." + ) + @appcmds.guild_only() + @appcmds.default_permissions(manage_guild=True) + async def comm_link_twitch_cmd(self, ctx: LionContext): + if not ctx.interaction: + return + assert ctx.guild is not None + + await ctx.interaction.response.defer(ephemeral=True) + + if not ctx.author.guild_permissions.manage_guild: + await ctx.error_reply("You need the `MANAGE_GUILD` permission to link this guild to a community.") + return + + # Ask the user to go through auth to get their userid + auth_cog = self.bot.get_cog('TwitchAuthCog') + flow = await auth_cog.start_auth( + scopes=[ + AuthScope.CHAT_EDIT, + AuthScope.CHAT_READ, + AuthScope.MODERATION_READ, + AuthScope.CHANNEL_BOT, + ] + ) + message = await ctx.reply( + f"Please [click here]({flow.auth.return_auth_url()}) to link your Twitch channel to this server." + ) + authrow = await flow.run() + await message.edit( + content="Authentication Complete! Beginning community profile merge..." + ) + + # results = await self.crocbot.fetch_users(ids=[authrow.userid]) + # if not results: + # logger.error( + # f"User {authrow} obtained from Twitch authentication does not exist." + # ) + # await ctx.error_reply("Sorry, something went wrong. Please try again later!") + # return + + # user = results[0] + try: + user = await first(self.bot.twitch.get_users(user_ids=[str(authrow.userid)])) + except Exception: + logger.error( + f"Looking up user {authrow} from Twitch authentication flow raised an error.", + exc_info=True + ) + await ctx.error_reply("Sorry, something went wrong. Please try again later!") + return + + if user is None: + logger.error( + f"User {authrow} obtained from Twitch authentication does not exist." + ) + await ctx.error_reply("Sorry, something went wrong. Please try again later!") + return + + # Retrieve author's profile if it exists + guild_comm = await Community.fetch_from_discordid(self.bot, ctx.guild.id) + + # Check if the twitch-side user has a profile + twitch_comm = await Community.fetch_from_twitchid(self.bot, user.id) + + if guild_comm and twitch_comm is None: + # All we need to do is attach the twitch row + await guild_comm.attach_twitch(user.id) + await message.edit( + content=f"Successfully linked Twitch channel **{user.display_name}**! There was no community data to merge." + ) + elif twitch_comm and guild_comm is None: + # Attach the discord row to the profile + await twitch_comm.attach_discord(ctx.guild.id) + await message.edit( + content=f"Successfully connected to Twitch channel **{user.display_name}**!" + ) + elif twitch_comm is None and guild_comm is None: + profile = await Community.create_from_discord(self.bot, ctx.guild) + await profile.attach_twitch(user.id) + + await message.edit( + content=f"Created a new community for this server and linked Twitch account **{user.display_name}**." + ) + elif guild_comm.communityid == twitch_comm.communityid: + await message.edit( + content=f"This server is already linked to the Twitch channel **{user.display_name}**!" + ) + else: + # Migrate the existing profile data to the new profiles + try: + results = await self.migrate_community(twitch_comm, guild_comm) + except Exception: + await ctx.error_reply( + "An issue was encountered while merging your community profiles!\n" + "Migration rolled back, no data has been lost.\n" + "The developer has been notified. Please try again later!" + ) + raise + + content = '\n'.join(( + "## Connecting Twitch account and merging community profiles...", + *results, + "**Successfully linked account and merged community data!**" + )) + await message.edit(content=content) diff --git a/src/modules/profiles/community.py b/src/modules/profiles/community.py new file mode 100644 index 0000000..0812f5b --- /dev/null +++ b/src/modules/profiles/community.py @@ -0,0 +1,123 @@ +from typing import Optional, Self + +import discord + +from meta import LionBot +from utils.lib import utc_now + +from . import logger +from .data import ProfileData + + + +class Community: + def __init__(self, bot: LionBot, community_row): + self.bot = bot + self.row: ProfileData.CommunityRow = community_row + + @property + def cog(self): + return self.bot.get_cog('ProfileCog') + + @property + def data(self) -> ProfileData: + return self.cog.data + + @property + def communityid(self): + return self.row.communityid + + def __repr__(self): + return f"" + + async def attach_discord(self, guildid: int): + """ + Attach a new discord guild to this community. + Assumes the discord guild is not already associated to a community. + """ + discord_row = await self.data.DiscordCommunityRow.create( + communityid=self.communityid, + guildid=guildid + ) + logger.info( + f"Attached discord guild {guildid} to community {self!r}" + ) + return discord_row + + async def attach_twitch(self, channelid: str): + """ + Attach a new Twitch user channel to this community. + """ + twitch_row = await self.data.TwitchCommunityRow.create( + communityid=self.communityid, + channelid=str(channelid) + ) + logger.info( + f"Attached twitch channel {channelid} to community {self!r}" + ) + return twitch_row + + async def discord_guilds(self) -> list[ProfileData.DiscordCommunityRow]: + """ + Fetch the Discord guild rows associated to this community. + """ + return await self.data.DiscordCommunityRow.fetch_where(communityid=self.communityid) + + async def twitch_channels(self) -> list[ProfileData.TwitchCommunityRow]: + """ + Fetch the Twitch user rows associated to this profile. + """ + return await self.data.TwitchCommunityRow.fetch_where(communityid=self.communityid) + + @classmethod + async def fetch(cls, bot: LionBot, community_id: int) -> Self: + community_row = await bot.get_cog('ProfileCog').data.CommunityRow.fetch(community_id) + if community_row is None: + raise ValueError("Provided community_id does not exist.") + return cls(bot, community_row) + + @classmethod + async def fetch_from_twitchid(cls, bot: LionBot, channelid: int | str) -> Optional[Self]: + data = bot.get_cog('ProfileCog').data + rows = await data.TwitchCommunityRow.fetch_where(channelid=str(channelid)) + if rows: + return await cls.fetch(bot, rows[0].communityid) + + @classmethod + async def fetch_from_discordid(cls, bot: LionBot, guildid: int) -> Optional[Self]: + data = bot.get_cog('ProfileCog').data + rows = await data.DiscordCommunityRow.fetch_where(guildid=guildid) + if rows: + return await cls.fetch(bot, rows[0].communityid) + + @classmethod + async def create(cls, bot: LionBot, **kwargs) -> Self: + """ + Create a new empty community with the given initial arguments. + + Communities should usually be created using `create_from_discord` or `create_from_twitch` + to correctly setup initial preferences (e.g. name, avatar). + """ + # Create a new community + data = bot.get_cog('ProfileCog').data + row = await data.CommunityRow.create(created_at=utc_now(), **kwargs) + return await cls.fetch(bot, row.communityid) + + @classmethod + async def create_from_discord(cls, bot: LionBot, guild: discord.Guild, **kwargs) -> Self: + """ + Create a new community using the given Discord guild as a base. + """ + self = await cls.create(bot, **kwargs) + await self.attach_discord(guild.id) + return self + + @classmethod + async def create_from_twitch(cls, bot: LionBot, user, **kwargs) -> Self: + """ + Create a new profile using the given Twitch channel user as a base. + The provided `user` must have an `id` attribute. + """ + self = await cls.create(bot, **kwargs) + await self.attach_twitch(str(user.id)) + return self diff --git a/src/modules/profiles/data.py b/src/modules/profiles/data.py new file mode 100644 index 0000000..f3e764c --- /dev/null +++ b/src/modules/profiles/data.py @@ -0,0 +1,158 @@ +from data import Registry, RowModel +from data.columns import Integer, String, Timestamp + + +class ProfileData(Registry): + class UserProfileRow(RowModel): + """ + Schema + ------ + CREATE TABLE user_profiles( + profileid SERIAL PRIMARY KEY, + nickname TEXT, + migrated INTEGER REFERENCES user_profiles (profileid) ON DELETE CASCADE ON UPDATE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + """ + _tablename_ = 'user_profiles' + _cache_ = {} + + profileid = Integer(primary=True) + nickname = String() + migrated = Integer() + created_at = Timestamp() + + + class DiscordProfileRow(RowModel): + """ + Schema + ------ + CREATE TABLE profiles_discord( + linkid SERIAL PRIMARY KEY, + profileid INTEGER NOT NULL REFERENCES user_profiles (profileid) ON DELETE CASCADE ON UPDATE CASCADE, + userid BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX profiles_discord_profileid ON profiles_discord (profileid); + CREATE UNIQUE INDEX profiles_discord_userid ON profiles_discord (userid); + """ + _tablename_ = 'profiles_discord' + _cache_ = {} + + linkid = Integer(primary=True) + profileid = Integer() + userid = Integer() + created_at = Integer() + + @classmethod + async def fetch_profile(cls, profileid: int): + rows = await cls.fetch_where(profiled=profileid) + return next(rows, None) + + + class TwitchProfileRow(RowModel): + """ + Schema + ------ + CREATE TABLE profiles_twitch( + linkid SERIAL PRIMARY KEY, + profileid INTEGER NOT NULL REFERENCES user_profiles (profileid) ON DELETE CASCADE ON UPDATE CASCADE, + userid TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX profiles_twitch_profileid ON profiles_twitch (profileid); + CREATE UNIQUE INDEX profiles_twitch_userid ON profiles_twitch (userid); + """ + _tablename_ = 'profiles_twitch' + _cache_ = {} + + linkid = Integer(primary=True) + profileid = Integer() + userid = String() + created_at = Timestamp() + + @classmethod + async def fetch_profile(cls, profileid: int): + rows = await cls.fetch_where(profiled=profileid) + return next(rows, None) + + class CommunityRow(RowModel): + """ + Schema + ------ + CREATE TABLE communities( + communityid SERIAL PRIMARY KEY, + migrated INTEGER REFERENCES user_profiles (profileid) ON DELETE CASCADE ON UPDATE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + """ + _tablename_ = 'communities' + _cache_ = {} + + communityid = Integer(primary=True) + migrated = Integer() + created_at = Timestamp() + + class DiscordCommunityRow(RowModel): + """ + Schema + ------ + CREATE TABLE communities_discord( + guildid BIGINT PRIMARY KEY, + communityid INTEGER NOT NULL REFERENCES communities (communityid) ON DELETE CASCADE ON UPDATE CASCADE, + linked_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + """ + _tablename_ = 'communities_discord' + _cache_ = {} + + guildid = Integer(primary=True) + communityid = Integer() + linked_at = Timestamp() + + @classmethod + async def fetch_community(cls, communityid: int): + rows = await cls.fetch_where(communityd=communityid) + return next(rows, None) + + class TwitchCommunityRow(RowModel): + """ + Schema + ------ + CREATE TABLE communities_twitch( + channelid TEXT PRIMARY KEY, + communityid INTEGER NOT NULL REFERENCES communities (communityid) ON DELETE CASCADE ON UPDATE CASCADE, + linked_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + """ + _tablename_ = 'communities_twitch' + _cache_ = {} + + channelid = String(primary=True) + communityid = Integer() + linked_at = Timestamp() + + @classmethod + async def fetch_community(cls, communityid: int): + rows = await cls.fetch_where(communityd=communityid) + return next(rows, None) + + class CommunityMemberRow(RowModel): + """ + Schema + ------ + CREATE TABLE community_members( + memberid SERIAL PRIMARY KEY, + communityid INTEGER NOT NULL REFERENCES communities (communityid) ON DELETE CASCADE ON UPDATE CASCADE, + profileid INTEGER NOT NULL REFERENCES user_profiles (profileid) ON DELETE CASCADE ON UPDATE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE UNIQUE INDEX community_members_communityid_profileid ON community_members (communityid, profileid); + """ + _tablename_ = 'community_members' + _cache_ = {} + + memberid = Integer(primary=True) + communityid = Integer() + profileid = Integer() + created_at = Timestamp() diff --git a/src/modules/profiles/profile.py b/src/modules/profiles/profile.py new file mode 100644 index 0000000..b8b2aff --- /dev/null +++ b/src/modules/profiles/profile.py @@ -0,0 +1,138 @@ +from typing import Optional, Self + +import discord + +from meta import LionBot +from utils.lib import utc_now + +from . import logger +from .data import ProfileData + + + +class UserProfile: + def __init__(self, bot: LionBot, profile_row): + self.bot = bot + self.profile_row: ProfileData.UserProfileRow = profile_row + + @property + def cog(self): + return self.bot.get_cog('ProfileCog') + + @property + def data(self) -> ProfileData: + return self.cog.data + + @property + def profileid(self): + return self.profile_row.profileid + + def __repr__(self): + return f"" + + async def get_name(self) -> Optional[str]: + return self.profile_row.nickname + + async def attach_discord(self, userid: int): + """ + Attach a new discord user to this profile. + Assumes the discord user does not itself have a profile. + """ + discord_row = await self.data.DiscordProfileRow.create( + profileid=self.profileid, + userid=userid + ) + logger.info( + f"Attached discord user {userid} to profile {self!r}" + ) + return discord_row + + async def attach_twitch(self, userid: str): + """ + Attach a new Twitch user to this profile. + """ + twitch_row = await self.data.TwitchProfileRow.create( + profileid=self.profileid, + userid=userid + ) + logger.info( + f"Attached twitch user {userid} to profile {self!r}" + ) + return twitch_row + + async def discord_accounts(self) -> list[ProfileData.DiscordProfileRow]: + """ + Fetch the Discord accounts associated to this profile. + """ + return await self.data.DiscordProfileRow.fetch_where( + profileid=self.profileid + ).order_by( + 'created_at' + ) + + async def twitch_accounts(self) -> list[ProfileData.TwitchProfileRow]: + """ + Fetch the Twitch accounts associated to this profile. + """ + return await self.data.TwitchProfileRow.fetch_where( + profileid=self.profileid + ).order_by( + 'created_at' + ) + + @classmethod + async def fetch(cls, bot: LionBot, profile_id: int) -> Self: + profile_row = await bot.get_cog('ProfileCog').data.UserProfileRow.fetch(profile_id) + if profile_row is None: + raise ValueError("Provided profile_id does not exist.") + return cls(bot, profile_row) + + @classmethod + async def fetch_from_twitchid(cls, bot: LionBot, userid: int | str) -> Optional[Self]: + data = bot.get_cog('ProfileCog').data + rows = await data.TwitchProfileRow.fetch_where(userid=str(userid)) + if rows: + return await cls.fetch(bot, rows[0].profileid) + + @classmethod + async def fetch_from_discordid(cls, bot: LionBot, userid: int) -> Optional[Self]: + data = bot.get_cog('ProfileCog').data + rows = await data.DiscordProfileRow.fetch_where(userid=(userid)) + if rows: + return await cls.fetch(bot, rows[0].profileid) + + @classmethod + async def create(cls, bot: LionBot, **kwargs) -> Self: + """ + Create a new empty profile with the given initial arguments. + + Profiles should usually be created using `create_from_discord` or `create_from_twitch` + to correctly setup initial profile preferences (e.g. name, avatar). + """ + # Create a new profile + data = bot.get_cog('ProfileCog').data + profile_row = await data.UserProfileRow.create(created_at=utc_now()) + profile = await cls.fetch(bot, profile_row.profileid) + return profile + + @classmethod + async def create_from_discord(cls, bot: LionBot, user: discord.Member | discord.User, **kwargs) -> Self: + """ + Create a new profile using the given Discord user as a base. + """ + kwargs.setdefault('nickname', user.name) + profile = await cls.create(bot, **kwargs) + await profile.attach_discord(user.id) + return profile + + @classmethod + async def create_from_twitch(cls, bot: LionBot, user, **kwargs) -> Self: + """ + Create a new profile using the given Twitch user as a base. + + Assumes the provided `user` has `id` and `name` attributes. + """ + kwargs.setdefault('nickname', user.name) + profile = await cls.create(bot, **kwargs) + await profile.attach_twitch(str(user.id)) + return profile