Cleaning irc_api folder and change for the Pypi package
This commit is contained in:
parent
30073ceafe
commit
5a2def921f
BIN
duck_hunting.db
BIN
duck_hunting.db
Binary file not shown.
36
fun_cmnds.py
36
fun_cmnds.py
|
|
@ -1,5 +1,5 @@
|
|||
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
|
||||
|
|
@ -66,7 +66,7 @@ def modify_hunter(name: str, **kwargs):
|
|||
|
||||
|
||||
|
||||
@api.every(30, desc="Fait apparaître un canard de manière aléatoire.")
|
||||
@commands.every(30, desc="Fait apparaître un canard de manière aléatoire.")
|
||||
def pop_duck(bot):
|
||||
if randint(1, 100) <= 20:
|
||||
if bot.troll_duck > 0 and randint(1, 4) <= 3:
|
||||
|
|
@ -87,8 +87,8 @@ def pop_duck(bot):
|
|||
bot.ducks_history.pop(0)
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("pan", desc="Tirer sur un canard.")
|
||||
@commands.channel(GAME_CHANNEL)
|
||||
@commands.command("pan", desc="Tirer sur un canard.")
|
||||
def fire(bot, msg):
|
||||
money, ammo, ammo_in_chamber, _, has_scope, grease, is_jammed = get_hunter(msg.author)
|
||||
|
||||
|
|
@ -121,7 +121,7 @@ def fire(bot, msg):
|
|||
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))
|
||||
if duck_type == 0:
|
||||
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))
|
||||
|
||||
|
|
@ -137,8 +137,8 @@ def fire(bot, msg):
|
|||
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)
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("recharge", alias=("reload",), desc="Fait monter une balle du chargeur dans la chambre")
|
||||
@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
|
||||
|
|
@ -156,8 +156,8 @@ def reload(bot, msg):
|
|||
bot.send(GAME_CHANNEL, f"Plus de munitions ! Va voir le magasin pour en acheter. 😉")
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("stat", alias=("info",), desc="Affiche les statistiques du joueur.")
|
||||
@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)
|
||||
|
||||
|
|
@ -176,8 +176,8 @@ def stat(bot, msg):
|
|||
bot.send(GAME_CHANNEL, text)
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("nettoyer", alias=("clean",), desc="Nettoie le fusil.")
|
||||
@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]
|
||||
|
||||
|
|
@ -193,23 +193,23 @@ def clear(bot, msg):
|
|||
|
||||
modify_hunter(msg.author, grease=grease, is_jammed=is_jammed)
|
||||
|
||||
@api.channel("Duck")
|
||||
@api.command("duck", desc="Fait apparaître un canard mécanique.")
|
||||
@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 !")
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("reset", desc="Réinitialise le profil du joueur")
|
||||
@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)
|
||||
|
||||
|
||||
@api.channel(GAME_CHANNEL)
|
||||
@api.command("achat", alias=("magasin", "shop"), desc="Acheter des munitions et accessoires.\nLancer la commande sans arguments pour voir le magasin.")
|
||||
@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")
|
||||
|
|
|
|||
|
|
@ -3,54 +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.on(lambda m: "mec" in m.text.lower() and "glados" in m.text.lower())
|
||||
@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.")
|
||||
|
||||
|
||||
@api.on(lambda m: "pote" in m.text.lower() and "glados" in m.text.lower())
|
||||
@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.")
|
||||
|
||||
|
||||
@api.on(lambda m: "gars" in m.text.lower() and "glados" in m.text.lower())
|
||||
@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.")
|
||||
|
||||
|
||||
@api.command("wiki", desc="wiki <recherche> [limite=1]\nFait une recherche wikipedia.")
|
||||
@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
|
||||
|
|
|
|||
326
irc_api/api.py
326
irc_api/api.py
|
|
@ -1,326 +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 == PREFIX + cmd or 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 WrongArg:
|
||||
"""If the transtyping has failed and the argument has no default value."""
|
||||
|
||||
|
||||
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, nick: str):
|
||||
"""Starts the bot and connect it to the given IRC and V5 servers."""
|
||||
# Start IRC
|
||||
self.irc.connexion(self.auth[0], self.auth[1], nick)
|
||||
|
||||
# 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)
|
||||
logging.info("auto call : %s", command.name)
|
||||
|
||||
self.threads.append(Thread(target=lambda bot: timed_func(bot), 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]
|
||||
answer = f"Aide sur la commande : {bot.prefix}{fct_name}\n"
|
||||
for line in bot.commands_help[fct_name].desc.splitlines():
|
||||
answer += f" │ {line}\n"
|
||||
else:
|
||||
answer = f"Liste des commandes ({PREFIX}aide <cmnd> pour plus d'info)\n"
|
||||
for cmnd_name in bot.commands_help.keys():
|
||||
answer += f" - {cmnd_name}\n"
|
||||
|
||||
bot.send(msg.to, answer)
|
||||
|
||||
|
||||
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
|
||||
|
||||
wrong_arg = WrongArg()
|
||||
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, wrong_arg)
|
||||
|
||||
# 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 wrong_arg 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, username: str, password: str, nick: 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 {username}:{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}"
|
||||
14
main.py
14
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,10 +18,10 @@ 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):
|
||||
class MyBot(Bot):
|
||||
# ducks_history
|
||||
# 0 : normal duck
|
||||
# 1 : super duck
|
||||
|
|
@ -32,11 +32,11 @@ class MyBot(api.Bot):
|
|||
|
||||
|
||||
glados = MyBot(
|
||||
(USER, PASSWORD),
|
||||
('irc.planet-casio.com', 6697),
|
||||
["#glados"],
|
||||
gv4_cmnds, fcmnds,
|
||||
prefix="!",
|
||||
auth=(USER, PASSWORD),
|
||||
channels=["#glados"],
|
||||
prefix="!"
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue