iio/DEFENDER
1
0
Fork 0
mirror of https://github.com/iio612/DEFENDER.git synced 2026-02-13 19:24:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Moving votekick to use asyncio, fix when shutdown defender

Browse Source
This commit is contained in:
adator
2025-11-20 16:17:27 +01:00
parent fe4b68e115
commit 6af1377823
7 changed files with 165 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
mods/votekick/mod_votekick.py
View File

@@ -15,7 +15,11 @@ import mods.votekick.schemas as schemas
import mods.votekick.utils as utils
from mods.votekick.votekick_manager import VotekickManager
import mods.votekick.threads as thds
from typing import Any, Optional
from typing import TYPE_CHECKING, Any, Optional
if TYPE_CHECKING:
from core.loader import Loader
class Votekick(IModule):
@@ -31,6 +35,14 @@ class Votekick(IModule):
'core_version':'Defender-6'
}
def __init__(self, context: 'Loader') -> None:
super().__init__(context)
self._mod_config: Optional[schemas.VoteChannelModel] = None
@property
def mod_config(self) -> ModConfModel:
return self._mod_config
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
@@ -53,57 +65,52 @@ class Votekick(IModule):
)
'''
self.Base.db_execute_query(table_logs)
self.Base.db_execute_query(table_vote)
self.ctx.Base.db_execute_query(table_logs)
self.ctx.Base.db_execute_query(table_vote)
return None
def load(self) -> None:
async def load(self) -> None:
self.ModConfig = self.ModConfModel()
self._mod_config = self.ModConfModel()
await self.sync_db()
# Add VoteKick Manager
self.VoteKickManager = VotekickManager(self)
# Add Utils module
self.ModUtils = utils
# Add Schemas module
self.Schemas = schemas
# Add Threads module
self.Threads = thds
self.ModUtils.join_saved_channels(self)
await utils.join_saved_channels(self)
metadata = self.Settings.get_cache('VOTEKICK')
metadata = self.ctx.Settings.get_cache('VOTEKICK')
if metadata is not None:
self.VoteKickManager.VOTE_CHANNEL_DB = metadata
# Créer les nouvelles commandes du module
self.Irc.build_command(1, self.module_name, 'vote', 'The kick vote module')
self.ctx.Irc.build_command(1, self.module_name, 'vote', 'The kick vote module')
def unload(self) -> None:
async def unload(self) -> None:
try:
# Cache the local DB with current votes.
if self.VoteKickManager.VOTE_CHANNEL_DB:
self.Settings.set_cache('VOTEKICK', self.VoteKickManager.VOTE_CHANNEL_DB)
self.ctx.Settings.set_cache('VOTEKICK', self.VoteKickManager.VOTE_CHANNEL_DB)
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
self.Protocol.send_part_chan(uidornickname=self.Config.SERVICE_ID, channel=chan.channel_name)
await self.ctx.Irc.Protocol.send_part_chan(uidornickname=self.ctx.Config.SERVICE_ID, channel=chan.channel_name)
self.VoteKickManager.VOTE_CHANNEL_DB = []
self.Logs.debug(f'Delete memory DB VOTE_CHANNEL_DB: {self.VoteKickManager.VOTE_CHANNEL_DB}')
self.ctx.Logs.debug(f'Delete memory DB VOTE_CHANNEL_DB: {self.VoteKickManager.VOTE_CHANNEL_DB}')
self.Irc.Commands.drop_command_by_module(self.module_name)
self.ctx.Commands.drop_command_by_module(self.module_name)
return None
except UnboundLocalError as ne:
self.Logs.error(f'{ne}')
self.ctx.Logs.error(f'{ne}')
except NameError as ue:
self.Logs.error(f'{ue}')
self.ctx.Logs.error(f'{ue}')
except Exception as err:
self.Logs.error(f'General Error: {err}')
self.ctx.Logs.error(f'General Error: {err}')
def cmd(self, data: list) -> None:
@@ -111,7 +118,7 @@ class Votekick(IModule):
return None
cmd = data.copy() if isinstance(data, list) else list(data).copy()
index, command = self.Irc.Protocol.get_ircd_protocol_poisition(cmd)
index, command = self.ctx.Irc.Protocol.get_ircd_protocol_poisition(cmd)
if index == -1:
return None
@@ -129,19 +136,19 @@ class Votekick(IModule):
return None
except KeyError as ke:
self.Logs.error(f"Key Error: {ke}")
self.ctx.Logs.error(f"Key Error: {ke}")
except IndexError as ie:
self.Logs.error(f"{ie} / {cmd} / length {str(len(cmd))}")
self.ctx.Logs.error(f"{ie} / {cmd} / length {str(len(cmd))}")
except Exception as err:
self.Logs.error(f"General Error: {err}")
self.ctx.Logs.error(f"General Error: {err}")
def hcmds(self, user:str, channel: Any, cmd: list, fullcmd: Optional[list] = None) -> None:
async def hcmds(self, user:str, channel: Any, cmd: list, fullcmd: Optional[list] = None) -> None:
# cmd is the command starting from the user command
# full cmd is sending the entire server response
command = str(cmd[0]).lower()
fullcmd = fullcmd
dnickname = self.Config.SERVICE_NICKNAME
dnickname = self.ctx.Config.SERVICE_NICKNAME
fromuser = user
fromchannel = channel
@@ -150,14 +157,14 @@ class Votekick(IModule):
case 'vote':
if len(cmd) == 1:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote activate #channel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote deactivate #channel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote +')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote -')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote cancel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote status')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote submit nickname')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote verdict')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote activate #channel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote deactivate #channel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote +')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote -')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote cancel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote status')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote submit nickname')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote verdict')
return None
option = str(cmd[1]).lower()
@@ -167,19 +174,19 @@ class Votekick(IModule):
case 'activate':
try:
# vote activate #channel
if self.Admin.get_admin(fromuser) is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' :Your are not allowed to execute this command')
if self.ctx.Admin.get_admin(fromuser) is None:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' :Your are not allowed to execute this command')
return None
sentchannel = str(cmd[2]).lower() if self.Channel.is_valid_channel(str(cmd[2]).lower()) else None
sentchannel = str(cmd[2]).lower() if self.ctx.Channel.is_valid_channel(str(cmd[2]).lower()) else None
if sentchannel is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" The correct command is {self.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" The correct command is {self.ctx.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
if self.VoteKickManager.activate_new_channel(sentchannel):
self.Channel.db_query_channel('add', self.module_name, sentchannel)
self.Protocol.send_join_chan(uidornickname=dnickname, channel=sentchannel)
self.Protocol.send2socket(f":{dnickname} SAMODE {sentchannel} +o {dnickname}")
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Channel.db_query_channel('add', self.module_name, sentchannel)
await self.ctx.Irc.Protocol.send_join_chan(uidornickname=dnickname, channel=sentchannel)
await self.ctx.Irc.Protocol.send2socket(f":{dnickname} SAMODE {sentchannel} +o {dnickname}")
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg="You can now use !submit <nickname> to decide if he will stay or not on this channel ",
channel=sentchannel
)
@@ -187,117 +194,117 @@ class Votekick(IModule):
return None
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option} #channel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option} #welcome')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option} #channel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option} #welcome')
case 'deactivate':
try:
# vote deactivate #channel
if self.Admin.get_admin(fromuser) is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" Your are not allowed to execute this command")
if self.ctx.Admin.get_admin(fromuser) is None:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" Your are not allowed to execute this command")
return None
sentchannel = str(cmd[2]).lower() if self.Channel.is_valid_channel(str(cmd[2]).lower()) else None
sentchannel = str(cmd[2]).lower() if self.ctx.Channel.is_valid_channel(str(cmd[2]).lower()) else None
if sentchannel is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" The correct command is {self.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" The correct command is {self.ctx.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
self.Protocol.send2socket(f":{dnickname} SAMODE {sentchannel} -o {dnickname}")
self.Protocol.send_part_chan(uidornickname=dnickname, channel=sentchannel)
await self.ctx.Irc.Protocol.send2socket(f":{dnickname} SAMODE {sentchannel} -o {dnickname}")
await self.ctx.Irc.Protocol.send_part_chan(uidornickname=dnickname, channel=sentchannel)
if self.VoteKickManager.drop_vote_channel_model(sentchannel):
self.Channel.db_query_channel('del', self.module_name, sentchannel)
await self.ctx.Channel.db_query_channel('del', self.module_name, sentchannel)
return None
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" /msg {dnickname} {command} {option} #channel")
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" Exemple /msg {dnickname} {command} {option} #welcome")
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" /msg {dnickname} {command} {option} #channel")
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" Exemple /msg {dnickname} {command} {option} #welcome")
case '+':
try:
# vote +
channel = fromchannel
if self.VoteKickManager.action_vote(channel, fromuser, '+'):
self.Protocol.send_priv_msg(nick_from=dnickname, msg="Vote recorded, thank you",channel=channel)
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname, msg="Vote recorded, thank you",channel=channel)
else:
self.Protocol.send_priv_msg(nick_from=dnickname, msg="You already submitted a vote", channel=channel)
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname, msg="You already submitted a vote", channel=channel)
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
case '-':
try:
# vote -
channel = fromchannel
if self.VoteKickManager.action_vote(channel, fromuser, '-'):
self.Protocol.send_priv_msg(nick_from=dnickname, msg="Vote recorded, thank you",channel=channel)
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname, msg="Vote recorded, thank you",channel=channel)
else:
self.Protocol.send_priv_msg(nick_from=dnickname, msg="You already submitted a vote", channel=channel)
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname, msg="You already submitted a vote", channel=channel)
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
case 'cancel':
try:
# vote cancel
if self.Admin.get_admin(fromuser) is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Your are not allowed to execute this command')
if self.ctx.Admin.get_admin(fromuser) is None:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' 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.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' You need to specify the channel => /msg {dnickname} vote_cancel #channel')
self.ctx.Logs.error(f"The channel is not known, defender can't cancel the vote")
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' You need to specify the channel => /msg {dnickname} vote_cancel #channel')
for vote in self.VoteKickManager.VOTE_CHANNEL_DB:
if vote.channel_name == channel:
if self.VoteKickManager.init_vote_system(channel):
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg="Vote system re-initiated",
channel=channel
)
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
case 'status':
try:
# vote status
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
self.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"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))}",
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"Channel: {chan.channel_name} | Target: {self.ctx.User.get_nickname(chan.target_user)} | For: {chan.vote_for} | Against: {chan.vote_against} | Number of voters: {str(len(chan.voter_users))}",
channel=channel
)
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
case 'submit':
try:
# vote submit nickname
if self.Admin.get_admin(fromuser) is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Your are not allowed to execute this command')
if self.ctx.Admin.get_admin(fromuser) is None:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' 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)
uid_submitted = self.ctx.User.get_uid(nickname_submitted)
user_submitted = self.ctx.User.get_user(nickname_submitted)
ongoing_user = None
# check if there is an ongoing vote
if self.VoteKickManager.is_vote_ongoing(channel):
votec = self.VoteKickManager.get_vote_channel_model(channel)
if votec:
ongoing_user = self.User.get_nickname(votec.target_user)
self.Protocol.send_priv_msg(nick_from=dnickname,
ongoing_user = self.ctx.User.get_nickname(votec.target_user)
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"There is an ongoing vote on {ongoing_user}",
channel=channel
)
@@ -305,24 +312,24 @@ class Votekick(IModule):
# check if the user exist
if user_submitted is None:
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"This nickname <{nickname_submitted}> do not exist",
channel=channel
)
return None
uid_cleaned = self.Loader.Utils.clean_uid(uid_submitted)
channel_obj = self.Channel.get_channel(channel)
uid_cleaned = self.ctx.Utils.clean_uid(uid_submitted)
channel_obj = self.ctx.Channel.get_channel(channel)
if channel_obj is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' This channel [{channel}] do not exist in the Channel Object')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' This channel [{channel}] do not exist in the Channel Object')
return None
clean_uids_in_channel: list = []
for uid in channel_obj.uids:
clean_uids_in_channel.append(self.Loader.Utils.clean_uid(uid))
clean_uids_in_channel.append(self.ctx.Utils.clean_uid(uid))
if not uid_cleaned in clean_uids_in_channel:
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"This nickname <{nickname_submitted}> is not available in this channel",
channel=channel
)
@@ -332,7 +339,7 @@ class Votekick(IModule):
pattern = fr'[o|B|S]'
operator_user = re.findall(pattern, user_submitted.umodes)
if operator_user:
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg="You cant vote for this user ! he/she is protected",
channel=channel
)
@@ -340,49 +347,49 @@ class Votekick(IModule):
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
chan.target_user = self.User.get_uid(nickname_submitted)
chan.target_user = self.ctx.User.get_uid(nickname_submitted)
self.Protocol.send_priv_msg(nick_from=dnickname,
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"{nickname_submitted} has been targeted for a vote",
channel=channel
)
self.Base.create_timer(60, self.Threads.timer_vote_verdict, (self, channel))
self.Protocol.send_priv_msg(nick_from=dnickname,
self.ctx.Base.create_asynctask(thds.timer_vote_verdict(self, channel))
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg="This vote will end after 60 secondes",
channel=channel
)
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option} nickname')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option} adator')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option} nickname')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option} adator')
case 'verdict':
try:
# vote verdict
if self.Admin.get_admin(fromuser) is None:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f'Your are not allowed to execute this command')
if self.ctx.Admin.get_admin(fromuser) is None:
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f'Your are not allowed to execute this command')
return None
votec = self.VoteKickManager.get_vote_channel_model(channel)
if votec:
target_user = self.User.get_nickname(votec.target_user)
target_user = self.ctx.User.get_nickname(votec.target_user)
if votec.vote_for >= votec.vote_against:
self.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll be kicked from the channel",
await self.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"User {self.ctx.Config.COLORS.bold}{target_user}{self.ctx.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll be kicked from the channel",
channel=channel
)
self.Protocol.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
await self.ctx.Irc.Protocol.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
else:
self.Protocol.send_priv_msg(
await self.ctx.Irc.Protocol.send_priv_msg(
nick_from=dnickname,
msg=f"User {self.Config.COLORS.bold}{target_user}{self.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll remain in the channel",
msg=f"User {self.ctx.Config.COLORS.bold}{target_user}{self.ctx.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll remain in the channel",
channel=channel
)
if self.VoteKickManager.init_vote_system(channel):
self.Protocol.send_priv_msg(
await self.ctx.Irc.Protocol.send_priv_msg(
nick_from=dnickname,
msg="System vote re initiated",
channel=channel
@@ -390,19 +397,19 @@ class Votekick(IModule):
return None
except Exception as err:
self.Logs.error(f'{err}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
self.ctx.Logs.error(f'{err}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} {command} {option}')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' Exemple /msg {dnickname} {command} {option}')
case _:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote activate #channel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote deactivate #channel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote +')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote -')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote cancel')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote status')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote submit nickname')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote verdict')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote activate #channel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote deactivate #channel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote +')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote -')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote cancel')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote status')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote submit nickname')
await self.ctx.Irc.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote verdict')
return None
case _:

23
mods/votekick/threads.py
View File

@@ -1,35 +1,38 @@
import asyncio
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mods.votekick.mod_votekick import Votekick
def timer_vote_verdict(uplink: 'Votekick', channel: str) -> None:
async def timer_vote_verdict(uplink: 'Votekick', channel: str) -> None:
dnickname = uplink.Config.SERVICE_NICKNAME
dnickname = uplink.ctx.Config.SERVICE_NICKNAME
if not uplink.VoteKickManager.is_vote_ongoing(channel):
return None
asyncio.sleep(60)
votec = uplink.VoteKickManager.get_vote_channel_model(channel)
if votec:
target_user = uplink.User.get_nickname(votec.target_user)
target_user = uplink.ctx.User.get_nickname(votec.target_user)
if votec.vote_for >= votec.vote_against and votec.vote_for != 0:
uplink.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"User {uplink.Config.COLORS.bold}{target_user}{uplink.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll be kicked from the channel",
await uplink.ctx.Irc.Protocol.send_priv_msg(nick_from=dnickname,
msg=f"User {uplink.ctx.Config.COLORS.bold}{target_user}{uplink.ctx.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll be kicked from the channel",
channel=channel
)
uplink.Protocol.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
await uplink.ctx.Irc.Protocol.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
else:
uplink.Protocol.send_priv_msg(
await uplink.ctx.Irc.Protocol.send_priv_msg(
nick_from=dnickname,
msg=f"User {uplink.Config.COLORS.bold}{target_user}{uplink.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll remain in the channel",
msg=f"User {uplink.ctx.Config.COLORS.bold}{target_user}{uplink.ctx.Config.COLORS.nogc} has {votec.vote_against} votes against and {votec.vote_for} votes for. For this reason, it\'ll remain in the channel",
channel=channel
)
if uplink.VoteKickManager.init_vote_system(channel):
uplink.Protocol.send_priv_msg(
await uplink.ctx.Irc.Protocol.send_priv_msg(
nick_from=dnickname,
msg="System vote re initiated",
channel=channel

18
mods/votekick/utils.py
View File

@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mods.votekick.mod_votekick import Votekick
def add_vote_channel_to_database(uplink: 'Votekick', channel: str) -> bool:
async def add_vote_channel_to_database(uplink: 'Votekick', channel: str) -> bool:
"""Adds a new channel to the votekick database if it doesn't already exist.
This function checks if the specified channel is already registered in the
@@ -18,16 +18,16 @@ def add_vote_channel_to_database(uplink: 'Votekick', channel: str) -> bool:
bool: True if the channel was successfully inserted into the database.
False if the channel already exists or the insertion failed.
"""
current_datetime = uplink.Utils.get_sdatetime()
current_datetime = uplink.ctx.Utils.get_sdatetime()
mes_donnees = {'channel': channel}
response = uplink.Base.db_execute_query("SELECT id FROM votekick_channel WHERE channel = :channel", mes_donnees)
response = await uplink.ctx.Base.db_execute_query("SELECT id FROM votekick_channel WHERE channel = :channel", mes_donnees)
is_channel_exist = response.fetchone()
if is_channel_exist is None:
mes_donnees = {'datetime': current_datetime, 'channel': channel}
insert = uplink.Base.db_execute_query(f"INSERT INTO votekick_channel (datetime, channel) VALUES (:datetime, :channel)", mes_donnees)
insert = await uplink.ctx.Base.db_execute_query(f"INSERT INTO votekick_channel (datetime, channel) VALUES (:datetime, :channel)", mes_donnees)
if insert.rowcount > 0:
return True
else:
@@ -35,7 +35,7 @@ def add_vote_channel_to_database(uplink: 'Votekick', channel: str) -> bool:
else:
return False
def delete_vote_channel_from_database(uplink: 'Votekick', channel: str) -> bool:
async def delete_vote_channel_from_database(uplink: 'Votekick', channel: str) -> bool:
"""Deletes a channel entry from the votekick database.
This function removes the specified channel from the `votekick_channel` table
@@ -49,7 +49,7 @@ def delete_vote_channel_from_database(uplink: 'Votekick', channel: str) -> bool:
bool: True if the channel was successfully deleted, False if no rows were affected.
"""
mes_donnes = {'channel': channel}
response = uplink.Base.db_execute_query("DELETE FROM votekick_channel WHERE channel = :channel", mes_donnes)
response = await uplink.ctx.Base.db_execute_query("DELETE FROM votekick_channel WHERE channel = :channel", mes_donnes)
affected_row = response.rowcount
@@ -61,14 +61,14 @@ def delete_vote_channel_from_database(uplink: 'Votekick', channel: str) -> bool:
async def join_saved_channels(uplink: 'Votekick') -> None:
param = {'module_name': uplink.module_name}
result = uplink.Base.db_execute_query(f"SELECT id, channel_name FROM {uplink.Config.TABLE_CHANNEL} WHERE module_name = :module_name", param)
result = await uplink.ctx.Base.db_execute_query(f"SELECT id, channel_name FROM {uplink.ctx.Config.TABLE_CHANNEL} WHERE module_name = :module_name", param)
channels = result.fetchall()
for channel in channels:
id_, chan = channel
uplink.VoteKickManager.activate_new_channel(chan)
await uplink.Protocol.send_sjoin(channel=chan)
await uplink.Protocol.send2socket(f":{uplink.Config.SERVICE_NICKNAME} SAMODE {chan} +o {uplink.Config.SERVICE_NICKNAME}")
await uplink.ctx.Irc.Protocol.send_sjoin(channel=chan)
await uplink.ctx.Irc.Protocol.send2socket(f":{uplink.ctx.Config.SERVICE_NICKNAME} SAMODE {chan} +o {uplink.ctx.Config.SERVICE_NICKNAME}")
return None

4
mods/votekick/votekick_manager.py
View File

@@ -10,8 +10,8 @@ class VotekickManager:
def __init__(self, uplink: 'Votekick'):
self.uplink = uplink
self.Logs = uplink.Logs
self.Utils = uplink.MainUtils
self.Logs = uplink.ctx.Logs
self.Utils = uplink.ctx.Utils
def activate_new_channel(self, channel_name: str) -> bool:
"""Activate a new channel in the votekick systeme