Croc/croccybot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rewrite: Shop and economy system.

Browse Source
This commit is contained in:
Conatum
2022-12-17 19:29:20 +02:00
parent 4ef1b58007
commit 4014e0a3a6
20 changed files with 1622 additions and 1178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bot/constants.py
View File

@@ -1,2 +1,4 @@
CONFIG_FILE = "config/bot.conf"
DATA_VERSION = 13
MAX_COINS = 2147483647 - 1

16
bot/core/lion.py
View File

@@ -100,6 +100,22 @@ class Lion:
if member.display_name != self.data.display_name:
await self.data.update(display_name=member.display_name)
async def get_member(self) -> Optional[discord.Member]:
"""
Retrieve the member object for this Lion, if possible.
If the guild or member cannot be retrieved, returns None.
"""
guild = self.bot.get_guild(self.guildid)
if guild is not None:
member = guild.get_member(self.userid)
if member is None:
try:
member = await guild.fetch_member(self.userid)
except discord.HTTPException:
pass
return member
class Lions(LionCog):
def __init__(self, bot: LionBot):

37
bot/modules/economy/cog.py
View File

@@ -12,6 +12,8 @@ from meta import LionCog, LionBot, LionContext
from meta.errors import ResponseTimedOut
from babel import LocalBabel
from core.data import CoreData
from utils.ui import LeoUI, LeoModal, Confirm, Pager
from utils.lib import error_embed, MessageArgs, utc_now
@@ -109,13 +111,27 @@ class EconomyData(Registry, name='economy'):
@classmethod
async def execute_transaction(
cls,
transaction_type: TransactionType,
guildid: int, actorid: int,
from_account: int, to_account: int, amount: int,
description: str,
note: Optional[str] = None, reference: Optional[str] = None, reminding: Optional[int] = None
from_account: int, to_account: int, amount: int, bonus: int = 0,
refunds: int = None
):
...
transaction = await cls.create(
transactiontype=transaction_type,
guildid=guildid, actorid=actorid, amount=amount, bonus=bonus,
from_account=from_account, to_account=to_account,
refunds=refunds
)
if from_account is not None:
await CoreData.Member.table.update_where(
guildid=guildid, userid=from_account
).set(coins=(CoreData.Member.coins - (amount + bonus)))
if to_account is not None:
await CoreData.Member.table.update_where(
guildid=guildid, userid=to_account
).set(coins=(CoreData.Member.coins + (amount + bonus)))
return transaction
class ShopTransaction(RowModel):
"""
@@ -131,6 +147,19 @@ class EconomyData(Registry, name='economy'):
transactionid = Integer(primary=True)
itemid = Integer()
@classmethod
async def purchase_transaction(
cls,
guildid: int, actorid: int,
userid: int, itemid: int, amount: int
):
row = await EconomyData.Transaction.execute_transaction(
TransactionType.PURCHASE,
guildid=guildid, actorid=actorid, from_account=userid, to_account=None,
amount=amount
)
return await cls.create(transactionid=row.transactionid, itemid=itemid)
class TaskTransaction(RowModel):
"""
Schema

4
bot/modules/pending-rewrite/economy/__init__.py
View File

@@ -1,4 +0,0 @@
from .module import module
from . import send_cmd
from . import shop_cmds

4
bot/modules/pending-rewrite/economy/module.py
View File

@@ -1,4 +0,0 @@
from LionModule import LionModule
module = LionModule("Economy")

91
bot/modules/pending-rewrite/economy/send_cmd.py
View File

@@ -1,91 +0,0 @@
import discord
import datetime
from cmdClient.checks import in_guild
from settings import GuildSettings
from core import Lion
from .module import module
@module.cmd(
"send",
group="Economy",
desc="Send some coins to another member."
)
@in_guild()
async def cmd_send(ctx):
"""
Usage``:
{prefix}send <user mention> <amount>
Description:
Send the given number of coins to the mentioned user.
Example:
{prefix}send {ctx.author.mention} 1000000
"""
# Extract target and amount
# Handle a slightly more flexible input than stated
splits = ctx.args.split()
digits = [split.isdigit() for split in splits[:2]]
mentions = ctx.msg.mentions
if len(splits) < 2 or not any(digits) or not (all(digits) or mentions):
return await _send_usage(ctx)
if all(digits):
# Both are digits, hopefully one is a member id, and one is an amount.
target, amount = ctx.guild.get_member(int(splits[0])), int(splits[1])
if not target:
amount, target = int(splits[0]), ctx.guild.get_member(int(splits[1]))
if not target:
return await _send_usage(ctx)
elif digits[0]:
amount, target = int(splits[0]), mentions[0]
elif digits[1]:
target, amount = mentions[0], int(splits[1])
# Fetch the associated lions
target_lion = Lion.fetch(ctx.guild.id, target.id)
source_lion = Lion.fetch(ctx.guild.id, ctx.author.id)
# Check sanity conditions
if amount > source_lion.coins:
return await ctx.error_reply(
"Sorry {}, you do not have enough LionCoins to do that.".format(ctx.author.mention)
)
if target == ctx.author:
return await ctx.embed_reply("What is this, tax evasion?")
if target == ctx.client.user:
return await ctx.embed_reply("Thanks, but Ari looks after all my needs!")
if target.bot:
return await ctx.embed_reply("We are still waiting for {} to open an account.".format(target.mention))
# Finally, send the amount and the ack message
target_lion.addCoins(amount)
source_lion.addCoins(-amount)
embed = discord.Embed(
title="Funds transferred",
description="You have sent **{}** LionCoins to {}!".format(amount, target.mention),
colour=discord.Colour.orange(),
timestamp=datetime.datetime.utcnow()
).set_footer(text=str(ctx.author), icon_url=ctx.author.avatar_url)
await ctx.reply(embed=embed, reference=ctx.msg)
GuildSettings(ctx.guild.id).event_log.log(
"{} sent {} `{}` LionCoins.".format(
ctx.author.mention,
target.mention,
amount
),
title="Funds transferred"
)
async def _send_usage(ctx):
return await ctx.error_reply(
"**Usage:** `{prefix}send <mention> <amount>`\n"
"**Example:** {prefix}send {ctx.author.mention} 1000000".format(
prefix=ctx.best_prefix,
ctx=ctx
)
)

