(WIP) Add user profile module.
This commit is contained in:
8
src/modules/profiles/__init__.py
Normal file
8
src/modules/profiles/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from .cog import ProfileCog
|
||||
|
||||
async def setup(bot):
|
||||
await bot.add_cog(ProfileCog(bot))
|
||||
436
src/modules/profiles/cog.py
Normal file
436
src/modules/profiles/cog.py
Normal file
@@ -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)
|
||||
123
src/modules/profiles/community.py
Normal file
123
src/modules/profiles/community.py
Normal file
@@ -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"<Community communityid={self.communityid} row={self.row}>"
|
||||
|
||||
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
|
||||
158
src/modules/profiles/data.py
Normal file
158
src/modules/profiles/data.py
Normal file
@@ -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()
|
||||
138
src/modules/profiles/profile.py
Normal file
138
src/modules/profiles/profile.py
Normal file
@@ -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"<UserProfile profileid={self.profileid} profile={self.profile_row}>"
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user