ConfrerieDuKassoulait/Tom
ConfrerieDuKassoulait/Tom
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages 1 Wiki Activity Actions

add types, refactor code

Browse source
This commit is contained in:
Mylloon 2024-01-24 20:54:57 +01:00
parent 589e062ad2
commit 114701c230
Signed by: Anri
GPG key ID: A82D63DFF8D1317F
6 changed files with 103 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/client.py Normal file
View file

@ -0,0 +1,31 @@
from twitchio import Message
from twitchio.ext import commands
from utils.commands import CommandesDB
from utils.core import load
class Client(commands.Bot):
def __init__(self):
self.keys = load(["ACCESS_TOKEN", "PREFIX", "CHANNEL"])
super().__init__(
token=self.keys["ACCESS_TOKEN"],
prefix=self.keys["PREFIX"],
initial_channels=self.keys["CHANNEL"],
)
async def event_ready(self) -> None:
CommandesDB().creationTable()
print(f"Logged in as {self.nick}")
async def event_message(self, message: Message) -> None:
if message.echo: # messages with echo set to True are messages sent by the bot
return
# Let the bot know we want to handle and invoke our commands
await self.handle_commands(message)
async def event_command_error(self, _, error) -> None:
if isinstance(error, commands.errors.CommandNotFound):
# Ignore unknown commands (useful because custom commands aren't known)
return
raise error

36
src/main.py
View file

@ -1,43 +1,13 @@
from os import listdir
from twitchio.ext import commands
from utils.commands import CommandesDB
from utils.core import load
class Client(commands.Bot):
def __init__(self):
self.keys = load(["ACCESS_TOKEN", "PREFIX", "CHANNEL"])
super().__init__(
token=self.keys["ACCESS_TOKEN"],
prefix=self.keys["PREFIX"],
initial_channels=self.keys["CHANNEL"],
)
async def event_ready(self):
CommandesDB().creationTable()
print(f"Logged in as {self.nick}")
async def event_message(self, message):
if message.echo: # Messages with echo set to True are messages sent by the bot
return
await self.handle_commands(
message
) # Let the bot know we want to handle and invoke our commands
async def event_command_error(self, _, error):
if isinstance(
error, commands.errors.CommandNotFound
): # Ignore unknown commands (useful because custom commands aren’t known)
return
raise error
from client import Client
client = Client()
# Load modules
for file in listdir("modules"):
if file.endswith(".py") and file.startswith("-") is False:
client.load_module(f"modules.{file[:-3]}")
print("Starting..", end=" ", flush=True)
client.run()

80
src/modules/commandes.py
View file