246
bot/modules/pending-rewrite/economy/shop_cmds.py
View File

@@ -1,246 +0,0 @@
import asyncio
import discord
from collections import defaultdict
from cmdClient.checks import in_guild
from .module import module
from .shop_core import ShopItem, ShopItemType, ColourRole
from wards import is_guild_admin
class ShopSession:
__slots__ = (
'key', 'response',
'_event', '_task'
)
_sessions = {}
def __init__(self, userid, channelid):
# Build unique context key for shop session
self.key = (userid, channelid)
self.response = None
# Cancel any existing sessions
if self.key in self._sessions:
self._sessions[self.key].cancel()
self._event = asyncio.Event()
self._task = None
# Add self to the session list
self._sessions[self.key] = self
@classmethod
def get(cls, userid, channelid) -> 'ShopSession':
"""
Get a ShopSession matching the given key, if it exists.
Otherwise, returns None.
"""
return cls._sessions.get((userid, channelid), None)
async def wait(self, timeout=None):
"""
Wait for a buy response. Return the set response or raise an appropriate exception.
"""
self._task = asyncio.create_task(self._event.wait())
try:
await asyncio.wait_for(self._task, timeout=timeout)
except asyncio.CancelledError:
# Session was cancelled, likely due to creation of a new session
raise
except asyncio.TimeoutError:
# Session timed out, likely due to reaching the timeout
raise
finally:
if self._sessions.get(self.key, None) == self:
self._sessions.pop(self.key, None)
return self.response
def set(self, response):
"""
Set response.
"""
self.response = response
self._event.set()
def cancel(self):
"""
Cancel a session.
"""
if self._task:
if self._sessions.get(self.key, None) == self:
self._sessions.pop(self.key, None)
self._task.cancel()
else:
raise ValueError("Cancelling a ShopSession that is already completed!")
@module.cmd(
'shop',
group="Economy",
desc="Open the guild shop.",
flags=('add==', 'remove==', 'clear')
)
@in_guild()
async def cmd_shop(ctx, flags):
"""
Usage``:
{prefix}shop
{prefix}shop --add <price>, <item>
{prefix}shop --remove
{prefix}shop --remove itemid, itemid, ...
{prefix}shop --remove itemname, itemname, ...
{prefix}shop --clear
Description:
Opens the guild shop. Visible items may be bought using `{prefix}buy`.
*Modifying the guild shop requires administrator permissions.*
"""
# TODO: (FUTURE) Register session (and cancel previous sessions) so we can track for buy
# Whether we are modifying the shop
modifying = any(value is not False for value in flags.values())
if modifying and not is_guild_admin(ctx.author):
return await ctx.error_reply(
"You need to be a guild admin to modify the shop!"
)
# Fetch all purchasable elements, this also populates the cache
shop_items = ShopItem.fetch_where(guildid=ctx.guild.id, deleted=False, purchasable=True)
if not shop_items and not modifying:
# TODO: (FUTURE) Add tip for guild admins about setting up
return await ctx.error_reply(
"Nothing to buy! Please come back later."
)
# Categorise items
item_cats = defaultdict(list)
for item in shop_items:
item_cats[item.item_type].append(item)
# If there is more than one category, ask for which category they want
# All FUTURE TODO stuff, to be refactored into a shop widget
# item_type = None
# if ctx.args:
# # Assume category has been entered
# ...
# elif len(item_cats) > 1:
# # TODO: Show selection menu
# item_type = ...
# ...
# else:
# # Pick the next one automatically
# item_type = next(iter(item_cats))
item_type = ShopItemType.COLOUR_ROLE
item_class = ColourRole
if item_type is not None:
items = [item for item in item_cats[item_type]]
embeds = item_class.cat_shop_embeds(
ctx.guild.id,
[item.itemid for item in items]
)
if modifying:
if flags['add']:
await item_class.parse_add(ctx, flags['add'])
elif flags['remove'] is not False:
await item_class.parse_remove(ctx, flags['remove'], items)
elif flags['clear']:
await item_class.parse_clear(ctx)
return
# Present shop pages
out_msg = await ctx.pager(embeds, add_cancel=True)
await ctx.cancellable(out_msg, add_reaction=False)
while True:
try:
response = await ShopSession(ctx.author.id, ctx.ch.id).wait(timeout=300)
except asyncio.CancelledError:
# User opened a new session
break
except asyncio.TimeoutError:
# Session timed out. Close the shop.
# TODO: (FUTURE) time out shop session by removing hint.
try:
embed = discord.Embed(
colour=discord.Colour.red(),
description="Shop closed! (Session timed out.)"
)
await out_msg.edit(
embed=embed
)
except discord.HTTPException:
pass
break
# Selection was made
# Sanity checks
# TODO: (FUTURE) Handle more flexible ways of selecting items
if int(response.args) >= len(items):
await response.error_reply(
"No item with this number exists! Please try again."
)
continue
item = items[int(response.args)]
if item.price > ctx.alion.coins:
await response.error_reply(
"Sorry, {} costs `{}` LionCoins, but you only have `{}`!".format(
item.display_name,
item.price,
ctx.alion.coins
)
)
continue
# Run the selection and keep the shop open in case they want to buy more.
# TODO: (FUTURE) The item may no longer exist
success = await item.buy(response)
if success and not item.allow_multi_select:
try:
await out_msg.delete()
except discord.HTTPException:
pass
break
@module.cmd(
'buy',
group="Hidden",
desc="Buy an item from the guild shop."
)
@in_guild()
async def cmd_buy(ctx):
"""
Usage``:
{prefix}buy <number>
Description:
Only usable while you have a shop open (see `{prefix}shop`).
Buys the selected item from the shop.
"""
# Check relevant session exists
session = ShopSession.get(ctx.author.id, ctx.ch.id)
if session is None:
return await ctx.error_reply(
"No shop open, nothing to buy!\n"
"Please open a shop with `{prefix}shop` first.".format(prefix=ctx.best_prefix)
)
# Check input is an integer
if not ctx.args.isdigit():
return await ctx.error_reply(
"**Usage:** `{prefix}buy <number>`, for example `{prefix}buy 1`.".format(prefix=ctx.best_prefix)
)
# Pass context back to session
session.set(ctx)
# TODO: (FUTURE) Buy command short-circuiting the shop command and acting on shop sessions
# TODO: (FUTURE) Inventory command showing the user's purchases

