Untitled
unknown
plain_text
17 days ago
16 kB
3
Indexable
Never
import traceback
import discord
import random
import asyncio
from random import choice
from discord import app_commands
from discord.ext import commands, tasks
#from discord.app_commands import cooldown, BucketType
import json
import time
import re
import os
from discord.ext.commands import CooldownMapping, BucketType
from discord.ui import View # Importing View class from discord.ui
from datetime import datetime, timedelta
import math
from math import ceil
from typing import Literal
from discord.ui import Button, View
import sqlite3
import string
#import datetime
import psutil
from typing import Optional
class ServerEconomy(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.db = sqlite3.connect('server.db')
self.cursor = self.db.cursor()
self.db1 = sqlite3.connect('economy.db')
self.cursor1 = self.db1.cursor()
self.setup_database()
self.moneyemoji = "<a:MoneyGIFNexus:1274650931992072205>"
self.ongoing_bankrobs = {}
def setup_database(self):
self.cursor.execute('''CREATE TABLE IF NOT EXISTS users
(user_id INTEGER PRIMARY KEY,
wallet INTEGER DEFAULT 0,
bank INTEGER DEFAULT 0)''')
self.cursor.execute('''CREATE TABLE IF NOT EXISTS server_pools
(server_id INTEGER PRIMARY KEY,
money INTEGER DEFAULT 0)''')
self.cursor.execute('''CREATE TABLE IF NOT EXISTS server_items
(server_id INTEGER,
item_name TEXT,
quantity INTEGER,
PRIMARY KEY (server_id, item_name))''')
self.cursor.execute('''CREATE TABLE IF NOT EXISTS server_managers
(server_id INTEGER PRIMARY KEY,
role_id INTEGER)''')
self.db.commit()
def get_server_money(self, server_id):
self.cursor.execute("SELECT money FROM server_pools WHERE server_id = ?", (server_id,))
result = self.cursor.fetchone()
return result[0] if result else 0
def get_server_items(self, server_id):
self.cursor.execute("SELECT item_name, quantity FROM server_items WHERE server_id = ?", (server_id,))
return dict(self.cursor.fetchall())
def update_server_money(self, server_id, amount):
self.cursor.execute("INSERT OR REPLACE INTO server_pools (server_id, money) VALUES (?, ?)",
(server_id, self.get_server_money(server_id) + amount))
self.db.commit()
def update_server_item(self, server_id, item_name, quantity):
self.cursor.execute("INSERT OR REPLACE INTO server_items (server_id, item_name, quantity) VALUES (?, ?, ?)",
(server_id, item_name, quantity))
self.db.commit()
def update_balance(self, user_id, amount, account_type='wallet'):
self.cursor.execute(f"UPDATE users SET {account_type} = {account_type} + ? WHERE user_id = ?", (amount, user_id))
self.db.commit()
def parse_amount(self, amount_str):
multipliers = {'k': 1000, 'm': 1000000, 'b': 1000000000}
match = re.match(r"(\d+)([kmb])?$", amount_str)
if match:
number, multiplier = match.groups()
number = int(number)
if multiplier:
number *= multipliers[multiplier.lower()]
return number
return 0
def check_manager_role(self, ctx):
server_id = ctx.guild.id
self.cursor.execute("SELECT role_id FROM server_managers WHERE server_id = ?", (server_id,))
manager_role_id = self.cursor.fetchone()
if not manager_role_id:
return False
return discord.utils.get(ctx.author.roles, id=manager_role_id[0]) is not None
@commands.hybrid_group(name="server", fallback="info")
async def server(self, ctx):
"""Server economy commands"""
if ctx.invoked_subcommand is None:
await ctx.send("Use `server bankrob` or `server payout` for more information.")
@server.command(name="manager", description="Set the role for server bankrob and payout permissions")
@commands.has_permissions(administrator=True)
@app_commands.describe(role="The role to set as server manager")
async def server_manager(self, ctx, role: discord.Role):
server_id = ctx.guild.id
self.cursor.execute("INSERT OR REPLACE INTO server_managers (server_id, role_id) VALUES (?, ?)",
(server_id, role.id))
self.db.commit()
embed = discord.Embed(
title="Server Manager Role Set",
description=f"{role.mention} has been set as the server manager role. Users with this role can now use server bankrob and server payout commands.",
color=discord.Color.green()
)
await ctx.send(embed=embed)
@server.command(name="pool")
async def server_pool(self, ctx):
server_id = ctx.guild.id
money = self.get_server_money(server_id)
items = self.get_server_items(server_id)
embed = discord.Embed(title=f"{ctx.guild.name}'s Server Pool", color=discord.Color.gold())
embed.add_field(name="Money", value=f"{self.moneyemoji} {money:,}", inline=False)
if items:
items_str = "\n".join([f"{item}: {quantity}" for item, quantity in items.items()])
embed.add_field(name="Items", value=items_str, inline=False)
else:
embed.add_field(name="Items", value="No items in the pool", inline=False)
await ctx.send(embed=embed)
@server.command(name="donate")
async def server_donate(self, ctx, quantity, item: Optional[str] = None):
server_id = ctx.guild.id
if item is None:
# Donating money
try:
amount = int(quantity.replace('k', '000').replace('m', '000000').replace('b', '000000000'))
except ValueError:
await ctx.send("Invalid amount format. Use a number or shorthand like 20k, 2m, 2b.")
return
confirm_embed = discord.Embed(
title="Donation Confirmation",
description=f"Are you sure you want to donate {self.moneyemoji} {amount:,} to the Server Pool?",
color=discord.Color.blue()
)
else:
# Donating item
try:
quantity = int(quantity)
except ValueError:
await ctx.send("Invalid quantity. Please use a number.")
return
confirm_embed = discord.Embed(
title="Donation Confirmation",
description=f"Are you sure you want to donate {quantity} {item} to the Server Pool?",
color=discord.Color.blue()
)
view = discord.ui.View()
view.add_item(discord.ui.Button(label="Yes", style=discord.ButtonStyle.green, custom_id="confirm"))
view.add_item(discord.ui.Button(label="No", style=discord.ButtonStyle.red, custom_id="cancel"))
message = await ctx.send(embed=confirm_embed, view=view)
def check(interaction):
return interaction.message.id == message.id and interaction.user.id == ctx.author.id
try:
interaction = await self.bot.wait_for("interaction", timeout=30.0, check=check)
if interaction.data["custom_id"] == "confirm":
if item is None:
# Update server pool
self.cursor.execute("UPDATE server_pools SET money = money + ? WHERE server_id = ?", (amount, server_id))
# Update user balance
self.cursor1.execute("UPDATE users SET wallet = wallet - ? WHERE user_id = ?", (amount, ctx.author.id))
await ctx.send(f"Successfully donated {self.moneyemoji} {amount:,} to the Server Pool.")
else:
# Update server items
self.cursor.execute("INSERT OR REPLACE INTO server_items (server_id, item_name, quantity) VALUES (?, ?, coalesce((SELECT quantity FROM server_items WHERE server_id = ? AND item_name = ?), 0) + ?)",
(server_id, item, server_id, item, quantity))
# Update user inventory
self.cursor1.execute("UPDATE inventory SET quantity = quantity - ? WHERE user_id = ? AND item_name = ?",
(quantity, ctx.author.id, item))
await ctx.send(f"Successfully donated {quantity} {item} to the Server Pool.")
self.db.commit()
self.db1.commit()
else:
await ctx.send("Donation cancelled.")
except asyncio.TimeoutError:
await ctx.send("Donation timed out.")
await message.edit(view=None)
@server.command(name="bankrob", description="Start a bankrob on the server pool")
@app_commands.describe(amount="The amount to rob from the server pool")
async def server_bankrob(self, ctx, amount: str):
if not self.check_manager_role(ctx):
await ctx.send("You don't have permission to use this command. You need the server manager role.")
return
server_id = ctx.guild.id
target_amount = self.parse_amount(amount)
server_money = self.get_server_money(server_id)
if server_money < target_amount:
await ctx.send("The server doesn't have that much money to rob!")
return
view = BankRobView(ctx, target_amount, self)
self.ongoing_bankrobs[server_id] = view
embed = discord.Embed(
title="Server Bankrob Started",
description=f"{ctx.author.mention} is starting a bankrob on the server pool for {self.moneyemoji} {target_amount:,}. Click the button below to join!",
color=discord.Color.red()
)
embed.set_thumbnail(url=ctx.guild.icon.url if ctx.guild.icon else None)
message = await ctx.send(embed=embed, view=view)
view.message = message
@server.command(name="payout", description="Pay a user from the server pool")
@app_commands.describe(user="The user to pay", quantity="The amount to pay", item="The item to give (optional)")
async def server_payout(self, ctx, user: discord.Member, quantity: str, item: Optional[str] = None):
if not self.check_manager_role(ctx):
await ctx.send("You don't have permission to use this command. You need the server manager role.")
return
server_id = ctx.guild.id
if item is None:
# Paying out money
amount = self.parse_amount(quantity)
server_money = self.get_server_money(server_id)
if server_money < amount:
await ctx.send("The server doesn't have enough money for this payout.")
return
confirm_embed = discord.Embed(
title="Payout Confirmation",
description=f"Are you sure you want to pay {self.moneyemoji} {amount:,} to {user.mention}?",
color=discord.Color.blue()
)
else:
# Paying out item
try:
quantity = int(quantity)
except ValueError:
await ctx.send("Invalid quantity. Please use a number.")
return
server_items = self.get_server_items(server_id)
if item not in server_items or server_items[item] < quantity:
await ctx.send("The server doesn't have enough of this item for the payout.")
return
confirm_embed = discord.Embed(
title="Payout Confirmation",
description=f"Are you sure you want to give {quantity} {item} to {user.mention}?",
color=discord.Color.blue()
)
view = discord.ui.View(timeout=30)
view.add_item(discord.ui.Button(label="Yes", style=discord.ButtonStyle.green, custom_id="confirm"))
view.add_item(discord.ui.Button(label="No", style=discord.ButtonStyle.red, custom_id="cancel"))
message = await ctx.send(embed=confirm_embed, view=view)
def check(interaction):
return interaction.message.id == message.id and interaction.user.id == ctx.author.id
try:
interaction = await self.bot.wait_for("interaction", timeout=30.0, check=check)
if interaction.data["custom_id"] == "confirm":
if item is None:
self.update_server_money(server_id, -amount)
self.update_balance(user.id, amount, 'wallet')
await ctx.send(f"Successfully paid {self.moneyemoji} {amount:,} to {user.mention}.")
else:
current_quantity = self.get_server_items(server_id)[item]
self.update_server_item(server_id, item, current_quantity - quantity)
self.add_item_to_inventory(user.id, item, quantity)
await ctx.send(f"Successfully gave {quantity} {item} to {user.mention}.")
else:
await ctx.send("Payout cancelled.")
except asyncio.TimeoutError:
await ctx.send("Payout timed out.")
await message.edit(view=None)
class BankRobView(discord.ui.View):
def __init__(self, ctx, target_amount, cog):
super().__init__(timeout=60)
self.ctx = ctx
self.target_amount = target_amount
self.cog = cog
self.participants = []
self.message = None
@discord.ui.button(label="JOIN BANKROB", style=discord.ButtonStyle.green, custom_id="join_bankrob")
async def join_button(self, interaction: discord.Interaction, button: discord.ui.Button):
if interaction.user not in self.participants:
self.participants.append(interaction.user)
await interaction.response.send_message("You have successfully joined the Bankrob.", ephemeral=True)
else:
await interaction.response.send_message("You have already joined the Bankrob.", ephemeral=True)
async def on_timeout(self):
for child in self.children:
child.disabled = True
await self.message.edit(view=self)
if len(self.participants) < 5:
await self.ctx.send("Not enough participants. The bankrob has failed!")
return
success_chance = random.random()
if success_chance < 0.5: # 50% chance of success
self.cog.update_server_money(self.ctx.guild.id, -self.target_amount)
share = self.target_amount // len(self.participants)
success_embed = discord.Embed(
title="Server Bankrob Successful!",
description=f"The bankrob was successful! Each participant gets {self.cog.moneyemoji} {share:,}",
color=discord.Color.green()
)
for participant in self.participants:
self.cog.update_balance(participant.id, share, 'wallet')
success_embed.add_field(name=participant.name, value=f"{self.cog.moneyemoji} {share:,}", inline=False)
await self.ctx.send(embed=success_embed)
else:
fail_embed = discord.Embed(
title="Server Bankrob Failed!",
description="The bankrob was unsuccessful. Better luck next time!",
color=discord.Color.red()
)
await self.ctx.send(embed=fail_embed)
async def setup(bot):
await bot.add_cog(ServerEconomy(bot))
Leave a Comment