@ -8,7 +8,8 @@ from utils.commands import CommandesDB, existeCommande, existeTouteCommande
from utils.core import listCommands, load
def prepare(client: Bot):
def prepare(client: Bot) -> None:
"""Add the module to the client"""
client.add_cog(Commandes(client))
@ -18,9 +19,9 @@ class Commandes(Cog):
def __init__(self, client: Bot):
self.client = client
self.prefix = load(["PREFIX"])["PREFIX"]
self.notModo = "tu n'es pas modérateur"
self.notExistingCommand = "cette commande n'existe pas"
self.alreadyExistingCommand = (
self.not_a_mod = "tu n'es pas modérateur"
self.unknown_command = "cette commande n'existe pas"
self.existing_command = (
f"cette commande existe déjà, {self.prefix}edit <name> <msg> pour l'éditer"
)
@ -35,89 +36,89 @@ class Commandes(Cog):
"""
if name is None or message is None:
return
author = cast(Chatter, ctx.author)
if author.is_mod:
if existeTouteCommande(self.client, name)[0] is False:
if existeTouteCommande(self.client, name) is False:
CommandesDB().ajoutCommande(name, message)
await ctx.send(f"@{ctx.author.name}, commande {name} ajoutée !")
else:
await ctx.send(f"@{ctx.author.name}, {self.alreadyExistingCommand}.")
await ctx.send(f"@{ctx.author.name}, {self.existing_command}.")
else:
await ctx.send(f"@{ctx.author.name}, {self.notModo}.")
await ctx.send(f"@{ctx.author.name}, {self.not_a_mod}.")
@new(name="remove", aliases=["delete", "suppr", "supprimer"], no_global_checks=True)
async def _remove(self, ctx: Context, commandName: str | None = None) -> None:
async def _remove(self, ctx: Context, name: str | None = None) -> None:
"""
Supprime une commande de la base de donnée du bot
`remove name`
"""
if commandName is None:
if name is None:
return
author = cast(Chatter, ctx.author)
if author.is_mod:
if existeCommande(commandName)[0]:
CommandesDB().suppressionCommande(commandName)
await ctx.send(
f"@{ctx.author.name}, commande {commandName} supprimée !"
)
if existeCommande(name)[0]:
CommandesDB().suppressionCommande(name)
await ctx.send(f"@{ctx.author.name}, commande {name} supprimée !")
else:
await ctx.send(f"@{ctx.author.name}, {self.notExistingCommand}.")
await ctx.send(f"@{ctx.author.name}, {self.unknown_command}.")
else:
await ctx.send(f"@{ctx.author.name}, {self.notModo}.")
await ctx.send(f"@{ctx.author.name}, {self.not_a_mod}.")
@new(name="list", aliases=["liste", "commands", "commandes", "help", "aide"])
async def _list(self, ctx: Context) -> None:
"""Affiche la liste des commandes de la base de donnée du bot"""
commandes = CommandesDB().listeCommande()
commandes: list[str] = [item[0] for item in CommandesDB().listeCommande()]
author = cast(Chatter, ctx.author)
for command in listCommands(self.client):
name = command.name
if command.no_global_checks:
if author.is_mod:
if command.no_global_checks and author.is_mod:
continue
if command.aliases:
name += " (alias: "
for aliase in command.aliases:
name += f"{self.prefix}{aliase}, "
name = f"{name[:-2]})"
commandes.append((name,))
commandes.append(name)
if len(commandes) > 0:
message = f"@{ctx.author.name}, liste des commandes -> "
for commande in commandes:
message += f"{self.prefix}{commande[0]}, "
await ctx.send(message[:-2])
await ctx.send(
f"@{ctx.author.name}, liste des commandes -> "
+ self.prefix
+ f", {self.prefix}".join(commandes)
)
else:
await ctx.send(
f"@{ctx.author.name}, aucune commande enrengistrée dans la base de donnée."
f"@{ctx.author.name}, aucune commande enregistrée dans la base de donnée."
)
@new(name="edit", no_global_checks=True)
async def _edit(
self,
ctx: Context,
commandName: str | None = None,
commandMessage: str | None = None,
name: str | None = None,
message: str | None = None,
) -> None:
"""
Modifie une commande de la base de donnée du bot
add name new_message
"""
if commandName is None or commandMessage is None:
if name is None or message is None:
return
author = cast(Chatter, ctx.author)
if author.is_mod:
if existeCommande(commandName)[0]:
CommandesDB().suppressionCommande(commandName)
CommandesDB().ajoutCommande(commandName, commandMessage)
await ctx.send(f"@{ctx.author.name}, commande {commandName} modifiée !")
if existeCommande(name)[0]:
CommandesDB().suppressionCommande(name)
CommandesDB().ajoutCommande(name, message)
await ctx.send(f"@{ctx.author.name}, commande {name} modifiée !")
else:
await ctx.send(f"@{ctx.author.name}, {self.notExistingCommand}.")
await ctx.send(f"@{ctx.author.name}, {self.unknown_command}.")
else:
await ctx.send(f"@{ctx.author.name}, {self.notModo}.")
await ctx.send(f"@{ctx.author.name}, {self.not_a_mod}.")
@Cog.event("event_message") # type: ignore
async def _event_message(self, message: Message) -> None:
@ -125,9 +126,8 @@ class Commandes(Cog):
return
if message.content.startswith(self.prefix):
command = existeCommande(
message.content[1:].split(" ")[0]
) # récupère le nom de la commande
if command[0]: # vérification si existe
# envois le contenu de la commande
# Récupère le nom de la commande
command = existeCommande(message.content[1:].split(" ")[0])
if command[0]: # vérification si elle existe
# Envois le contenu de la commande
await message.channel.send(f"@{message.author.name}, {command[1]}")

19
src/utils/commands.py
View file

@ -1,3 +1,4 @@
from twitchio.ext.commands import Bot
from utils.core import listCommands
from utils.db import Database
@ -6,7 +7,7 @@ class CommandesDB(Database):
def __init__(self):
super().__init__(r"db/bot.sqlite3")
def creationTable(self):
def creationTable(self) -> None:
"""Créer la table qui stocker les commandes."""
requete = """
CREATE TABLE IF NOT EXISTS commandes (
@ -16,7 +17,7 @@ class CommandesDB(Database):
"""
self.requete(requete)
def ajoutCommande(self, name: str, message: str):
def ajoutCommande(self, name: str, message: str) -> None:
"""Ajoute une commande."""
requete = """
INSERT INTO commandes (
@ -28,7 +29,7 @@ class CommandesDB(Database):
"""
self.requete(requete, [name, message])
def suppressionCommande(self, name: str):
def suppressionCommande(self, name: str) -> None:
"""Supprime une commande."""
requete = """
DELETE FROM commandes
@ -36,12 +37,12 @@ class CommandesDB(Database):
"""
self.requete(requete, name)
def listeCommande(self):
def listeCommande(self) -> list[tuple]:
"""Retourne la liste des commandes."""
return self.affichageResultat(self.requete("SELECT * FROM commandes;"))
return self.getResults(self.requete("SELECT * FROM commandes;"))
def existeCommande(command: str):
def existeCommande(command: str) -> tuple[bool, str | None]:
"""Vérifie qu'une commande existe dans la base de donnée."""
commandes = CommandesDB().listeCommande()
for commande in commandes:
@ -50,7 +51,7 @@ def existeCommande(command: str):
return (False, None)
def existeTouteCommande(client, commande: str):
def existeTouteCommande(client: Bot, commande: str) -> bool:
"""Vérifie qu'une commande existe dans la base de donnée et dans le bot en lui-même."""
commandes = []
for command in CommandesDB().listeCommande():
@ -60,6 +61,4 @@ def existeTouteCommande(client, commande: str):
if command.aliases:
for comm in command.aliases:
commandes.append(comm)
if commande in commandes:
return (True,)
return (False,)
return commande in commandes

5
src/utils/core.py
View file

@ -2,9 +2,10 @@ from os import environ
from sys import exit
from dotenv import load_dotenv
from twitchio.ext.commands import Bot, Command
def load(variables):
def load(variables: list[str]) -> dict:
"""Load env variables"""
keys = {}
load_dotenv()
@ -22,7 +23,7 @@ def load(variables):
return keys
def listCommands(client):
def listCommands(client: Bot) -> list[Command]:
cogs = client.cogs.values()
commands = []
for cog in cogs:

28
src/utils/db.py
View file

@ -1,25 +1,27 @@
import sqlite3
from sqlite3 import Connection, Cursor, Error, connect
class Database:
def __init__(self, urlDatabase: str):
connexion = self.createConnection(urlDatabase)
_cursor = tuple[Cursor, int | None] | None
def __init__(self, url: str):
connexion = self.createConnection(url)
if connexion is None:
raise Exception("Can't connect to database")
self.connexion = connexion
def createConnection(self, path):
def createConnection(self, path: str) -> Connection | None:
"""Connexion à une base de donnée SQLite"""
if not self.isFileExists(path):
if not self._fileExists(path):
open(path, "x")
connnexion = None
try:
connnexion = sqlite3.connect(path)
except sqlite3.Error as e:
connnexion = connect(path)
except Error as e:
print(e)
return connnexion
def isFileExists(self, path):
def _fileExists(self, path: str) -> bool:
"""Vérifie qu'un fichier existe"""
try:
open(path, "r")
@ -28,8 +30,10 @@ class Database:
else:
return True
def requete(self, requete, valeurs=None):
"""Envois une requête vers la base de données"""
def requete(
self, requete: str, valeurs: str | list | tuple | None = None
) -> _cursor:
"""Envoie une requête vers la base de données"""
try:
curseur = self.connexion.cursor()
if valeurs:
@ -40,10 +44,10 @@ class Database:
curseur.execute(requete)
self.connexion.commit()
return (curseur, curseur.lastrowid)
except sqlite3.Error as e:
except Error as e:
print(e)
def affichageResultat(self, curseur):
def getResults(self, curseur: _cursor) -> list[tuple]:
"""Affiche le résultat d'une requête"""
tableau = []
if curseur is None: