ruby/Ruby-Cogs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor Shop Cog to improve inventory management and trading functionalities. Enhance shop management commands for better item handling and user interaction. Update documentation for clarity on new features and usage.
Some checks are pending
Run pre-commit / Run pre-commit (push) Waiting to run
Details

Browse source
This commit is contained in:
Valerie 2025-05-24 05:36:18 -04:00
parent 7fc2053951
commit d507065184
9 changed files with 3006 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
casino/__init__.py Normal file
View file

@ -0,0 +1,9 @@
from .casino import Casino
__red_end_user_data_statement__ = "This cog stores discord IDs as needed for operation."
async def setup(bot):
cog = Casino(bot)
await bot.add_cog(cog)
await cog.initialise()

30
casino/cache.py Normal file
View file

@ -0,0 +1,30 @@
from typing import Dict, Optional
import discord
from redbot.core import Config
class OldMessageTypeManager:
def __init__(self, config: Config, enable_cache: bool = True):
self._config: Config = config
self.enable_cache = enable_cache
self._cached_guild: Dict[int, bool] = {}
async def get_guild(self, guild: discord.Guild) -> bool:
ret: bool
gid: int = guild.id
if self.enable_cache and gid in self._cached_guild:
ret = self._cached_guild[gid]
else:
ret = await self._config.guild_from_id(gid).use_old_style()
self._cached_guild[gid] = ret
return ret
async def set_guild(self, guild: discord.Guild, set_to: Optional[bool]) -> None:
gid: int = guild.id
if set_to is not None:
await self._config.guild_from_id(gid).use_old_style.set(set_to)
self._cached_guild[gid] = set_to
else:
await self._config.guild_from_id(gid).use_old_style.clear()
self._cached_guild[gid] = self._config.defaults["GUILD"]["use_old_style"]

1382
casino/casino.py Normal file
View file

File diff suppressed because it is too large Load diff

413
casino/data.py Normal file
View file

