mirror of
https://github.com/iio612/DEFENDER.git
synced 2026-02-13 11:14:23 +00:00
496 lines
24 KiB
Python
496 lines
24 KiB
Python
from core.irc import Irc
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
|
|
# Activer le systeme sur un salon (activate #salon)
|
|
# Le service devra se connecter au salon
|
|
# Le service devra se mettre en op
|
|
# Soumettre un nom de user (submit nickname)
|
|
# voter pour un ban (vote_for)
|
|
# voter contre un ban (vote_against)
|
|
|
|
|
|
|
|
class Votekick():
|
|
|
|
@dataclass
|
|
class VoteChannelModel:
|
|
channel_name: str
|
|
target_user: str
|
|
voter_users: list
|
|
vote_for: int
|
|
vote_against: int
|
|
|
|
VOTE_CHANNEL_DB:list[VoteChannelModel] = []
|
|
|
|
def __init__(self, ircInstance:Irc) -> None:
|
|
|
|
# Module name (Mandatory)
|
|
self.module_name = 'mod_' + str(self.__class__.__name__).lower()
|
|
|
|
# Add Irc Object to the module
|
|
self.Irc = ircInstance
|
|
|
|
# Add Global Configuration to the module
|
|
self.Config = ircInstance.Config
|
|
|
|
# Add Base object to the module
|
|
self.Base = ircInstance.Base
|
|
|
|
# Add logs object to the module
|
|
self.Logs = ircInstance.Base.logs
|
|
|
|
# Add User object to the module
|
|
self.User = ircInstance.User
|
|
|
|
# Add Channel object to the module
|
|
self.Channel = ircInstance.Channel
|
|
|
|
# Créer les nouvelles commandes du module
|
|
self.commands_level = {
|
|
0: ['vote']
|
|
}
|
|
|
|
# Init the module
|
|
self.__init_module()
|
|
|
|
# Log the module
|
|
self.Logs.debug(f'Module {self.module_name} loaded ...')
|
|
|
|
def __init_module(self) -> None:
|
|
|
|
# Add admin object to retrieve admin users
|
|
self.Admin = self.Irc.Admin
|
|
|
|
self.__set_commands(self.commands_level)
|
|
self.__create_tables()
|
|
self.join_saved_channels()
|
|
|
|
return None
|
|
|
|
def __set_commands(self, commands:dict[int, list[str]]) -> None:
|
|
"""### Rajoute les commandes du module au programme principal
|
|
|
|
Args:
|
|
commands (list): Liste des commandes du module
|
|
"""
|
|
for level, com in commands.items():
|
|
for c in commands[level]:
|
|
if not c in self.Irc.commands:
|
|
self.Irc.commands_level[level].append(c)
|
|
self.Irc.commands.append(c)
|
|
|
|
return None
|
|
|
|
def __create_tables(self) -> None:
|
|
"""Methode qui va créer la base de donnée si elle n'existe pas.
|
|
Une Session unique pour cette classe sera crée, qui sera utilisé dans cette classe / module
|
|
Args:
|
|
database_name (str): Nom de la base de données ( pas d'espace dans le nom )
|
|
|
|
Returns:
|
|
None: Aucun retour n'es attendu
|
|
"""
|
|
|
|
table_logs = '''CREATE TABLE IF NOT EXISTS votekick_logs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
datetime TEXT,
|
|
server_msg TEXT
|
|
)
|
|
'''
|
|
|
|
table_vote = '''CREATE TABLE IF NOT EXISTS votekick_channel (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
datetime TEXT,
|
|
channel TEXT
|
|
)
|
|
'''
|
|
|
|
self.Base.db_execute_query(table_logs)
|
|
self.Base.db_execute_query(table_vote)
|
|
return None
|
|
|
|
def unload(self) -> None:
|
|
try:
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} PART {chan.channel_name}")
|
|
|
|
self.VOTE_CHANNEL_DB = []
|
|
self.Logs.debug(f'Delete memory DB VOTE_CHANNEL_DB: {self.VOTE_CHANNEL_DB}')
|
|
|
|
return None
|
|
except UnboundLocalError as ne:
|
|
self.Logs.error(f'{ne}')
|
|
except NameError as ue:
|
|
self.Logs.error(f'{ue}')
|
|
except:
|
|
self.Logs.error('Error on the module')
|
|
|
|
def init_vote_system(self, channel: str) -> bool:
|
|
|
|
response = False
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
chan.target_user = ''
|
|
chan.voter_users = []
|
|
chan.vote_against = 0
|
|
chan.vote_for = 0
|
|
response = True
|
|
|
|
return response
|
|
|
|
def insert_vote_channel(self, ChannelObject: VoteChannelModel) -> bool:
|
|
result = False
|
|
found = False
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == ChannelObject.channel_name:
|
|
found = True
|
|
|
|
if not found:
|
|
self.VOTE_CHANNEL_DB.append(ChannelObject)
|
|
self.Logs.debug(f"The channel has been added {ChannelObject}")
|
|
# self.db_add_vote_channel(ChannelObject.channel_name)
|
|
|
|
return result
|
|
|
|
def db_add_vote_channel(self, channel:str) -> bool:
|
|
"""Cette fonction ajoute les salons ou seront autoriser les votes
|
|
|
|
Args:
|
|
channel (str): le salon à enregistrer.
|
|
"""
|
|
current_datetime = self.Base.get_datetime()
|
|
mes_donnees = {'channel': channel}
|
|
|
|
response = self.Base.db_execute_query("SELECT id FROM votekick_channel WHERE channel = :channel", mes_donnees)
|
|
|
|
isChannelExist = response.fetchone()
|
|
|
|
if isChannelExist is None:
|
|
mes_donnees = {'datetime': current_datetime, 'channel': channel}
|
|
insert = self.Base.db_execute_query(f"INSERT INTO votekick_channel (datetime, channel) VALUES (:datetime, :channel)", mes_donnees)
|
|
if insert.rowcount > 0:
|
|
return True
|
|
else:
|
|
return False
|
|
else:
|
|
return False
|
|
|
|
def db_delete_vote_channel(self, channel: str) -> bool:
|
|
"""Cette fonction supprime les salons de join de Defender
|
|
|
|
Args:
|
|
channel (str): le salon à enregistrer.
|
|
"""
|
|
mes_donnes = {'channel': channel}
|
|
response = self.Base.db_execute_query("DELETE FROM votekick_channel WHERE channel = :channel", mes_donnes)
|
|
|
|
affected_row = response.rowcount
|
|
|
|
if affected_row > 0:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def join_saved_channels(self) -> None:
|
|
|
|
param = {'module_name': self.module_name}
|
|
result = self.Base.db_execute_query(f"SELECT id, channel_name FROM {self.Config.table_channel} WHERE module_name = :module_name", param)
|
|
|
|
channels = result.fetchall()
|
|
unixtime = self.Base.get_unixtime()
|
|
|
|
for channel in channels:
|
|
id, chan = channel
|
|
self.insert_vote_channel(self.VoteChannelModel(channel_name=chan, target_user='', voter_users=[], vote_for=0, vote_against=0))
|
|
self.Irc.send2socket(f":{self.Config.SERVEUR_ID} SJOIN {unixtime} {chan} + :{self.Config.SERVICE_ID}")
|
|
self.Irc.send2socket(f":{self.Config.SERVICE_NICKNAME} SAMODE {chan} +o {self.Config.SERVICE_NICKNAME}")
|
|
|
|
return None
|
|
|
|
def is_vote_ongoing(self, channel: str) -> bool:
|
|
|
|
response = False
|
|
for vote in self.VOTE_CHANNEL_DB:
|
|
if vote.channel_name == channel:
|
|
if vote.target_user:
|
|
response = True
|
|
|
|
return response
|
|
|
|
def timer_vote_verdict(self, channel: str) -> None:
|
|
|
|
dnickname = self.Config.SERVICE_NICKNAME
|
|
|
|
if not self.is_vote_ongoing(channel):
|
|
return None
|
|
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
target_user = self.User.get_nickname(chan.target_user)
|
|
if chan.vote_for > chan.vote_against:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {chan.vote_against} votes against and {chan.vote_for} votes for. For this reason, it\'ll be kicked from the channel')
|
|
self.Irc.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
|
|
self.Channel.delete_user_from_channel(channel, self.User.get_uid(target_user))
|
|
elif chan.vote_for <= chan.vote_against:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {chan.vote_against} votes against and {chan.vote_for} votes for. For this reason, it\'ll remain in the channel')
|
|
|
|
# Init the system
|
|
if self.init_vote_system(channel):
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :System vote re initiated')
|
|
|
|
return None
|
|
|
|
def cmd(self, data:list) -> None:
|
|
try:
|
|
cmd = list(data).copy()
|
|
return None
|
|
|
|
except KeyError as ke:
|
|
self.Base.logs.error(f"Key Error: {ke}")
|
|
except IndexError as ie:
|
|
self.Base.logs.error(f"{ie} / {cmd} / length {str(len(cmd))}")
|
|
except Exception as err:
|
|
self.Base.logs.error(f"General Error: {err}")
|
|
|
|
def _hcmds(self, user:str, channel: any, cmd: list, fullcmd: list = []) -> None:
|
|
# cmd is the command starting from the user command
|
|
# full cmd is sending the entire server response
|
|
|
|
command = str(cmd[0]).lower()
|
|
dnickname = self.Config.SERVICE_NICKNAME
|
|
fromuser = user
|
|
fromchannel = channel
|
|
|
|
match command:
|
|
case 'vote':
|
|
option = str(cmd[1]).lower()
|
|
|
|
if len(command) == 1:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote activate #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote deactivate #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote +')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote -')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote cancel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote status')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote submit nickname')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote verdict')
|
|
|
|
match option:
|
|
|
|
case 'activate':
|
|
try:
|
|
# vote activate #channel
|
|
if self.Admin.get_Admin(fromuser) is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Your are not allowed to execute this command')
|
|
return None
|
|
|
|
sentchannel = str(cmd[2]).lower() if self.Base.Is_Channel(str(cmd[2]).lower()) else None
|
|
if sentchannel is None:
|
|
self.Irc.send2socket(f":{dnickname} NOTICE {fromuser} :The correct command is {self.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
|
|
|
|
self.insert_vote_channel(
|
|
self.VoteChannelModel(
|
|
channel_name=sentchannel,
|
|
target_user='',
|
|
voter_users=[],
|
|
vote_for=0,
|
|
vote_against=0
|
|
)
|
|
)
|
|
|
|
self.Base.db_query_channel('add', self.module_name, sentchannel)
|
|
|
|
self.Irc.send2socket(f":{dnickname} JOIN {sentchannel}")
|
|
self.Irc.send2socket(f":{dnickname} SAMODE {sentchannel} +o {dnickname}")
|
|
self.Irc.send2socket(f":{dnickname} PRIVMSG {sentchannel} :You can now use !submit <nickname> to decide if he will stay or not on this channel ")
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option} #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option} #welcome')
|
|
|
|
case 'deactivate':
|
|
try:
|
|
# vote deactivate #channel
|
|
if self.Admin.get_Admin(fromuser) is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Your are not allowed to execute this command')
|
|
return None
|
|
|
|
sentchannel = str(cmd[2]).lower() if self.Base.Is_Channel(str(cmd[2]).lower()) else None
|
|
if sentchannel is None:
|
|
self.Irc.send2socket(f":{dnickname} NOTICE {fromuser} :The correct command is {self.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
|
|
|
|
self.Irc.send2socket(f":{dnickname} SAMODE {sentchannel} -o {dnickname}")
|
|
self.Irc.send2socket(f":{dnickname} PART {sentchannel}")
|
|
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == sentchannel:
|
|
self.VOTE_CHANNEL_DB.remove(chan)
|
|
self.Base.db_query_channel('del', self.module_name, chan.channel_name)
|
|
|
|
self.Logs.debug(f"The Channel {sentchannel} has been deactivated from the vote system")
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option} #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option} #welcome')
|
|
|
|
case '+':
|
|
try:
|
|
# vote +
|
|
channel = fromchannel
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
if fromuser in chan.voter_users:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :You already submitted a vote')
|
|
else:
|
|
chan.vote_for += 1
|
|
chan.voter_users.append(fromuser)
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :Vote recorded, thank you')
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option}')
|
|
|
|
case '-':
|
|
try:
|
|
# vote -
|
|
channel = fromchannel
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
if fromuser in chan.voter_users:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :You already submitted a vote')
|
|
else:
|
|
chan.vote_against += 1
|
|
chan.voter_users.append(fromuser)
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :Vote recorded, thank you')
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option}')
|
|
|
|
case 'cancel':
|
|
try:
|
|
# vote cancel
|
|
if self.Admin.get_Admin(fromuser) is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Your are not allowed to execute this command')
|
|
return None
|
|
|
|
if channel is None:
|
|
self.Logs.error(f"The channel is not known, defender can't cancel the vote")
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :You need to specify the channel => /msg {dnickname} vote_cancel #channel')
|
|
|
|
for vote in self.VOTE_CHANNEL_DB:
|
|
if vote.channel_name == channel:
|
|
self.init_vote_system(channel)
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :Vote system re-initiated')
|
|
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option}')
|
|
|
|
case 'status':
|
|
try:
|
|
# vote status
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :Channel: {chan.channel_name} | Target: {self.User.get_nickname(chan.target_user)} | For: {chan.vote_for} | Against: {chan.vote_against} | Number of voters: {str(len(chan.voter_users))}')
|
|
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option}')
|
|
|
|
case 'submit':
|
|
try:
|
|
# vote submit nickname
|
|
if self.Admin.get_Admin(fromuser) is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Your are not allowed to execute this command')
|
|
return None
|
|
|
|
nickname_submitted = cmd[2]
|
|
uid_submitted = self.User.get_uid(nickname_submitted)
|
|
user_submitted = self.User.get_User(nickname_submitted)
|
|
|
|
# check if there is an ongoing vote
|
|
if self.is_vote_ongoing(channel):
|
|
for vote in self.VOTE_CHANNEL_DB:
|
|
if vote.channel_name == channel:
|
|
ongoing_user = self.User.get_nickname(vote.target_user)
|
|
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :There is an ongoing vote on {ongoing_user}')
|
|
return False
|
|
|
|
# check if the user exist
|
|
if user_submitted is None:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :This nickname <{nickname_submitted}> do not exist')
|
|
return False
|
|
|
|
uid_cleaned = self.Base.clean_uid(uid_submitted)
|
|
ChannelInfo = self.Channel.get_Channel(channel)
|
|
if ChannelInfo is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :This channel [{channel}] do not exist in the Channel Object')
|
|
return False
|
|
|
|
clean_uids_in_channel: list = []
|
|
for uid in ChannelInfo.uids:
|
|
clean_uids_in_channel.append(self.Base.clean_uid(uid))
|
|
|
|
if not uid_cleaned in clean_uids_in_channel:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :This nickname <{nickname_submitted}> is not available in this channel')
|
|
return False
|
|
|
|
# check if Ircop or Service or Bot
|
|
pattern = fr'[o|B|S]'
|
|
operator_user = re.findall(pattern, user_submitted.umodes)
|
|
if operator_user:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :You cant vote for this user ! he/she is protected')
|
|
return False
|
|
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
chan.target_user = self.User.get_uid(nickname_submitted)
|
|
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :{nickname_submitted} has been targeted for a vote')
|
|
|
|
self.Base.create_timer(60, self.timer_vote_verdict, (channel, ))
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :This vote will end after 60 secondes')
|
|
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option} nickname')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option} adator')
|
|
|
|
case 'verdict':
|
|
try:
|
|
# vote verdict
|
|
if self.Admin.get_Admin(fromuser) is None:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Your are not allowed to execute this command')
|
|
return None
|
|
|
|
for chan in self.VOTE_CHANNEL_DB:
|
|
if chan.channel_name == channel:
|
|
target_user = self.User.get_nickname(chan.target_user)
|
|
if chan.vote_for > chan.vote_against:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {chan.vote_against} votes against and {chan.vote_for} votes for. For this reason, it\'ll be kicked from the channel')
|
|
self.Irc.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
|
|
elif chan.vote_for <= chan.vote_against:
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {chan.vote_against} votes against and {chan.vote_for} votes for. For this reason, it\'ll remain in the channel')
|
|
|
|
# Init the system
|
|
if self.init_vote_system(channel):
|
|
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :System vote re initiated')
|
|
except Exception as err:
|
|
self.Logs.error(f'{err}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} {command} {option}')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :Exemple /msg {dnickname} {command} {option}')
|
|
|
|
case _:
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote activate #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote deactivate #channel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote +')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote -')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote cancel')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote status')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote submit nickname')
|
|
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote verdict') |