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

Updating new classes into votekick module

Browse Source
This commit is contained in:
adator
2025-08-16 21:27:00 +02:00
parent f5212deacf
commit 8d23827e1e
2 changed files with 74 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
mods/votekick/mod_votekick.py
View File

@@ -1,4 +1,4 @@
'''
"""
File : mod_votekick.py
Version : 1.0.0
Description : Manages votekick sessions for multiple channels.
@@ -7,11 +7,11 @@
Created : 2025-08-16
Last Updated: 2025-08-16
-----------------------------------------
'''
from typing import TYPE_CHECKING, Any
"""
import re
from dataclasses import dataclass, field
import mods.votekick.votekick_manager
import mods.votekick.schemas as schemas
from mods.votekick.votekick_manager import VotekickManager
from typing import TYPE_CHECKING, Any, Optional
if TYPE_CHECKING:
from core.irc import Irc
@@ -19,52 +19,49 @@ if TYPE_CHECKING:
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:
def __init__(self, uplink: 'Irc') -> None:
# Module name (Mandatory)
self.module_name = 'mod_' + str(self.__class__.__name__).lower()
# Add Irc Object to the module
self.Irc = ircInstance
self.Irc = uplink
# Add Loader Object to the module (Mandatory)
self.Loader = ircInstance.Loader
self.Loader = uplink.Loader
# Add server protocol Object to the module (Mandatory)
self.Protocol = ircInstance.Protocol
self.Protocol = uplink.Protocol
# Add Global Configuration to the module
self.Config = ircInstance.Config
self.Config = uplink.Config
# Add Base object to the module
self.Base = ircInstance.Base
self.Base = uplink.Base
# Add logs object to the module
self.Logs = ircInstance.Base.logs
self.Logs = uplink.Base.logs
# Add User object to the module
self.User = ircInstance.User
self.User = uplink.User
# Add Channel object to the module
self.Channel = ircInstance.Channel
self.Channel = uplink.Channel
# Add Utils.
self.Utils = ircInstance.Utils
self.Utils = uplink.Utils
metadata = ircInstance.Loader.Settings.get_cache('VOTEKICK')
# Add Schemas module
self.Schemas = schemas
# Add VoteKick Manager
self.VoteKickManager = VotekickManager(self)
metadata = uplink.Loader.Settings.get_cache('VOTEKICK')
if metadata is not None:
self.VOTE_CHANNEL_DB = metadata
self.VoteKickManager.VOTE_CHANNEL_DB = metadata
# self.VOTE_CHANNEL_DB = metadata
# Créer les nouvelles commandes du module
self.Irc.build_command(1, self.module_name, 'vote', 'The kick vote module')
@@ -87,8 +84,6 @@ class Votekick:
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
@@ -115,13 +110,13 @@ class Votekick:
def unload(self) -> None:
try:
# Cache the local DB with current votes.
self.Loader.Settings.set_cache('VOTEKICK', self.VOTE_CHANNEL_DB)
self.Loader.Settings.set_cache('VOTEKICK', self.VoteKickManager.VOTE_CHANNEL_DB)
for chan in self.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)
self.VOTE_CHANNEL_DB = []
self.Logs.debug(f'Delete memory DB VOTE_CHANNEL_DB: {self.VOTE_CHANNEL_DB}')
self.VoteKickManager.VOTE_CHANNEL_DB = []
self.Logs.debug(f'Delete memory DB VOTE_CHANNEL_DB: {self.VoteKickManager.VOTE_CHANNEL_DB}')
return None
except UnboundLocalError as ne:
@@ -134,7 +129,7 @@ class Votekick:
def init_vote_system(self, channel: str) -> bool:
response = False
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
chan.target_user = ''
chan.voter_users = []
@@ -144,21 +139,21 @@ class Votekick:
return response
def insert_vote_channel(self, ChannelObject: VoteChannelModel) -> bool:
def insert_vote_channel(self, channel_obj: schemas.VoteChannelModel) -> bool:
result = False
found = False
for chan in self.VOTE_CHANNEL_DB:
if chan.channel_name == ChannelObject.channel_name:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel_obj.channel_name:
found = True
if not found:
self.VOTE_CHANNEL_DB.append(ChannelObject)
self.Logs.debug(f"The channel has been added {ChannelObject}")
self.VoteKickManager.VOTE_CHANNEL_DB.append(channel_obj)
self.Logs.debug(f"The channel has been added {channel_obj}")
# self.db_add_vote_channel(ChannelObject.channel_name)
return result
def db_add_vote_channel(self, channel:str) -> bool:
def db_add_vote_channel(self, channel: str) -> bool:
"""Cette fonction ajoute les salons ou seront autoriser les votes
Args:
@@ -169,9 +164,9 @@ class Votekick:
response = self.Base.db_execute_query("SELECT id FROM votekick_channel WHERE channel = :channel", mes_donnees)
isChannelExist = response.fetchone()
is_channel_exist = response.fetchone()
if isChannelExist is None:
if is_channel_exist 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:
@@ -203,11 +198,10 @@ class Votekick:
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))
id_, chan = channel
self.insert_vote_channel(self.Schemas.VoteChannelModel(channel_name=chan, target_user='', voter_users=[], vote_for=0, vote_against=0))
self.Protocol.sjoin(channel=chan)
self.Protocol.send2socket(f":{self.Config.SERVICE_NICKNAME} SAMODE {chan} +o {self.Config.SERVICE_NICKNAME}")
@@ -216,7 +210,7 @@ class Votekick:
def is_vote_ongoing(self, channel: str) -> bool:
response = False
for vote in self.VOTE_CHANNEL_DB:
for vote in self.VoteKickManager.VOTE_CHANNEL_DB:
if vote.channel_name == channel:
if vote.target_user:
response = True
@@ -230,7 +224,7 @@ class Votekick:
if not self.is_vote_ongoing(channel):
return None
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
target_user = self.User.get_nickname(chan.target_user)
if chan.vote_for > chan.vote_against:
@@ -258,9 +252,12 @@ class Votekick:
return None
def cmd(self, data:list) -> None:
def cmd(self, data: list) -> None:
cmd = list(data).copy()
try:
cmd = list(data).copy()
return None
except KeyError as ke:
@@ -270,11 +267,12 @@ class Votekick:
except Exception as err:
self.Logs.error(f"General Error: {err}")
def hcmds(self, user:str, channel: Any, cmd: list, fullcmd: list = []) -> None:
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
fromuser = user
fromchannel = channel
@@ -310,7 +308,7 @@ class Votekick:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f" The correct command is {self.Config.SERVICE_PREFIX}{command} {option} #CHANNEL")
self.insert_vote_channel(
self.VoteChannelModel(
self.Schemas.VoteChannelModel(
channel_name=sentchannel,
target_user='',
voter_users=[],
@@ -346,9 +344,9 @@ class Votekick:
self.Protocol.send2socket(f":{dnickname} SAMODE {sentchannel} -o {dnickname}")
self.Protocol.send_part_chan(uidornickname=dnickname, channel=sentchannel)
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == sentchannel:
self.VOTE_CHANNEL_DB.remove(chan)
self.VoteKickManager.VOTE_CHANNEL_DB.remove(chan)
self.Channel.db_query_channel('del', self.module_name, chan.channel_name)
self.Logs.debug(f"The Channel {sentchannel} has been deactivated from the vote system")
@@ -361,7 +359,7 @@ class Votekick:
try:
# vote +
channel = fromchannel
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
if fromuser in chan.voter_users:
self.Protocol.send_priv_msg(nick_from=dnickname,
@@ -384,7 +382,7 @@ class Votekick:
try:
# vote -
channel = fromchannel
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
if fromuser in chan.voter_users:
self.Protocol.send_priv_msg(nick_from=dnickname,
@@ -414,7 +412,7 @@ class Votekick:
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')
for vote in self.VOTE_CHANNEL_DB:
for vote in self.VoteKickManager.VOTE_CHANNEL_DB:
if vote.channel_name == channel:
self.init_vote_system(channel)
self.Protocol.send_priv_msg(nick_from=dnickname,
@@ -430,7 +428,7 @@ class Votekick:
case 'status':
try:
# vote status
for chan in self.VOTE_CHANNEL_DB:
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))}",
@@ -451,10 +449,11 @@ class Votekick:
nickname_submitted = cmd[2]
uid_submitted = self.User.get_uid(nickname_submitted)
user_submitted = self.User.get_User(nickname_submitted)
ongoing_user = None
# check if there is an ongoing vote
if self.is_vote_ongoing(channel):
for vote in self.VOTE_CHANNEL_DB:
for vote in self.VoteKickManager.VOTE_CHANNEL_DB:
if vote.channel_name == channel:
ongoing_user = self.User.get_nickname(vote.target_user)
@@ -462,7 +461,7 @@ class Votekick:
msg=f"There is an ongoing vote on {ongoing_user}",
channel=channel
)
return False
return None
# check if the user exist
if user_submitted is None:
@@ -470,16 +469,16 @@ class Votekick:
msg=f"This nickname <{nickname_submitted}> do not exist",
channel=channel
)
return False
return None
uid_cleaned = self.Base.clean_uid(uid_submitted)
ChannelInfo = self.Channel.get_channel(channel)
if ChannelInfo is None:
channel_obj = self.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')
return False
return None
clean_uids_in_channel: list = []
for uid in ChannelInfo.uids:
for uid in channel_obj.uids:
clean_uids_in_channel.append(self.Base.clean_uid(uid))
if not uid_cleaned in clean_uids_in_channel:
@@ -487,7 +486,7 @@ class Votekick:
msg=f"This nickname <{nickname_submitted}> is not available in this channel",
channel=channel
)
return False
return None
# check if Ircop or Service or Bot
pattern = fr'[o|B|S]'
@@ -497,9 +496,9 @@ class Votekick:
msg="You cant vote for this user ! he/she is protected",
channel=channel
)
return False
return None
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
chan.target_user = self.User.get_uid(nickname_submitted)
@@ -526,7 +525,7 @@ class Votekick:
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f'Your are not allowed to execute this command')
return None
for chan in self.VOTE_CHANNEL_DB:
for chan in self.VoteKickManager.VOTE_CHANNEL_DB:
if chan.channel_name == channel:
target_user = self.User.get_nickname(chan.target_user)
if chan.vote_for > chan.vote_against:
@@ -562,4 +561,8 @@ class Votekick:
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')
self.Protocol.send_notice(nick_from=dnickname, nick_to=fromuser,msg=f' /msg {dnickname} vote verdict')
return None
case _:
return None