@ -0,0 +1,413 @@
import asyncio
import logging
from copy import deepcopy
import discord
from redbot.core import Config, bank
from collections import namedtuple
from .cache import OldMessageTypeManager
from .utils import is_input_unsupported, min_int, max_int
user_defaults = {
"Pending_Credits": 0,
"Membership": {"Name": "Basic", "Assigned": False},
"Played": {
"Allin": 0,
"Blackjack": 0,
"Coin": 0,
"Craps": 0,
"Cups": 0,
"Dice": 0,
"Hilo": 0,
"War": 0,
"Double": 0,
},
"Won": {
"Allin": 0,
"Blackjack": 0,
"Coin": 0,
"Craps": 0,
"Cups": 0,
"Dice": 0,
"Hilo": 0,
"War": 0,
"Double": 0,
},
"Cooldowns": {
"Allin": 0,
"Blackjack": 0,
"Coin": 0,
"Craps": 0,
"Cups": 0,
"Dice": 0,
"Hilo": 0,
"War": 0,
"Double": 0,
},
}
guild_defaults = {
"use_old_style": False,
"Memberships": {},
"Settings": {
"Global": False,
"Casino_Name": "Redjumpman's",
"Casino_Open": True,
"Payout_Switch": False,
"Payout_Limit": 10000,
},
"Games": {
"Allin": {
"Access": 0,
"Cooldown": 43200,
"Min": None,
"Max": None,
"Multiplier": None,
"Open": True,
},
"Blackjack": {
"Access": 0,
"Cooldown": 5,
"Min": 50,
"Max": 500,
"Multiplier": 2.0,
"Open": True,
},
"Coin": {
"Access": 0,
"Cooldown": 5,
"Max": 10,
"Min": 10,
"Multiplier": 1.5,
"Open": True,
},
"Craps": {
"Access": 0,
"Cooldown": 5,
"Max": 500,
"Min": 50,
"Multiplier": 2.0,
"Open": True,
},
"Cups": {
"Access": 0,
"Cooldown": 5,
"Max": 100,
"Min": 25,
"Multiplier": 1.8,
"Open": True,
},
"Dice": {
"Access": 0,
"Cooldown": 5,
"Max": 100,
"Min": 25,
"Multiplier": 1.8,
"Open": True,
},
"Hilo": {
"Access": 0,
"Cooldown": 5,
"Min": 25,
"Max": 75,
"Multiplier": 1.7,
"Open": True,
},
"Double": {
"Access": 0,
"Cooldown": 5,
"Min": 10,
"Max": 250,
"Multiplier": None,
"Open": True,
},
"War": {"Access": 0, "Cooldown": 5, "Min": 25, "Max": 75, "Multiplier": 1.5, "Open": True},
},
}
member_defaults = deepcopy(user_defaults)
global_defaults = deepcopy(guild_defaults)
global_defaults["Settings"]["Global"] = True
_DataNamedTuple = namedtuple("Casino", "foo")
_DataObj = _DataNamedTuple(foo=None)
log = logging.getLogger("red.jumper-plugins.casino")
class Database:
config: Config = Config.get_conf(_DataObj, 5074395001, force_registration=True)
def __init__(self):
self.config.register_guild(**guild_defaults)
self.config.register_global(schema_version=1, **global_defaults)
self.config.register_member(**member_defaults)
self.config.register_user(**user_defaults)
self.old_message_cache = OldMessageTypeManager(config=self.config, enable_cache=True)
self.migration_task: asyncio.Task = None
self.cog_ready_event: asyncio.Event = asyncio.Event()
async def data_schema_migration(self, from_version: int, to_version: int):
if from_version == to_version:
self.cog_ready_event.set()
return
if from_version < 2 <= to_version:
try:
async with self.config.all() as casino_data:
temp = deepcopy(casino_data)
global_payout = casino_data["Settings"]["Payout_Limit"]
if is_input_unsupported(global_payout):
casino_data["Settings"]["Payout_Limit"] = await bank.get_max_balance()
for g, g_data in temp["Games"].items():
for g_data_key, g_data_value in g_data.items():
if g_data_key in ["Access", "Cooldown", "Max", "Min", "Multiplier"]:
if is_input_unsupported(g_data_value):
if g_data_value < min_int:
g_data_value_new = min_int
else:
g_data_value_new = max_int
casino_data["Games"][g][g_data_key] = g_data_value_new
async with self.config._get_base_group(self.config.GUILD).all() as casino_data:
temp = deepcopy(casino_data)
for guild_id, guild_data in temp.items():
if (
"Settings" in temp[guild_id]
and "Payout_Limit" in temp[guild_id]["Settings"]
):
guild_payout = casino_data[guild_id]["Settings"]["Payout_Limit"]
if is_input_unsupported(guild_payout):
casino_data[guild_id]["Settings"][
"Payout_Limit"
] = await bank.get_max_balance(
guild_payout, guild=discord.Object(id=int(guild_id))
)
if "Games" in temp[guild_id]:
for g, g_data in temp[guild_id]["Games"].items():
for g_data_key, g_data_value in g_data.items():
if g_data_key in [
"Access",
"Cooldown",
"Max",
"Min",
"Multiplier",
]:
if is_input_unsupported(g_data_value):
if g_data_value < min_int:
g_data_value_new = min_int
else:
g_data_value_new = max_int
casino_data[guild_id]["Games"][g][
g_data_key
] = g_data_value_new
await self.config.schema_version.set(2)
except Exception as e:
log.exception(
"Fatal Exception during Data migration to Scheme 2, Casino cog will not be loaded.",
exc_info=e,
)
raise
self.cog_ready_event.set()
async def casino_is_global(self):
"""Checks to see if the casino is storing data on
a per server basis or globally."""
return await self.config.Settings.Global()
async def get_data(self, ctx, player=None):
"""
:param ctx: Context object
:param player: Member or user object
:return: Database that corresponds to the given data.
Returns the appropriate config category based on the given
data, and wheater or not the casino is global.
"""
if await self.casino_is_global():
if player is None:
return self.config
else:
return self.config.user(player)
else:
if player is None:
return self.config.guild(ctx.guild)
else:
return self.config.member(player)
async def get_all(self, ctx, player):
"""
:param ctx: Context Object
:param player: Member or user object
:return: Tuple with two dictionaries
Returns a dictionary representation of casino's settings data
and the player data.
"""
settings = await self.get_data(ctx)
player_data = await self.get_data(ctx, player=player)
return await settings.all(), await player_data.all()
async def _wipe_casino(self, ctx):
"""
Wipes all the casino data available
:param ctx: context object
:return: None
This wipes everything, including member/user data.
"""
await self.config.clear_all()
msg = "{0.name} ({0.id}) wiped all casino data.".format(ctx.author)
await ctx.send(msg)
async def _reset_settings(self, ctx):
"""
Resets only the settings data.
"""
data = await self.get_data(ctx)
await data.Settings.clear()
msg = ("{0.name} ({0.id}) reset all casino settings.").format(ctx.author)
await ctx.send(msg)
async def _reset_memberships(self, ctx):
"""
Resets all the information pertaining to memberships
"""
data = await self.get_data(ctx)
await data.Memberships.clear()
msg = ("{0.name} ({0.id}) cleared all casino memberships.").format(ctx.author)
await ctx.send(msg)
async def _reset_games(self, ctx):
"""
Resets all game settings, such as multipliers and bets.
"""
data = await self.get_data(ctx)
await data.Games.clear()
msg = ("{0.name} ({0.id}) restored casino games to default settings.").format(ctx.author)
await ctx.send(msg)
async def _reset_all_settings(self, ctx):
"""
Resets all settings, but retains all player data.
"""
await self._reset_settings(ctx)
await self._reset_memberships(ctx)
await self._reset_games(ctx)
await self._reset_cooldowns(ctx)
async def _reset_player_stats(self, ctx, player):
"""
:param ctx: Context object
:param player: user or member object
:return: None
Resets a player's win / played stats.
"""
data = await self.get_data(ctx, player=player)
await data.Played.clear()
await data.Won.clear()
msg = ("{0.name} ({0.id}) reset all stats for {1.name} ({1.id}).").format(ctx.author, player)
await ctx.send(msg)
async def _reset_player_all(self, ctx, player):
"""
:param ctx: context object
:param player: user or member object
:return: None
Resets all data belonging to the user, including stats and memberships.
"""
data = await self.get_data(ctx, player=player)
await data.clear()
msg = ("{0.name} ({0.id}) reset all data for {1.name} ({1.id}).").format(ctx.author, player)
await ctx.send(msg)
async def _reset_player_cooldowns(self, ctx, player):
"""
:param ctx: context object
:param player: user or member object
:return: None
Resets all game cooldowns for a player.
"""
data = await self.get_data(ctx, player=player)
await data.Cooldowns.clear()
msg = ("{0.name} ({0.id}) reset all cooldowns for {1.name} ({1.id}).").format(ctx.author, player)
await ctx.send(msg)
async def _reset_cooldowns(self, ctx):
"""
Resets all game cooldowns for every player in the database.
"""
if await self.casino_is_global():
for player in await self.config.all_users():
user = discord.Object(id=player)
await self.config.user(user).Cooldowns.clear()
msg = ("{0.name} ({0.id}) reset all global cooldowns.").format(ctx.author)
else:
for player in await self.config.all_members(ctx.guild):
user = discord.Object(id=player)
await self.config.member(user).Cooldowns.clear()
msg = ("{0.name} ({0.id}) reset all cooldowns on {1.name}.").format(ctx.author, ctx.guild)
await ctx.send(msg)
async def change_mode(self, mode):
"""
:param mode: String, must be local or global.
:return: None
Toggles how data is stored for casino between local and global.
When switching modes, all perviously stored data will be deleted.
"""
if mode == "global":
await self.config.clear_all_members()
await self.config.clear_all_guilds()
await self.config.Settings.Global.set(True)
else:
await self.config.clear_all_users()
await self.config.clear_all_globals()
await self.config.Settings.Global.set(False)
async def _update_cooldown(self, ctx, game, time):
player_data = await self.get_data(ctx, player=ctx.author)
await player_data.set_raw("Cooldowns", game, value=time)
async def _get_player_membership(self, ctx, player):
"""
:param ctx: context object
:param player: user or member object
:return: Membership name and a dictionary with the perks
Performs a lookup on the user and the created memberhips for casino.
If the user has a memberhip that was deleted, it will return the
default basic membership. It will also set their new membership to the
default.
"""
basic = {"Reduction": 0, "Access": 0, "Color": "grey", "Bonus": 1}
player_data = await self.get_data(ctx, player=player)
name = await player_data.Membership.Name()
if name == "Basic":
return name, basic
data = await self.get_data(ctx)
memberships = await data.Memberships.all()
try:
return name, memberships[name]
except KeyError:
await player_data.Membership.set({"Name": "Basic", "Assigned": False})
return "Basic", basic