377
bot/modules/pending-rewrite/economy/shop_core/ColourRole.py
View File

@@ -1,377 +0,0 @@
import re
import asyncio
from typing import List
import datetime
import discord
from cmdClient.lib import SafeCancellation
from meta import client
from utils.lib import multiselect_regex, parse_ranges
from .ShopItem import ShopItem, ShopItemType
from .data import shop_items, shop_item_info, colour_roles
@ShopItem.register_item_class
class ColourRole(ShopItem):
item_type = ShopItemType.COLOUR_ROLE
allow_multi_select = False
buy_hint = (
"Buy a colour by typing, e.g.,`{prefix}buy 0`.\n"
"**Note: You may only own one colour at a time!**"
).format(prefix=client.prefix)
@property
def display_name(self):
return "<@&{}>".format(self.data.roleid)
@property
def roleid(self) -> int:
return self.data.roleid
@property
def role(self) -> discord.Role:
return self.guild.get_role(self.roleid)
@classmethod
async def create(cls, guildid, price, roleid, **kwargs):
"""
Create a new ColourRole item.
"""
shop_row = shop_items.insert(
guildid=guildid,
item_type=cls.item_type,
price=price
)
colour_roles.insert(
itemid=shop_row['itemid'],
roleid=roleid
)
return cls(shop_row['itemid'])
# Shop interface
@classmethod
def _cat_shop_embed_items(cls, items: List['ColourRole'], hint: str = buy_hint, **kwargs) -> List[discord.Embed]:
"""
Embed a list of items specifically for displaying in the shop.
Subclasses will usually extend or override this, if only to add metadata.
"""
if items:
# TODO: prefix = items[0].guild_settings.prefix.value
embeds = cls._cat_embed_items(items, **kwargs)
for embed in embeds:
embed.title = "{} shop!".format(cls.item_type.desc)
embed.description += "\n\n" + hint
else:
embed = discord.Embed(
title="{} shop!".format(cls.item_type.desc),
description="No colours available at the moment! Please come back later."
)
embeds = [embed]
return embeds
async def buy(self, ctx):
"""
Action when a member buys a color role.
Uses Discord as the source of truth (rather than the future inventory).
Removes any existing colour roles, and adds the purchased role.
Also notifies the user and logs the transaction.
"""
# TODO: Exclusivity should be handled at a higher level
# TODO: All sorts of error handling
member = ctx.author
# Fetch the set colour roles
colour_rows = shop_item_info.fetch_rows_where(
guildid=self.guildid,
item_type=self.item_type
)
roleids = (row.roleid for row in colour_rows)
roles = (self.guild.get_role(roleid) for roleid in roleids)
roles = set(role for role in roles if role)
# Compute the roles to add and remove
to_add = self.guild.get_role(self.roleid)
member_has = roles.intersection(member.roles)
if to_add in member_has:
await ctx.error_reply("You already have this colour!")
return False
to_remove = list(member_has)
# Role operations
if to_add:
try:
await member.add_roles(to_add, reason="Updating purchased colour role")
except discord.HTTPException:
# TODO: Add to log
to_add = None
pass
if to_remove:
try:
await member.remove_roles(*to_remove, reason="Updating purchased colour role")
except discord.HTTPException:
# TODO: Add to log
pass
# Only charge the member if everything went well
if to_add:
ctx.alion.addCoins(-self.price)
# Build strings for logging and response
desc = None # Description of reply message to the member
log_str = None # Description of event log message
log_error = False # Whether to log an error
if to_add:
if to_remove:
if len(to_remove) > 1:
rem_str = ', '.join(role.mention for role in to_remove[:-1]) + 'and' + to_remove[-1].mention
else:
rem_str = to_remove[0].mention
desc = "You have exchanged {} for {}. Enjoy!".format(rem_str, to_add.mention)
log_str = "{} exchanged {} for {}.".format(
member.mention,
rem_str,
to_add.mention
)
else:
desc = "You have bought {}. Enjoy!".format(to_add.mention)
log_str = "{} bought {}.".format(member.mention, to_add.mention)
else:
desc = (
"Something went wrong! Please try again later.\n"
"(I don't have the server permissions to give this role to you!)"
)
log_str = (
"{} bought `{}`, but I couldn't add the role!\n"
"Please ensure that I have permission to manage this role.\n"
"(I need to have the `manage_roles` permission, "
"and my top role needs to be above the colour roles.)"
).format(member.mention, self.roleid)
log_error = True
# Build and send embeds
reply_embed = discord.Embed(
colour=to_add.colour if to_add else discord.Colour.red(),
description=desc,
timestamp=datetime.datetime.utcnow()
)
if to_add:
reply_embed.set_footer(
text="New Balance: {} LC".format(ctx.alion.coins)
)
log_embed = discord.Embed(
title="Colour Role Purchased" if not log_error else "Error purchasing colour role.",
colour=discord.Colour.red() if log_error else discord.Colour.orange(),
description=log_str
)
try:
await ctx.reply(embed=reply_embed)
except discord.HTTPException:
pass
event_log = ctx.guild_settings.event_log.value
if event_log:
try:
await event_log.send(embed=log_embed)
except discord.HTTPException:
pass
if not to_add:
return False
return True
# Shop admin interface
@classmethod
async def parse_add(cls, ctx, args):
"""
Parse a request to add colour roles.
Syntax: `<price>, <role>`, with different lines treated as different entries.
Assumes the author is an admin.
"""
if not args:
raise SafeCancellation("No arguments given, nothing to do!")
lines = args.splitlines()
to_add = [] # List of (price, role) tuples to add
for line in lines:
# Syntax: <price>, <role>
splits = line.split(',')
splits = [split.strip() for split in splits]
if len(splits) < 2 or not splits[0].isdigit():
raise SafeCancellation("**Syntax:** `--add <price>, <role>`")
price = int(splits[0])
role = await ctx.find_role(splits[1], create=True, interactive=True, allow_notfound=False)
to_add.append((price, role))
# Add the roles to data
for price, role in to_add:
# TODO: Batch update would be best
await cls.create(ctx.guild.id, price, role.id)
# Report back
if len(to_add) > 1:
await ctx.reply(
embed=discord.Embed(
title="Shop Updated",
description="Added `{}` new colours to the shop!".format(len(to_add))
)
)
else:
await ctx.reply(
embed=discord.Embed(
title="Shop Updated",
description="Added {} to the shop for `{}` coins.".format(to_add[0][1].mention, to_add[0][0])
)
)
@classmethod
async def parse_remove(cls, ctx, args, items):
"""
Parse a request to remove colour roles.
Syntax: `<ids>` or `<command separated roles>`
Assumes the author is an admin.
"""
if not items:
raise SafeCancellation("Colour shop is empty, nothing to delete!")
to_delete = []
if args:
if re.search(multiselect_regex, args):
# ids were selected
indexes = parse_ranges(args)
to_delete = [items[index] for index in indexes if index < len(items)]
if not to_delete:
# Assume comma separated list of roles
splits = args.split(',')
splits = [split.strip() for split in splits]
available_roles = (item.role for item in items)
available_roles = [role for role in available_roles if role]
roles = [
await ctx.find_role(rolestr, collection=available_roles, interactive=True, allow_notfound=False)
for rolestr in splits
]
roleids = set(role.id for role in roles)
to_delete = [item for item in items if item.roleid in roleids]
else:
# Interactive multi-selector
itemids = [item.itemid for item in items]
embeds = cls.cat_shop_embeds(
ctx.guild.id,
itemids,
hint=("Please select colour(s) to remove, or `c` to cancel.\n"
"(Respond with e.g. `1, 2, 3` or `1-3`.)")
)
out_msg = await ctx.pager(embeds)
def check(msg):
valid = msg.channel == ctx.ch and msg.author == ctx.author
valid = valid and (re.search(multiselect_regex, msg.content) or msg.content.lower() == 'c')
return valid
try:
message = await ctx.client.wait_for('message', check=check, timeout=60)
except asyncio.TimeoutError:
await out_msg.delete()
await ctx.error_reply("Session timed out. No colour roles were removed.")
return
try:
await out_msg.delete()
await message.delete()
except discord.HTTPException:
pass
if message.content.lower() == 'c':
return
to_delete = [
items[index]
for index in parse_ranges(message.content) if index < len(items)
]
if not to_delete:
raise SafeCancellation("Nothing to delete!")
# Build an ack string before we delete
rolestr = to_delete[0].role.mention if to_delete[0].role else "`{}`".format(to_delete[0].roleid)
# Delete the items
shop_items.delete_where(itemid=[item.itemid for item in to_delete])
# Update the info cache
[shop_item_info.row_cache.pop(item.itemid, None) for item in to_delete]
# Ack and log
if len(to_delete) > 1:
try:
await ctx.reply(
embed=discord.Embed(
title="Colour Roles removed",
description="You have removed `{}` colour roles.".format(len(to_delete)),
colour=discord.Colour.orange()
)
)
except discord.HTTPException:
pass
event_log = ctx.guild_settings.event_log.value
if event_log:
try:
await event_log.send(
embed=discord.Embed(
title="Colour Roles deleted",
description="{} removed `{}` colour roles from the shop.".format(
ctx.author.mention,
len(to_delete)
),
timestamp=datetime.datetime.utcnow()
)
)
except discord.HTTPException:
pass
else:
try:
await ctx.reply(
embed=discord.Embed(
title="Colour Role removed",
description="You have removed the colour role {}.".format(rolestr),
colour=discord.Colour.orange()
)
)
except discord.HTTPException:
pass
event_log = ctx.guild_settings.event_log.value
if event_log:
try:
await event_log.send(
embed=discord.Embed(
title="Colour Role deleted",
description="{} removed the colour role {} from the shop.".format(
ctx.author.mention,
rolestr
),
timestamp=datetime.datetime.utcnow()
)
)
except discord.HTTPException:
pass
@classmethod
async def parse_clear(cls, ctx):
"""
Parse a request to clear colour roles.
Assumes the author is an admin.
"""
if await ctx.ask("Are you sure you want to remove all colour roles from the shop?"):
shop_items.delete_where(guildid=ctx.guild.id, item_type=cls.item_type)
await ctx.reply("All colour roles removed from the shop.")
ctx.guild_settings.event_log.log("{} cleared the colour role shop.".format(ctx.author.mention))

