refractor lib code
This commit is contained in:
@@ -1,81 +1,44 @@
|
||||
import logging
|
||||
import tempfile
|
||||
from os import path, makedirs, rename, remove, walk
|
||||
from random import random
|
||||
from asyncio import sleep
|
||||
from os import path, makedirs, rename, remove
|
||||
from random import randrange
|
||||
|
||||
import disnake
|
||||
import inter as inter
|
||||
from disnake import FFmpegPCMAudio
|
||||
from enum import Enum
|
||||
from disnake.ext import commands
|
||||
|
||||
import lib
|
||||
|
||||
|
||||
# Files = commands.option_enum(lib.Commands.list_files(inter.id))
|
||||
def Files(str, fold, Enum) -> None:
|
||||
fl = []
|
||||
for filenames in walk(fold):
|
||||
fl.extend(filenames)
|
||||
break
|
||||
files = {}
|
||||
for x in fl[2]:
|
||||
files[x] = x
|
||||
from lib.DB import read_db, check_exist_audio, add_audio
|
||||
|
||||
|
||||
class Audio(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot # defining bot as global var in class
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener() # this is a decorator for events/listeners
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_voice_state_update(self, member, before, after):
|
||||
role = await lib.Commands.read_json(member.guild.id, 'tigger_role')
|
||||
audio = await lib.DB.read_db(member.guild.id, member.id, 'usertracks')
|
||||
audio_files = await lib.Commands.list_files(f'audio/{member.id}')
|
||||
logging.info(f'Func check audio state:\n'
|
||||
f'\t\t\t\t\tUser: {member}\n'
|
||||
f'\t\t\t\t\tTrigger role: {role}\n'
|
||||
f'\t\t\t\t\tAudio list: {audio}\n'
|
||||
f'\t\t\t\t\tAudio file list: {audio_files}\n'
|
||||
f'\t\t\t\t\t-----------------')
|
||||
if before.channel is None and role in member.roles:
|
||||
pass
|
||||
# track = random.randint(0, len(f) - 1)
|
||||
# audio_source = FFmpegPCMAudio(f'audio/{f[track]}')
|
||||
# logging.error(f'{track}\t\t{f[track]}')
|
||||
# if not bot.voice_clients:
|
||||
# await sleep(1)
|
||||
# _channel = after.channel
|
||||
# vc = await after.channel.connect()
|
||||
# if not vc.is_playing():
|
||||
# vc.play(audio_source, after=None)
|
||||
# while vc.is_playing():
|
||||
# await sleep(0.5)
|
||||
# await sleep(1)
|
||||
# await vc.disconnect()
|
||||
# if before.channel is None and member.id == _memb:
|
||||
# track = random.randint(0, len(f) - 1)
|
||||
# audio_source = FFmpegPCMAudio(f'audio/{_memb}/bear2_enemy_scav3.wav')
|
||||
# logging.error(f'{track}\t\t\t{f[track]}')
|
||||
# if not bot.voice_clients:
|
||||
# await sleep(1)
|
||||
# _channel = after.channel
|
||||
# vc = await after.channel.connect()
|
||||
# if not vc.is_playing():
|
||||
# vc.play(audio_source, after=None)
|
||||
# while vc.is_playing():
|
||||
# await sleep(0.5)
|
||||
# await sleep(1)
|
||||
# await vc.disconnect()
|
||||
|
||||
# @commands.slash_command(name="select_audio")
|
||||
# async def select_audio(self, inter: disnake.ApplicationCommandInteraction,
|
||||
# files: Files):
|
||||
# pass
|
||||
from lib.Comands import read_json
|
||||
role = await read_json(member.guild.id, 'tigger_role')
|
||||
audio = await read_db(member.guild.id, member.id, 'usertracks')
|
||||
from lib.Comands import list_files
|
||||
audio_files = await list_files(f'audio/{member.id}')
|
||||
f = await list_files()
|
||||
if role is not None and before.channel is None and role in member.roles:
|
||||
track = randrange(0, len(f) - 1, 1)
|
||||
audio_source = FFmpegPCMAudio(f'audio/{f[track]}')
|
||||
if not self.bot.voice_clients:
|
||||
await sleep(1)
|
||||
_channel = after.channel
|
||||
vc = await after.channel.connect()
|
||||
if not vc.is_playing():
|
||||
vc.play(audio_source, after=None)
|
||||
while vc.is_playing():
|
||||
await sleep(0.5)
|
||||
await sleep(1)
|
||||
await vc.disconnect()
|
||||
|
||||
@commands.command(name="upload_audio")
|
||||
async def upload_audio(self, ctx, user=None):
|
||||
@@ -88,7 +51,7 @@ class Audio(commands.Cog):
|
||||
try:
|
||||
makedirs(f'audio/{user.id}')
|
||||
except error as _error:
|
||||
logging.info(f"Failed to create dir", _error)
|
||||
pass
|
||||
for at in ctx.message.attachments:
|
||||
import mimetypes
|
||||
|
||||
@@ -102,14 +65,14 @@ class Audio(commands.Cog):
|
||||
await ctx.reply(f'Audio duration is {duration}, but max is 15')
|
||||
remove(f'{tempfile.tempdir}/{user.id}/{at.filename}')
|
||||
else:
|
||||
a = await lib.DB.read_db(ctx.guild.id, user.id, 'usertracks')
|
||||
a = await read_db(ctx.guild.id, user.id, 'usertracks')
|
||||
if a:
|
||||
audiolist = a + ", " + f'{at.filename}'
|
||||
else:
|
||||
audiolist = f'{at.filename}'
|
||||
|
||||
await lib.DB.check_exist_audio(ctx, ctx.guild.id, user.id, 'usertracks', at.filename)
|
||||
await lib.DB.add_audio(ctx.guild.id, user.id, audiolist)
|
||||
await check_exist_audio(ctx, ctx.guild.id, user.id, 'usertracks', at.filename)
|
||||
await add_audio(ctx.guild.id, user.id, audiolist)
|
||||
rename(f'{tempfile.tempdir}/{user.id}/{at.filename}', f'audio/{user.id}/{at.filename}')
|
||||
elif guess[0].split('/')[0] != 'audio':
|
||||
await ctx.reply(f'It not audio {at.filename}\n it`s {guess[0]}')
|
||||
|
||||
@@ -5,6 +5,8 @@ from disnake import Option, OptionType, Colour
|
||||
from disnake.ext import commands
|
||||
|
||||
import lib
|
||||
from lib.Comands import read_json, set_prefix, write_json
|
||||
from lib.DB import fill_bd, prepare_db, work_with_db
|
||||
|
||||
|
||||
class General(commands.Cog):
|
||||
@@ -14,23 +16,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():
|
||||
await lib.DB.prepare_db(g.guild.id)
|
||||
await prepare_db(g.guild.id)
|
||||
for g in self.bot.get_all_members():
|
||||
await lib.DB.fill_bd(g.name, g.id, g.bot, g.nick, g.guild.id)
|
||||
await fill_bd(g.name, g.id, g.bot, g.nick, g.guild.id)
|
||||
|
||||
logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self, guild):
|
||||
for g in guild.members:
|
||||
await lib.DB.fill_bd(g.name, g.id, g.bot, g.nick, guild.id)
|
||||
await fill_bd(g.name, g.id, g.bot, g.nick, guild.id)
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_member_join(self, member):
|
||||
await lib.DB.fill_bd(member.name, member.id, member.bot, member.nick, member.guild.id)
|
||||
|
||||
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
|
||||
bot_role = read_json(member.guild.id, 'bot_role') # Get bot role
|
||||
guest_role = read_json(member.guild.id, 'guest_role') # Get guest role
|
||||
|
||||
if bot_role and guest_role:
|
||||
if member.bot == 0:
|
||||
@@ -44,7 +46,7 @@ class General(commands.Cog):
|
||||
async def on_member_update(self, before: disnake.Member, after: disnake.Member):
|
||||
sql_update_query = f"""UPDATE "{after.guild.id}" set nick = ? where userid = ?"""
|
||||
data_tuple = (after.nick, before.id)
|
||||
await lib.DB.work_with_db(sql_update_query, data_tuple)
|
||||
await work_with_db(sql_update_query, data_tuple)
|
||||
|
||||
@commands.slash_command(
|
||||
name="info",
|
||||
@@ -99,13 +101,13 @@ class General(commands.Cog):
|
||||
)
|
||||
@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 write_json(ctx.guild.id, "guest_role", role.id)
|
||||
await ctx.send(f"Setted up dsss bot role to: `{role.name}`", ephemeral=True)
|
||||
|
||||
@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)
|
||||
await set_prefix(ctx.guild.id, prefix)
|
||||
await ctx.reply(f"Prefix set to: `{prefix}`")
|
||||
|
||||
@commands.slash_command(
|
||||
@@ -117,7 +119,7 @@ class General(commands.Cog):
|
||||
)
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def slash_set_prefix(self, inter, prefix: str):
|
||||
await lib.Commands.set_prefix(inter, prefix)
|
||||
await set_prefix(inter.guild.id, prefix)
|
||||
await inter.responce.send_message(f"Prefix set to: `{prefix}`", ephemeral=True)
|
||||
|
||||
@slash_set_prefix.error
|
||||
@@ -133,7 +135,7 @@ class General(commands.Cog):
|
||||
)
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def set_trigger_role(self, inter, role):
|
||||
await lib.Commands.write_json(inter.guild.id, "tigger_role", role.id)
|
||||
await write_json(inter.guild.id, "tigger_role", role.id)
|
||||
await inter.response.send_message(f"Role to trigger set to : `{role.name}`", ephemeral=True)
|
||||
|
||||
@commands.slash_command(
|
||||
@@ -145,7 +147,7 @@ class General(commands.Cog):
|
||||
)
|
||||
@commands.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 write_json(ctx.guild.id, "bot_role", role.id)
|
||||
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
|
||||
|
||||
@commands.slash_command()
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
import logging
|
||||
import os
|
||||
|
||||
import psutil
|
||||
|
||||
import disnake
|
||||
import psutil
|
||||
from disnake.ext import commands
|
||||
|
||||
|
||||
@@ -27,6 +26,7 @@ class Bot_info(commands.Cog, name='Bot Info'):
|
||||
emb.add_field(name="Bot ping", value=f'Bot ping: {round(self.bot.latency * 1000)}', inline=True)
|
||||
emb.add_field(name="Memory Usage", value=f"{_process.memory_info()[0]/2**20} Mb", inline=True)
|
||||
emb.add_field(name="CPU Usage", value=f"{_process.cpu_percent()}%", inline=True)
|
||||
emb.add_field(name="Prefix", value=f"{self.bot.prefix}")
|
||||
emb.add_field(name="Bot owner", value=f"<@{self.bot.owner_id}>")
|
||||
emb.set_footer(text="Information requested by: {}".format(inter.author.display_name))
|
||||
|
||||
|
||||
229
lib.py
229
lib.py
@@ -1,229 +0,0 @@
|
||||
import json
|
||||
import logging
|
||||
import sqlite3
|
||||
from os import walk, listdir, path
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class DB:
|
||||
@staticmethod
|
||||
async 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
|
||||
async def work_with_db(db_func: str, data_turple: tuple):
|
||||
"""
|
||||
Writing to db per server userinfo
|
||||
:param db_func:
|
||||
:param 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
|
||||
async 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: tuple[str, int, str, bool] = (name, userid, nick, isbot)
|
||||
await DB.work_with_db(sqlite_insert_with_param, data_tuple)
|
||||
|
||||
@staticmethod
|
||||
async def add_audio(guild: int, user: int, audio: str, track: str = 'usertracks'):
|
||||
"""
|
||||
Adding audio into folder and DB
|
||||
|
||||
:param guild: Guild id
|
||||
:param user:
|
||||
:param audio:
|
||||
:param track: usertracks or defaulttracks
|
||||
"""
|
||||
# audio = f'{DB.read_db(guild, user, track)}, {audio}'
|
||||
|
||||
logging.error(f"Guild id is: {guild}\n"
|
||||
f"\t\t\t\t\tUser id is: {user}\n"
|
||||
f"\t\t\t\t\taudio is: {audio}\n"
|
||||
f"\t\t\t\t\tcolumn is {track}\n"
|
||||
f"\t\t\t\t\t---------------")
|
||||
sql_update_query = f"""UPDATE "{guild}" set {track} = ? where userid = ?"""
|
||||
data_tuple = (audio, user)
|
||||
logging.info(f'Func add_audio:\n'
|
||||
f'\t\t\t\t\tQuery: {sql_update_query}\n'
|
||||
f'\t\t\t\t\tTurple: {data_tuple}\n'
|
||||
f'\t\t\t\t\tAudio: {audio}\n'
|
||||
f'\t\t\t\t\t-----------------')
|
||||
await DB.work_with_db(sql_update_query, data_tuple)
|
||||
|
||||
@staticmethod
|
||||
async def read_db(guild: int, user: int, column: str):
|
||||
_col_dict = {'userid': 0,
|
||||
'username': 1,
|
||||
'nick': 2,
|
||||
'isbot': 3,
|
||||
'defaulttracks': 4,
|
||||
'usertracks': 5}
|
||||
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()
|
||||
|
||||
logging.info(f'Func read_db:\n'
|
||||
f'\t\t\t\t\tTrack: {column}\n'
|
||||
f'\t\t\t\t\tRecord: {record}\n'
|
||||
f'\t\t\t\t\t-----------------')
|
||||
|
||||
return record[_col_dict[column]]
|
||||
|
||||
except sqlite3.Error as _error:
|
||||
logging.error("Failed to read sqlite table", _error)
|
||||
|
||||
@staticmethod
|
||||
async def check_exist_audio(ctx, guild: int, user: int, column: str, audio: str):
|
||||
_list_str = await DB.read_db(guild, user, column)
|
||||
print(type(_list_str))
|
||||
if _list_str is not None:
|
||||
_list = _list_str.split(',')
|
||||
if audio in _list:
|
||||
await ctx.reply("File in list")
|
||||
logging.info(f'File {audio} in list')
|
||||
else:
|
||||
logging.info(f'File {audio} is not in list')
|
||||
else:
|
||||
_list = 'None'
|
||||
logging.info(f'check_exist_audio\n'
|
||||
f'\t\t\t\t\t{_list}'
|
||||
f'\t\t\t\t\t{audio}')
|
||||
|
||||
|
||||
class CogsPrepare:
|
||||
"""
|
||||
Loads, unloads Cogs files
|
||||
"""
|
||||
|
||||
@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 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
|
||||
def check_json():
|
||||
if not path.isfile('prefix.json'):
|
||||
with open('prefix.json', 'w+', encoding='utf-8') as f:
|
||||
f.write("")
|
||||
|
||||
@staticmethod
|
||||
async def set_prefix(inter, prefix: str):
|
||||
await Commands.write_json(inter.guild.id, "prefix", prefix)
|
||||
|
||||
@staticmethod
|
||||
async 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 TypeError:
|
||||
_json = {}
|
||||
if guild: # If the guild exists
|
||||
try:
|
||||
guild_conf = _json[f"{guild}"]
|
||||
try:
|
||||
parameter = guild_conf[f"{param}"]
|
||||
except:
|
||||
pass
|
||||
except:
|
||||
pass
|
||||
|
||||
return parameter
|
||||
|
||||
@staticmethod
|
||||
async def write_json(guild: int, param_name: str, param: str or int):
|
||||
with open('prefix.json', 'r', encoding='utf-8') as f:
|
||||
try:
|
||||
_json = json.load(f)
|
||||
except json.decoder.JSONDecodeError:
|
||||
_json = {}
|
||||
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', encoding='utf-8') as f:
|
||||
json.dump(_json, f, indent=4)
|
||||
|
||||
@staticmethod
|
||||
async def determine_prefix(bot, msg):
|
||||
"""
|
||||
Determite perserver bot prefix
|
||||
:param bot: Disnake Bot object
|
||||
:param msg: Disnake msg object
|
||||
:return: prefix
|
||||
"""
|
||||
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
|
||||
parameter: str
|
||||
try:
|
||||
from json import load
|
||||
_json = load(fp) # Load the custom prefixes
|
||||
except:
|
||||
_json = {}
|
||||
|
||||
try:
|
||||
parameter = _json[f"{msg.guild.id}"]["prefix"] # Read prefix from json if is setted up
|
||||
except:
|
||||
parameter = '$'
|
||||
|
||||
return parameter
|
||||
30
lib/CogsPrepare.py
Normal file
30
lib/CogsPrepare.py
Normal file
@@ -0,0 +1,30 @@
|
||||
from os import listdir
|
||||
|
||||
"""
|
||||
Loads, unloads Cogs files
|
||||
"""
|
||||
|
||||
|
||||
def cog_list():
|
||||
cogs_list = []
|
||||
for _filename in listdir('./cogs'):
|
||||
if _filename.endswith('.py'):
|
||||
cogs_list.append(_filename[:-3])
|
||||
return cogs_list
|
||||
|
||||
|
||||
async def cogs_dict():
|
||||
cog_dict = {}
|
||||
for _cog in cog_list():
|
||||
cog_dict.update({f'{_cog}': f'{_cog}'})
|
||||
return cog_dict
|
||||
|
||||
|
||||
async def work_with_cogs(what_do, bot):
|
||||
for _filename in 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}')
|
||||
79
lib/Comands.py
Normal file
79
lib/Comands.py
Normal file
@@ -0,0 +1,79 @@
|
||||
import json
|
||||
from os import walk, path
|
||||
|
||||
|
||||
def check_json():
|
||||
if not path.isfile('prefix.json'):
|
||||
with open('prefix.json', 'w+', encoding='utf-8') as f:
|
||||
f.write("")
|
||||
|
||||
|
||||
async def set_prefix(guildid: int, prefix: str) -> None:
|
||||
await write_json(guildid, "prefix", prefix)
|
||||
|
||||
|
||||
async def list_files(fold: str = 'audio'):
|
||||
fl = []
|
||||
for filenames in walk(fold):
|
||||
fl.extend(filenames)
|
||||
break
|
||||
files = {}
|
||||
for x in fl[2]:
|
||||
files[x] = x
|
||||
return fl[2]
|
||||
|
||||
|
||||
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 TypeError:
|
||||
_json = {}
|
||||
if guild: # If the guild exists
|
||||
try:
|
||||
guild_conf = _json[f"{guild}"]
|
||||
try:
|
||||
parameter = guild_conf[f"{param}"]
|
||||
except:
|
||||
pass
|
||||
except:
|
||||
pass
|
||||
return parameter
|
||||
|
||||
|
||||
async def write_json(guild: int, param_name: str, param: str or int):
|
||||
with open('prefix.json', 'r', encoding='utf-8') as f:
|
||||
try:
|
||||
_json = json.load(f)
|
||||
except json.decoder.JSONDecodeError:
|
||||
_json = {}
|
||||
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', encoding='utf-8') as f:
|
||||
json.dump(_json, f, indent=4)
|
||||
|
||||
|
||||
async def determine_prefix(bot, msg):
|
||||
"""
|
||||
Determite per-server bot prefix
|
||||
:param bot: Disnake Bot object
|
||||
:param msg: Disnake msg object
|
||||
:return: prefix for server, default is $
|
||||
"""
|
||||
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
|
||||
parameter: str
|
||||
try:
|
||||
from json import load
|
||||
_json = load(fp) # Load the custom prefixes
|
||||
except:
|
||||
_json = {}
|
||||
try:
|
||||
parameter = _json[f"{msg.guild.id}"]["prefix"] # Read prefix from json if is setted up
|
||||
except:
|
||||
parameter = '$'
|
||||
return parameter
|
||||
85
lib/DB.py
Normal file
85
lib/DB.py
Normal file
@@ -0,0 +1,85 @@
|
||||
import sqlite3
|
||||
|
||||
|
||||
async 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, [defaulttracks] TEXT, [usertracks] TEXT)
|
||||
''')
|
||||
cursor.execute(create_table)
|
||||
cursor.close()
|
||||
except sqlite3.Error as _error:
|
||||
pass
|
||||
|
||||
|
||||
async def work_with_db(db_func: str, data_turple: tuple):
|
||||
"""
|
||||
Writing to db per server userinfo
|
||||
:param db_func:
|
||||
:param 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:
|
||||
pass
|
||||
|
||||
|
||||
async 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: tuple[str, int, str, bool] = (name, userid, nick, isbot)
|
||||
await work_with_db(sqlite_insert_with_param, data_tuple)
|
||||
|
||||
|
||||
async def add_audio(guild: int, user: int, audio: str, track: str = 'usertracks'):
|
||||
"""
|
||||
Adding audio into folder and DB
|
||||
:param guild: Guild id
|
||||
:param user:
|
||||
:param audio:
|
||||
:param track: usertracks or defaulttracks
|
||||
"""
|
||||
# audio = f'{DB.read_db(guild, user, track)}, {audio}'
|
||||
sql_update_query = f"""UPDATE "{guild}" set {track} = ? where userid = ?"""
|
||||
data_tuple = (audio, user)
|
||||
await work_with_db(sql_update_query, data_tuple)
|
||||
|
||||
|
||||
async def read_db(guild: int, user: int, column: str):
|
||||
_col_dict = {'userid': 0,
|
||||
'username': 1,
|
||||
'nick': 2,
|
||||
'isbot': 3,
|
||||
'defaulttracks': 4,
|
||||
'usertracks': 5}
|
||||
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[_col_dict[column]]
|
||||
except sqlite3.Error as _error:
|
||||
pass
|
||||
|
||||
|
||||
async def check_exist_audio(ctx, guild: int, user: int, column: str, audio: str):
|
||||
_list_str = await read_db(guild, user, column)
|
||||
print(type(_list_str))
|
||||
if _list_str is not None:
|
||||
_list = _list_str.split(',')
|
||||
if audio in _list:
|
||||
await ctx.reply("File in list")
|
||||
|
||||
else:
|
||||
pass
|
||||
|
||||
else:
|
||||
_list = 'None'
|
||||
0
lib/__init__.py
Normal file
0
lib/__init__.py
Normal file
14
test.py
14
test.py
@@ -1,31 +1,29 @@
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
|
||||
import disnake
|
||||
from disnake import OptionChoice, OptionType, Option
|
||||
from disnake.ext import commands
|
||||
|
||||
import lib
|
||||
from lib.CogsPrepare import cog_list, work_with_cogs
|
||||
from lib.Comands import check_json, determine_prefix
|
||||
|
||||
bot_owner = 386629192743256065
|
||||
lib.Commands.check_json()
|
||||
check_json()
|
||||
intents = disnake.Intents(messages=True, guilds=True, message_content=True)
|
||||
intents.members = True
|
||||
intents.voice_states = True
|
||||
|
||||
bot = commands.Bot(command_prefix=lib.Commands.determine_prefix,
|
||||
bot = commands.Bot(command_prefix=determine_prefix,
|
||||
intents=intents,
|
||||
status=disnake.Status.idle,
|
||||
reload=True,
|
||||
test_guilds=[929446191270330410, 987120286933602354])
|
||||
|
||||
threading.current_thread().name = "main"
|
||||
logging.basicConfig(stream=sys.stdout, filemode='w', level='INFO',
|
||||
format='%(asctime)s - %(levelname)s - %(threadName)s - %(message)s')
|
||||
|
||||
|
||||
for filename in lib.CogsPrepare.cog_list():
|
||||
for filename in cog_list():
|
||||
try:
|
||||
bot.load_extension(f'cogs.{filename}')
|
||||
logging.info(f'Loaded cog {filename}')
|
||||
@@ -65,7 +63,7 @@ async def on_ready():
|
||||
)
|
||||
async def slash_cogs(inter, what_do):
|
||||
if inter.author.id == bot_owner:
|
||||
await lib.CogsPrepare.work_with_cogs(what_do, bot)
|
||||
await work_with_cogs(what_do, bot)
|
||||
await inter.response.send_message(f'Cogs are {what_do}ed', ephemeral=True)
|
||||
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user