117
casino/deck.py Normal file
View file

@ -0,0 +1,117 @@
import random
from collections import deque
from itertools import product, chain
class Deck:
"""Creates a Deck of playing cards."""
suites = (":clubs:", ":diamonds:", ":hearts:", ":spades:")
face_cards = ("King", "Queen", "Jack", "Ace")
bj_vals = {"Jack": 10, "Queen": 10, "King": 10, "Ace": 1}
war_values = {"Jack": 11, "Queen": 12, "King": 13, "Ace": 14}
def __init__(self):
self._deck = deque()
def __len__(self):
return len(self._deck)
def __str__(self):
return "Standard deck of cards with {} cards remaining.".format(len(self._deck))
def __repr__(self):
return "Deck{!r}".format(self._deck)
@property
def deck(self):
if len(self._deck) < 1:
self.new()
return self._deck
def shuffle(self):
random.shuffle(self._deck)
def war_count(self, card):
try:
return self.war_values[card[1]]
except KeyError:
return card[1]
def bj_count(self, hand: list, hole=False):
hand = self._hand_type(hand)
if hole:
card = hand[0][1]
count = self.bj_vals[card] if isinstance(card, str) else card
return count if count > 1 else 11
count = sum([self.bj_vals[y] if isinstance(y, str) else y for x, y in hand])
if any("Ace" in pair for pair in hand) and count <= 11:
count += 10
return count
@staticmethod
def fmt_hand(hand: list):
return ["{} {}".format(y, x) for x, y in hand]
@staticmethod
def fmt_card(card):
return "{1} {0}".format(*card)
@staticmethod
def hand_check(hand: list, card):
return any(x[1] == card for x in hand)
def split(self, position: int):
self._deck.rotate(-position)
@staticmethod
def _true_hand(hand: list):
return [x.split(" ") for x in hand]
def draw(self, top=True):
self._check()
if top:
card = self._deck.popleft()
else:
card = self._deck.pop()
return card
def _check(self, num=1):
if num > 52:
raise ValueError("Can not exceed deck limit.")
if len(self._deck) < num:
self.new()
def _hand_type(self, hand: list):
if isinstance(hand[0], tuple):
return hand
try:
return self._true_hand(hand)
except ValueError:
raise ValueError("Invalid hand input.")
def deal(self, num=1, top=True, hand=None):
self._check(num=num)
if hand is None:
hand = []
for x in range(0, num):
if top:
hand.append(self._deck.popleft())
else:
hand.append(self._deck.pop())
return hand
def burn(self, num):
self._check(num=num)
for x in range(0, num):
del self._deck[0]
def new(self):
cards = product(self.suites, chain(range(2, 11), ("King", "Queen", "Jack", "Ace")))
self._deck = deque(cards)
self.shuffle()

322
casino/engine.py Normal file
View file

