all classes moved into file

some commands moved into cogs
This commit is contained in:
2022-07-02 12:46:43 +03:00
parent 0588a0c01c
commit 19e1d4226b
8 changed files with 278 additions and 331 deletions

View File

@@ -1,5 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="PROJECT_PROFILE" value="Default" />
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>

View File

@@ -12,5 +12,5 @@ class Audio(commands.Cog):
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
def setup(bot): # a extension must have a setup function
def setup(bot): # an extension must have a setup function
bot.add_cog(Audio(bot)) # adding a cog

View File

@@ -1,36 +0,0 @@
import logging
import discord
from discord.ext import commands
class Test_Cog(commands.Cog):
def __init__(self, bot):
self.bot = bot # defining bot as global var in class
@commands.Cog.listener() # this is a decorator for events/listeners
async def on_ready(self):
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
@commands.command() # this is for making a command
async def bot_info(self, ctx, bot):
# await ctx.send(f'Pong! {round(self.bot.latency * 1000)}')
emb = discord.Embed(
title=f"General information",
description=f"General information on about {bot}",
icon=bot.avatar_url
)
emb.set_thumbnail(url=bot.avatar_url)
emb.add_field(name="General info",
value=f"Username: {bot}\n"
f"Nickname: {bot.nick}\n"
f"Joined at: {bot.joined_at.strftime('%A, %B %d %Y @ %H:%M:%S')}", inline=False)
# emb.add_field(name="Audio list", value=f"{audios}", inline=True)
# emb.add_field(name="Roles list", value=f"{roles}", inline=True)
emb.set_footer(text="Information requested by: {}".format(ctx.author.display_name))
await ctx.reply(embed=emb, ephemeral=True)
def setup(bot): # a extension must have a setup function
bot.add_cog(Test_Cog(bot)) # adding a cog

View File

@@ -1,10 +1,12 @@
import logging
import discord
from discord.ext import commands
from dislash import Option, OptionType
import dislash
from test import DB
import lib
import discord
from discord.ext import commands
from dislash import Option, OptionType, slash_command
class General(commands.Cog):
@@ -13,18 +15,23 @@ class General(commands.Cog):
@commands.Cog.listener() # this is a decorator for events/listeners
async def on_ready(self):
for g in self.bot.get_all_members():
lib.DB.prepare_db(g.guild.id)
for g in self.bot.get_all_members():
lib.DB.fill_bd(g.name, g.id, g.bot, g.nick, g.guild.id)
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
@cog_ext.cog_slash(
@slash_command(
name="info",
description="Read list of tracks for user",
options=[
Option("user", "Specify any user", OptionType.USER),
]
)
async def info(ctx, user=None):
async def info(self, ctx, user=None):
user = user or ctx.author
audio = DB.read_db(ctx.guild.id, user.id)
audio = lib.DB.read_db(ctx.guild.id, user.id)
rolelist = [r.mention for r in user.roles if r != ctx.guild.default_role]
if rolelist:
roles = "\n".join(rolelist)
@@ -52,6 +59,63 @@ class General(commands.Cog):
await ctx.reply(embed=emb, ephemeral=True)
@slash_command(
name="set_guest_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_guest_role(self, ctx, role):
await lib.Commands.write_json(ctx.guild.id, "guest_role", role.id)
await ctx.send(f"Setted up dsss bot role to: `{role.name}`", ephemeral=True)
def setup(bot): # a extension must have a setup function
@commands.command(name="set_prefix")
@commands.has_permissions(administrator=True)
async def command_set_prefix(self, ctx, prefix: str):
await lib.Commands.set_prefix(ctx, prefix)
@slash_command(
name="set_prefix",
description="Setting up bot prefix",
options=[
Option("prefix", "Specify prefix", OptionType.STRING, required=True),
]
)
@dislash.has_permissions(administrator=True)
async def slash_set_prefix(self, ctx, prefix: str):
await lib.Commands.set_prefix(ctx, prefix)
@slash_set_prefix.error
@command_set_prefix.error
async def set_prefix_error(self, ctx, prefix):
await ctx.reply('You don`t have permissions', ephemeral=True)
@slash_command(
name="set_trigger_role",
description="Setting up role to trigger bot",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_trigger_role(self, ctx, role):
await lib.Commands.write_json(ctx.guild.id, "tigger_role", role.id)
await ctx.send(f"Role to trigger set to : `{role.name}`", ephemeral=True)
@slash_command(
name="set_bot_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@dislash.has_permissions(administrator=True)
async def set_bot_role(self, ctx, role):
await lib.Commands.write_json(ctx.guild.id, "bot_role", role.id)
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
def setup(bot): # an extension must have a setup function
bot.add_cog(General(bot)) # adding a cog

31
cogs/info.py Normal file
View File

@@ -0,0 +1,31 @@
import logging
import discord
from discord.ext import commands
from dislash import slash_command
class Bot_info(commands.Cog):
def __init__(self, bot):
self.bot = bot # defining bot as global var in class
@commands.Cog.listener() # this is a decorator for events/listeners
async def on_ready(self):
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
@slash_command(name="info_bot") # this is for making a command
async def info_bot(self, ctx):
# await ctx.send(f'Pong! {round(self.bot.latency * 1000)}')
emb = discord.Embed(
title=f"General information",
description=f"General information on about bot",
)
emb.add_field(name="Bot ping", value=f'Bot ping: {round(self.bot.latency * 1000)}', inline=True)
# emb.add_field(name="Roles list", value=f"{roles}", inline=True)
emb.set_footer(text="Information requested by: {}".format(ctx.author.display_name))
await ctx.reply(embed=emb, ephemeral=True)
def setup(bot): # an extension must have a setup function
bot.add_cog(Bot_info(bot)) # adding a cog

144
lib.py Normal file
View File

@@ -0,0 +1,144 @@
import json
import logging
import sqlite3
from os import walk, listdir
class DB:
@staticmethod
def prepare_db(guild: int):
try:
connect = sqlite3.connect('user.db')
cursor = connect.cursor()
create_table = (f'''CREATE TABLE IF NOT EXISTS "{guild}"
([userid] INTEGER PRIMARY KEY, [username] TEXT, [nick] TEXT, [isbot] BOOL, [track] TEXT)
''')
cursor.execute(create_table)
cursor.close()
except sqlite3.Error as _error:
logging.error("failed to connect db", _error)
@staticmethod
def work_with_db(db_func, data_turple):
try:
connect = sqlite3.connect('user.db')
cursor = connect.cursor()
cursor.execute(db_func, data_turple)
connect.commit()
cursor.close()
except sqlite3.Error as _error:
logging.error("failed to connect db", _error)
@staticmethod
def fill_bd(name: str, userid: int, isbot: bool, nick: str, guild: int):
sqlite_insert_with_param = (f"""INSERT OR IGNORE INTO "{guild}"
(username, userid, nick, isbot)
VALUES (?, ?, ?, ?)""")
data_tuple = (name, userid, nick, isbot)
DB.work_with_db(sqlite_insert_with_param, data_tuple)
@staticmethod
def add_audio(guild: int, user: int, audio: str):
sql_update_query = f"""UPDATE "{guild}" set track = ? where userid = ?"""
data_tuple = (audio, user)
DB.work_with_db(sql_update_query, data_tuple)
@staticmethod
def read_db(guild: int, user: int):
try:
sql_con = sqlite3.connect("user.db")
cursor = sql_con.cursor()
sql_read = f"""SELECT * FROM "{guild}" where userid = {user}"""
cursor.execute(sql_read)
record = cursor.fetchone()
return record[4]
except sqlite3.Error as _error:
logging.error("Failed to read sqlite table", _error)
class CogsPrepare:
@staticmethod
def cog_list():
cogs_list = []
for _filename in listdir('./cogs'):
if _filename.endswith('.py'):
cogs_list.append(_filename[:-3])
return cogs_list
@staticmethod
def cogs_dict():
cog_dict = {}
for _cog in CogsPrepare.cog_list():
cog_dict.update({f'{_cog}': f'{_cog}'})
return cog_dict
@staticmethod
async def work_with_cogs(what_do, bot):
for _filename in CogsPrepare.cog_list():
if what_do == "load":
bot.load_extension(f'cogs.{_filename}')
elif what_do == 'unload':
bot.unload_extension(f'cogs.{_filename}')
elif what_do == 'reload':
bot.reload_extension(f'cogs.{_filename}')
logging.info(f'{what_do}ing cog {_filename}')
class Commands:
@staticmethod
async def set_prefix(ctx, prefix: str):
await Commands.write_json(ctx.guild.id, "prefix", prefix)
await ctx.send(f"Prefix set to: `{prefix}`", ephemeral=True)
@staticmethod
def list_files(fold: str):
fl = []
for filenames in walk(fold):
fl.extend(filenames)
break
files = {}
for x in fl[2]:
files[x] = x
return files
@staticmethod
async def read_json(guild: int, param: str):
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
parameter = None
try:
_json = json.load(fp) # Load the custom prefixes
except:
_json = {}
if guild: # If the guild exists
try:
guild_conf = _json[f"{guild}"]
try:
parameter = guild_conf[f"{param}"]
except:
pass
except:
pass
fp.close()
return parameter
@staticmethod
async def write_json(guild: int, param_name: str, param: str or int):
with open('prefix.json', 'r') as _f:
try:
_json = json.load(_f)
except json.decoder.JSONDecodeError:
_json = {}
_f.close()
try:
_guild = _json[f'{guild}']
except KeyError:
_json.update({f'{guild}': {}})
_guild = _json[f'{guild}']
_guild.update({f'{param_name}': f'{param}'})
with open('prefix.json', 'w') as _f:
json.dump(_json, _f, indent=4)
_f.close()

View File

@@ -6,7 +6,6 @@ import discord
from asyncio import sleep
from os import walk
from discord import user, member
from discord import FFmpegPCMAudio
from discord.ext import commands

314
test.py
View File

@@ -1,12 +1,11 @@
import json
import logging
import sqlite3
import sys
import threading
from os import walk, path, makedirs, remove, rename, listdir
import discord
import dislash
import lib
from os import path, makedirs, remove, rename
from discord.ext import commands
from dislash import InteractionClient, Option, OptionType, OptionChoice
@@ -20,12 +19,12 @@ def determine_prefix(bot, msg):
parameter = DEFAULT_PREFIX
try:
_json = json.load(fp) # Load the custom prefixes
except:
except error:
_json = {}
if guild: # If the guild exists
try:
parameter = _json[f"{guild.id}"]["prefix"]
except:
parameter = _json[f"{guild.id}"]["prefix"] # Read prefix from json if is setted up
except error:
pass
return parameter
@@ -36,164 +35,19 @@ bot = commands.Bot(command_prefix=determine_prefix, guild_subscriptions=True, in
inter_client = InteractionClient(bot, sync_commands=True)
class DB:
@staticmethod
def prepare_db(guild: int):
try:
connect = sqlite3.connect('user.db')
cursor = connect.cursor()
create_table = (f'''CREATE TABLE IF NOT EXISTS "{guild}"
([userid] INTEGER PRIMARY KEY, [username] TEXT, [nick] TEXT, [isbot] BOOL, [track] TEXT)
''')
cursor.execute(create_table)
cursor.close()
except sqlite3.Error as error:
logging.error("failed to connecct db", error)
@staticmethod
def work_with_db(db_func, data_turple):
try:
connect = sqlite3.connect('user.db')
cursor = connect.cursor()
cursor.execute(db_func, data_turple)
connect.commit()
cursor.close()
except sqlite3.Error as error:
logging.error("failed to connecct db", error)
@staticmethod
def fill_bd(name: str, userid: int, isbot: bool, nick: str, guild: int):
sqlite_insert_with_param = (f"""INSERT OR IGNORE INTO "{guild}"
(username, userid, nick, isbot)
VALUES (?, ?, ?, ?)""")
data_tuple = (name, userid, nick, isbot)
DB.work_with_db(sqlite_insert_with_param, data_tuple)
@staticmethod
def add_audio(guild: int, user: int, audio: str):
sql_update_query = f"""UPDATE "{guild}" set track = ? where userid = ?"""
data_tuple = (audio, user)
DB.work_with_db(sql_update_query, data_tuple)
@staticmethod
def read_db(guild: int, user: int):
try:
sql_con = sqlite3.connect("user.db")
cursor = sql_con.cursor()
sql_read = f"""SELECT * FROM "{guild}" where userid = {user}"""
cursor.execute(sql_read)
record = cursor.fetchone()
return record[4]
except sqlite3.Error as error:
logging.error("Failed to read sqlite table", error)
class CogsPrepare:
@staticmethod
def cog_list():
cogs_list = []
for _filename in listdir('./cogs'):
if _filename.endswith('.py'):
cogs_list.append(_filename[:-3])
return cogs_list
@staticmethod
async def work_with_cogs(what_do, cogs):
for _filename in cogs:
if what_do == "load":
bot.load_extension(f'cogs.{_filename}')
elif what_do == 'unload':
bot.unload_extension(f'cogs.{_filename}')
elif what_do == 'reload':
bot.reload_extension(f'cogs.{_filename}')
logging.info(f'{what_do}ing cog {_filename}')
@staticmethod
async def cog_load(ctx, cog):
try:
bot.load_extension(f'cogs.{cog}')
await ctx.reply(f'Loaded cog {cog}', ephemeral=True)
except commands.ExtensionNotFound:
await ctx.send(f"Error: {cog} couldn't be found to load.", ephemeral=True)
except commands.ExtensionFailed:
await ctx.send(f'Error: {cog} failed to load properly.', ephemeral=True)
except commands.ExtensionError:
await ctx.send(f'Error: unknown error with {cog}', ephemeral=True)
class Commands:
@staticmethod
async def set_prefix(ctx, prefix: str):
await Commands.write_json(ctx.guild.id, "prefix", prefix)
await ctx.send(f"Prefix set to: `{prefix}`", ephemeral=True)
@staticmethod
def list_files(fold: str):
fl = []
for filenames in walk(fold):
fl.extend(filenames)
break
files = {}
for x in fl[2]:
files[x] = x
return files
@staticmethod
async def read_json(guild: int, param: str):
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
parameter = None
try:
_json = json.load(fp) # Load the custom prefixes
except:
_json = {}
if guild: # If the guild exists
try:
guild_conf = _json[f"{guild}"]
try:
parameter = guild_conf[f"{param}"]
except:
pass
except:
pass
fp.close()
return parameter
@staticmethod
async def write_json(guild: int, param_name: str, param: str or int):
with open('prefix.json', 'r') as _f:
try:
_json = json.load(_f)
except json.decoder.JSONDecodeError:
_json = {}
_f.close()
try:
_guild = _json[f'{guild}']
except KeyError:
_json.update({f'{guild}': {}})
_guild = _json[f'{guild}']
_guild.update({f'{param_name}': f'{param}'})
with open('prefix.json', 'w') as _f:
json.dump(_json, _f, indent=4)
_f.close()
threading.current_thread().name = "main"
logging.basicConfig(stream=sys.stdout, filemode='w', level='INFO',
format='%(asctime)s - %(levelname)s - %(threadName)s - %(message)s')
if not path.isfile('prefix.json'):
with open('prefix.json', 'w+') as f:
f.write("")
f.close()
for filename in CogsPrepare.cog_list():
for filename in lib.CogsPrepare.cog_list():
try:
bot.load_extension(f'cogs.{filename}')
logging.info(f'Loaded cog {filename}')
@@ -201,8 +55,8 @@ for filename in CogsPrepare.cog_list():
except commands.ExtensionNotFound:
logging.error(f"Error: {filename} couldn't be found to load.")
except commands.ExtensionFailed:
logging.error(f'Error: {filename} failed to load properly.')
except commands.ExtensionFailed as error:
logging.error(f'Error: {filename} failed to load properly.', error)
except commands.ExtensionError:
logging.error(f'Error: unknown error with {filename}')
@@ -210,10 +64,6 @@ for filename in CogsPrepare.cog_list():
@bot.event
async def on_ready():
for g in bot.get_all_members():
DB.prepare_db(g.guild.id)
for g in bot.get_all_members():
DB.fill_bd(g.name, g.id, g.bot, g.nick, g.guild.id)
logging.info(f'Bot started')
logging.info('We have logged in as {0.user}'.format(bot))
@@ -221,15 +71,15 @@ async def on_ready():
@bot.event
async def on_guild_join(guild):
for g in guild.members:
DB.fill_bd(g.name, g.id, g.bot, g.nick, guild.id)
lib.DB.fill_bd(g.name, g.id, g.bot, g.nick, guild.id)
@bot.event
async def on_member_join(member):
DB.fill_bd(member.name, member.id, member.bot, member.nick, member.guild.id)
lib.DB.fill_bd(member.name, member.id, member.bot, member.nick, member.guild.id)
bot_role = Commands.read_json(member.guild.id, 'bot_role') # Get bot role
guest_role = Commands.read_json(member.guild.id, 'guest_role') # Get guest role
bot_role = lib.Commands.read_json(member.guild.id, 'bot_role') # Get bot role
guest_role = lib.Commands.read_json(member.guild.id, 'guest_role') # Get guest role
if bot_role and guest_role:
if member.bot == 0:
@@ -244,7 +94,7 @@ async def on_member_join(member):
async def on_member_update(before: discord.Member, after: discord.Member):
sql_update_query = f"""UPDATE "{after.guild.id}" set nick = ? where userid = ?"""
data_tuple = (after.nick, before.id)
DB.work_with_db(sql_update_query, data_tuple)
lib.DB.work_with_db(sql_update_query, data_tuple)
@bot.command(name="upload_audio")
@@ -256,13 +106,13 @@ async def upload_audio(ctx, user=None):
if not path.isdir(f'tmp/{user.id}'):
try:
makedirs(f'tmp/{user.id}')
except error as error:
logging.info(f"Failed to create dir", error)
except error as _error:
logging.info(f"Failed to create dir", _error)
if not path.isdir(f'audio/{user.id}'):
try:
makedirs(f'audio/{user.id}')
except error as error:
logging.info(f"Failed to create dir", error)
except error as _error:
logging.info(f"Failed to create dir", _error)
for at in ctx.message.attachments:
import mimetypes
@@ -275,13 +125,13 @@ async def upload_audio(ctx, user=None):
if duration > 15:
await ctx.reply(f'Audio duration is {duration}, but max is 15', ephemeral=True)
else:
a = DB.read_db(ctx.guild.id, user.id)
a = lib.DB.read_db(ctx.guild.id, user.id)
if a is None:
audiolist = f'{user.id}/{at.filename}'
else:
audiolist = DB.read_db(ctx.guild.id, user.id) + ", " + f'{user.id}/{at.filename}'
audiolist = lib.DB.read_db(ctx.guild.id, user.id) + ", " + f'{user.id}/{at.filename}'
DB.add_audio(ctx.guild.id, user.id, audiolist)
lib.DB.add_audio(ctx.guild.id, user.id, audiolist)
rename(f'tmp/{user.id}/{at.filename}', f'audio/{user.id}/{at.filename}')
else:
await ctx.reply(f'Failed to find MIME type of {at.filename}', ephemeral=True)
@@ -292,94 +142,6 @@ async def upload_audio(ctx, user=None):
await ctx.reply(f'You`re not admin. You can add audio only for your own account', ephemeral=True)
@bot.command(name="set_prefix")
@commands.has_permissions(administrator=True)
async def command_set_prefix(ctx, prefix: str):
await Commands.set_prefix(ctx, prefix)
@inter_client.slash_command(
name="set_prefix",
description="Settin up bot prefix",
options=[
Option("prefix", "Specify prefix", OptionType.STRING, required=True),
]
)
@dislash.has_permissions(administrator=True)
async def slash_set_prefix(ctx, prefix: str):
await Commands.set_prefix(ctx, prefix)
@slash_set_prefix.error
@command_set_prefix.error
async def set_prefix_error(ctx, prefix):
await ctx.reply('You don`t have permissions', ephemeral=True)
@inter_client.slash_command(
name="info",
description="Read list of tracks for user",
options=[
Option("user", "Specify any user", OptionType.USER),
]
)
async def info(ctx, user=None):
user = user or ctx.author
audio = DB.read_db(ctx.guild.id, user.id)
rolelist = [r.mention for r in user.roles if r != ctx.guild.default_role]
if rolelist:
roles = "\n".join(rolelist)
else:
roles = "Not added any role"
if audio is None:
audios = "Not selected audio"
else:
audios = "" + "\n".join(sorted(audio.split(", ")))
emb = discord.Embed(
title=f"General information",
description=f"General information on server about {user}",
icon=user.avatar_url
)
emb.set_thumbnail(url=user.avatar_url)
emb.add_field(name="General info",
value=f"Username: {user}\n"
f"Nickname: {user.nick}\n"
f"Joined at: {user.joined_at.strftime('%A, %B %d %Y @ %H:%M:%S')}", inline=False)
emb.add_field(name="Audio list", value=f"{audios}", inline=True)
emb.add_field(name="Roles list", value=f"{roles}", inline=True)
emb.set_footer(text="Information requested by: {}".format(ctx.author.display_name))
await ctx.reply(embed=emb, ephemeral=True)
@inter_client.slash_command(
name="set_trigger_role",
description="Setting up role to trigger bot",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_trigger_role(ctx, role):
await Commands.write_json(ctx.guild.id, "tigger_role", role.id)
await ctx.send(f"Role to trigger set to : `{role.name}`", ephemeral=True)
@inter_client.slash_command(
name="set_bot_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_bot_role(ctx, role):
await Commands.write_json(ctx.guild.id, "bot_role", role.id)
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
@inter_client.slash_command(
name="cog",
description="Work with cogs",
@@ -394,34 +156,16 @@ async def set_bot_role(ctx, role):
OptionChoice("unload", "unload"),
OptionChoice("reload", "reload")
]
),
Option('cog', "Specify cog if need", OptionType.STRING)
)
]
)
async def slash_cogs(ctx, what_do, cogs=None):
if cogs is None:
cogs = CogsPrepare.cog_list()
async def slash_cogs(ctx, what_do):
if ctx.author.id == bot_owner:
await CogsPrepare.work_with_cogs(what_do, cogs)
if cogs is CogsPrepare.cog_list():
await ctx.reply(f'Cogs are {what_do}ed', ephemeral=True)
else:
await ctx.reply(f'Cog {cogs[0]} is {what_do}ed', ephemeral=True)
await lib.CogsPrepare.work_with_cogs(what_do, bot)
await ctx.reply(f'Cogs are {what_do}ed', ephemeral=True)
else:
await ctx.reply('You`re not bot owner', ephemeral=True)
@inter_client.slash_command(
name="set_guest_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.ROLE, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_guest_role(ctx, role):
await Commands.write_json(ctx.guild.id, "guest_role", role.id)
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
bot.run('OTQ3OTUzOTAxNzgzNjIxNjYy.GTXbMv.KrztaTO7-ivsPEAVjsyikSQ-GP-ANwULmDraig')