Compare commits
2 Commits
89077db935
...
5a2def921f
Author | SHA1 | Date |
---|---|---|
Shadow15510 | 5a2def921f | |
Shadow15510 | 30073ceafe |
Binary file not shown.
259
fun_cmnds.py
259
fun_cmnds.py
|
@ -1,24 +1,251 @@
|
|||
from irc_api import api
|
||||
from irc_api.api import auto_help
|
||||
from irc_api import commands
|
||||
from irc_api.commands import auto_help
|
||||
|
||||
from random import randint, choice
|
||||
import sqlite3
|
||||
|
||||
@api.every(30, desc="Fait apparaître un canard de manière aléatoire")
|
||||
|
||||
GAME_CHANNEL = "#glados"
|
||||
|
||||
|
||||
def get_hunter(name: str):
|
||||
def get():
|
||||
table = sqlite3.connect("duck_hunting.db")
|
||||
c = table.cursor()
|
||||
data = c.execute(f"""
|
||||
SELECT * FROM hunter
|
||||
WHERE name="{name}"
|
||||
""").fetchall()
|
||||
table.close()
|
||||
return data
|
||||
|
||||
data = get()
|
||||
if not data:
|
||||
add_hunter(name)
|
||||
data = get()
|
||||
|
||||
return list(data[0][1:])
|
||||
|
||||
|
||||
def add_hunter(name: str):
|
||||
table = sqlite3.connect("duck_hunting.db")
|
||||
c = table.cursor()
|
||||
c.execute(f"""
|
||||
INSERT INTO hunter (name)
|
||||
VALUES ("{name}")
|
||||
""")
|
||||
table.commit()
|
||||
table.close()
|
||||
|
||||
|
||||
def del_hunter(name: str):
|
||||
table = sqlite3.connect("duck_hunting.db")
|
||||
c = table.cursor()
|
||||
c.execute(f"""
|
||||
DELETE FROM hunter
|
||||
WHERE name="{name}"
|
||||
""")
|
||||
table.commit()
|
||||
table.close()
|
||||
|
||||
|
||||
def modify_hunter(name: str, **kwargs):
|
||||
args = ""
|
||||
for var, value in kwargs.items():
|
||||
args += f"{var}={value}, "
|
||||
|
||||
table = sqlite3.connect("duck_hunting.db")
|
||||
c = table.cursor()
|
||||
c.execute(f"""
|
||||
UPDATE hunter
|
||||
SET {args[:-2]}
|
||||
WHERE name="{name}"
|
||||
""")
|
||||
table.commit()
|
||||
table.close()
|
||||
|
||||
|
||||
|
||||
@commands.every(30, desc="Fait apparaître un canard de manière aléatoire.")
|
||||
def pop_duck(bot):
|
||||
if randint(1, 10) == 1:
|
||||
ducks = ("Coin coin ! 🦆", "Couac ! 🦆")
|
||||
bot.nb_ducks += 1
|
||||
bot.send("#glados", choice(ducks))
|
||||
if randint(1, 10) == 1 and bot.nb_ducks > 1:
|
||||
bot.send("#glados", "Le canard est parti…")
|
||||
bot.nb_ducks -= 1
|
||||
if randint(1, 100) <= 20:
|
||||
if bot.troll_duck > 0 and randint(1, 4) <= 3:
|
||||
bot.troll_duck -= 1
|
||||
bot.ducks_history.append(2)
|
||||
ducks = ("crr… Coin coin ! 🦆", "Couac crr… ! 🦆")
|
||||
elif randint(1, 10) == 1:
|
||||
ducks = ("COIN COIN ! 🦆", "COUAC ! 🦆")
|
||||
bot.ducks_history.append(1)
|
||||
else:
|
||||
ducks = ("Coin coin ! 🦆", "Couac ! 🦆")
|
||||
bot.ducks_history.append(0)
|
||||
|
||||
bot.send(GAME_CHANNEL, choice(ducks))
|
||||
if randint(1, 100) <= 15 and len(bot.ducks_history) > 1:
|
||||
ducks = ("Le canard est parti…", "Le canard s'est envolé…")
|
||||
bot.send(GAME_CHANNEL, "Le canard est parti…")
|
||||
bot.ducks_history.pop(0)
|
||||
|
||||
|
||||
@api.channel("#glados")
|
||||
@api.command("pan", desc="Tirer sur un canard")
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("pan", desc="Tirer sur un canard.")
|
||||
def fire(bot, msg):
|
||||
if bot.nb_ducks > 0:
|
||||
bot.nb_ducks -= 1
|
||||
bot.send("#glados", "Touché !")
|
||||
money, ammo, ammo_in_chamber, _, has_scope, grease, is_jammed = get_hunter(msg.author)
|
||||
|
||||
if ammo_in_chamber <= 0:
|
||||
bot.send(GAME_CHANNEL, "Chambre vide ! Il faut recharger. 😉")
|
||||
return
|
||||
|
||||
if is_jammed:
|
||||
bot.send(GAME_CHANNEL, "Votre fusil est enrayé !")
|
||||
return
|
||||
|
||||
if grease > 0:
|
||||
grease -= 1
|
||||
elif grease == 0 and randint(1, 100) < 33:
|
||||
is_jammed = 1
|
||||
|
||||
if len(bot.ducks_history) > 0:
|
||||
is_hit = randint(1, 100)
|
||||
if has_scope:
|
||||
has_scope -= 1
|
||||
is_hit = randint(1, 15)
|
||||
|
||||
if is_hit <= 80:
|
||||
duck_type = bot.ducks_history.pop(-1)
|
||||
money += 2
|
||||
if duck_type == 2:
|
||||
hit = ("Oh un canard en plastique", "Peeww… CLONG")
|
||||
bot.send(GAME_CHANNEL, "Canard mécanique ! [munition -1]\n" + choice(hit))
|
||||
elif duck_type == 1 or is_hit < 10:
|
||||
money += 2
|
||||
hit = ("C'était un super canard !", "Ouh la belle bête !", "ÇA c'est du canard.")
|
||||
bot.send(GAME_CHANNEL, "Super canard touché ! [argent +4, munition -1]\n" + choice(hit))
|
||||
elif duck_type == 0:
|
||||
hit = ("Touché !", "Et un canard en moins !", "You dit it bro!", "Joli tir !", "Peeww !")
|
||||
bot.send(GAME_CHANNEL, "Canard touché ! [argent +2, munition -1]\n" + choice(hit))
|
||||
|
||||
|
||||
else:
|
||||
fail = ("Et le canard est toujours vivant…", "Mais… Pourquoi tu tires à côté ?", "Il est pas passé loin de la vieille celui-là !")
|
||||
bot.send(GAME_CHANNEL, "Tir raté… [munition -1]\n" + choice(fail))
|
||||
|
||||
else:
|
||||
bot.send("#glados", "Euh va check ta vue stp…")
|
||||
no_ducks = ("Euh, va faire vérifier ta vue s'il-te-plaît.", "Il est où le canard ? Il est bien caché ?", "Et une grand mère en moins ! Une ! Wait…", "Un canard ne hurle pas comme ça.", "Joli tir… Mais ce n'était pas un canard…")
|
||||
bot.send(GAME_CHANNEL, "Aucun canard en vue. [munition -1]\n" + choice(no_ducks))
|
||||
|
||||
modify_hunter(msg.author, money=money, ammo=ammo, ammo_in_chamber=ammo_in_chamber - 1, has_scope=has_scope, grease=grease, is_jammed=is_jammed)
|
||||
|
||||
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("recharge", alias=("reload",), desc="Fait monter une balle du chargeur dans la chambre")
|
||||
def reload(bot, msg):
|
||||
ammo, ammo_in_chamber, max_ammo = get_hunter(msg.author)[1: 4]
|
||||
ammo_to_fill = max_ammo - ammo_in_chamber
|
||||
|
||||
if ammo_to_fill == 0:
|
||||
bot.send(GAME_CHANNEL, "Ton arme est déjà chargée.")
|
||||
elif ammo > 0:
|
||||
if ammo >= ammo_to_fill:
|
||||
ammo_filled = ammo_to_fill
|
||||
else:
|
||||
ammo_filled = ammo
|
||||
modify_hunter(msg.author, ammo=ammo - ammo_filled, ammo_in_chamber=ammo_in_chamber + ammo_filled)
|
||||
bot.send(GAME_CHANNEL, f"{msg.author} a rechargé son arme.")
|
||||
else:
|
||||
bot.send(GAME_CHANNEL, f"Plus de munitions ! Va voir le magasin pour en acheter. 😉")
|
||||
|
||||
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("stat", alias=("info",), desc="Affiche les statistiques du joueur.")
|
||||
def stat(bot, msg):
|
||||
money, ammo, ammo_in_chamber, max_ammo, has_scope, grease, is_jammed = get_hunter(msg.author)
|
||||
|
||||
additionnal_infos = ""
|
||||
if has_scope:
|
||||
additionnal_infos += "[lunette installée] "
|
||||
if is_jammed:
|
||||
additionnal_infos += "[fusil enrayé] "
|
||||
|
||||
text = f"Statistiques de {msg.author} {additionnal_infos[:-1]}\n"
|
||||
text += f" │ argent : {money}\n"
|
||||
text += f" │ chargeur : {ammo_in_chamber} / {max_ammo}\n"
|
||||
text += f" │ munitions : {ammo}\n"
|
||||
text += f" │ graisse : {grease}"
|
||||
|
||||
bot.send(GAME_CHANNEL, text)
|
||||
|
||||
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("nettoyer", alias=("clean",), desc="Nettoie le fusil.")
|
||||
def clear(bot, msg):
|
||||
grease, is_jammed = get_hunter(msg.author)[5: 7]
|
||||
|
||||
if is_jammed:
|
||||
if grease:
|
||||
bot.send(GAME_CHANNEL, f"{msg.author} a nettoyé son fusil !")
|
||||
grease -= 1
|
||||
is_jammed = 0
|
||||
else:
|
||||
bot.send(GAME_CHANNEL, "Vous devez acheter de la graisse pour pouvoir nettoyer votre fusil.")
|
||||
else:
|
||||
bot.send(GAME_CHANNEL, "Votre fusil est déjà en bon état.")
|
||||
|
||||
modify_hunter(msg.author, grease=grease, is_jammed=is_jammed)
|
||||
|
||||
@commands.channel("Duck")
|
||||
@commands.command("duck", desc="Fait apparaître un canard mécanique.")
|
||||
def fake_duck(bot, msg):
|
||||
bot.troll_duck += 1
|
||||
bot.send(msg.author, "Un canard mécanique apparaîtra… ou pas !")
|
||||
|
||||
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("reset", desc="Réinitialise le profil du joueur")
|
||||
def reset(bot, msg):
|
||||
hunter = get_hunter(msg.author)[:3]
|
||||
bot.send(GAME_CHANNEL, f"Le profil de {msg.author} a été réinitialisé.")
|
||||
del_hunter(msg.author)
|
||||
|
||||
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("achat", alias=("magasin", "shop"), desc="Acheter des munitions et accessoires.\nLancer la commande sans arguments pour voir le magasin.")
|
||||
def achat(bot, msg, article: str=""):
|
||||
def get_spec(spec):
|
||||
fields = ("argent", "munition", "munition dans la chambre", "capacité chargeur", "lunette de visée", "graisse")
|
||||
text = "["
|
||||
for field_index, value in spec.items():
|
||||
text += f"{fields[field_index]} {('-', '+')[abs(value) == value]}{abs(value)}, "
|
||||
return text[:-2] + "]"
|
||||
|
||||
hunter = get_hunter(msg.author)
|
||||
db_fields = ("money", "ammo", "ammo_in_chamber", "max_ammo", "has_scope", "grease")
|
||||
|
||||
availables_articles = {
|
||||
"chargeur": ("Ajoute cinq balles à ton inventaire", "un chargeur", {0: -4, 1: 5}),
|
||||
"balle": ("Ajoute une balle à ton inventaire", "une balle", {0: -1, 1: 1}),
|
||||
"capacité": ("Augmente la capacité du chargeur d'une balle", "un chargeur haute capacité", {0: -5, 3: 1}),
|
||||
"viseur": ("Votre prochain tir touchera forcément un canard.", "une lunette de visée", {0: -3, 4: 1}),
|
||||
"graisse": ("Permet de nettoyer son arme et de limiter le risque d'enrayement.", "une boîte de graisse pour fusil", {0: -3, 5: 5})
|
||||
}
|
||||
|
||||
if article:
|
||||
article = article.lower()
|
||||
if article in availables_articles.keys():
|
||||
args = availables_articles[article]
|
||||
|
||||
# not enough money
|
||||
if hunter[0] < abs(args[2][0]):
|
||||
bot.send(GAME_CHANNEL, f"Vous n'avez pas assez d'argent pour acheter {args[1]} {get_spec(args[2])}.")
|
||||
else:
|
||||
bot.send(GAME_CHANNEL, f"Vous avez acheté {args[1]} {get_spec(args[2])}.")
|
||||
for index, value in args[2].items():
|
||||
hunter[index] += value
|
||||
modify_hunter(msg.author, **{field: hunter[i] for i, field in enumerate(db_fields)})
|
||||
else:
|
||||
msg = "Liste des articles disponibles :\n"
|
||||
msg += "".join([
|
||||
f" | {article} : {args[0]} {get_spec(args[2])}\n"
|
||||
for article, args in availables_articles.items()
|
||||
])
|
||||
bot.send(GAME_CHANNEL, msg)
|
||||
|
|
|
@ -3,39 +3,54 @@ import re
|
|||
import requests
|
||||
import time
|
||||
|
||||
from irc_api import api
|
||||
from irc_api.api import auto_help
|
||||
from irc_api import commands
|
||||
# from irc_api.commands import auto_help
|
||||
|
||||
|
||||
@api.on(lambda m: isinstance(re.match(r"(.*)(bonjour|coucou|salut|hey|hello|hi|yo) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
@commands.on(lambda m: isinstance(re.match(r"(.*)(bonjour|coucou|salut|hey|hello|hi|yo) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
def greetings_hello(bot, msg):
|
||||
"""Dit bonjour à l'utilisateur"""
|
||||
greetings = ("Bonjour", "Coucou", "Salut", "Hey", "Hello", "Hi", "Yo")
|
||||
bot.send(msg.to, f"{choice(greetings)} {msg.author}.")
|
||||
|
||||
|
||||
@api.on(lambda m: isinstance(re.match(r"(.*)(au revoir|(à|a) plus|a\+|(à|a) toute|@\+|bye|see you|see you soon) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
@commands.on(lambda m: isinstance(re.match(r"(.*)(au revoir|(à|a) plus|a\+|(à|a) toute|@\+|bye|see you|see you soon) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
def greetings_bye(bot, msg):
|
||||
"""Dit au revoir à l'utilisateur."""
|
||||
greetings = ("Au revoir,", "À plus,", "Bye bye,", "See you soon,", "Bye,")
|
||||
bot.send(msg.to, f"{choice(greetings)} {msg.author}.")
|
||||
|
||||
|
||||
@api.on(lambda m: isinstance(re.match(r"(.*)(merci|merci beaucoup|thx|thanks|thank you) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
@commands.on(lambda m: isinstance(re.match(r"(.*)(merci|merci beaucoup|thx|thanks|thank you) glados(.*)", m.text, re.IGNORECASE), re.Match))
|
||||
def thanks(bot, msg):
|
||||
"""Répond aux remerciements de l'utilisateur."""
|
||||
thanks_choice = ("Mais je vous en prie.", "Tout le plaisir est pour moi.", "C'est tout naturel.")
|
||||
bot.send(msg.to, choice(thanks_choice))
|
||||
|
||||
|
||||
@api.on(lambda m: isinstance(re.match(r"(.*)quelle heure(.*)?", m.text, re.IGNORECASE), re.Match))
|
||||
@commands.on(lambda m: isinstance(re.match(r"(.*)quelle heure(.*)?", m.text, re.IGNORECASE), re.Match))
|
||||
def hour(bot, msg):
|
||||
"""Donne l'heure actuelle"""
|
||||
now = time.strftime("%Hh%M", time.localtime())
|
||||
bot.send(msg.to, f"Il est très exactement {now}.")
|
||||
|
||||
|
||||
@api.command("wiki", desc="wiki <recherche> [limite=1]\nFait une recherche wikipedia.")
|
||||
@commands.on(lambda m: "mec" in m.text.lower() and "glados" in m.text.lower())
|
||||
def react_on_mec(bot, msg):
|
||||
bot.send(msg.to, "Chuis pas ton mec, mon pote.")
|
||||
|
||||
|
||||
@commands.on(lambda m: "pote" in m.text.lower() and "glados" in m.text.lower())
|
||||
def react_on_pote(bot, msg):
|
||||
bot.send(msg.to, "Chuis pas ton pote, mon gars.")
|
||||
|
||||
|
||||
@commands.on(lambda m: "gars" in m.text.lower() and "glados" in m.text.lower())
|
||||
def react_on_gars(bot, msg):
|
||||
bot.send(msg.to, "Chuis pas ton gars, mec.")
|
||||
|
||||
|
||||
@commands.command("wiki", desc="wiki <recherche> [limite=1]\nFait une recherche wikipedia.")
|
||||
def wiki(bot, msg, text: str, limit: int=1):
|
||||
if not (1 < limit <= 10):
|
||||
limit = 1
|
||||
|
|
322
irc_api/api.py
322
irc_api/api.py
|
@ -1,322 +0,0 @@
|
|||
import logging
|
||||
import re
|
||||
from irc_api.irc import IRC, History
|
||||
from threading import Thread
|
||||
import time
|
||||
|
||||
|
||||
PREFIX = ""
|
||||
|
||||
|
||||
def command(name, alias=(), desc=""):
|
||||
if not alias or not name in alias:
|
||||
alias += (name,)
|
||||
def decorator(func):
|
||||
return Command(
|
||||
name=name,
|
||||
func=func,
|
||||
events=[lambda m: True in [m.text.startswith(PREFIX + cmd) for cmd in alias]],
|
||||
desc=desc,
|
||||
cmnd_type=1
|
||||
)
|
||||
return decorator
|
||||
|
||||
|
||||
def on(event, desc=""):
|
||||
def decorator(func_or_cmnd):
|
||||
if isinstance(func_or_cmnd, Command):
|
||||
func_or_cmnd.events.append(event)
|
||||
return func_or_cmnd
|
||||
else:
|
||||
return Command(
|
||||
name=func_or_cmnd.__name__,
|
||||
func=func_or_cmnd,
|
||||
events=[event],
|
||||
desc=desc,
|
||||
cmnd_type=0
|
||||
)
|
||||
return decorator
|
||||
|
||||
|
||||
def channel(channel_name, desc=""):
|
||||
def decorator(func_or_cmnd):
|
||||
if isinstance(func_or_cmnd, Command):
|
||||
func_or_cmnd.events.append(lambda m: m.to == channel_name)
|
||||
return func_or_cmnd
|
||||
else:
|
||||
return Command(
|
||||
name=func_or_cmnd.__name__,
|
||||
func=func_or_cmnd,
|
||||
events=[lambda m: m.to == channel_name],
|
||||
desc=desc,
|
||||
cmnd_type=0
|
||||
)
|
||||
return decorator
|
||||
|
||||
|
||||
def user(user_name, desc=""):
|
||||
def decorator(func_or_cmnd):
|
||||
if isinstance(func_or_cmnd, Command):
|
||||
func_or_cmnd.events.append(lambda m: m.author == user_name)
|
||||
return func_or_cmnd
|
||||
else:
|
||||
return Command(
|
||||
name=func_or_cmnd.__name__,
|
||||
func=func_or_cmnd,
|
||||
events=[lambda m: m.author == user_name],
|
||||
desc=desc,
|
||||
cmnd_type=0
|
||||
)
|
||||
return decorator
|
||||
|
||||
|
||||
def every(time, desc=""):
|
||||
def decorator(func):
|
||||
return Command(
|
||||
name=func.__name__,
|
||||
func=func,
|
||||
events=time,
|
||||
desc=desc,
|
||||
cmnd_type=2
|
||||
)
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
class Command:
|
||||
def __init__(self, name, func, events, desc, cmnd_type):
|
||||
self.name = name
|
||||
self.func = func
|
||||
self.events = events
|
||||
self.cmnd_type = cmnd_type
|
||||
|
||||
if desc:
|
||||
self.desc = desc
|
||||
else:
|
||||
self.desc = "..."
|
||||
if func.__doc__:
|
||||
self.desc = func.__doc__
|
||||
|
||||
self.bot = None
|
||||
|
||||
def __call__(self, msg, *args):
|
||||
return self.func(self.bot, msg, *args)
|
||||
|
||||
|
||||
class Bot:
|
||||
"""Run the connexion between IRC's server and V5 one.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
irc : IRC, public
|
||||
IRC wrapper which handle communication with IRC server.
|
||||
v5 : V5, public
|
||||
V5 wrapper which handle communication with V5 server.
|
||||
channels : list, public
|
||||
The channels the bot will listen.
|
||||
|
||||
Methods
|
||||
-------
|
||||
start : NoneType, public
|
||||
Runs the bot and connects it to IRC and V5 servers.
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
auth: tuple,
|
||||
irc_params: tuple,
|
||||
channels: list=["#general"],
|
||||
*commands_modules,
|
||||
**kwargs
|
||||
):
|
||||
"""Initialize the Bot instance.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
irc_params : tuple
|
||||
Contains the IRC server informations (host, port)
|
||||
channels : list
|
||||
Contains the names of the channels on which the bot will connect.
|
||||
prefix : str, optionnal
|
||||
The prefix on which the bot will react.
|
||||
"""
|
||||
global PREFIX
|
||||
if kwargs.get('prefix'):
|
||||
PREFIX = kwargs.get('prefix')
|
||||
self.prefix = PREFIX
|
||||
|
||||
self.irc = IRC(*irc_params)
|
||||
self.history = History(kwargs.get('limit'))
|
||||
self.channels = channels
|
||||
self.auth = auth
|
||||
self.callbacks = {}
|
||||
self.commands_help = {}
|
||||
self.threads = []
|
||||
|
||||
if commands_modules:
|
||||
self.add_commands_modules(*commands_modules)
|
||||
|
||||
def start(self):
|
||||
"""Starts the bot and connect it to the given IRC and V5 servers."""
|
||||
# Start IRC
|
||||
self.irc.connexion(self.auth[0], self.auth[1])
|
||||
|
||||
# Join channels
|
||||
for channel in self.channels:
|
||||
self.irc.join(channel)
|
||||
|
||||
# mainloop
|
||||
while True:
|
||||
message = self.irc.receive()
|
||||
self.history.add(message)
|
||||
logging.info("received %s", message)
|
||||
if message is not None:
|
||||
for callback in self.callbacks.values():
|
||||
if not False in [event(message) for event in callback.events]:
|
||||
logging.info("matched %s", callback.name)
|
||||
|
||||
# @api.on
|
||||
if callback.cmnd_type == 0:
|
||||
callback(message)
|
||||
|
||||
# @api.command
|
||||
elif callback.cmnd_type == 1:
|
||||
args = check_args(callback.func, *parse(message.text)[1:])
|
||||
if isinstance(args, list):
|
||||
callback(message, *args)
|
||||
else:
|
||||
self.send(
|
||||
message.to,
|
||||
"Erreur : les arguments donnés ne correspondent pas."
|
||||
)
|
||||
|
||||
def send(self, target: str, message: str):
|
||||
"""Send a message to the specified target (channel or user).
|
||||
|
||||
Parameters
|
||||
----------
|
||||
target : str
|
||||
The target of the message. It can be a channel or user (private message).
|
||||
message : str
|
||||
The content of the message to send.
|
||||
"""
|
||||
for line in message.splitlines():
|
||||
self.irc.send(f"PRIVMSG {target} :{line}")
|
||||
|
||||
def add_command(self, command, add_to_help=False):
|
||||
command.bot = self
|
||||
|
||||
if command.cmnd_type == 2:
|
||||
def timed_func(bot):
|
||||
while True:
|
||||
command.func(bot)
|
||||
time.sleep(command.events)
|
||||
|
||||
self.threads.append(Thread(target=timed_func, args=(self,)))
|
||||
self.threads[-1].start()
|
||||
else:
|
||||
self.callbacks[command.name] = command
|
||||
|
||||
if add_to_help and command.cmnd_type == 1:
|
||||
self.commands_help[command.name] = command
|
||||
|
||||
def add_commands(self, *commands, **kwargs):
|
||||
"""Add a list of commands to the bot.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
commands : list
|
||||
A list of command's instances.
|
||||
"""
|
||||
add_to_help = "auto_help" in [cmnd.name for cmnd in commands]
|
||||
for command in commands:
|
||||
self.add_command(command, add_to_help=add_to_help)
|
||||
|
||||
def add_commands_modules(self, *commands_modules):
|
||||
for commands_module in commands_modules:
|
||||
add_to_help = "auto_help" in dir(commands_module)
|
||||
for cmnd_name in dir(commands_module):
|
||||
cmnd = getattr(commands_module, cmnd_name)
|
||||
if isinstance(cmnd, Command):
|
||||
self.add_command(cmnd, add_to_help=add_to_help)
|
||||
|
||||
def remove_command(self, command_name: str):
|
||||
if command_name in self.callbacks.keys():
|
||||
self.callbacks.pop(command_name)
|
||||
|
||||
|
||||
@command("aide", alias=("aide", "help", "doc", "assistance"))
|
||||
def auto_help(bot, msg, fct_name: str=""):
|
||||
"""Aide des commandes disponibles."""
|
||||
if fct_name and fct_name in bot.commands_help.keys():
|
||||
cmnd = bot.commands_help[fct_name]
|
||||
if cmnd.cmnd_type == 1:
|
||||
bot.send(msg.to, f"Aide sur la commande : {bot.prefix}{fct_name}")
|
||||
else:
|
||||
bot.send(msg.to, f"Aide sur la commande : {fct_name}")
|
||||
|
||||
for line in bot.commands_help[fct_name].desc.splitlines():
|
||||
bot.send(msg.to, f" │ {line}")
|
||||
else:
|
||||
bot.send(msg.to, f"Liste des commandes ({PREFIX}aide <cmnd> pour plus d'info)")
|
||||
for cmnd_name in bot.commands_help.keys():
|
||||
bot.send(msg.to, f" - {cmnd_name}")
|
||||
|
||||
|
||||
def parse(message):
|
||||
pattern = re.compile(r"((\"[^\"]+\"\ *)|(\'[^\']+\'\ *)|([^\ ]+\ *))", re.IGNORECASE)
|
||||
args_to_return = []
|
||||
for match in re.findall(pattern, message):
|
||||
match = match[0].strip().rstrip()
|
||||
if (match.startswith("\"") and match.endswith("\"")) \
|
||||
or (match.startswith("'") and match.endswith("'")):
|
||||
args_to_return.append(match[1: -1])
|
||||
else:
|
||||
args_to_return.append(match)
|
||||
return args_to_return
|
||||
|
||||
|
||||
def convert(data, new_type, default=None):
|
||||
try:
|
||||
return new_type(data)
|
||||
except:
|
||||
return default
|
||||
|
||||
|
||||
def check_args(func, *input_args):
|
||||
# gets the defaults values given in arguments
|
||||
defaults = getattr(func, "__defaults__")
|
||||
if not defaults:
|
||||
defaults = []
|
||||
|
||||
# gets the arguments and their types
|
||||
annotations = getattr(func, "__annotations__")
|
||||
if not annotations:
|
||||
return []
|
||||
|
||||
# nb of required arguments
|
||||
required_args = len(annotations) - len(defaults)
|
||||
|
||||
# if the number of given arguments just can't match
|
||||
if len(input_args) < required_args:
|
||||
return None
|
||||
|
||||
converted_args = []
|
||||
for index, arg_type in enumerate(annotations.values()):
|
||||
# construction of a tuple (type, default_value) for each expected argument
|
||||
if index + 1 > required_args:
|
||||
check_args = (arg_type, defaults[index - required_args])
|
||||
else:
|
||||
check_args = (arg_type, None)
|
||||
|
||||
# transtypes each given arguments to its target type
|
||||
if len(input_args) > index:
|
||||
converted_args.append(convert(input_args[index], *check_args))
|
||||
else:
|
||||
converted_args.append(check_args[1])
|
||||
|
||||
# if an argument has no default value and transtyping has failed
|
||||
if None in converted_args:
|
||||
return None
|
||||
|
||||
return converted_args
|
||||
|
229
irc_api/irc.py
229
irc_api/irc.py
|
@ -1,229 +0,0 @@
|
|||
"""
|
||||
irc (GLaDOS)
|
||||
============
|
||||
|
||||
Description
|
||||
-----------
|
||||
Manage the IRC layer of GLaDOS.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
|
||||
from functools import wraps
|
||||
from queue import Queue
|
||||
from threading import Thread
|
||||
|
||||
|
||||
class IRC:
|
||||
"""Manage connexion to an IRC server, authentication and callbacks.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
connected : bool, public
|
||||
If the bot is connected to an IRC server or not.
|
||||
callbacks : list, public
|
||||
List of the registred callbacks.
|
||||
|
||||
socket : ssl.SSLSocket, private
|
||||
The IRC's socket.
|
||||
inbox : Queue, private
|
||||
Queue of the incomming messages.
|
||||
handler : Thread, private
|
||||
|
||||
Methods
|
||||
-------
|
||||
start : NoneType, public
|
||||
Starts the IRC layer and manage authentication.
|
||||
run : NoneType, public
|
||||
Mainloop, allows to handle public messages.
|
||||
send : NoneType, public
|
||||
Sends a message to a given channel.
|
||||
receive : Message, public
|
||||
Same as ``run`` for private messages.
|
||||
join : NoneType, public
|
||||
Allows to join a given channel.
|
||||
on : function, public
|
||||
Add a callback on a given message.
|
||||
|
||||
handle : NoneType, private
|
||||
Handles the ping and store incoming messages into the inbox attribute.
|
||||
send : NoneType, private
|
||||
Send message to a target.
|
||||
recv : str, private
|
||||
Get the oldest incoming message and returns it.
|
||||
waitfor : str, private
|
||||
Wait for a raw message that matches the given condition.
|
||||
"""
|
||||
def __init__(self, host: str, port: int):
|
||||
"""Initialize an IRC wrapper.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
host : str
|
||||
The adress of the IRC server.
|
||||
port : int
|
||||
The port of the IRC server.
|
||||
"""
|
||||
|
||||
# Public attributes
|
||||
self.connected = False # Simple lock
|
||||
|
||||
# Private attributes
|
||||
self.__socket = ssl.create_default_context().wrap_socket(
|
||||
socket.create_connection((host, port)),
|
||||
server_hostname=host
|
||||
)
|
||||
self.__inbox = Queue()
|
||||
self.__handler = Thread(target=self.__handle)
|
||||
|
||||
# Public methods
|
||||
def connexion(self, nick: str, password: str):
|
||||
"""Start the IRC layer. Manage authentication as well.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
nick : str
|
||||
The username for login and nickname once connected.
|
||||
password : str
|
||||
The password for authentification.
|
||||
"""
|
||||
self.__handler.start()
|
||||
|
||||
self.send(f"USER {nick} * * :{nick}")
|
||||
self.send(f"NICK {nick}")
|
||||
self.waitfor(lambda m: "NOTICE" in m and "/AUTH" in m)
|
||||
self.send(f"AUTH {nick}:{password}")
|
||||
self.waitfor(lambda m: "You are now logged in" in m)
|
||||
|
||||
self.connected = True
|
||||
|
||||
def receive(self):
|
||||
"""Receive a private message.
|
||||
|
||||
Returns
|
||||
-------
|
||||
msg : Message
|
||||
The incoming processed private message.
|
||||
"""
|
||||
while True:
|
||||
message = self.__inbox.get()
|
||||
if " PRIVMSG " in message:
|
||||
msg = Message(message)
|
||||
if msg:
|
||||
return msg
|
||||
|
||||
def join(self, channel: str):
|
||||
"""Join a channel.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
channel : str
|
||||
The name of the channel to join.
|
||||
"""
|
||||
self.send(f"JOIN {channel}")
|
||||
logging.info("joined %s", channel)
|
||||
|
||||
def send(self, raw: str):
|
||||
"""Wrap and encode raw message to send.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
raw : str
|
||||
The raw message to send.
|
||||
"""
|
||||
self.__socket.send(f"{raw}\r\n".encode())
|
||||
|
||||
def waitfor(self, condition):
|
||||
"""Wait for a raw message that matches the condition.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
condition : function
|
||||
``condition`` is a function that must taking a raw message in parameter and returns a
|
||||
boolean.
|
||||
|
||||
Returns
|
||||
-------
|
||||
msg : str
|
||||
The last message received that doesn't match the condition.
|
||||
"""
|
||||
msg = self.__inbox.get()
|
||||
while not condition(msg):
|
||||
msg = self.__inbox.get()
|
||||
return msg
|
||||
|
||||
# Private methods
|
||||
def __handle(self):
|
||||
"""Handle raw messages from irc and manage ping."""
|
||||
while True:
|
||||
# Get incoming messages
|
||||
data = self.__socket.recv(4096).decode()
|
||||
|
||||
# Split multiple lines
|
||||
for msg in data.split('\r\n'):
|
||||
# Manage ping
|
||||
if msg.startswith("PING"):
|
||||
self.send(msg.replace("PING", "PONG"))
|
||||
|
||||
# Or add a new message to inbox
|
||||
elif len(msg):
|
||||
self.__inbox.put(msg)
|
||||
logging.debug("received %s", msg)
|
||||
|
||||
|
||||
class History:
|
||||
def __init__(self, limit: int):
|
||||
self.__content = []
|
||||
if limit:
|
||||
self.__limit = limit
|
||||
else:
|
||||
self.__limit = 100
|
||||
|
||||
def __len__(self):
|
||||
return len(self.__content)
|
||||
|
||||
def add(self, elmnt):
|
||||
if len(self.__content) == self.__limit:
|
||||
self.__content.pop(0)
|
||||
self.__content.append(elmnt)
|
||||
|
||||
def get(self):
|
||||
return self.__content
|
||||
|
||||
|
||||
class Message:
|
||||
"""Parse the raw message in three fields : author, the channel, and text.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
pattern : re.Pattern, public
|
||||
The message parsing pattern.
|
||||
author : str, public
|
||||
The message's author.
|
||||
to : str, public
|
||||
The message's origin (channel or DM).
|
||||
text : str, public
|
||||
The message's content.
|
||||
"""
|
||||
pattern = re.compile(
|
||||
r"^:(?P<author>[\w.~|\-\[\]]+)(?:!(?P<host>\S+))? PRIVMSG (?P<to>\S+) :(?P<text>.+)"
|
||||
)
|
||||
|
||||
def __init__(self, raw: str):
|
||||
match = re.search(Message.pattern, raw)
|
||||
if match:
|
||||
self.author = match.group("author")
|
||||
self.to = match.group("to")
|
||||
self.text = match.group("text")
|
||||
logging.debug("sucessfully parsed %s into %s", raw, self.__str__())
|
||||
else:
|
||||
self.author = ""
|
||||
self.to = ""
|
||||
self.text = ""
|
||||
logging.warning("failed to parse %s into valid message", raw)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.author} to {self.to}: {self.text}"
|
24
main.py
24
main.py
|
@ -10,7 +10,7 @@ Create a bot's instance and manages it.
|
|||
|
||||
import logging
|
||||
import re
|
||||
from irc_api import api
|
||||
from irc_api.bot import Bot
|
||||
|
||||
|
||||
import glados_cmnds as gv4_cmnds
|
||||
|
@ -18,20 +18,26 @@ import fun_cmnds as fcmnds
|
|||
from secrets import USER, PASSWORD
|
||||
|
||||
|
||||
LOG_FORMAT = "%(asctime)s [%(levelname)s] <%(filename)s> %(funcName)s: %(message)s"
|
||||
logging.basicConfig(format=LOG_FORMAT, level=logging.DEBUG)
|
||||
LOG_FORMAT = "[%(levelname)s] %(message)s"
|
||||
logging.basicConfig(format=LOG_FORMAT, level=logging.INFO)
|
||||
|
||||
class MyBot(api.Bot):
|
||||
nb_ducks = 0
|
||||
class MyBot(Bot):
|
||||
# ducks_history
|
||||
# 0 : normal duck
|
||||
# 1 : super duck
|
||||
# 2 : mecanics duck
|
||||
ducks_history = []
|
||||
# nb_ducks = 0
|
||||
troll_duck = 0
|
||||
|
||||
|
||||
glados = MyBot(
|
||||
(USER, PASSWORD),
|
||||
('irc.planet-casio.com', 6697),
|
||||
["#glados"],
|
||||
gv4_cmnds, fcmnds,
|
||||
prefix="!",
|
||||
auth=(USER, PASSWORD),
|
||||
channels=["#glados"],
|
||||
prefix="!"
|
||||
)
|
||||
|
||||
|
||||
glados.start()
|
||||
glados.start("Duck")
|
||||
|
|
Loading…
Reference in New Issue