@ -0,0 +1,322 @@
# Standard Library
import calendar
from functools import wraps
# Casino
from typing import Optional
from redbot.core.utils.chat_formatting import humanize_number
from . import utils
from .data import Database
# Red
from redbot.core import bank
from redbot.core.errors import BalanceTooHigh
from redbot.core.i18n import Translator
# Discord
import discord
_ = Translator("Casino", __file__)
def game_engine(name=None, choice=None, choices=None):
def wrapper(coro):
@wraps(coro)
async def wrapped(*args, **kwargs):
try:
user_choice = args[3]
except IndexError:
user_choice = None
engine = GameEngine(name, user_choice, choice, args[1], args[2])
if await engine.check_conditions():
result = await coro(*args, **kwargs)
await engine.game_teardown(result)
return wrapped
return wrapper
class GameEngine(Database):
"""A class that handles setup and teardown for games.
This is a helper class to make games easier to create games and to
provide a level of consistency. This class is only to be used
in conjunction with the game_engine decorator.
You only need to specify the name, and depending on the game, a choice or
a list of choices to choose from. The decorater will obtain the rest of the
attributes.
Attributes
-----------
game: str
The name of the game.
choice: str
The decision the player chose for the game. When a decision is not
required, leave it None.
choices: list
A list of choices the player must pick from. If a list of choices is not
required, leave it None.
ctx: object
The Red context object necessary for sending/waiting for messages.
player: object
User or member object necessary for interacting with the player.
guild: object
The guild object from the Red Context object. This is used to pull data
from config.
bet: int
The amount the player has wagered.
"""
__slots__ = ("game", "choice", "choices", "ctx", "bet", "player", "guild")
def __init__(self, game, choice, choices, ctx, bet):
self.game = game
self.choice = choice
self.choices = choices
self.bet = bet
self.ctx = ctx
self.player = ctx.author
self.guild = ctx.guild
super().__init__()
async def check_conditions(self):
"""
Performs all the necessary checks for a game. Every game must validate under these specific
checks. The following conditions are checked:
- Checking to see if the casino is open
- Checking to see if the game is open
- Checking to see if the player has a high enough access level to play the game.
- Validating that the player's choice is in the list of declared choices.
- Checking that the bet is within the range of the set min and max.
- Checking to see that has enough currency in the bank account to cover the bet.
- Checking to see if the game is on cooldown.
Cooldowns must be checked last so that the game doesn't trigger a cooldown if another
condition has failed.
"""
settings, player_data = await super().get_all(self.ctx, self.player)
access = self.access_calculator(settings["Memberships"], player_data["Membership"]["Name"])
if not settings["Settings"]["Casino_Open"]:
error = _("The Casino is closed.")
elif not settings["Games"][self.game]["Open"]:
error = _("{} is closed.".format(self.game))
elif settings["Games"][self.game]["Access"] > access:
error = _(
"{} requires an access level of {}. Your current access level is {}. Obtain "
"a higher membership to play this game."
).format(self.game, settings["Games"][self.game]["Access"], access)
elif self.choices is not None and self.choice not in self.choices:
error = _("Incorrect response. Accepted responses are:\n{}.").format(utils.fmt_join(self.choices))
elif not self.bet_in_range(
settings["Games"][self.game]["Min"], settings["Games"][self.game]["Max"]
):
error = _(
"Your bet must be between "
"{} and {}.".format(
settings["Games"][self.game]["Min"], settings["Games"][self.game]["Max"]
)
)
elif not await bank.can_spend(self.player, self.bet):
error = _("You do not have enough credits to cover the bet.")
else:
error = await self.check_cooldown(settings["Games"][self.game], player_data)
if error:
await self.ctx.send(error)
return False
else:
await bank.withdraw_credits(self.player, self.bet)
await self.update_stats(stat="Played")
return True
async def update_stats(self, stat: str):
"""
:param stat: string
Must be Played or Won
:return: None
Updates either a player's win or played stat.
"""
instance = await self.get_data(self.ctx, player=self.player)
current = await instance.get_raw(stat, self.game)
await instance.set_raw(stat, self.game, value=current + 1)
async def check_cooldown(self, game_data, player_data):
"""
:param game_data: Dictionary
Contains all the data pertaining to a particular game.
:param player_data: Object
User or member Object
:return: String or None
Returns a string when a cooldown is remaining on a game, otherwise it will
return None
Checks the time a player last played a game, and compares it with the set cooldown
for that game. If a user is still on cooldown, then a string detailing the time
remaining will be returned. Otherwise this will update their cooldown, and return None.
"""
user_time = player_data["Cooldowns"][self.game]
now = calendar.timegm(self.ctx.message.created_at.utctimetuple())
base = game_data["Cooldown"]
membership = await super()._get_player_membership(self.ctx, self.player)
reduction = membership[1]["Reduction"]
if now >= user_time - reduction:
await super()._update_cooldown(self.ctx, self.game, now + base)
else:
seconds = int((user_time + reduction - now))
remaining = utils.time_formatter(seconds)
msg = _("{} is still on a cooldown. You still have: {} remaining.").format(self.game, remaining)
return msg
async def game_teardown(self, result):
data = await super().get_data(self.ctx)
settings = await data.all()
message_obj: Optional[discord.Message]
win, amount, msg, message_obj = result
if not win:
embed = await self.build_embed(msg, settings, win, total=amount, bonus="(+0)")
if (not await self.old_message_cache.get_guild(self.ctx.guild)) and message_obj:
return await message_obj.edit(content=self.player.mention, embed=embed, view=None)
else:
return await self.ctx.send(self.player.mention, embed=embed, view=None)
player_data = await super().get_data(self.ctx, player=self.player)
await self.update_stats(stat="Won")
if self.limit_check(settings, amount):
embed = await self.build_embed(msg, settings, win, total=amount, bonus="(+0)")
return await self.limit_handler(
embed,
amount,
player_data,
settings["Settings"]["Payout_Limit"],
message=message_obj,
)
total, bonus = await self.deposit_winnings(amount, player_data, settings)
embed = await self.build_embed(msg, settings, win, total=total, bonus=bonus)
if (not await self.old_message_cache.get_guild(self.ctx.guild)) and message_obj:
return await message_obj.edit(content=self.player.mention, embed=embed, view=None)
else:
return await self.ctx.send(self.player.mention, embed=embed, view=None)
async def limit_handler(self, embed, amount, player_instance, limit, message):
await player_instance.Pending_Credits.set(int(amount))
if (not await self.old_message_cache.get_guild(self.ctx.guild)) and message:
await message.edit(content=self.player.mention, embed=embed)
else:
await self.ctx.send(self.player.mention, embed=embed)
msg = _(
"{} Your winnings exceeded the maximum credit limit allowed ({}). The amount "
"of {} credits will be pending on your account until reviewed. Until an "
"Administrator or higher authority has released the pending currency, "
"**DO NOT** attempt to place a bet that will exceed the payout limit. You "
"may only have **ONE** pending payout at a "
"time."
).format(self.player.name, limit, amount)
await self.player.send(msg)
async def deposit_winnings(self, amount, player_instance, settings):
multiplier = settings["Games"][self.game]["Multiplier"]
if self.game == "Allin" or self.game == "Double":
try:
await bank.deposit_credits(self.player, amount)
return amount, "(+0)"
except BalanceTooHigh as e:
return await bank.set_balance(self.player, e.max_balance), "(+0)"
initial = round(amount * multiplier)
total, amt, msg = await self.calculate_bonus(initial, player_instance, settings)
try:
await bank.deposit_credits(self.player, total)
except BalanceTooHigh as e:
await bank.set_balance(self.player, e.max_balance)
return total, msg
def bet_in_range(self, minimum, maximum):
if self.game == "Allin":
return True
if minimum <= self.bet <= maximum:
return True
else:
return False
async def build_embed(self, msg, settings, win, total, bonus):
balance = await bank.get_balance(self.player)
currency = await bank.get_currency_name(self.guild)
bal_msg = _("**Remaining Balance:** {} {}").format(humanize_number(balance), currency)
embed = discord.Embed()
embed.title = _("{} Casino | {}").format(settings["Settings"]["Casino_Name"], self.game)
if isinstance(msg, discord.Embed):
for field in msg.fields:
embed.add_field(**field.__dict__)
else:
embed.description = msg
if win:
embed.colour = 0x00FF00
end = _("Congratulations, you just won {} {} {}!\n{}").format(
humanize_number(total), currency, bonus, bal_msg
)
else:
embed.colour = 0xFF0000
end = _("Sorry, you didn't win anything.\n{}").format(bal_msg)
embed.add_field(name="-" * 65, value=end)
return embed
@staticmethod
def access_calculator(memberships, user_membership):
if user_membership == "Basic":
return 0
try:
access = memberships[user_membership]["Access"]
except KeyError:
return 0
else:
return access
@staticmethod
async def calculate_bonus(amount, player_instance, settings):
membership = await player_instance.Membership.Name()
try:
bonus_multiplier = settings["Memberships"][membership]["Bonus"]
except KeyError:
bonus_multiplier = 1
total = round(amount * bonus_multiplier)
bonus = total - amount
return total, amount, "(+{})".format(humanize_number(bonus) if bonus_multiplier > 1 else 0)
@staticmethod
def limit_check(settings, amount):
if settings["Settings"]["Payout_Switch"]:
if amount > settings["Settings"]["Payout_Limit"]:
return True
else:
return False
else:
return False