265
bot/modules/pending-rewrite/economy/shop_core/ShopItem.py
View File

@@ -1,265 +0,0 @@
import discord
import datetime
from typing import List
from meta import client
from utils.lib import FieldEnum
from data import Row
from settings import GuildSettings
from .data import shop_items, shop_item_info
class ShopItemType(FieldEnum):
COLOUR_ROLE = 'COLOUR_ROLE', 'Colour'
class ShopItem:
"""
Abstract base class representing an item in a guild shop.
"""
__slots__ = ('itemid', '_guild')
# Mapping of item types to class handlers
_item_classes = {} # ShopItemType -> ShopItem subclass
# Item type handled by the current subclass
item_type = None # type: ShopItemType
# Format string to use for each item of this type in the shop embed
shop_fmt = "`[{num:<{num_len}}]` | `{item.price:<{price_len}} LC` {item.display_name}"
# Shop input modifiers
allow_multi_select = True
buy_hint = None
def __init__(self, itemid, *args, **kwargs):
self.itemid = itemid # itemid in shop_items representing this info
self._guild = None # cached Guild
# Meta
@classmethod
def register_item_class(cls, itemcls):
"""
Decorator to register a class as a handler for a given item type.
The item type must be set as the `item_type` class attribute.
"""
cls._item_classes[itemcls.item_type] = itemcls
return itemcls
@classmethod
async def create(cls, guildid, price, *args, **kwargs):
"""
Create a new ShopItem of this type.
Must be implemented by each item class.
"""
raise NotImplementedError
@classmethod
def fetch_where(cls, **kwargs):
"""
Fetch ShopItems matching the given conditions.
Automatically filters by `item_type` if set in class, and not provided.
"""
if cls.item_type is not None and 'item_type' not in kwargs:
kwargs['item_type'] = cls.item_type
rows = shop_item_info.fetch_rows_where(**kwargs)
return [
cls._item_classes[ShopItemType(row.item_type)](row.itemid)
for row in rows
]
@classmethod
def fetch(cls, itemid):
"""
Fetch a single ShopItem by itemid.
"""
row = shop_item_info.fetch(itemid)
if row:
return cls._item_classes[ShopItemType(row.item_type)](row.itemid)
else:
return None
# Data and transparent data properties
@property
def data(self) -> Row:
"""
Return the cached data row for this item.
This is not guaranteed to be up to date.
This is also not guaranteed to keep existing during a session.
"""
return shop_item_info.fetch(self.itemid)
@property
def guildid(self) -> int:
return self.data.guildid
@property
def price(self) -> int:
return self.data.price
@property
def purchasable(self) -> bool:
return self.data.purchasable
# Computed properties
@property
def guild(self) -> discord.Guild:
if not self._guild:
self._guild = client.get_guild(self.guildid)
return self._guild
@property
def guild_settings(self) -> GuildSettings:
return GuildSettings(self.guildid)
# Display properties
@property
def display_name(self) -> str:
"""
Short name to display after purchasing the item, and by default in the shop.
"""
raise NotImplementedError
# Data manipulation methods
def refresh(self) -> Row:
"""
Refresh the stored data row.
"""
shop_item_info.row_cache.pop(self.itemid, None)
return self.data
def _update(self, **kwargs):
"""
Updates the data with the provided kwargs.
Subclasses are expected to override this is they provide their own updatable data.
This method does *not* refresh the data row. This is expect to be handled by `update`.
"""
handled = ('price', 'purchasable')
update = {key: kwargs[key] for key in handled}
if update:
shop_items.update_where(
update,
itemid=self.itemid
)
async def update(self, **kwargs):
"""
Update the shop item with the given kwargs.
"""
self._update()
self.refresh()
# Formatting
@classmethod
def _cat_embed_items(cls, items: List['ShopItem'], blocksize: int = 20,
fmt: str = shop_fmt, **kwargs) -> List[discord.Embed]:
"""
Build a list of embeds for the current item type from a list of items.
These embeds may be used anywhere multiple items may be shown,
including confirmations and shop pages.
Subclasses may extend or override.
"""
embeds = []
if items:
# Cut into blocks
item_blocks = [items[i:i+blocksize] for i in range(0, len(items), blocksize)]
for i, item_block in enumerate(item_blocks):
# Compute lengths
num_len = len(str((i * blocksize + len(item_block) - 1)))
max_price = max(item.price for item in item_block)
price_len = len(str(max_price))
# Format items
string_block = '\n'.join(
fmt.format(
item=item,
num=i * blocksize + j,
num_len=num_len,
price_len=price_len
) for j, item in enumerate(item_block)
)
# Build embed
embed = discord.Embed(
description=string_block,
timestamp=datetime.datetime.utcnow()
)
if len(item_blocks) > 1:
embed.set_footer(text="Page {}/{}".format(i+1, len(item_blocks)))
embeds.append(embed)
else:
# Empty shop case, should generally be avoided
embed = discord.Embed(
description="Nothing to show!"
)
embeds.append(embed)
return embeds
# Shop interface
@classmethod
def _cat_shop_embed_items(cls, items: List['ShopItem'], **kwargs) -> List[discord.Embed]:
"""
Embed a list of items specifically for displaying in the shop.
Subclasses will usually extend or override this, if only to add metadata.
"""
if items:
# TODO: prefix = items[0].guild_settings.prefix.value
prefix = client.prefix
embeds = cls._cat_embed_items(items, **kwargs)
for embed in embeds:
embed.title = "{} shop!".format(cls.item_type.desc)
embed.description = "{}\n\n{}".format(
embed.description,
"Buy items with `{prefix}buy <numbers>`, e.g. `{prefix}buy 1, 2, 3`.".format(
prefix=prefix
)
)
else:
embed = discord.Embed(
title="{} shop!".format(cls.item_type.desc),
description="This shop is empty! Please come back later."
)
embeds = [embed]
return embeds
@classmethod
def cat_shop_embeds(cls, guildid: int, itemids: List[int] = None, **kwargs) -> List[discord.Embed]:
"""
Format the items of this type (i.e. this category) as one or more embeds.
Subclasses may extend or override.
"""
if itemids is None:
# Get itemids if not provided
# TODO: Not using the row cache here, make sure we don't need an extended caching form
rows = shop_item_info.fetch_rows_where(
guildid=guildid,
item_type=cls.item_type,
purchasable=True,
deleted=False
)
itemids = [row.itemid for row in rows]
elif not all(itemid in shop_item_info.row_cache for itemid in itemids):
# Ensure cache is populated
shop_item_info.fetch_rows_where(itemid=itemids)
return cls._cat_shop_embed_items([cls(itemid) for itemid in itemids], **kwargs)
async def buy(self, ctx):
"""
Action to trigger when a member buys this item.
"""
raise NotImplementedError
# Shop admin interface
@classmethod
async def parse_new(self, ctx):
"""
Parse new shop items.
"""
raise NotImplementedError

4
bot/modules/pending-rewrite/economy/shop_core/__init__.py
View File

@@ -1,4 +0,0 @@
from . import data
from .ShopItem import ShopItem, ShopItemType
from .ColourRole import ColourRole

19
bot/modules/pending-rewrite/economy/shop_core/data.py
View File

@@ -1,19 +0,0 @@
from cachetools import LRUCache
from data import Table, RowTable
shop_items = Table('shop_items')
colour_roles = Table('shop_items_colour_roles', attach_as='colour_roles')
shop_item_info = RowTable(
'shop_item_info',
('itemid',
'guildid', 'item_type', 'price', 'purchasable', 'deleted', 'created_at',
'roleid', # Colour roles
),
'itemid',
cache=LRUCache(1000)
)

0
bot/modules/pending-rewrite/economy/shopadmin_cmd.py
View File

213
bot/modules/shop/cog.py
View File

@@ -1,3 +1,66 @@
"""
The initial Shopping Cog acts as the entry point to the shopping system.
It provides the commands:
/shop open [type:<acmpl str>]
/shop buy <item:<acmpl str>>
It also provides the "/editshop" group,
which each shop type should hook into (via placeholder groups)
to provide a shop creation and edit interface.
A "Shop" represents a colour shop, role shop, boost shop, etc.
Each Shop has a different editing interface (the ShopCog),
and a different user interface page (the Store).
When a user runs /shop open, the "StoreManager" opens.
This PagedUI serves as a top-level manager for each shop store,
even though it has no UI elements itself.
The user sees the first Store page, and can freely use the store UI
to interact with that Shop.
They may also use the "shop row" buttons to switch between shops.
Now, onto the individual shop systems.
When the user first opens a shop, a Customer is created.
The Customer represents the member themselves (i.e. the lion),
along with their inventory (a list of raw items that they own).
For fast access (for e.g. autocomplete), the Customer may be
cached, as long as it has a unique registry map.
Each initialised Shop has a collection of ShopItems.
The ShopItems represent individual objects of that shop,
and usually contain state particular to that shop.
The ShopItems shouldn't need to remember their origin shop,
or the Customer.
The Shop itself handles purchases, checking whether a customer can purchase
a given item, and running the purchase logic for an item.
Note: Timing out Acmpl state to handle caching shared states between acmpl?
Note: shop_item_info table which joins the others? Include guild_item_id,
and cache the items.
/shop open:
- Create the Customer
- Initialise the Shops, with the current Guild and the Customer
- Pass the shop stores to the StoreManager, along with the command interaction
- Run display on the StoreManager, which displays the first Store UI
- The StoreManager gets a store button from each store, then passes those back to the individuals
- This would be best done with a UILayout modification, but we don't have that yet
- The Store displays its UI, which is Customer-dependent, relying on the Shop for most computations.
- (The Store might not even need to keep the Customer, just using Shop methods to access them.)
- The Customer may make a purchase/refund/etc, which gets mapped back to the Shop.
- After the Customer has made an action, the Store refreshes its UI, from the Shop data.
/shop buy <item>:
- Instantiates the Customer and Shops (via acmpl state?), which should _not_ require any data lookups.
- Acmpl shows an intelligent list of matching items, nicely formatted
- (Shopitem itself can be responsible for formatting?)
- (Shop can be responsible for showing which items the user can purchase?)
- Command gets the shopitemid (possibly partitioned by guild), gets that item, gets the item Shop.
- Command then gets the Shop to purchase that item.
"""
import asyncio
import logging
from typing import Optional
@@ -7,12 +70,13 @@ from discord.ext import commands as cmds
from discord import app_commands as appcmds
from meta import LionBot, LionCog, LionContext
from utils import ui
from utils.lib import error_embed
from . import babel
from .shops.colours import ColourShop
from .shops.base import Customer, ShopCog
from .data import ShopData
logger = logging.getLogger(__name__)
@@ -20,90 +84,163 @@ _p = babel._p
class Shopping(LionCog):
Shops = [ColourShop]
# List of active Shop cogs
ShopCogs = ShopCog.active
def __init__(self, bot: LionBot):
self.bot = bot
self.data = None
self.shops = []
self.data = bot.db.load_registry(ShopData())
self.active_cogs = []
async def cog_load(self):
for Shop in self.Shops:
shop = Shop(self.bot, self.data)
await shop.load_into(self)
self.shops.append(shop)
await self.data.init()
for SCog in self.ShopCogs:
shop_cog = SCog(self.bot, self.data)
await shop_cog.load_into(self)
self.active_cogs.append(shop_cog)
async def cog_unload(self):
for shop in self.shops:
await shop.unload()
@cmds.hybrid_group(name='editshop')
@cmds.hybrid_group(
name=_p('group:editshop', 'editshop')
)
async def editshop_group(self, ctx: LionContext):
return
@cmds.hybrid_group(name='shop')
@cmds.hybrid_group(
name=_p('group:shop', 'shop')
)
async def shop_group(self, ctx: LionContext):
return
@shop_group.command(name='open')
@shop_group.command(
name=_p('cmd:shop_open', 'open'),
description=_p('cmd:shop_open|desc', "Open the server shop.")
)
async def shop_open_cmd(self, ctx: LionContext):
"""
Opens the shop UI for the current guild.
"""
...
t = self.bot.translator.t
# Typechecker guards
if not ctx.guild:
return
if not ctx.interaction:
return
await ctx.interaction.response.defer(ephemeral=True, thinking=True)
# Create the Customer
customer = await Customer.fetch(self.bot, self.data, ctx.guild.id, ctx.author.id)
# Create the Shops
shops = [await cog.make_shop_for(customer) for cog in self.active_cogs]
# TODO: Filter by shops which actually have items
if not shops:
await ctx.reply(
embed=error_embed(
t(_p('cmd:shop_open|error:no_shops', "There is nothing to buy!"))
),
ephemeral=True
)
return
# Extract the Stores
stores = [shop.make_store(ctx.interaction) for shop in shops]
# Build the StoreManager from the Stores
manager = StoreManager(self.bot, self.data, stores)
# Display the StoreManager
await manager.run(ctx.interaction)
await manager.wait()
# TODO: shortcut shop buy command
class StoreManager(ui.LeoUI):
def __init__(self, bot, data, shops):
def __init__(self, bot, data, stores, **kwargs):
super().__init__(**kwargs)
self.bot = bot
self.data = data
self.shops = shops
self.stores = stores
self.page_num = 0
# Original interaction that opened this shop
self._original: Optional[discord.Interaction] = None
# tuple of Buttons to each active store
self._store_row = self.make_buttons()
self._widgets = self.prepare_widgets()
async def redraw(self):
"""
Ask the current shop widget to redraw.
"""
...
self.page_num %= len(self.stores)
await self.stores[self.page_num].refresh()
await self.stores[self.page_num].redraw()
def make_buttons(self):
"""
Make a tuple of shop buttons.
If there is only one shop, return an empty tuple.
If there is only one shop, returns an empty tuple.
"""
if len(self.shops) <= 1:
return ()
buttons = []
for i, shop in enumerate(self.shops):
@ui.AButton(label=shop.name)
async def pressed_switch_shop(press: discord.Interaction, pressed):
await press.response.defer()
await self.change_page(i)
buttons.append(pressed_switch_shop)
return tuple(buttons)
t = self.bot.translator.t
def prepare_widgets(self):
widgets = []
for shop in self.shops:
widget = shop.make_widget()
# TODO: Update this when we have a UILayout class
# widget.layout.set_row('shops', self._store_row, affinity=1)
widget.shop_row = self._store_row
widgets.append(widget)
return widgets
buttons = []
if len(self.stores) > 1:
for i, store in enumerate(self.stores):
@ui.AButton(label=store.shop.name)
async def pressed_switch_shop(press: discord.Interaction, pressed):
await press.response.defer()
await self.change_page(i)
buttons.append(pressed_switch_shop)
@ui.AButton(
label=_p('ui:stores|button:close|label', "Close"),
emoji=self.bot.config.emojis.getemoji('cancel')
)
async def pressed_close(press: discord.Interaction, pressed):
await press.response.defer()
if not self._original.is_expired():
embed = discord.Embed(
title=t(_p('ui:stores|button:close|response|title', "Shop Closed")),
colour=discord.Colour.orange()
)
await self._original.edit_original_response(embed=embed, view=None)
await self.close()
buttons.append(pressed_close)
for button in buttons:
self.add_item(button)
return tuple(buttons)
async def change_page(self, i):
"""
Change to the given page number.
"""
...
self.page_num = i
self.page_num %= len(self.stores)
await self.redraw()
async def run(self, interaction):
self._original = interaction
for store in self.stores:
self.children.append(store)
store.set_store_row(self._store_row)
await self.redraw()
async def monitor(self):
"""
When one of the stores closes, we want all the stores to close,
along with this parent UI.
"""
# TODO
...