592
casino/games.py Normal file
View file

@ -0,0 +1,592 @@
from __future__ import annotations
# Standard Library
import asyncio
import random
# Casino
from .deck import Deck
from .engine import game_engine
# Red
from redbot.core import bank
from redbot.core.i18n import Translator
from redbot.core.errors import BalanceTooHigh
from redbot.core.utils.chat_formatting import box
from redbot.core.utils.predicates import MessagePredicate
# Discord
import discord
_ = Translator("Casino", __file__)
deck = Deck()
# Any game created must return a tuple of 3 arguments.
# The outcome (True or False)
# The final bet or amount (int)
# A msg that is either a string or an embed
# If the msg is a string it is added to the description of the final embed.
# If the msg is an embed, it's fields are added to the final embed.
class Core:
"""
A simple class to hold the basic original Casino mini games.
Games
-----------
Allin
Bet all your credits. All or nothing gamble.
Coin
Coin flip game. Pick heads or tails.
Cups
Three cups are shuffled. Pick the one covering the ball.
Dice
Roll a pair of die. 2, 7, 11, or 12 wins.
Hilo
Guess if the dice result will be high, low, or 7.
Craps
Win with a comeout roll of 7 or 11, lose on 2, 3, or 12.
If you roll any other number you must match it on your
second roll to win.
"""
def __init__(self, old_message_cache):
self.old_message_cache = old_message_cache
@game_engine("Allin")
async def play_allin(self, ctx, bet, multiplier):
message = await ctx.send(
_("You put all your chips into the machine and pull the lever...")
)
await asyncio.sleep(3)
outcome = random.randint(0, multiplier + 1)
if outcome == 0:
msg = "▂▃▅▇█▓▒░ [♠] [♥] [♦] [♣] ░▒▓█▇▅▃▂\n"
msg += _(" CONGRATULATIONS YOU WON\n")
msg += _("░▒▓█▇▅▃▂ ⚅ J A C K P O T ⚅ ▂▃▅▇█▓▒░")
msg = box(msg, lang="py")
bet *= multiplier
else:
msg = _("Nothing happens. You stare at the machine contemplating your decision.")
return outcome == 0, bet, msg, message
@game_engine("Coin", (_("heads"), _("tails")))
async def play_coin(self, ctx, bet, choice):
message = await ctx.send(_("The coin flips into the air..."))
await asyncio.sleep(2)
outcome = random.choice((_("heads"), _("tails")))
msg = _("The coin landed on {}!").format(outcome)
return choice.lower() in outcome, bet, msg, message
@game_engine("Cups", ("1", "2", "3"))
async def play_cups(self, ctx, bet, choice):
message = await ctx.send(_("The cups start shuffling along the table..."))
await asyncio.sleep(3)
outcome = random.randint(1, 3)
msg = _("The coin was under cup {}!").format(outcome)
return int(choice) == outcome, bet, msg, message
@game_engine("Dice")
async def play_dice(self, ctx, bet):
message = await ctx.send(
_("The dice strike the back of the table and begin to tumble into place...")
)
await asyncio.sleep(2)
die_one, die_two = self.roll_dice()
outcome = die_one + die_two
msg = _("The dice landed on {} and {} ({}).").format(die_one, die_two, outcome)
return outcome in (2, 7, 11, 12), bet, msg, message
@game_engine("Hilo", (_("low"), _("lo"), _("high"), _("hi"), _("seven"), _("7")))
async def play_hilo(self, ctx, bet, choice):
message = await ctx.send(_("The dice hit the table and slowly fall into place..."))
await asyncio.sleep(2)
result = sum(self.roll_dice())
if result < 7:
outcome = (_("low"), _("lo"))
elif result > 7:
outcome = (_("high"), _("hi"))
else:
outcome = (_("seven"), "7")
msg = _("The outcome was {} ({[0]})!").format(result, outcome)
if result == 7 and outcome[1] == "7":
bet *= 5
return choice.lower() in outcome, bet, msg, message
@game_engine(name="Craps")
async def play_craps(self, ctx, bet):
return await self._craps_game(ctx, bet)
async def _craps_game(self, ctx, bet, comeout=False, message=None):
msg1 = _("The dice strike against the back of the table...")
if (not await self.old_message_cache.get_guild(ctx.guild)) and message:
await asyncio.sleep(3)
await message.edit(content=msg1)
else:
message = await ctx.send(msg1)
await asyncio.sleep(2)
d1, d2 = self.roll_dice()
result = d1 + d2
msg = _("You rolled a {} and {}.")
if comeout:
if result == comeout:
return True, bet, msg.format(d1, d2), message
return False, bet, msg.format(d1, d2), message
if result == 7:
bet *= 3
return True, bet, msg.format(d1, d2), message
elif result == 11:
return True, bet, msg.format(d1, d2), message
elif result in (2, 3, 12):
return False, bet, msg.format(d1, d2), message
msg2 = _(
"{}\nI'll roll the dice one more time. This time you will need exactly {} to win."
).format(msg.format(d1, d2), d1 + d2)
if not await self.old_message_cache.get_guild(ctx.guild):
await message.edit(content=msg2)
else:
await ctx.send(msg2)
return await self._craps_game(ctx, bet, comeout=result, message=message)
@staticmethod
def roll_dice():
return random.randint(1, 6), random.randint(1, 6)
class BlackjackView(discord.ui.View):
def __init__(self, ctx, include_double: bool):
self.ctx = ctx
super().__init__()
self.result = None
if include_double is False:
self.double_button.disabled = True
@discord.ui.button(label="Hit", emoji="\N{RAISED FIST}")
async def hit_button(self, interaction: discord.Interaction, button):
self.result = "hit"
await interaction.response.defer()
self.stop()
@discord.ui.button(label="Stay", emoji="\N{RAISED HAND}")
async def stay_button(self, interaction: discord.Interaction, button):
self.result = "stay"
await interaction.response.defer()
self.stop()
@discord.ui.button(label="Double", emoji="\N{VICTORY HAND}\N{VARIATION SELECTOR-16}")
async def double_button(self, interaction: discord.Interaction, button):
self.result = "double"
await interaction.response.defer()
self.stop()
async def interaction_check(self, interaction: discord.Interaction):
if interaction.user.id != self.ctx.author.id:
await interaction.response.send_message(
"You are not authorized to interact with this.", ephemeral=True
)
return False
return True
class Blackjack:
"""A simple class to hold the game logic for Blackjack.
Blackjack requires inheritance from data to verify the user
can double down.
"""
def __init__(self, old_message_cache):
self.old_message_cache = old_message_cache
super().__init__()
@game_engine(name="Blackjack")
async def play(self, ctx, bet):
ph, dh, amt, msg = await self.blackjack_game(ctx, bet)
result = await self.blackjack_results(ctx, amt, ph, dh, message=msg)
return result
@game_engine(name="Blackjack")
async def mock(self, ctx, bet, ph, dh):
result = await self.blackjack_results(ctx, bet, ph, dh)
return result
async def blackjack_game(self, ctx, amount):
ph = deck.deal(num=2)
ph_count = deck.bj_count(ph)
dh = deck.deal(num=2)
# End game if player has 21
if ph_count == 21:
return ph, dh, amount, None
embed = self.bj_embed(ctx, ph, dh, ph_count, initial=True)
view = BlackjackView(ctx, include_double=True)
msg = await ctx.send(ctx.author.mention, embed=embed, view=view)
await view.wait()
if view.result == "stay":
dh = self.dealer(dh)
return ph, dh, amount, msg
if view.result == "double":
return await self.double_down(ctx, ph, dh, amount, message=msg)
else:
ph, dh, message = await self.bj_loop(ctx, ph, dh, ph_count, message=msg)
dh = self.dealer(dh)
return ph, dh, amount, msg
async def double_down(self, ctx, ph, dh, amount, message):
try:
await bank.withdraw_credits(ctx.author, amount)
except ValueError:
await ctx.send(
_("{} You can not cover the bet. Please choose hit or stay.").format(
ctx.author.mention
)
)
view = BlackjackView(ctx, include_double=False)
ph_count = deck.bj_count(ph)
embed = self.bj_embed(ctx, ph, dh, ph_count, initial=False)
if not await self.old_message_cache.get_guild(ctx.guild):
await message.edit(content=ctx.author.mention, embed=embed, view=view)
else:
await ctx.send(content=ctx.author.mention, embed=embed, view=view)
await view.wait()
if view.result == "stay":
dh = self.dealer(dh)
return ph, dh, amount, message
elif view.result == "hit":
ph, dh, message = await self.bj_loop(
ctx, ph, dh, deck.bj_count(ph), message=message
)
dh = self.dealer(dh)
return ph, dh, amount, message
else:
deck.deal(hand=ph)
dh = self.dealer(dh)
amount *= 2
return ph, dh, amount, message
async def blackjack_results(self, ctx, amount, ph, dh, message=None):
dc = deck.bj_count(dh)
pc = deck.bj_count(ph)
if dc > 21 >= pc or dc < pc <= 21:
outcome = _("Winner!")
result = True
elif pc > 21:
outcome = _("BUST!")
result = False
elif dc == pc <= 21:
outcome = _("Pushed")
try:
await bank.deposit_credits(ctx.author, amount)
except BalanceTooHigh as e:
await bank.set_balance(ctx.author, e.max_balance)
result = False
else:
outcome = _("House Wins!")
result = False
embed = self.bj_embed(ctx, ph, dh, pc, outcome=outcome)
return result, amount, embed, message
async def bj_loop(self, ctx, ph, dh, count, message: discord.Message):
while count < 21:
ph = deck.deal(hand=ph)
count = deck.bj_count(hand=ph)
if count >= 21:
break
embed = self.bj_embed(ctx, ph, dh, count)
view = BlackjackView(ctx, include_double=False)
if not await self.old_message_cache.get_guild(ctx.guild):
await message.edit(content=ctx.author.mention, embed=embed, view=view)
else:
await ctx.send(content=ctx.author.mention, embed=embed, view=view)
await view.wait()
if view.result == "stay":
break
await asyncio.sleep(1)
# Return player hand & dealer hand when count >= 21 or the player picks stay.
return ph, dh, message
@staticmethod
def dealer(dh):
count = deck.bj_count(dh)
# forces hit if ace in first two cards without 21
if deck.hand_check(dh, "Ace") and count != 21:
deck.deal(hand=dh)
count = deck.bj_count(dh)
# defines maximum hit score X
while count < 17:
deck.deal(hand=dh)
count = deck.bj_count(dh)
return dh
@staticmethod
def bj_embed(ctx, ph, dh, count1, initial=False, outcome=None):
hand = _("{}\n**Score:** {}")
footer = _("Cards in Deck: {}")
start = _("**Options:** hit, stay, or double")
after = _("**Options:** hit or stay")
options = "**Outcome:** " + outcome if outcome else start if initial else after
count2 = deck.bj_count(dh, hole=True) if not outcome else deck.bj_count(dh)
hole = " ".join(deck.fmt_hand([dh[0]]))
dealer_hand = hole if not outcome else ", ".join(deck.fmt_hand(dh))
embed = discord.Embed(colour=0xFF0000)
embed.add_field(
name=_("{}'s Hand").format(ctx.author.name),
value=hand.format(", ".join(deck.fmt_hand(ph)), count1),
)
embed.add_field(
name=_("{}'s Hand").format(ctx.bot.user.name), value=hand.format(dealer_hand, count2)
)
embed.add_field(name="\u200b", value=options, inline=False)
embed.set_footer(text=footer.format(len(deck)))
return embed
class WarView(discord.ui.View):
def __init__(self, ctx):
self.ctx = ctx
super().__init__()
self.result = None
@discord.ui.button(label=_("War"))
async def war_button(self, interaction: discord.Interaction, button):
self.result = "war"
await interaction.response.defer()
self.stop()
@discord.ui.button(label=_("Surrender"), emoji="\N{WAVING WHITE FLAG}\N{VARIATION SELECTOR-16}")
async def surrender_button(self, interaction: discord.Interaction, button):
self.result = "surrender"
await interaction.response.defer()
self.stop()
async def interaction_check(self, interaction: discord.Interaction):
if interaction.user.id != self.ctx.author.id:
await interaction.response.send_message(
"You are not authorized to interact with this.", ephemeral=True
)
return False
return True
class War:
"""A simple class for the war card game."""
def __init__(self, old_message_cache):
self.old_message_cache = old_message_cache
@game_engine("War")
async def play(self, ctx, bet):
outcome, player_card, dealer_card, amount, msg = await self.war_game(ctx, bet)
return await self.war_results(outcome, player_card, dealer_card, amount, message=msg)
async def war_game(self, ctx, bet):
player_card, dealer_card, pc, dc = self.war_draw()
message = await ctx.send(
_(
"The dealer shuffles the deck and deals 2 cards face down. One for the "
"player and one for the dealer..."
)
)
await asyncio.sleep(2)
text = _("**FLIP!**")
if not await self.old_message_cache.get_guild(ctx.guild):
await message.edit(content=text)
else:
await ctx.send(text)
await asyncio.sleep(1)
if pc != dc:
if pc >= dc:
outcome = "Win"
else:
outcome = "Loss"
return outcome, player_card, dealer_card, bet, message
content = _(
"The player and dealer are both showing a **{}**!\nTHIS MEANS "
"WAR! You may choose to surrender and forfeit half your bet, or "
"you can go to war.\nIf you go to war your bet will be doubled, "
"but the multiplier is only applied to your original bet, the rest will "
"be pushed."
).format(deck.fmt_card(player_card))
view = WarView(ctx)
if not await self.old_message_cache.get_guild(ctx.guild):
await message.edit(content=content, view=view)
else:
await ctx.send(content=content, view=view)
await view.wait()
choice = view.result
if choice is None or choice.title() in (_("Surrender"), _("Ffs")):
outcome = "Surrender"
bet /= 2
return outcome, player_card, dealer_card, bet, message
else:
player_card, dealer_card, pc, dc = self.burn_and_draw()
msg1 = _("The dealer burns three cards and deals two cards face down...")
msg2 = _("**FLIP!**")
if not await self.old_message_cache.get_guild(ctx.guild):
action = message.edit
else:
action = ctx.send
await action(content=msg1)
await asyncio.sleep(3)
await action(content=msg2)
if pc >= dc:
outcome = "Win"
else:
outcome = "Loss"
return outcome, player_card, dealer_card, bet, message
@staticmethod
async def war_results(outcome, player_card, dealer_card, amount, message=None):
msg = _("**Player Card:** {}\n**Dealer Card:** {}\n").format(
deck.fmt_card(player_card), deck.fmt_card(dealer_card)
)
if outcome == "Win":
msg += _("**Result**: Winner")
return True, amount, msg, message
elif outcome == "Loss":
msg += _("**Result**: Loser")
return False, amount, msg, message
else:
msg += _("**Result**: Surrendered")
return False, amount, msg, message
@staticmethod
def get_count(pc, dc):
return deck.war_count(pc), deck.war_count(dc)
def war_draw(self):
player_card, dealer_card = deck.deal(num=2)
pc, dc = self.get_count(player_card, dealer_card)
return player_card, dealer_card, pc, dc
def burn_and_draw(self):
deck.burn(3)
player_card, dealer_card = deck.deal(num=2)
pc, dc = self.get_count(player_card, dealer_card)
return player_card, dealer_card, pc, dc
class DoubleView(discord.ui.View):
def __init__(self, ctx):
self.ctx = ctx
super().__init__()
self.result = None
@discord.ui.button(label=_("Double"))
async def war_button(self, interaction: discord.Interaction, button):
self.result = "double"
await interaction.response.defer()
self.stop()
@discord.ui.button(label=_("Cash out"), emoji="\N{BANKNOTE WITH DOLLAR SIGN}")
async def surrender_button(self, interaction: discord.Interaction, button):
self.result = None
# set this to None so we exit even if the user doesn't interact
await interaction.response.defer()
self.stop()
async def interaction_check(self, interaction: discord.Interaction):
if interaction.user.id != self.ctx.author.id:
await interaction.response.send_message(
"You are not authorized to interact with this.", ephemeral=True
)
return False
return True
class Double:
"""A simple class for the Double Or Nothing game."""
def __init__(self, old_message_cache):
self.old_message_cache = old_message_cache
@game_engine("Double")
async def play(self, ctx, bet):
count, amount, message = await self.double_game(ctx, bet)
return await self.double_results(ctx, count, amount, message=message)
async def double_game(self, ctx, bet):
count = 0
message = None
while bet > 0:
count += 1
flip = random.randint(0, 1)
if flip == 0:
bet = 0
break
else:
bet *= 2
view = DoubleView(ctx)
embed = self.double_embed(ctx, count, bet)
if (not await self.old_message_cache.get_guild(ctx.guild)) and message:
await message.edit(content=ctx.author.mention, embed=embed, view=view)
else:
message = await ctx.send(ctx.author.mention, embed=embed, view=view)
await view.wait()
if not view.result:
break
await asyncio.sleep(1)
return count, bet, message
async def double_results(self, ctx, count, amount, message=None):
if amount > 0:
outcome = _("Cashed Out!")
result = True
else:
outcome = _("You Lost It All!")
result = False
embed = self.double_embed(ctx, count, amount, outcome=outcome)
return result, amount, embed, message
@staticmethod
def double_embed(ctx, count, amount, outcome=None):
double = _("{}\n**DOUBLE!:** x{}")
zero = _("{}\n**NOTHING!**")
choice = _("**Options:** double or cash out")
options = "**Outcome:** " + outcome if outcome else choice
if amount == 0:
score = zero.format(amount)
else:
score = double.format(amount, count)
embed = discord.Embed(colour=0xFF0000)
embed.add_field(name=_("{}'s Score").format(ctx.author.name), value=score)
embed.add_field(name="\u200b", value=options, inline=False)
if not outcome:
embed.add_field(
name="\u200b", value="Remember, you can cash out at anytime.", inline=False
)
embed.set_footer(text="Try again and test your luck!")
return embed