75
bot/modules/shop/data.py
View File

@@ -1,4 +1,4 @@
from enums import Enum
from enum import Enum
from cachetools import TTLCache
@@ -59,6 +59,35 @@ class ShopData(Registry, name='shop'):
itemid = Integer(primary=True)
roleid = Integer()
class ShopItemInfo(RowModel):
"""
A view joining the shop item sub-type information,
and including the guild id.
Schema
------
CREATE VIEW shop_item_info AS
SELECT
*,
row_number() OVER (PARTITION BY guildid ORDER BY itemid) AS guild_itemid
FROM
shop_items
LEFT JOIN shop_items_colour_roles USING (itemid)
ORDER BY itemid ASC;
"""
_tablename_ = 'shop_item_info'
_readonly_ = True
itemid = Integer(primary=True)
guild_itemid = Integer()
guildid = Integer()
item_type: Column[ShopItemType] = Column()
price = Integer()
purchasable = Bool()
deleted = Bool()
created_at = Timestamp()
roleid = Integer()
class MemberInventory(RowModel):
"""
Schema
@@ -80,8 +109,46 @@ class ShopData(Registry, name='shop'):
transactionid = Integer()
itemid = Integer()
async def fetch_inventory(self, guildid, userid) -> list['ShopData.MemberInventory']:
class MemberInventoryInfo(RowModel):
"""
Composite view joining the member inventory with shop item information.
Schema
------
CREATE VIEW member_inventory_info AS
SELECT
inv.inventoryid AS inventoryid,
inv.guildid AS guildid,
inv.userid AS userid,
inv.transactionid AS transactionid,
items.itemid AS itemid,
items.item_type AS item_type,
items.price AS price,
items.purchasable AS purchasable,
items.deleted AS deleted
FROM
member_inventory inv
LEFT JOIN shop_item_info items USING (itemid)
ORDER BY itemid ASC;
"""
_tablename_ = 'member_inventory_info'
_readonly_ = True
inventoryid = Integer(primary=True)
guildid = Integer()
userid = Integer()
transactionid = Integer()
itemid = Integer()
guild_itemid = Integer()
item_type: Column[ShopItemType] = Column()
price = Integer()
purchasable = Bool()
deleted = Bool()
roleid = Integer()
@classmethod
async def fetch_inventory_info(cls, guildid, userid) -> list['ShopData.MemberInventoryInfo']:
"""
Fetch the given member's inventory.
Fetch the information rows for the given members inventory.
"""
return await self.fetch_where(guildid=guildid, userid=userid)
return await cls.fetch_where(guildid=guildid, userid=userid)

2
bot/modules/shop/shops/__init__.py
View File

@@ -0,0 +1,2 @@
from . import base
from . import colours

218
bot/modules/shop/shops/base.py
View File

@@ -1,81 +1,225 @@
from typing import Type, TYPE_CHECKING
from weakref import WeakValueDictionary
import discord
from discord.ui.button import Button
from meta import LionBot
from meta import LionBot, LionCog
from utils import ui
from babel.translator import LazyStr
from ..data import ShopData
if TYPE_CHECKING:
from core.lion import Lion
class MemberInventory:
class ShopCog(LionCog):
"""
Minimal base class for a ShopCog.
"""
_shop_cls_: Type['Shop']
active: list[Type['ShopCog']] = []
def __init__(self, bot: LionBot, data: ShopData):
self.bot = bot
self.data = data
async def load_into(self, cog: LionCog):
"""
Load this ShopCog into the parent Shopping Cog.
Usually just attaches the editshop placeholder group, if applicable.
May also load the cog itself into the client,
if the ShopCog needs to provide global features
or commands.
"""
raise NotImplementedError
async def make_shop_for(self, customer: 'Customer'):
"""
Make a Shop instance for the provided Customer.
"""
shop = self._shop_cls_(self.bot, self.data, customer)
await shop.refresh()
return shop
@classmethod
def register(self, shop):
"""
Helper decorator to register a given ShopCog as active.
"""
self.active.append(shop)
return shop
class Customer:
"""
An interface to the member's inventory.
"""
def __init__(self, bot, shop_data, lion, inventory):
self.bot = bot
self.lion = lion
self.guildid = lion.guildid
self.userid = lion.userid
# A list of InventoryItems held by this user
_cache_ = WeakValueDictionary()
def __init__(self, bot: LionBot, shop_data: ShopData, lion, inventory: ShopData.MemberInventory):
self.bot = bot
self.data = shop_data
self.lion: 'Lion' = lion
# A list of InventoryItems held by this customer
self.inventory = inventory
@property
def guildid(self):
return self.lion.guildid
@property
def userid(self):
return self.lion.userid
@property
def balance(self):
return self.lion.data['coins']
@classmethod
async def fetch(cls, bot: LionBot, shop_data: ShopData, guildid: int, userid: int):
lion = await bot.core.lions.fetch(guildid, userid)
inventory = await shop_data.fetch_where(guildid=guildid, userid=userid)
inventory = await shop_data.MemberInventoryInfo.fetch_inventory_info(guildid, userid)
return cls(bot, shop_data, lion, inventory)
async def refresh(self):
"""
Refresh the data for this member.
"""
self.lion = self.bot.core.lions.fetch(self.guild.id, self.user.id)
data = self.bot.get_cog('Shopping').data
self.inventory_items = await data.InventoryItem.fetch_where(userid=self.userid, guildid=self.guildid)
self.lion = await self.bot.core.lions.fetch(self.guildid, self.userid)
await self.lion.data.refresh()
self.inventory = await self.data.MemberInventoryInfo.fetch_inventory_info(self.guildid, self.userid)
return self
class ShopItem:
"""
ABC representing a purchasable guild shop item.
"""
def __init__(self, data):
self.data = data
Base class representing a purchasable guild shop item.
async def purchase(self, userid):
"""
Called when a member purchases this item.
"""
...
In its most basic form, this is just a direct interface to the data,
with some formatting methods.
"""
def __init__(self, bot: LionBot, data: ShopData.ShopItemInfo):
self.bot = bot
self.data = data
class Shop:
"""
Base class representing a Shop for a particular member.
Base class representing a Shop for a particular Customer.
"""
def __init__(self, bot: LionBot, shop_data: ShopData, member: discord.Member):
# The name of this shop class, as a lazystring
_name_: LazyStr
# Store class describing the shop UI.
_store_cls_: Type['Store']
def __init__(self, bot: LionBot, shop_data: ShopData, customer: Customer):
self.bot = bot
self.data = shop_data
self.member = member
self.guild = member.guild
self.customer = customer
# A list of ShopItems that are currently visible to the member
self.items = []
# A map itemid: ShopItem of items viewable by the customer
self.items = {}
# Current inventory for the member
self.inventory = None
def purchasable(self):
"""
Retrieve a list of items purchasable by the customer.
"""
raise NotImplementedError
async def refresh(self):
...
"""
Refresh the shop and customer data.
"""
raise NotImplementedError
@property
def name(self):
"""
The localised name of this shop.
Usually just a context-aware translated version of cls._name_
"""
t = self.bot.translator.t
return t(self._name_)
async def purchase(self, itemid):
"""
Have the shop customer purchase the given (global) itemid.
Checks that the item is actually purchasable by the customer.
This method must be overridden in base classes.
"""
raise NotImplementedError
def make_store(self):
"""
Initialise and return a new Store UI for this shop.
"""
return self._store_cls_(self)
class Store(ui.LeoUI):
"""
Base UI for the different shops.
ABC for the UI used to interact with a given shop.
This must always be an ephemeral UI,
so extra permission checks are not required.
(Note that each Shop instance is specific to a single customer.)
"""
def __init__(self, bot: LionBot, data, shops):
self.bot = bot
self.data = data
self.shops = shops
def __init__(self, shop: Shop, interaction: discord.Interaction, **kwargs):
super().__init__(**kwargs)
# The shop this Store is an interface for
# Client, shop, and customer data is taken from here
# The Shop also manages most Customer object interaction, including purchases.
self.shop = shop
# The row of Buttons used to access different shops, if any
# Transient, will be deprecated by direct access to UILayout.
self.store_row = ()
# Current embed page
self.embed: Optional[discord.Embed] = None
# Current interaction to use
self.interaction: discord.Interaction = interaction
def set_store_row(self, row):
self.store_row = row
for item in row:
self.add_item(item)
async def refresh(self):
"""
Refresh all UI elements.
"""
raise NotImplementedError
async def redraw(self):
"""
Redraw the store UI.
"""
if self.interaction.is_expired():
# This is actually possible,
# If the user keeps using the UI,
# but never closes it until the origin interaction expires
raise ValueError("This interaction has expired!")
return
if self.embed is None:
await self.refresh()
await self.interaction.edit_original_response(embed=self.embed, view=self)
async def make_embed(self):
"""
Embed page for this shop.
"""
raise NotImplementedError

1195
bot/modules/shop/shops/colours.py
View File

File diff suppressed because it is too large Load Diff