13
casino/info.json Normal file
View file

@ -0,0 +1,13 @@
{
"author" : ["Redjumpman (Redjumpman#1337)"],
"install_msg" : "Thank you for installing casino. Be sure to check out the wiki here: https://github.com/Redjumpman/Jumper-Plugins/wiki/Casino-RedV3\nThis cog may put a heavy load on your bot if used with 10k users or more.",
"name" : "Casino",
"short" : "Casino style mini games.",
"requirements" : ["tabulate"],
"description" : "Play up to 7 unique games and earn currency.",
"permissions" : ["Manage Messages", "Embed Links"],
"tags" : ["Games", "Economy", "Fun", "Casino"],
"min_python_version": [3, 8, 1],
"min_bot_version": "3.2.0",
"end_user_data_statement": "This cog stores discord IDs as needed for operation."
}

128
casino/utils.py Normal file
View file

@ -0,0 +1,128 @@
import re
import math
from typing import Union, Dict, List, Sequence
utf8_re = re.compile(r"^[\U00000000-\U0010FFFF]*$")
min_int, max_int = 1 - (2 ** 64), (2 ** 64) - 1
def is_input_unsupported(data: Union[Dict, List, str, int, float]):
if type(data) is dict:
for k, v in data.items():
if is_input_unsupported(k) or is_input_unsupported(v):
return True
if type(data) is list:
for i in data:
if is_input_unsupported(i):
return True
if type(data) is str and not utf8_re.match(data):
return True
if type(data) is int:
if not (min_int <= data <= max_int):
return True
if type(data) is float:
if math.isnan(data) or math.isinf(data):
return True
if not (min_int <= data <= max_int):
return True
class PluralDict(dict):
"""This class is used to plural strings
You can plural strings based on the value input when using this class as a dictionary.
"""
def __missing__(self, key):
if "(" in key and key.endswith(")"):
key, rest = key.split("(", 1)
value = super().__getitem__(key)
suffix = rest.rstrip(")").split(",")
if len(suffix) == 1:
suffix.insert(0, "")
return suffix[0] if value <= 1 else suffix[1]
raise KeyError(key)
def time_converter(units):
return sum(int(x) * 60 ** i for i, x in enumerate(reversed(units.split(":"))))
def color_lookup(color="grey"):
colors = {
"blue": 0x3366FF,
"red": 0xFF0000,
"green": 0x00CC33,
"orange": 0xFF6600,
"purple": 0xA220BD,
"yellow": 0xFFFF00,
"teal": 0x009999,
"magenta": 0xBA2586,
"turquoise": 0x00FFFF,
"grey": 0x666666,
"pink": 0xFE01D1,
"white": 0xFFFFFF,
}
color = colors[color]
return color
def fmt_join(words: Sequence, ending: str = "or"):
if not words:
return ""
elif len(words) == 1:
return words[0]
else:
return "{} {} {}".format(", ".join(map(str, words[:-1])), ending, words[-1])
def cooldown_formatter(seconds, custom_msg="0"):
m, s = divmod(seconds, 60)
h, m = divmod(m, 60)
if h > 0:
msg = "{0}h"
if m > 0 and s > 0:
msg += ", {1}m, and {2}s"
elif s > 0 and m == 0:
msg += " and {2}s"
elif s == 0 and m == 0:
pass
else:
msg += " and {1}m"
elif h == 0 and m > 0:
msg = "{1}m" if s == 0 else "{1}m and {2}s"
elif m == 0 and h == 0 and s > 0:
msg = "{2}s"
else:
msg = custom_msg
return msg.format(h, m, s)
def time_formatter(seconds):
# Calculate the time and input into a dict to plural the strings later.
m, s = divmod(seconds, 60)
h, m = divmod(m, 60)
data = PluralDict({"hour": h, "minute": m, "second": s})
# Determine the remaining time.
if h > 0:
fmt = "{hour} hour{hour(s)}"
if data["minute"] > 0 and data["second"] > 0:
fmt += ", {minute} minute{minute(s)}, and {second} second{second(s)}"
if data["second"] > 0 == data["minute"]:
fmt += ", and {second} second{second(s)}"
msg = fmt.format_map(data)
elif h == 0 and m > 0:
if data["second"] == 0:
fmt = "{minute} minute{minute(s)}"
else:
fmt = "{minute} minute{minute(s)}, and {second} second{second(s)}"
msg = fmt.format_map(data)
elif m == 0 and h == 0 and s > 0:
fmt = "{second} second{second(s)}"
msg = fmt.format_map(data)
else:
msg = "None"
return msg